AutoRpa/rpa_contract/tests/test_flow_gates.py

137 lines
4.0 KiB
Python

from pathlib import Path
from contract_flow_steps import (
ContractFlowContext,
begin_date_for,
check_customer_diligence,
sign_date_for,
_as_index_list,
_next_unused_goods_row,
_pick_contract_row,
_resolve_contract_body_path,
_submit_error_from_body,
)
from errors import DiligenceError
import pytest
def _ctx(**overrides) -> ContractFlowContext:
data = dict(
contract_url="http://example",
username="u",
password="p",
supplier_name="示例",
order_date="2026-09-28",
source_type="",
)
data.update(overrides)
return ContractFlowContext(**data)
class _FakeBrowser:
def __init__(self, text: str):
self._text = text
def read_body_text(self) -> str:
return self._text
def test_sign_and_begin_fall_back_to_order_date():
ctx = _ctx()
assert sign_date_for(ctx) == "2026-09-28"
assert begin_date_for(ctx) == "2026-09-28"
def test_sign_and_begin_use_contract_dates_when_set():
ctx = _ctx(contract_sign_date="2026-10-01", contract_start_date="2026-10-02")
assert sign_date_for(ctx) == "2026-10-01"
assert begin_date_for(ctx) == "2026-10-02"
def test_diligence_raises_when_page_says_not_done():
with pytest.raises(DiligenceError, match="客户未尽调"):
check_customer_diligence(_FakeBrowser("客户未尽调,不能起草"), _ctx())
def test_diligence_passes_when_text_absent():
check_customer_diligence(_FakeBrowser("客户已尽调"), _ctx())
def test_same_product_keeps_own_row():
assert _next_unused_goods_row([0, 1, 2], []) == 0
assert _next_unused_goods_row([0, 1, 2], [0]) == 1
assert _next_unused_goods_row([0, 1, 2], [0, 1]) == 2
assert _next_unused_goods_row([0], [0]) == -1
assert _next_unused_goods_row([], [0]) == -1
def test_goods_row_indexes_from_script_string():
assert _as_index_list("0,1,2") == [0, 1, 2]
assert _as_index_list("") == []
assert _as_index_list(None) == []
assert _as_index_list([0, 1]) == [0, 1]
def test_pick_contract_row_fd_id_beats_latest():
rows = [
{"no": "HT-NEW", "title": "别人的合同", "customer": "X", "href": "fdId=aaa"},
{
"no": "HT-OURS",
"title": "重庆亿昊药业有限公司采购协议20261002",
"customer": "DW0100046361",
"href": "kmAgreementApply.do?method=view&fdId=ours",
},
]
assert (
_pick_contract_row(
rows,
fd_id="ours",
title="重庆亿昊药业有限公司采购协议20261002",
customer_code="DW0100046361",
)
== "HT-OURS"
)
def test_pick_contract_row_title_and_customer():
rows = [
{"no": "HT-NEW", "title": "别人的合同", "customer": "X", "href": "fdId=aaa"},
{
"no": "HT-OURS",
"title": "重庆亿昊药业有限公司采购协议20261002",
"customer": "DW0100046361",
"href": "fdId=ours",
},
]
assert (
_pick_contract_row(
rows,
title="重庆亿昊药业有限公司采购协议20261002",
customer_code="DW0100046361",
)
== "HT-OURS"
)
def test_pick_contract_row_does_not_take_first_unmatched():
rows = [{"no": "HT-NEW", "title": "别人的合同", "customer": "X", "href": "fdId=aaa"}]
assert _pick_contract_row(rows, title="重庆亿昊", customer_code="DW0100046361") == ""
def test_submit_error_from_doc_missing_page():
body = "您的请求已提交,系统返回如下信息:\n操作失败!\n该文档不存在"
assert _submit_error_from_body(body) == "该文档不存在"
assert _submit_error_from_body("客户编码 提交") == ""
def test_submit_without_file_uses_placeholder_body():
ctx = _ctx(submit_mode="submit", source_type="")
path = _resolve_contract_body_path(ctx)
assert Path(path).is_file()
assert Path(path).stat().st_size > 64
def test_dry_run_without_file_skips_body():
ctx = _ctx(submit_mode="dry_run", source_type="")
assert _resolve_contract_body_path(ctx) == ""