zkxw/backend/tests/test_party_customer_sync.py

"""当事人三角色客户同步 / 撞库合并"""
import unittest
from unittest.mock import MagicMock, patch

from app.services.party_customer_sync import (
    apply_role_profile_overwrite,
    merge_customer_fields,
    sync_party_customers,
    ROLE_MOTHER,
    ROLE_FATHER,
    ROLE_NEWBORN,
)


class TestMergeCustomerFields(unittest.TestCase):
    def test_fill_empty_only_and_union_tags(self):
        target = MagicMock(
            customer_name="已有",
            phone_no="111",
            age=None,
            sex="2",
            remark="",
            customer_tags="vip",
        )
        source = MagicMock(
            phone_no="222",
            age=30,
            sex="1",
            remark="备注A",
            customer_tags="a,vip",
        )
        merge_customer_fields(target, source)
        self.assertEqual(target.phone_no, "111")  # 不覆盖
        self.assertEqual(target.age, 30)
        self.assertEqual(target.sex, "1")  # 目标为未知才补
        self.assertEqual(target.remark, "备注A")
        tags = set((target.customer_tags or "").split(","))
        self.assertEqual(tags, {"vip", "a"})
        self.assertEqual(target.customer_name, "已有")


class TestSyncPartyCustomers(unittest.TestCase):
    def test_role_constants(self):
        self.assertEqual(ROLE_MOTHER, "mother")
        self.assertEqual(ROLE_FATHER, "father")
        self.assertEqual(ROLE_NEWBORN, "newborn")

    def test_no_name_clears_role_id(self):
        db = MagicMock()
        party = {
            "motherName": "",
            "customerId": "old-m",
            "fatherName": "",
            "fatherCustomerId": "old-f",
        }
        out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
        self.assertEqual(out.get("customer_id") or out.get("customerId"), "")
        self.assertEqual(out.get("father_customer_id") or out.get("fatherCustomerId"), "")

    def test_three_names_without_id_card_creates_three(self):
        db = MagicMock()
        created = []

        def fake_create(db_, **kwargs):
            obj = MagicMock()
            obj.id = f"c{len(created)+1}"
            created.append(kwargs)
            return obj

        party = {
            "motherName": "妈",
            "fatherName": "爸",
            "newbornName": "宝",
            "newbornSex": "0",
        }
        with patch("app.services.party_customer_sync.customer_crud.create_customer", side_effect=fake_create):
            with patch("app.services.party_customer_sync.find_customer_by_id_card", return_value=None):
                with patch("app.services.party_customer_sync.get_customer", return_value=None):
                    out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
        self.assertEqual(len(created), 3)
        self.assertTrue(out.get("customer_id") or out.get("customerId"))
        self.assertTrue(out.get("father_customer_id") or out.get("fatherCustomerId"))
        self.assertTrue(out.get("newborn_customer_id") or out.get("newbornCustomerId"))
        sexes = {c["customer_name"]: c.get("sex") for c in created}
        self.assertEqual(sexes["妈"], "1")
        self.assertEqual(sexes["爸"], "0")
        self.assertEqual(sexes["宝"], "0")

    def test_snake_case_input_clears_and_keeps_other_fields(self):
        db = MagicMock()
        party = {
            "mother_name": "",
            "customer_id": "old-m",
            "remark": "keep-me",
        }
        out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
        self.assertEqual(out.get("customer_id"), "")
        self.assertEqual(out.get("remark"), "keep-me")

    def test_name_and_id_card_keeps_linked_id_when_not_found(self):
        db = MagicMock()
        linked = MagicMock()
        linked.id = "cust-a"
        linked.customer_name = "旧名"
        linked.phone_no = None
        linked.id_card_no = None
        party = {
            "mother_name": "妈",
            "mother_phone": "13900000000",
            "mother_id_card": "110101199001011234",
            "customer_id": "cust-a",
        }
        with patch("app.services.party_customer_sync.find_customer_by_id_card", return_value=None):
            with patch("app.services.party_customer_sync.get_customer", return_value=linked):
                with patch("app.services.party_customer_sync.customer_crud.create_customer") as create:
                    out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
        create.assert_not_called()
        self.assertEqual(out.get("customer_id"), "cust-a")
        self.assertEqual(linked.customer_name, "妈")
        self.assertEqual(linked.phone_no, "13900000000")
        self.assertEqual(linked.id_card_no, "110101199001011234")

    def test_id_card_collision_merges_and_repoints(self):
        db = MagicMock()
        customer_a = MagicMock(
            id="A",
            customer_name="临时",
            phone_no="111",
            age=None,
            sex="2",
            remark="",
            customer_tags="tmp",
        )
        customer_b = MagicMock(
            id="B",
            customer_name="已有",
            phone_no="",
            age=None,
            sex="2",
            remark="",
            customer_tags="vip",
        )
        party = {
            "motherName": "妈",
            "motherIdCard": "110101199001011234",
            "customerId": "A",
        }

        def fake_get(db_, customer_id, tenant_no=None):
            mapping = {"A": customer_a, "B": customer_b}
            return mapping.get(customer_id)

        with patch("app.services.party_customer_sync.find_customer_by_id_card", return_value=customer_b):
            with patch("app.services.party_customer_sync.get_customer", side_effect=fake_get):
                with patch("app.services.party_customer_sync.repoint_customer_refs") as repoint:
                    with patch(
                        "app.services.party_customer_sync.customer_crud.soft_delete_customer"
                    ) as soft_delete:
                        out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
        self.assertEqual(out.get("customer_id") or out.get("customerId"), "B")
        repoint.assert_called_once()
        args, kwargs = repoint.call_args
        if args and len(args) >= 4:
            self.assertEqual(args[1], "A")
            self.assertEqual(args[2], "B")
            self.assertEqual(args[3], "zkxw")
        else:
            self.assertEqual(kwargs.get("from_id"), "A")
            self.assertEqual(kwargs.get("to_id"), "B")
            self.assertEqual(kwargs.get("tenant_no"), "zkxw")
        soft_delete.assert_called_once()
        sd_args, sd_kwargs = soft_delete.call_args
        deleted_id = sd_args[1] if len(sd_args) > 1 else sd_kwargs.get("customer_id")
        self.assertEqual(deleted_id, "A")
        self.assertEqual(customer_b.phone_no, "111")
        self.assertEqual(customer_b.customer_name, "妈")
        tags = set((customer_b.customer_tags or "").split(","))
        self.assertEqual(tags, {"vip", "tmp"})


