zkxw/backend/tests/test_presign_order.py

"""预签约订单：意向金状态机与转正约束单测"""

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()