461 lines
15 KiB
Python
461 lines
15 KiB
Python
# 合同班次：领取戳、合单、尽调失败、系统故障短码。
|
||
|
||
from contextlib import contextmanager
|
||
from datetime import date, datetime, timedelta
|
||
from decimal import Decimal
|
||
|
||
from errors import DiligenceError, NodeError
|
||
from job import PAGE_FAILED, outcome_from_run, run_batch
|
||
from persist import (
|
||
KIND_DILIGENCE,
|
||
KIND_FAIL,
|
||
KIND_OK,
|
||
LOOKUP_ONLY_RESULT,
|
||
ContractOutcome,
|
||
build_save,
|
||
)
|
||
from repo import MemoryContractRepo, PurchaseContractRow
|
||
from run_batch import (
|
||
DEFAULT_CLAIM_LIMIT,
|
||
contract_lookup_title,
|
||
main as batch_main,
|
||
production_runner,
|
||
)
|
||
|
||
|
||
# 构造一条待录合同的表 3 行。
|
||
def _row(**overrides) -> PurchaseContractRow:
|
||
data = dict(
|
||
fid=1,
|
||
oms_id=2026100501,
|
||
supplier_id="SUP001",
|
||
supplier_name="示例供应商",
|
||
contract_sign_date=date(2026, 10, 1),
|
||
contract_start_date=date(2026, 10, 2),
|
||
contract_end_date=date(2027, 9, 30),
|
||
payment_term=90,
|
||
goods_id="G0001",
|
||
goods_name="阿莫西林胶囊",
|
||
reg_no="国药准字H20201234",
|
||
spec="0.25g*24粒",
|
||
unit="盒",
|
||
qty=Decimal("10.0000"),
|
||
unit_price=Decimal("100.0000"),
|
||
purchaser_id="P001",
|
||
purchaser_name="张三",
|
||
planner_id="PL001",
|
||
planner_name="李四",
|
||
contract_entry_status=None,
|
||
contract_no=None,
|
||
process_status=0,
|
||
org_code="901",
|
||
goods_mdm_code="QX0100167583",
|
||
supplier_mdm_code="DW0100046361",
|
||
)
|
||
data.update(overrides)
|
||
return PurchaseContractRow(**data)
|
||
|
||
|
||
# process_status=0 领走后变成 3，协议重录行和已完成行不领。
|
||
def test_claim_pending_sets_processing_and_skips_protocol_rerun():
|
||
repo = MemoryContractRepo(
|
||
[
|
||
_row(),
|
||
_row(fid=2, oms_id=2026100502, process_status=1),
|
||
_row(fid=3, oms_id=2026100503, contract_entry_status=0, contract_no="HT-OLD"),
|
||
]
|
||
)
|
||
claimed = repo.claim_pending()
|
||
assert [row.fid for row in claimed] == [1]
|
||
assert repo.table3[1].process_status == 3
|
||
assert repo.table3[3].process_status == 0
|
||
|
||
|
||
# process_status=3 且 process_time 超过租约的行可以再领。
|
||
def test_claim_reclaims_expired_status_3():
|
||
now = datetime(2026, 10, 22, 12, 0, 0)
|
||
repo = MemoryContractRepo(
|
||
[_row(process_status=3, process_time=now - timedelta(minutes=31))]
|
||
)
|
||
claimed = repo.claim_pending(now=now)
|
||
assert [row.fid for row in claimed] == [1]
|
||
assert repo.table3[1].process_status == 3
|
||
|
||
|
||
# process_status=3 且未过租约的行不领。
|
||
def test_claim_skips_fresh_status_3():
|
||
now = datetime(2026, 10, 22, 12, 0, 0)
|
||
repo = MemoryContractRepo(
|
||
[_row(process_status=3, process_time=now - timedelta(minutes=1))]
|
||
)
|
||
claimed = repo.claim_pending(now=now)
|
||
assert claimed == []
|
||
assert repo.table3[1].process_status == 3
|
||
|
||
|
||
# 协议重录行即使停在过期 3 也不领。
|
||
def test_claim_skips_protocol_rerun_even_if_expired():
|
||
now = datetime(2026, 10, 22, 12, 0, 0)
|
||
repo = MemoryContractRepo(
|
||
[
|
||
_row(
|
||
process_status=3,
|
||
process_time=now - timedelta(minutes=31),
|
||
contract_entry_status=0,
|
||
contract_no="HT-OLD",
|
||
)
|
||
]
|
||
)
|
||
assert repo.claim_pending(now=now) == []
|
||
assert repo.table3[1].process_status == 3
|
||
|
||
|
||
# 同一 oms_id 多行只插一条表 4，表 3 全部改成成功。
|
||
def test_save_group_inserts_one_table4_row_per_oms_id():
|
||
repo = MemoryContractRepo(
|
||
[
|
||
_row(),
|
||
_row(fid=2, oms_id=2026100501, goods_id="G0002"),
|
||
_row(fid=3, oms_id=2026100502, goods_id="G0003"),
|
||
]
|
||
)
|
||
repo.claim_pending()
|
||
outcome = ContractOutcome(KIND_OK, "HT20261005001", None)
|
||
payload = build_save(list(repo.table3.values()), outcome)
|
||
repo.save_group(payload)
|
||
assert [row["oms_id"] for row in repo.table4] == [2026100501, 2026100502]
|
||
assert {row["contract_no"] for row in repo.table4} == {"HT20261005001"}
|
||
assert {row["action_code"] for row in repo.table4} == {1}
|
||
assert {row["contract_status"] for row in repo.table4} == {1}
|
||
assert {row["process_status"] for row in repo.table4} == {0}
|
||
assert {row["create_user_name"] for row in repo.table4} == {"系统管理员"}
|
||
assert {row["update_user_name"] for row in repo.table4} == {"系统管理员"}
|
||
assert {row["push_status"] for row in repo.table4} == {0}
|
||
assert repo.table3[1].process_status == 1
|
||
assert repo.table3[3].process_status == 1
|
||
|
||
|
||
# 同一供应商合单一次录入，协议重录行不进本班。
|
||
def test_job_merges_same_supplier_and_writes_table4():
|
||
repo = MemoryContractRepo(
|
||
[
|
||
_row(),
|
||
_row(fid=2, oms_id=2026100502, goods_id="G0002"),
|
||
_row(fid=3, oms_id=2026100503, contract_entry_status=0, contract_no="HT-OLD"),
|
||
]
|
||
)
|
||
calls: list = []
|
||
|
||
# 记下草稿并返回成功合同号。
|
||
def runner(draft):
|
||
calls.append(draft)
|
||
return ContractOutcome(KIND_OK, "HT20261005001", None)
|
||
|
||
run_batch(repo, runner=runner)
|
||
assert len(calls) == 1
|
||
assert [line.goods_code for line in calls[0].goods_lines] == [
|
||
"QX0100167583",
|
||
"QX0100167583",
|
||
]
|
||
assert [line.mdm_code for line in calls[0].goods_lines] == [
|
||
"QX0100167583",
|
||
"QX0100167583",
|
||
]
|
||
assert calls[0].customer_code == "DW0100046361"
|
||
assert [row["oms_id"] for row in repo.table4] == [2026100501, 2026100502]
|
||
assert {row["contract_no"] for row in repo.table4} == {"HT20261005001"}
|
||
assert {row["action_code"] for row in repo.table4} == {1}
|
||
assert {row["contract_status"] for row in repo.table4} == {1}
|
||
assert {row["process_status"] for row in repo.table4} == {0}
|
||
assert {row["create_user_name"] for row in repo.table4} == {"系统管理员"}
|
||
assert {row["push_status"] for row in repo.table4} == {0}
|
||
assert repo.table3[1].process_status == 1
|
||
assert repo.table3[2].process_status == 1
|
||
assert repo.table3[3].process_status == 0
|
||
|
||
|
||
# 尽调失败写客户未尽调，表 3 改成 2，表 4 起草失败。
|
||
def test_job_marks_diligence_fail():
|
||
repo = MemoryContractRepo([_row()])
|
||
|
||
# 返回未尽调结果。
|
||
def runner(draft):
|
||
return ContractOutcome(KIND_DILIGENCE, None, "客户未尽调")
|
||
|
||
run_batch(repo, runner=runner)
|
||
assert repo.table3[1].process_status == 2
|
||
assert "客户未尽调" in repo.table3[1].process_result
|
||
assert repo.table4[0]["contract_status"] == 3
|
||
assert repo.table4[0]["process_status"] == 2
|
||
assert repo.table4[0]["update_user_name"] == "系统管理员"
|
||
assert repo.table4[0]["contract_no"] is None
|
||
assert repo.table4[0]["contract_fail_reason"] == "客户未尽调"
|
||
|
||
|
||
# 缺供应商 MDM 码时不调页面，直接业务失败。
|
||
def test_job_skips_runner_when_mdm_missing():
|
||
repo = MemoryContractRepo([_row(supplier_mdm_code="")])
|
||
called = {"n": 0}
|
||
|
||
# 本用例不应被调用。
|
||
def runner(draft):
|
||
called["n"] += 1
|
||
return ContractOutcome(KIND_OK, "HT-X", None)
|
||
|
||
run_batch(repo, runner=runner)
|
||
assert called["n"] == 0
|
||
assert repo.table3[1].process_status == 2
|
||
assert "缺少供应商MDM码" in repo.table3[1].process_result
|
||
|
||
|
||
# 页面尽调失败映射成 KIND_DILIGENCE。
|
||
def test_outcome_from_run_maps_diligence():
|
||
# 抛出未尽调。
|
||
def run():
|
||
raise DiligenceError("客户未尽调")
|
||
|
||
out = outcome_from_run(run)
|
||
assert out.kind == KIND_DILIGENCE
|
||
assert out.reason == "客户未尽调"
|
||
assert out.contract_no is None
|
||
|
||
|
||
# 提交成功但没读到合同号时记失败。
|
||
def test_outcome_from_run_maps_empty_number_to_fail():
|
||
out = outcome_from_run(lambda: "")
|
||
assert out.kind == KIND_FAIL
|
||
assert "未读取到真实合同编号" in (out.reason or "")
|
||
|
||
|
||
# 读到合同号时记成功。
|
||
def test_outcome_from_run_ok():
|
||
out = outcome_from_run(lambda: "HT9")
|
||
assert out == ContractOutcome(KIND_OK, "HT9", None)
|
||
|
||
|
||
# 页面 NodeError 仍写业务文案。
|
||
def test_outcome_from_run_maps_node_error():
|
||
# 抛出客商加载失败。
|
||
def run():
|
||
raise NodeError("客商加载失败")
|
||
|
||
out = outcome_from_run(run)
|
||
assert out.kind == KIND_FAIL
|
||
assert "客商加载失败" in (out.reason or "")
|
||
|
||
|
||
# 未捕获异常只写 PAGE_FAILED，不含原文。
|
||
def test_outcome_from_run_maps_unexpected_to_short_code():
|
||
# 抛出带 ORA 原文的异常。
|
||
def run():
|
||
raise RuntimeError("ORA-00001 unique constraint (RPA.UK)")
|
||
|
||
out = outcome_from_run(run)
|
||
assert out.kind == KIND_FAIL
|
||
assert out.reason == PAGE_FAILED
|
||
assert "ORA-" not in (out.reason or "")
|
||
assert out.error_text == "ORA-00001 unique constraint (RPA.UK)"
|
||
|
||
|
||
# runner 抛未捕获异常时表 3/表 4 原因列只写 PAGE_FAILED。
|
||
def test_job_maps_unexpected_error_to_short_code():
|
||
repo = MemoryContractRepo([_row()])
|
||
|
||
# 模拟页面未捕获异常。
|
||
def runner(draft):
|
||
raise RuntimeError("traceback here")
|
||
|
||
run_batch(repo, runner=runner)
|
||
assert repo.table3[1].process_status == 2
|
||
assert repo.table3[1].process_result == PAGE_FAILED
|
||
assert repo.table4[0]["contract_fail_reason"] == PAGE_FAILED
|
||
assert len(repo.shift_errors) == 1
|
||
assert repo.shift_errors[0].shift_name == "合同"
|
||
assert repo.shift_errors[0].biz_key == "SUP001"
|
||
assert repo.shift_errors[0].error_code == PAGE_FAILED
|
||
assert repo.shift_errors[0].error_text == "traceback here"
|
||
|
||
|
||
def test_default_limit_caps_claim_at_10():
|
||
"""不传 --limit 时合同班最多领 10 行。"""
|
||
repo = MemoryContractRepo(
|
||
[
|
||
_row(fid=i, oms_id=2026100500 + i, supplier_id=f"SUP{i}")
|
||
for i in range(1, 16)
|
||
]
|
||
)
|
||
seen: dict[str, int | None] = {}
|
||
original = repo.claim_pending
|
||
|
||
def claim_pending(limit=None, **kwargs):
|
||
"""记下入口传给领取的上限和实际领走的行数。"""
|
||
seen["limit"] = limit
|
||
claimed = original(limit=limit, **kwargs)
|
||
seen["count"] = len(claimed)
|
||
return claimed
|
||
|
||
repo.claim_pending = claim_pending
|
||
|
||
def runner(_draft):
|
||
"""不打开页面，直接记失败。"""
|
||
return ContractOutcome(KIND_FAIL, None, "skip")
|
||
|
||
batch_main([], repo=repo, runner=runner)
|
||
assert seen["limit"] == DEFAULT_CLAIM_LIMIT
|
||
assert seen["count"] == 10
|
||
assert seen["count"] <= DEFAULT_CLAIM_LIMIT
|
||
|
||
|
||
def test_unique_existing_contract_does_not_draft_again():
|
||
"""对方已有唯一合同号、本地没写上时，重跑不起草第二份。"""
|
||
repo = MemoryContractRepo([_row()])
|
||
drafted: list[str] = []
|
||
|
||
def finder(_browser, draft):
|
||
"""标题不含货主和起止日。对方只有一份合同时返回该号。"""
|
||
title = contract_lookup_title(draft)
|
||
assert title == "示例供应商采购协议20261001"
|
||
assert "901" not in title
|
||
assert "20261002" not in title
|
||
assert "20270930" not in title
|
||
return ["HT-EXISTING"]
|
||
|
||
def creator(_browser, _draft):
|
||
"""本用例不应起草。"""
|
||
drafted.append("created")
|
||
return "HT-NEW"
|
||
|
||
def runner(draft):
|
||
"""用注入的查重，不打开浏览器。"""
|
||
return production_runner(None, draft, find_contracts=finder, create=creator)
|
||
|
||
summary = run_batch(repo, runner=runner)
|
||
assert drafted == []
|
||
assert summary.ok == 1
|
||
assert repo.table4[0]["contract_no"] == "HT-EXISTING"
|
||
assert repo.table3[1].process_status == 1
|
||
|
||
|
||
def test_many_contracts_stay_in_process():
|
||
"""命中多条时停在处理中，不新建，不取最新一条。"""
|
||
repo = MemoryContractRepo([_row()])
|
||
drafted: list[str] = []
|
||
|
||
def finder(_browser, _draft):
|
||
"""两行都在，调用方不能挑最新的那条。"""
|
||
return ["HT-OLD", "HT-NEW"]
|
||
|
||
def creator(_browser, _draft):
|
||
"""命中多条时不应起草。"""
|
||
drafted.append("created")
|
||
return "HT-PICKED"
|
||
|
||
def runner(draft):
|
||
"""用注入的查重，不打开浏览器。"""
|
||
return production_runner(None, draft, find_contracts=finder, create=creator)
|
||
|
||
run_batch(repo, runner=runner)
|
||
assert drafted == []
|
||
assert repo.table4 == []
|
||
assert repo.table3[1].process_status == 3
|
||
assert repo.table3[1].process_result == "我的合同命中多条"
|
||
assert repo.claim_pending() == []
|
||
|
||
|
||
def test_missing_number_stays_in_process_and_next_shift_only_looks_up():
|
||
"""提交后读不到号停在处理中。下一班只查号，不再起草。"""
|
||
repo = MemoryContractRepo([_row()])
|
||
drafted: list[int] = []
|
||
found = {1: []}
|
||
|
||
def finder(_browser, draft):
|
||
"""第二班必须带着只查号标记。"""
|
||
if drafted:
|
||
assert draft.lookup_only is True
|
||
return found[1]
|
||
|
||
def creator(_browser, _draft):
|
||
"""只有第一班会进来，而且读不到号。"""
|
||
drafted.append(1)
|
||
return ""
|
||
|
||
def runner(draft):
|
||
"""用注入的查重和起草，不打开浏览器。"""
|
||
return production_runner(None, draft, find_contracts=finder, create=creator)
|
||
|
||
run_batch(repo, runner=runner)
|
||
assert drafted == [1]
|
||
assert repo.table4 == []
|
||
assert repo.table3[1].process_status == 3
|
||
assert repo.table3[1].process_result == LOOKUP_ONLY_RESULT
|
||
repo.table3[1].process_time = datetime.now() - timedelta(minutes=31)
|
||
found[1] = ["HT-LATER"]
|
||
run_batch(repo, runner=runner)
|
||
assert drafted == [1]
|
||
assert repo.table4[0]["contract_no"] == "HT-LATER"
|
||
assert repo.table3[1].process_status == 1
|
||
|
||
|
||
class HoldingContractRepo(MemoryContractRepo):
|
||
"""内存仓库加上可计数的 session，用来断言页面期间连接已归还。"""
|
||
|
||
def __init__(self, rows):
|
||
"""在内存行上加 held 计数。"""
|
||
super().__init__(rows)
|
||
self.held = 0
|
||
|
||
@contextmanager
|
||
def session(self):
|
||
"""借连接时 held 加一，退出时减回。"""
|
||
self.held += 1
|
||
try:
|
||
yield
|
||
finally:
|
||
self.held -= 1
|
||
|
||
|
||
def test_page_runs_after_connection_is_returned():
|
||
"""两组页面执行时连接已归还，落表时才再借。"""
|
||
repo = HoldingContractRepo(
|
||
[
|
||
_row(fid=1, supplier_id="SUP-A"),
|
||
_row(fid=2, oms_id=2026100502, supplier_id="SUP-B"),
|
||
]
|
||
)
|
||
during_page: list[int] = []
|
||
during_save: list[int] = []
|
||
original_claim = repo.claim_pending
|
||
original_save = repo.save_group
|
||
original_errors = repo.save_shift_errors
|
||
|
||
def claim_pending(limit=None, **kwargs):
|
||
"""领取时必须占着本班的连接。"""
|
||
assert repo.held == 1
|
||
return original_claim(limit=limit, **kwargs)
|
||
|
||
def save_group(payload):
|
||
"""落表时必须占着本班的连接。"""
|
||
during_save.append(repo.held)
|
||
return original_save(payload)
|
||
|
||
def save_shift_errors(items):
|
||
"""审计走连接池，不能在 session 里写。"""
|
||
assert repo.held == 0
|
||
return original_errors(items)
|
||
|
||
def runner(_draft):
|
||
"""页面执行时连接必须已经归还。"""
|
||
during_page.append(repo.held)
|
||
return ContractOutcome(KIND_OK, "HT-REUSE", None)
|
||
|
||
repo.claim_pending = claim_pending
|
||
repo.save_group = save_group
|
||
repo.save_shift_errors = save_shift_errors
|
||
summary = run_batch(repo, runner=runner)
|
||
assert during_page == [0, 0]
|
||
assert during_save == [1, 1]
|
||
assert repo.held == 0
|
||
assert summary.groups == 2
|
||
assert summary.ok == 2
|
||
|