178 lines
6.9 KiB
Python
178 lines
6.9 KiB
Python
"""当事人三角色客户同步 / 撞库合并"""
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from app.services.party_customer_sync import (
|
|
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"})
|