class TestApplyRoleProfileOverwrite(unittest.TestCase):
    def test_nonempty_overwrites_empty_skips(self):
        target = MagicMock(
            customer_name="旧名",
            phone_no="111",
            id_card_no="A",
            sex="1",
            age=20,
            birth_ym="199001",
            ethnic="汉族",
            email="old@x.com",
            address="旧址",
        )
        apply_role_profile_overwrite(
            target,
            MagicMock(
                customer_name="新名",
                phone_no="222",
                id_card_no="",
                sex="1",
                age=None,
                birth_ym="199102",
                ethnic="",
                email="new@x.com",
                address=None,
            ),
        )
        self.assertEqual(target.customer_name, "新名")
        self.assertEqual(target.phone_no, "222")
        self.assertEqual(target.id_card_no, "A")
        self.assertEqual(target.birth_ym, "199102")
        self.assertEqual(target.ethnic, "汉族")
        self.assertEqual(target.email, "new@x.com")
        self.assertEqual(target.address, "旧址")

    def test_mother_sync_overwrites_extended_fields_on_linked(self):
        db = MagicMock()
        linked = MagicMock(
            id="cust-a",
            customer_name="旧名",
            phone_no="111",
            id_card_no="110101199001011234",
            sex="1",
            age=30,
            birth_ym="199001",
            ethnic="回族",
            email="old@x.com",
            address="旧址",
        )
        party = {
            "mother_name": "妈",
            "mother_phone": "13900000000",
            "mother_id_card": "110101199001011234",
            "mother_birth_ym": "199002",
            "mother_ethnic": "汉族",
            "mother_email": "new@x.com",
            "mother_address": "新址",
            "customer_id": "cust-a",
        }
        with patch("app.services.party_customer_sync.find_customer_by_id_card", return_value=None):
            with patch("app.services.party_customer_sync.get_customer", return_value=linked):
                with patch("app.services.party_customer_sync.customer_crud.create_customer") as create:
                    out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
        create.assert_not_called()
        self.assertEqual(out.get("customer_id"), "cust-a")
        self.assertEqual(linked.customer_name, "妈")
        self.assertEqual(linked.phone_no, "13900000000")
        self.assertEqual(linked.birth_ym, "199002")
        self.assertEqual(linked.ethnic, "汉族")
        self.assertEqual(linked.email, "new@x.com")
        self.assertEqual(linked.address, "新址")