# 接口 2：从表 2 取 push_status=0，组装文档报文。假连接录 SQL，不连真库，不发 HTTP。 from __future__ import annotations from decimal import Decimal from app.outbound.cart_settlement import ( MAX_BATCH, batch_limit, build_cart_settlement_envelope, load_cart_settlement_envelope, ) from app.outbound.oracle_repo import OracleCartSettlementRepo from tests.fake_oracle import FakeDb # 构造一条表 2 行，供组装层单测用。 def _row(**overrides): row = { "fid": 100001, "operate_type": 1, "oms_id": 20260928000001, "platform_type": 48, "supplier_name": "示例供应商", "bar_code": "06901234567890", "purchase_price": Decimal("128.5000"), "purchase_qty": 10, "purchase_total": Decimal("1285.0000"), "status": 1, "fail_reason": None, "pay_order_no": None, "pay_order_dtl_id": None, "org_code": "901", "push_status": 0, "multi_platform_flag": 0, } row.update(overrides) return row # 把表 2 行放进假库。list_unpushed 走真 SQL，调用记在 db.calls。 def _open(rows: list[dict]) -> tuple[OracleCartSettlementRepo, FakeDb]: db = FakeDb() for row in rows: db.carts[row["fid"]] = row return OracleCartSettlementRepo(db.connection()), db # 取出最近一次 execute 的 SQL 和绑定。 def _last_execute(db: FakeDb) -> tuple[str, dict]: execute_calls = [call for call in db.calls if call[0] == "execute"] assert execute_calls _, sql, params = execute_calls[-1] return sql, params or {} # 未推行按文档信封字段组装，金额四位小数字符串。 def test_assembles_unpushed_rows_as_document_envelope(): repo, db = _open( [ _row(), _row( fid=100002, operate_type=2, platform_type=66, status=None, pay_order_no="PAY20260928000001", pay_order_dtl_id="1233", ), ] ) envelope = load_cart_settlement_envelope(repo) assert envelope == { "data": [ { "fid": 100001, "operateType": 1, "omsId": 20260928000001, "platformType": 48, "supplierName": "示例供应商", "barCode": "06901234567890", "purchasePrice": "128.5000", "purchaseQty": 10, "purchaseTotal": "1285.0000", "status": 1, "failReason": None, "payOrderNo": None, "payOrderDtlId": None, "orgCode": "901", }, { "fid": 100002, "operateType": 2, "omsId": 20260928000001, "platformType": 66, "supplierName": "示例供应商", "barCode": "06901234567890", "purchasePrice": "128.5000", "purchaseQty": 10, "purchaseTotal": "1285.0000", "status": None, "failReason": None, "payOrderNo": "PAY20260928000001", "payOrderDtlId": "1233", "orgCode": "901", }, ] } sql, params = _last_execute(db) assert "rpa_cart_settlement_goods_data" in sql.lower() assert "push_status = 0" in sql.lower() assert params["limit"] == 500 # 失败行金额为 NULL 时该行置空，同批其它行照常组装。 def test_null_amounts_do_not_fail_the_batch(): repo, _db = _open( [ _row( fid=1, status=2, purchase_price=None, purchase_total=None, fail_reason="库存不足", ), _row(fid=2, oms_id=20260928000002), ] ) envelope = load_cart_settlement_envelope(repo) failed, ok = envelope["data"] assert failed["status"] == 2 assert failed["purchasePrice"] is None assert failed["purchaseTotal"] is None assert failed["failReason"] == "库存不足" assert ok["purchasePrice"] == "128.5000" assert ok["purchaseTotal"] == "1285.0000" # push_status=1 的行不进报文。 def test_skips_already_pushed_rows(): repo, db = _open( [ _row(fid=1, push_status=1), _row(fid=2, oms_id=20260928000002, push_status=0), ] ) envelope = load_cart_settlement_envelope(repo) assert [item["fid"] for item in envelope["data"]] == [2] sql, _params = _last_execute(db) assert "push_status = 0" in sql.lower() # 同一 omsId 的行必须落在同一批 data。 def test_keeps_same_oms_id_in_one_envelope(): rows = [ _row(fid=1, oms_id=1, platform_type=48, multi_platform_flag=1), _row(fid=2, oms_id=1, platform_type=66, multi_platform_flag=1), _row(fid=3, oms_id=2, platform_type=48), ] envelope = build_cart_settlement_envelope(rows, limit=2) fids = [item["fid"] for item in envelope["data"]] assert fids == [1, 2] # 批次已有行时，装不下的 omsId 整组留下。 def test_does_not_split_oms_id_group_across_batch_limit(): rows = [ _row(fid=1, oms_id=1, platform_type=48), _row(fid=2, oms_id=2, platform_type=48, multi_platform_flag=1), _row(fid=3, oms_id=2, platform_type=66, multi_platform_flag=1), ] envelope = build_cart_settlement_envelope(rows, limit=2) assert [item["fid"] for item in envelope["data"]] == [1] # 默认上限 500；超过则按组截到 500。 def test_batch_cap_is_500(): assert MAX_BATCH == 500 rows = [_row(fid=i, oms_id=i) for i in range(1, 502)] envelope = build_cart_settlement_envelope(rows) assert len(envelope["data"]) == 500 assert envelope["data"][0]["fid"] == 1 assert envelope["data"][-1]["fid"] == 500 # 报文不含 push_status 等表内列。 def test_payload_omits_table_only_columns(): envelope = build_cart_settlement_envelope([_row()]) item = envelope["data"][0] assert "pushStatus" not in item assert "push_status" not in item assert "multiPlatformFlag" not in item assert "createTime" not in item # 组装入口把 limit 传给仓库，让 SQL 侧截组。 def test_load_passes_limit_to_list_unpushed(): seen: dict = {} # 只记 list_unpushed 收到的 limit。 class SpyRepo: # 记下上限并返回空列表。 def list_unpushed(self, limit: int = 500) -> list[dict]: seen["limit"] = limit return [] load_cart_settlement_envelope(SpyRepo(), limit=3) assert seen["limit"] == 3 # OMS_OUTBOUND_BATCH_LIMIT 覆盖默认 500。 def test_batch_limit_reads_env(monkeypatch): monkeypatch.setenv("OMS_OUTBOUND_BATCH_LIMIT", "3") assert batch_limit() == 3 # 非法或非正数上限回退 500。 def test_batch_limit_invalid_falls_back_to_500(monkeypatch): monkeypatch.setenv("OMS_OUTBOUND_BATCH_LIMIT", "abc") assert batch_limit() == 500 monkeypatch.setenv("OMS_OUTBOUND_BATCH_LIMIT", "0") assert batch_limit() == 500 # 未显式传 limit 时按环境变量截组。 def test_load_uses_env_batch_limit(monkeypatch): monkeypatch.setenv("OMS_OUTBOUND_BATCH_LIMIT", "1") repo, db = _open( [ _row(fid=1, oms_id=1), _row(fid=2, oms_id=2), ] ) envelope = load_cart_settlement_envelope(repo) assert [item["fid"] for item in envelope["data"]] == [1] _sql, params = _last_execute(db) assert params["limit"] == 1