AutoRpa/rpa_selection/tests/test_cli.py

from __future__ import annotations

import sys
from pathlib import Path
from types import SimpleNamespace

import pytest

_CART_ROOT = Path(__file__).resolve().parents[2] / "rpa_cart"
if str(_CART_ROOT) not in sys.path:
    sys.path.append(str(_CART_ROOT))

import cart_ybm  # noqa: E402
import cart_yaoex  # noqa: E402

import main  # noqa: E402
from job import run_batch
from mall_config import YAOEX_MALL_NAME, YBM_MALL_NAME
from repo import MappingRow, MemorySelectRepo
from rules import Candidate, normalize_vendor
from search_yaoex import YaoexAuth


def test_old_excel_and_cart_flags_are_gone():
    with pytest.raises(SystemExit):
        main.parse_args(["--execute-cart"])
    with pytest.raises(SystemExit):
        main.parse_args(["--input", "x.xlsx"])
    with pytest.raises(SystemExit):
        main.parse_args(["--dry-run"])


def test_main_writes_cart_rows_from_mapping(monkeypatch):
    repo = MemorySelectRepo(
        [
            MappingRow(
                fid=1,
                oms_id=2026100201,
                bar_code="06901234567890",
                sale_price=100.0,
                plan_purchase_qty=10,
                approval_no="国药准字H20100001",
                goods_name="药A",
                goods_spec="0.25g",
                manufacturer="示例",
                org_code="907",
                deal_flag=0,
            )
        ]
    )

    def fake_login_yaoex(*_args, **_kwargs):
        return YaoexAuth(token="t", user_id="u", buyer_code="b")

    def fake_login_ybm(*_args, **_kwargs):
        return {"sid": "1"}

    def fake_search_yaoex(*_args, **_kwargs):
        vendor = "广东壹号药业有限公司"
        return [
            Candidate(
                mall_name=YAOEX_MALL_NAME,
                sku_id="S1",
                name="药A",
                spec="0.25g",
                price=90.0,
                expiry_raw="13",
                expiry_months=13.0,
                vendor_raw=vendor,
                vendor_norm=normalize_vendor(vendor),
                stock=20.0,
            )
        ]

    def fake_search_ybm(*_args, **_kwargs):
        vendor = "某第三方"
        return [
            Candidate(
                mall_name=YBM_MALL_NAME,
                sku_id="Y1",
                name="药A",
                spec="0.25g",
                price=70.0,
                expiry_raw="13",
                expiry_months=13.0,
                vendor_raw=vendor,
                vendor_norm=normalize_vendor(vendor),
                stock=20.0,
            )
        ]

    # 挡住真实加车。入口应在落表后按拆定的 sku 调到这里。
    def fake_add(_auth, offer, qty, **_kwargs):
        from cart_add import KIND_OK, AddOutcome

        assert offer.sku_id == "S1"
        assert qty == 10
        return AddOutcome(kind=KIND_OK)

    monkeypatch.setattr(main, "login_yaoex", fake_login_yaoex)
    monkeypatch.setattr(main, "login_ybm", fake_login_ybm)
    monkeypatch.setattr(main, "search_yaoex_candidates", fake_search_yaoex)
    monkeypatch.setattr(main, "search_ybm_candidates", fake_search_ybm)
    monkeypatch.setattr(cart_yaoex, "add_yaoex_cart", fake_add)
    monkeypatch.setattr(cart_ybm, "add_ybm_cart", fake_add)

    main.main(["--fresh-login"], repo=repo)

    assert repo.mapping[2026100201].deal_flag == 1
    assert len(repo.cart) == 1
    assert repo.cart[0]["platform_type"] == 66
    assert repo.cart[0]["status"] == 1
    assert repo.cart[0]["bar_code"] == "06901234567890"
    assert repo.cart[0]["update_user_name"] == "加购班"


def test_default_limit_caps_claim_at_50(monkeypatch):
    """不传 --limit 时最多领 50 条，多出来的行保持待领。"""
    rows = [
        MappingRow(
            fid=i,
            oms_id=2026100200 + i,
            bar_code="06901234567890",
            sale_price=100.0,
            plan_purchase_qty=10,
            approval_no="国药准字H20100001",
            goods_name="药A",
            goods_spec="0.25g",
            manufacturer="示例",
            org_code="907",
            deal_flag=0,
        )
        for i in range(1, 61)
    ]
    repo = MemorySelectRepo(rows)
    seen: dict[str, int | None] = {}

    def fake_run_batch(bound_repo, **kwargs):
        """只按入口传入的上限领取，不搜平台。"""
        seen["limit"] = kwargs.get("limit")
        claimed = bound_repo.claim_pending(limit=kwargs.get("limit"))
        seen["count"] = len(claimed)
        return SimpleNamespace(
            claimed=len(claimed),
            added=0,
            skipped=0,
            failed=0,
            cart_added=0,
            cart_failed=0,
        )

    def no_browser(*_args, **_kwargs):
        """挡住浏览器登录。"""
        raise RuntimeError("no browser")

    monkeypatch.setattr(main, "run_batch", fake_run_batch)
    monkeypatch.setattr(main, "login_yaoex", no_browser)
    monkeypatch.setattr(main, "login_ybm", no_browser)
    main.main([], repo=repo)
    assert seen["limit"] == main.DEFAULT_CLAIM_LIMIT
    assert seen["count"] == 50
    assert seen["count"] <= main.DEFAULT_CLAIM_LIMIT


def test_run_batch_empty_claim_is_noop():
    repo = MemorySelectRepo([])
    summary = run_batch(
        repo,
        platforms=[YAOEX_MALL_NAME],
        searchers={},
        login_errors={},
    )
    assert summary.claimed == 0
    assert repo.cart == []