from __future__ import annotations from repo import MappingRow, MemorySelectRepo from rules import Candidate, normalize_vendor from job import run_batch HANGZHOU = "杭州小药药自营旗舰店" GUANGDONG = "广东壹号药业有限公司" def _row(**overrides) -> MappingRow: data = { "fid": 1, "oms_id": 2026100201, "bar_code": "06901234567890", "sale_price": 100.0, "plan_purchase_qty": 10, "approval_no": "国药准字H20100001", "goods_name": "阿莫西林", "goods_spec": "0.25g", "manufacturer": "示例制药", "org_code": "901", "deal_flag": 0, } data.update(overrides) return MappingRow(**data) def _ok_candidate(mall: str = "1号药城", **overrides) -> Candidate: vendor = overrides.pop("vendor", GUANGDONG if mall == "1号药城" else HANGZHOU) data = { "mall_name": mall, "sku_id": "S1", "name": "阿莫西林", "spec": "0.25g", "price": 90.0, "expiry_raw": "13", "expiry_months": 13.0, "vendor_raw": vendor, "vendor_norm": normalize_vendor(vendor), "stock": 20.0, } data.update(overrides) if "vendor_raw" in overrides and "vendor_norm" not in overrides: data["vendor_norm"] = normalize_vendor(overrides["vendor_raw"]) return Candidate(**data) def test_claim_pending_sets_processing_flag(): repo = MemorySelectRepo([_row(), _row(fid=2, oms_id=2026100202, deal_flag=1)]) claimed = repo.claim_pending() assert [row.oms_id for row in claimed] == [2026100201] assert repo.mapping[2026100201].deal_flag == 3 assert repo.mapping[2026100202].deal_flag == 1 def test_job_searches_barcode_not_approval_no(): repo = MemorySelectRepo([_row()]) seen: list[str] = [] def search_yaoex(query_key: str): seen.append(query_key) return [_ok_candidate()] summary = run_batch( repo, platforms=["1号药城"], searchers={"1号药城": search_yaoex}, login_errors={}, ) assert seen == ["06901234567890"] assert summary.added == 1 assert repo.mapping[2026100201].deal_flag == 1 assert len(repo.cart) == 1 assert repo.cart[0]["bar_code"] == "06901234567890" assert repo.cart[0]["status"] == 1 assert repo.cart[0]["purchase_qty"] == 10 def test_job_skips_search_when_barcode_missing(): repo = MemorySelectRepo([_row(bar_code="")]) called = {"n": 0} def search_yaoex(_query_key: str): called["n"] += 1 return [_ok_candidate()] summary = run_batch( repo, platforms=["1号药城"], searchers={"1号药城": search_yaoex}, login_errors={}, ) assert called["n"] == 0 assert summary.skipped == 1 assert repo.mapping[2026100201].deal_flag == 2 assert repo.cart[0]["status"] == 2 assert "条形码" in (repo.cart[0]["fail_reason"] or "") def test_job_limit_claims_first_n(): repo = MemorySelectRepo( [ _row(fid=1, oms_id=1, bar_code="111"), _row(fid=2, oms_id=2, bar_code="222"), _row(fid=3, oms_id=3, bar_code="333"), ] ) claimed = repo.claim_pending(limit=2) assert [row.oms_id for row in claimed] == [1, 2] assert repo.mapping[3].deal_flag == 0 def test_both_malls_login_failed_marks_system_fault(): repo = MemorySelectRepo([_row()]) summary = run_batch( repo, platforms=["1号药城", "药帮忙"], searchers={}, login_errors={"1号药城": "chrome down", "药帮忙": "timeout"}, ) assert summary.failed == 2 assert repo.mapping[2026100201].deal_flag == 2 assert repo.mapping[2026100201].deal_result.startswith("系统故障") assert len(repo.cart) == 2 assert all(row["status"] == 2 for row in repo.cart) def test_split_across_malls_writes_two_success_rows(): repo = MemorySelectRepo([_row()]) def search_yaoex(_key: str): return [_ok_candidate("1号药城", stock=20.0)] def search_ybm(_key: str): return [_ok_candidate("药帮忙", vendor=HANGZHOU, sku_id="HZ", stock=3.0)] summary = run_batch( repo, platforms=["1号药城", "药帮忙"], searchers={"1号药城": search_yaoex, "药帮忙": search_ybm}, login_errors={}, ) assert summary.added == 2 assert repo.mapping[2026100201].deal_flag == 1 ybm = next(row for row in repo.cart if row["platform_type"] == 48) yaoex = next(row for row in repo.cart if row["platform_type"] == 66) assert ybm["purchase_qty"] == 3 assert yaoex["purchase_qty"] == 7 assert ybm["multi_platform_flag"] == 1 def test_peer_login_failed_but_other_fills_is_success(): repo = MemorySelectRepo([_row()]) def search_yaoex(_key: str): return [_ok_candidate(stock=20.0)] summary = run_batch( repo, platforms=["1号药城", "药帮忙"], searchers={"1号药城": search_yaoex}, login_errors={"药帮忙": "timeout"}, ) assert summary.added == 1 assert repo.mapping[2026100201].deal_flag == 1 assert len(repo.cart) == 1 assert repo.cart[0]["platform_type"] == 66 def test_peer_login_failed_and_short_stock_is_system_fault(): repo = MemorySelectRepo([_row()]) def search_yaoex(_key: str): return [_ok_candidate(stock=4.0)] summary = run_batch( repo, platforms=["1号药城", "药帮忙"], searchers={"1号药城": search_yaoex}, login_errors={"药帮忙": "timeout"}, ) assert summary.failed == 2 assert summary.added == 0 assert repo.mapping[2026100201].deal_flag == 2 assert repo.mapping[2026100201].deal_result.startswith("系统故障") assert all(row["status"] == 2 for row in repo.cart) def test_combined_short_stock_is_business_fail(): repo = MemorySelectRepo([_row()]) def search_yaoex(_key: str): return [_ok_candidate(stock=2.0)] def search_ybm(_key: str): return [_ok_candidate("药帮忙", vendor=HANGZHOU, sku_id="HZ", stock=2.0)] summary = run_batch( repo, platforms=["1号药城", "药帮忙"], searchers={"1号药城": search_yaoex, "药帮忙": search_ybm}, login_errors={}, ) assert summary.added == 0 assert summary.skipped == 2 assert repo.mapping[2026100201].deal_flag == 2 assert "库存不足" in repo.mapping[2026100201].deal_result assert all(row["status"] == 2 for row in repo.cart)