zkxw/backend/tests/test_sample_customer_service_register.py

"""样本客户服务登记：空不覆盖 merge 规则。"""
import unittest
from datetime import date, datetime

from app.services.sample_customer_service_register import (
    has_any_customer_service_payload,
    merge_customer_service_fields,
)


class TestMergeCustomerServiceFields(unittest.TestCase):
    def test_merge_skips_empty_and_keeps_existing(self):
        existing = {"sms_send_status": "已发", "effective_address": "合肥"}
        incoming = {"sms_send_status": "", "effective_address": "亳州", "household_registered": "是"}
        merged = merge_customer_service_fields(existing, incoming)
        self.assertEqual(merged["sms_send_status"], "已发")
        self.assertEqual(merged["effective_address"], "亳州")
        self.assertEqual(merged["household_registered"], "是")

    def test_merge_all_sections_partial(self):
        existing = {"sms_send_status": "已发", "microbe_followup_status": None}
        incoming = {"microbe_followup_status": "回访成功"}
        merged = merge_customer_service_fields(existing, incoming)
        self.assertEqual(merged["sms_send_status"], "已发")
        self.assertEqual(merged["microbe_followup_status"], "回访成功")

    def test_merge_skips_missing_keys(self):
        existing = {"sms_send_date": None, "effective_address": "A"}
        incoming = {"effective_address": "B"}
        merged = merge_customer_service_fields(existing, incoming)
        self.assertEqual(merged["effective_address"], "B")
        self.assertIsNone(merged.get("sms_send_date"))

    def test_merge_skips_none_incoming(self):
        existing = {"sms_send_status": "已发", "effective_address": "合肥"}
        incoming = {"sms_send_status": None, "effective_address": "亳州"}
        merged = merge_customer_service_fields(existing, incoming)
        self.assertEqual(merged["sms_send_status"], "已发")
        self.assertEqual(merged["effective_address"], "亳州")

    def test_merge_skips_whitespace_only_incoming(self):
        existing = {"microbe_followup_status": "回访成功", "household_registered": "是"}
        incoming = {"microbe_followup_status": "   ", "household_registered": "否"}
        merged = merge_customer_service_fields(existing, incoming)
        self.assertEqual(merged["microbe_followup_status"], "回访成功")
        self.assertEqual(merged["household_registered"], "否")

    def test_merge_passes_through_dates(self):
        d = date(2026, 10, 13)
        dt = datetime(2026, 10, 13, 14, 30, 0)
        existing = {"sms_send_date": None}
        incoming = {"sms_send_date": d, "einvoice_send_time": dt}
        merged = merge_customer_service_fields(existing, incoming)
        self.assertEqual(merged["sms_send_date"], d)
        self.assertEqual(merged["einvoice_send_time"], dt)


class TestHasAnyCustomerServicePayload(unittest.TestCase):
    def test_empty_false(self):
        self.assertFalse(has_any_customer_service_payload({}))
        self.assertFalse(has_any_customer_service_payload({"sms_send_status": ""}))
        self.assertFalse(has_any_customer_service_payload({"sms_send_status": "  "}))

    def test_nonempty_true(self):
        self.assertTrue(has_any_customer_service_payload({"sms_send_status": "已发"}))
        self.assertTrue(has_any_customer_service_payload({"effective_address": "合肥"}))


if __name__ == "__main__":
    unittest.main()