98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
"""关联核验必须使用真实类型字典，不能依靠恰好相同的跨表 ID。"""
|
|
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from app.services.tplus.relationship_audit import TPlusRelationshipAudit
|
|
|
|
|
|
def _audit(types=None):
|
|
source = Mock()
|
|
rows = {
|
|
"source_version": [{"source_version": "16.0.4165.4"}],
|
|
"voucher_types": types
|
|
if types is not None
|
|
else [
|
|
{"id": 9104, "code": "SA04", "name": "销货单"},
|
|
{"id": 9092, "code": "ARAP01", "name": "收款单"},
|
|
],
|
|
"state_enums": [{"id": 189, "code": "01", "name": "生效", "enum_name": "VoucherState"}],
|
|
"receipt_kinds": [
|
|
{
|
|
"voucher_type_code": "ARAP01",
|
|
"voucher_type_name": "收款单",
|
|
"receive_flag": 1,
|
|
"row_count": 10,
|
|
}
|
|
],
|
|
"receipt_line_kinds": [
|
|
{"voucher_type_id": 9104, "voucher_type_name": "销货单", "row_count": 12}
|
|
],
|
|
"sales_links": [
|
|
{
|
|
"row_count": 12,
|
|
"delivery_matches": 12,
|
|
"detail_matches": 7,
|
|
"direct_order_matches": 4,
|
|
"invalid_order_ids": 0,
|
|
}
|
|
],
|
|
"order_coverage": [
|
|
{"row_count": 12, "without_order": 2, "single_order": 8, "multiple_orders": 2}
|
|
],
|
|
}
|
|
audit = TPlusRelationshipAudit(source)
|
|
audit.fetch = Mock(side_effect=lambda name, **kw: rows[name])
|
|
return audit
|
|
|
|
|
|
def test_类型标识从字典读取而不是硬编码104():
|
|
audit = _audit()
|
|
result = audit.run()
|
|
args = next(call.kwargs for call in audit.fetch.call_args_list if call.args[0] == "sales_links")
|
|
assert args == {"sale_delivery_type_id": 9104, "receive_flag": 1}
|
|
assert result["sales_links"]["delivery_matches"] == 12
|
|
|
|
|
|
def test_枚举名称按源库保留且2022不通过2008验收():
|
|
result = _audit().run()
|
|
assert result["state_enums"][0]["name"] == "生效"
|
|
assert result["is_target_engine"] is False
|
|
assert result["compatibility_status"] == "PENDING"
|
|
|
|
|
|
def test_同类型字典有歧义就停止该链路统计不猜ID():
|
|
audit = _audit(
|
|
[
|
|
{"id": 1, "code": "SA04", "name": "A"},
|
|
{"id": 2, "code": "SA04", "name": "B"},
|
|
{"id": 3, "code": "ARAP01", "name": "收款单"},
|
|
]
|
|
)
|
|
result = audit.run()
|
|
assert result["relationship_status"] == "UNRESOLVED"
|
|
assert result["sales_links"] is None
|
|
assert not any(call.args[0] == "sales_links" for call in audit.fetch.call_args_list)
|
|
|
|
|
|
def test_多订单和成本日期仍需业务确认不自动选择():
|
|
result = _audit().run()
|
|
assert result["order_coverage"]["multiple_orders"] == 2
|
|
keys = {item["key"] for item in result["pending_confirmations"]}
|
|
assert {"cost_basis", "receipt_date", "ship_date", "allocation"} <= keys
|
|
|
|
|
|
def test_SQL查询有固定白名单且不接受自定义表名():
|
|
audit = TPlusRelationshipAudit(Mock())
|
|
with pytest.raises(ValueError, match="核验查询"):
|
|
audit.fetch("SELECT * FROM customers")
|
|
|
|
|
|
def test_主表来源单据类型为空不妨碍按是否收款核验():
|
|
# 字典明确 idvouchertype 是“来源单据类型”，不是当前收款单的类型。
|
|
audit = _audit([{"id": 9104, "code": "SA04", "name": "销货单"}])
|
|
assert audit.run()["relationship_status"] == "OBSERVED"
|
|
assert "p.isReceiveFlag = :receive_flag" in audit.queries["sales_links"]
|
|
assert "p.idvouchertype =" not in audit.queries["sales_links"]
|