197 lines
6.3 KiB
Python
197 lines
6.3 KiB
Python
# 接口 4：从表 4 取 push_status=0，组装文档报文。假连接录 SQL，不连真库，不发 HTTP。
|
||
|
||
from __future__ import annotations
|
||
|
||
from app.outbound.contract_agreement import (
|
||
MAX_BATCH,
|
||
batch_limit,
|
||
build_contract_agreement_envelope,
|
||
load_contract_agreement_envelope,
|
||
)
|
||
from app.outbound.oracle_repo import OracleContractAgreementRepo
|
||
from tests.fake_oracle import FakeDb
|
||
|
||
|
||
# 构造一条表 4 行，供组装层单测用。
|
||
def _row(**overrides):
|
||
row = {
|
||
"fid": 300001,
|
||
"oms_id": 20260928000001,
|
||
"action_code": 1,
|
||
"contract_status": 1,
|
||
"contract_fail_reason": None,
|
||
"contract_no": "HT1",
|
||
"agreement_status": None,
|
||
"agreement_fail_reason": None,
|
||
"agreement_id": None,
|
||
"agreement_dtl_id": None,
|
||
"org_code": "901",
|
||
"push_status": 0,
|
||
}
|
||
row.update(overrides)
|
||
return row
|
||
|
||
|
||
# 把表 4 行放进假库。list_unpushed 走真 SQL，调用记在 db.calls。
|
||
def _open(rows: list[dict]) -> tuple[OracleContractAgreementRepo, FakeDb]:
|
||
db = FakeDb()
|
||
for row in rows:
|
||
db.agreements[row["fid"]] = row
|
||
return OracleContractAgreementRepo(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=300002,
|
||
action_code=2,
|
||
contract_status=None,
|
||
contract_no="HT1",
|
||
agreement_status=1,
|
||
agreement_id="AGR1",
|
||
agreement_dtl_id="AGRD1",
|
||
),
|
||
_row(fid=300003, push_status=1, contract_no="HT-SKIP"),
|
||
]
|
||
)
|
||
envelope = load_contract_agreement_envelope(repo)
|
||
assert envelope == {
|
||
"data": [
|
||
{
|
||
"fid": 300001,
|
||
"omsId": 20260928000001,
|
||
"actionCode": 1,
|
||
"contractStatus": 1,
|
||
"contractFailReason": None,
|
||
"contractNo": "HT1",
|
||
"agreementStatus": None,
|
||
"agreementFailReason": None,
|
||
"agreementId": None,
|
||
"agreementDtlId": None,
|
||
"orgCode": "901",
|
||
},
|
||
{
|
||
"fid": 300002,
|
||
"omsId": 20260928000001,
|
||
"actionCode": 2,
|
||
"contractStatus": None,
|
||
"contractFailReason": None,
|
||
"contractNo": "HT1",
|
||
"agreementStatus": 1,
|
||
"agreementFailReason": None,
|
||
"agreementId": "AGR1",
|
||
"agreementDtlId": "AGRD1",
|
||
"orgCode": "901",
|
||
},
|
||
]
|
||
}
|
||
assert isinstance(envelope["data"][0]["omsId"], int)
|
||
sql, params = _last_execute(db)
|
||
assert "rpa_contract_agreement_result" in sql.lower()
|
||
assert "push_status = 0" in sql.lower()
|
||
assert params["limit"] == 500
|
||
|
||
|
||
# 合同成功行和协议行可以同批。
|
||
def test_assembles_contract_success_and_agreement_rows():
|
||
repo, db = _open(
|
||
[
|
||
_row(contract_status=2),
|
||
_row(
|
||
fid=300002,
|
||
action_code=2,
|
||
contract_status=None,
|
||
contract_no="HT1",
|
||
agreement_status=1,
|
||
agreement_id="AGR1",
|
||
agreement_dtl_id="AGRD1",
|
||
),
|
||
]
|
||
)
|
||
envelope = load_contract_agreement_envelope(repo)
|
||
assert envelope["data"][0]["actionCode"] == 1
|
||
assert envelope["data"][0]["contractStatus"] == 2
|
||
assert envelope["data"][1]["actionCode"] == 2
|
||
assert envelope["data"][1]["agreementStatus"] == 1
|
||
assert envelope["data"][1]["agreementId"] == "AGR1"
|
||
assert envelope["data"][1]["agreementDtlId"] == "AGRD1"
|
||
sql, _params = _last_execute(db)
|
||
assert "rpa_contract_agreement_result" in sql.lower()
|
||
|
||
|
||
# 空批时超限的同一 contractNo 仍整组进入。
|
||
def test_oversized_contract_no_group_enters_when_batch_empty():
|
||
rows = [
|
||
_row(fid=1, oms_id=1, contract_no="HT-BIG"),
|
||
_row(fid=2, oms_id=2, contract_no="HT-BIG"),
|
||
_row(fid=3, oms_id=3, contract_no="HT-BIG"),
|
||
]
|
||
envelope = build_contract_agreement_envelope(rows, limit=2)
|
||
assert [item["fid"] for item in envelope["data"]] == [1, 2, 3]
|
||
|
||
|
||
# 批次已有行时，装不下的 contractNo 整组留下。
|
||
def test_does_not_split_contract_no_when_batch_already_has_rows():
|
||
rows = [
|
||
_row(fid=1, oms_id=1, contract_no="HT-A"),
|
||
_row(fid=2, oms_id=2, contract_no="HT-B"),
|
||
_row(fid=3, oms_id=3, contract_no="HT-B"),
|
||
]
|
||
envelope = build_contract_agreement_envelope(rows, limit=2)
|
||
assert [item["fid"] for item in envelope["data"]] == [1]
|
||
|
||
|
||
# 空 contractNo 按行独立，互不捆绑。
|
||
def test_empty_contract_no_rows_are_independent():
|
||
rows = [
|
||
_row(fid=1, oms_id=1, contract_no=None, contract_status=3),
|
||
_row(fid=2, oms_id=2, contract_no="", contract_status=3),
|
||
_row(fid=3, oms_id=3, contract_no="HT-C"),
|
||
]
|
||
envelope = build_contract_agreement_envelope(rows, limit=1)
|
||
assert [item["fid"] for item in envelope["data"]] == [1]
|
||
|
||
|
||
# 默认上限 500。
|
||
def test_batch_cap_is_500():
|
||
assert MAX_BATCH == 500
|
||
|
||
|
||
# 报文不含 push_status。
|
||
def test_payload_omits_push_status():
|
||
item = build_contract_agreement_envelope([_row()])["data"][0]
|
||
assert "pushStatus" not in item
|
||
assert "push_status" 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_contract_agreement_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
|