104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import cart_job
|
|
import main
|
|
from cart_repo import STAMP_SELECTION, MemoryCartRepo
|
|
from main import DEFAULT_CART_LIMIT, parse_args, pay_export_path
|
|
|
|
|
|
def test_parse_args_defaults():
|
|
args = parse_args([])
|
|
assert args.command == "pay"
|
|
|
|
|
|
def test_parse_args_pay_rejects_mode_switches():
|
|
with pytest.raises(SystemExit):
|
|
parse_args(["--xls", "/tmp/orders.xls", "--dry-run"])
|
|
|
|
|
|
def test_pay_export_path_groups_by_order_day(tmp_path):
|
|
when = datetime(2026, 10, 24, 15, 30, 45)
|
|
path = pay_export_path(
|
|
"2026-10-24 00:00:00",
|
|
"2026-10-24 23:59:59",
|
|
when,
|
|
root=tmp_path,
|
|
)
|
|
assert path == tmp_path / "2026-10-24" / "153045.xls"
|
|
|
|
|
|
def test_pay_export_path_keeps_same_second_export(tmp_path):
|
|
when = datetime(2026, 10, 24, 15, 30, 45)
|
|
first = pay_export_path(
|
|
"2026-10-24 00:00:00",
|
|
"2026-10-24 23:59:59",
|
|
when,
|
|
root=tmp_path,
|
|
)
|
|
first.parent.mkdir(parents=True)
|
|
first.write_bytes(b"a")
|
|
second = pay_export_path(
|
|
"2026-10-24 00:00:00",
|
|
"2026-10-24 23:59:59",
|
|
when,
|
|
root=tmp_path,
|
|
)
|
|
assert second == tmp_path / "2026-10-24" / "153045_2.xls"
|
|
|
|
|
|
def test_parse_args_cart_subcommand():
|
|
"""显式 --limit 仍覆盖默认 20。"""
|
|
args = parse_args(["cart", "--limit", "5", "--platform", "yaoex"])
|
|
assert args.command == "cart"
|
|
assert args.limit == 5
|
|
assert args.platform == "yaoex"
|
|
|
|
|
|
def test_cart_default_limit_caps_claim(monkeypatch):
|
|
"""不传 --limit 时加购最多领 20 行。"""
|
|
import main
|
|
|
|
rows = [
|
|
{
|
|
"fid": i,
|
|
"oms_id": 2026100200 + i,
|
|
"platform_type": 66,
|
|
"supplier_name": "广东壹号药业有限公司",
|
|
"bar_code": "06901234567890",
|
|
"purchase_qty": 1,
|
|
"status": 1,
|
|
"fail_reason": None,
|
|
"push_status": 0,
|
|
"update_user_name": STAMP_SELECTION,
|
|
"operate_type": 1,
|
|
}
|
|
for i in range(1, 31)
|
|
]
|
|
repo = MemoryCartRepo(rows)
|
|
seen: dict[str, int | None] = {}
|
|
|
|
def fake_run_cart_batch(bound_repo, **kwargs):
|
|
"""只按入口传入的上限领取，不加车。"""
|
|
seen["limit"] = kwargs.get("limit")
|
|
claimed = bound_repo.claim_pending(
|
|
limit=kwargs.get("limit"),
|
|
platform_types=kwargs.get("platform_types"),
|
|
)
|
|
seen["count"] = len(claimed)
|
|
return SimpleNamespace(claimed=len(claimed), added=0, failed=0)
|
|
|
|
def no_browser(**_kwargs):
|
|
"""挡住商城登录。"""
|
|
raise RuntimeError("no browser")
|
|
|
|
monkeypatch.setattr(cart_job, "run_cart_batch", fake_run_cart_batch)
|
|
monkeypatch.setattr(main, "resolve_yaoex_auth", no_browser)
|
|
monkeypatch.setattr(main, "resolve_ybm_cookies", no_browser)
|
|
main.main(["cart"], repo=repo)
|
|
assert seen["limit"] == DEFAULT_CART_LIMIT
|
|
assert seen["count"] == 20
|
|
assert seen["count"] <= DEFAULT_CART_LIMIT
|