270 lines
9.5 KiB
Python
270 lines
9.5 KiB
Python
"""预签约订单：意向金状态机与转正约束单测"""
|
|
|
|
import unittest
|
|
|
|
from app.services.protocol_transfer_status import can_confirm_receipt, can_submit_receipt
|
|
|
|
|
|
class TestPresignTransferStatus(unittest.TestCase):
|
|
def test_presign_transfer_same_status_machine(self):
|
|
self.assertTrue(can_submit_receipt("dfj"))
|
|
self.assertFalse(can_submit_receipt("jfqrz"))
|
|
self.assertFalse(can_submit_receipt("yjf"))
|
|
self.assertTrue(can_confirm_receipt("jfqrz"))
|
|
self.assertFalse(can_confirm_receipt("dfj"))
|
|
|
|
def test_can_withdraw_only_jfqrz(self):
|
|
from app.services.protocol_transfer_status import can_withdraw_receipt
|
|
|
|
self.assertTrue(can_withdraw_receipt("jfqrz"))
|
|
self.assertFalse(can_withdraw_receipt("dfj"))
|
|
self.assertFalse(can_withdraw_receipt("yjf"))
|
|
self.assertFalse(can_withdraw_receipt(None))
|
|
self.assertFalse(can_withdraw_receipt(""))
|
|
|
|
|
|
class TestFinanceTransferFields(unittest.TestCase):
|
|
def test_sample_type_names_helper_exists(self):
|
|
from app.crud.presign_order import _sample_type_names
|
|
|
|
self.assertTrue(callable(_sample_type_names))
|
|
|
|
|
|
class TestConvertRules(unittest.TestCase):
|
|
def test_convert_requires_active_status(self):
|
|
from app.crud.presign_order import assert_presign_convertible
|
|
|
|
with self.assertRaises(ValueError):
|
|
assert_presign_convertible(order_status="converted")
|
|
with self.assertRaises(ValueError):
|
|
assert_presign_convertible(order_status="void")
|
|
assert_presign_convertible(order_status="active")
|
|
assert_presign_convertible(order_status="draft")
|
|
|
|
|
|
class TestConvertStaffMapping(unittest.TestCase):
|
|
"""转正式：采集专员→采集专员+市场专员；采集经理→地区经理。"""
|
|
|
|
def test_collector_fills_collector_and_sales(self):
|
|
from app.crud.presign_order import resolve_convert_staff_fields
|
|
|
|
out = resolve_convert_staff_fields(
|
|
collector_user_id="c1",
|
|
sales_user_id="s1",
|
|
region_manager_user_id="m1",
|
|
fallback_user_id="u0",
|
|
)
|
|
self.assertEqual(out["collector_user_id"], "c1")
|
|
self.assertEqual(out["sales_user_id"], "c1")
|
|
self.assertEqual(out["region_manager_user_id"], "m1")
|
|
|
|
def test_collector_fallback_to_operator_when_empty(self):
|
|
from app.crud.presign_order import resolve_convert_staff_fields
|
|
|
|
out = resolve_convert_staff_fields(
|
|
collector_user_id=None,
|
|
sales_user_id="s1",
|
|
region_manager_user_id="",
|
|
fallback_user_id="u0",
|
|
)
|
|
self.assertEqual(out["collector_user_id"], "u0")
|
|
self.assertEqual(out["sales_user_id"], "u0")
|
|
self.assertIsNone(out["region_manager_user_id"])
|
|
|
|
|
|
class TestProtocolDepositAssociation(unittest.TestCase):
|
|
"""正式协议意向金通过 source_presign_order_id 关联预签，不落快照表。"""
|
|
|
|
def test_list_deposit_transfers_for_protocol_uses_presign_source(self):
|
|
from app.crud.protocol_sample_payment import list_deposit_transfers_for_protocol
|
|
|
|
# 无来源预签时返回空列表
|
|
class _FakeDb:
|
|
pass
|
|
|
|
self.assertEqual(
|
|
list_deposit_transfers_for_protocol(
|
|
_FakeDb(), source_presign_order_id=None, tenant_no="t1"
|
|
),
|
|
[],
|
|
)
|
|
self.assertEqual(
|
|
list_deposit_transfers_for_protocol(
|
|
_FakeDb(), source_presign_order_id=" ", tenant_no="t1"
|
|
),
|
|
[],
|
|
)
|
|
|
|
|
|
class TestPresignWithdrawReceipt(unittest.TestCase):
|
|
def test_withdraw_sets_dfj_keeps_finance_remark(self):
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
from app.crud import presign_order as po
|
|
|
|
t = SimpleNamespace(
|
|
id="t1",
|
|
pay_status="jfqrz",
|
|
finance_remark="keep-me",
|
|
finance_confirm_attachments="http://x",
|
|
update_by=None,
|
|
update_time=None,
|
|
)
|
|
db = MagicMock()
|
|
orig = po._get_transfer_or_raise
|
|
orig_update = po._update_transfer_expected_status
|
|
po._get_transfer_or_raise = lambda *a, **k: t
|
|
|
|
def atomic_update(*args, **kwargs):
|
|
for column, value in kwargs["values"].items():
|
|
setattr(t, column.key, value)
|
|
db.commit()
|
|
return t
|
|
|
|
po._update_transfer_expected_status = atomic_update
|
|
try:
|
|
out = po.withdraw_receipt(
|
|
db, transfer_id="t1", tenant_no="zkxw", user_id="u1"
|
|
)
|
|
self.assertEqual(out.pay_status, "dfj")
|
|
self.assertEqual(out.finance_remark, "keep-me")
|
|
self.assertEqual(out.finance_confirm_attachments, "http://x")
|
|
db.commit.assert_called()
|
|
finally:
|
|
po._get_transfer_or_raise = orig
|
|
po._update_transfer_expected_status = orig_update
|
|
|
|
def test_withdraw_rejects_non_jfqrz(self):
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
from app.crud import presign_order as po
|
|
|
|
t = SimpleNamespace(id="t1", pay_status="yjf", finance_remark="x")
|
|
db = MagicMock()
|
|
orig = po._get_transfer_or_raise
|
|
po._get_transfer_or_raise = lambda *a, **k: t
|
|
try:
|
|
with self.assertRaises(ValueError) as ctx:
|
|
po.withdraw_receipt(
|
|
db, transfer_id="t1", tenant_no="zkxw", user_id="u1"
|
|
)
|
|
self.assertIn("撤回", str(ctx.exception))
|
|
finally:
|
|
po._get_transfer_or_raise = orig
|
|
|
|
|
|
class TestPresignProtocolNoAllocate(unittest.TestCase):
|
|
"""预签约协议编号：保存发号、忽略客户端改号、空号补发。"""
|
|
|
|
def test_assert_biz_no_rejects_presign_occupied(self):
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
from app.crud.protocol_donation_convert import assert_biz_no_available
|
|
from app.models.models import BizCollectionProtocol, BizPresignOrder
|
|
from app.models.research_donation import BizResearchDonation
|
|
|
|
db = MagicMock()
|
|
|
|
def query(model):
|
|
q = MagicMock()
|
|
if model is BizPresignOrder:
|
|
q.filter.return_value.all.return_value = [
|
|
SimpleNamespace(id="presign-1")
|
|
]
|
|
else:
|
|
q.filter.return_value.all.return_value = []
|
|
return q
|
|
|
|
db.query.side_effect = query
|
|
with self.assertRaises(ValueError) as ctx:
|
|
assert_biz_no_available(db, tenant_no="t1", biz_no="XBCC20261012000001")
|
|
self.assertIn("协议编号已存在", str(ctx.exception))
|
|
|
|
# 排除自身预签后可通过
|
|
assert_biz_no_available(
|
|
db,
|
|
tenant_no="t1",
|
|
biz_no="XBCC20261012000001",
|
|
exclude_presign_order_id="presign-1",
|
|
)
|
|
# 静默使用导入，避免 lint 误报
|
|
_ = (BizCollectionProtocol, BizResearchDonation)
|
|
|
|
def test_create_allocates_and_ignores_client_protocol_no(self):
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.crud import presign_order as po
|
|
|
|
db = MagicMock()
|
|
with patch.object(
|
|
po, "allocate_next_biz_no", return_value="XBCC20261012000009"
|
|
) as alloc, patch.object(po, "replace_sample_types"):
|
|
obj = po.create_presign_order(
|
|
db,
|
|
tenant_no="t1",
|
|
user_id="u1",
|
|
fields={"protocolNo": "CLIENT-NO", "motherName": "张三"},
|
|
commit=False,
|
|
)
|
|
self.assertEqual(obj.protocol_no, "XBCC20261012000009")
|
|
alloc.assert_called_once()
|
|
self.assertEqual(alloc.call_args.kwargs.get("kind"), "collection")
|
|
|
|
def test_update_ignores_client_change_and_backfills_empty(self):
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.crud import presign_order as po
|
|
|
|
kept = SimpleNamespace(
|
|
id="o1",
|
|
protocol_no="KEEP-NO",
|
|
order_status="active",
|
|
pay_mode=None,
|
|
contract_amount=None,
|
|
sample_collect_time=None,
|
|
update_by=None,
|
|
update_time=None,
|
|
)
|
|
empty = SimpleNamespace(
|
|
id="o2",
|
|
protocol_no=None,
|
|
order_status="active",
|
|
pay_mode=None,
|
|
contract_amount=None,
|
|
sample_collect_time=None,
|
|
update_by=None,
|
|
update_time=None,
|
|
)
|
|
db = MagicMock()
|
|
with patch.object(po, "_get_order_or_raise", return_value=kept):
|
|
po.update_presign_order(
|
|
db,
|
|
order_id="o1",
|
|
tenant_no="t1",
|
|
user_id="u1",
|
|
fields={"protocolNo": "HACK", "motherName": "李四"},
|
|
commit=False,
|
|
)
|
|
self.assertEqual(kept.protocol_no, "KEEP-NO")
|
|
self.assertEqual(kept.mother_name, "李四")
|
|
|
|
with patch.object(po, "_get_order_or_raise", return_value=empty), patch.object(
|
|
po, "allocate_next_biz_no", return_value="XBCC20261012000010"
|
|
) as alloc:
|
|
po.update_presign_order(
|
|
db,
|
|
order_id="o2",
|
|
tenant_no="t1",
|
|
user_id="u1",
|
|
fields={"motherName": "王五"},
|
|
commit=False,
|
|
)
|
|
self.assertEqual(empty.protocol_no, "XBCC20261012000010")
|
|
alloc.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|