"""当事人三角色客户同步 / 撞库合并""" 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, "新址")