# -*- coding: utf-8 -*-
"""RegistrationStore 与本地 Grok 客服 MCP 共用数据文件的并发测试。"""

from __future__ import annotations

import json
import tempfile
import threading
import unittest
from pathlib import Path
from unittest import mock

import customer_service_policy as policy
from registration_store import RegistrationStore, process_registration_reply


class RegistrationStoreConcurrencyTest(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp_dir.cleanup)
        self.path = Path(self.temp_dir.name) / "registration_leads.json"
        self.path_patch = mock.patch.object(
            policy,
            "REGISTRATIONS_PATH",
            self.path,
        )
        self.path_patch.start()
        self.addCleanup(self.path_patch.stop)

    def _mcp_register(self, session_id: str) -> dict:
        return policy.record_registration(
            session_id=session_id,
            customer_message="帮我预约一下，我最近空腹血糖偏高",
            contact_name="测试客户",
        )

    def test_existing_store_refreshes_after_mcp_write(self) -> None:
        store = RegistrationStore(str(self.path))
        result = self._mcp_register("00112233445566778899aabbccddeeff")

        self.assertTrue(result["registered"])
        leads = store.list_leads()
        self.assertEqual(1, len(leads))
        self.assertEqual("pending_human_confirmation", leads[0]["status"])
        self.assertEqual(1, store.pending_count())

    def test_ui_mutation_reloads_and_preserves_new_mcp_lead(self) -> None:
        store = RegistrationStore(str(self.path))
        self._mcp_register("00112233445566778899aabbccddeeff")

        ui_lead = store.add_or_update(
            session_id="ui-session",
            contact="UI 客户",
            symptom="待补充",
            status="pending_symptom",
        )

        payload = json.loads(self.path.read_text(encoding="utf-8"))
        sessions = {item["session_id"] for item in payload["leads"]}
        self.assertEqual(
            {"00112233445566778899aabbccddeeff", "ui-session"},
            sessions,
        )
        self.assertIn(ui_lead["id"], {item["id"] for item in payload["leads"]})

    def test_parallel_ui_and_mcp_writes_do_not_lose_records(self) -> None:
        store = RegistrationStore(str(self.path))
        barrier = threading.Barrier(2)
        errors: list[BaseException] = []

        def mcp_writer() -> None:
            try:
                barrier.wait()
                for index in range(10):
                    self._mcp_register(f"{index + 1:032x}")
            except BaseException as exc:  # pragma: no cover - assertion below
                errors.append(exc)

        def ui_writer() -> None:
            try:
                barrier.wait()
                for index in range(10):
                    store.add_or_update(
                        session_id=f"ui-session-{index}",
                        contact=f"UI 客户 {index}",
                        symptom="待补充",
                        status="pending_symptom",
                    )
            except BaseException as exc:  # pragma: no cover - assertion below
                errors.append(exc)

        threads = [
            threading.Thread(target=mcp_writer),
            threading.Thread(target=ui_writer),
        ]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join(timeout=10)

        self.assertFalse(any(thread.is_alive() for thread in threads))
        self.assertEqual([], errors)
        leads = store.list_leads()
        self.assertEqual(20, len(leads))
        self.assertEqual(20, len({item["session_id"] for item in leads}))

    def test_save_failure_is_not_swallowed(self) -> None:
        store = RegistrationStore(str(self.path))
        with mock.patch.object(
            policy,
            "_atomic_write_json",
            side_effect=policy.LocalStoreError("模拟保存失败"),
        ):
            with self.assertRaisesRegex(policy.LocalStoreError, "模拟保存失败"):
                store.add_or_update(
                    session_id="ui-session",
                    contact="UI 客户",
                    symptom="待补充",
                    status="pending_symptom",
                )

        self.assertFalse(self.path.exists())

    def test_pending_count_refreshes_all_pending_statuses(self) -> None:
        self.path.write_text(
            json.dumps(
                {
                    "leads": [
                        {"id": "a", "status": "pending_symptom"},
                        {"id": "b", "status": "pending_human_confirmation"},
                        {"id": "c", "status": "booked"},
                        {"id": "d", "status": "done"},
                    ]
                },
                ensure_ascii=False,
            ),
            encoding="utf-8",
        )
        store = RegistrationStore(str(self.path))
        self.assertEqual(3, store.pending_count())

    def test_legacy_registration_helper_never_claims_booking_success(self) -> None:
        store = RegistrationStore(str(self.path))
        reply, lead = process_registration_reply(
            session_id="legacy-session",
            user_text="请帮我预约一下，我最近空腹血糖偏高",
            reply_text="我先帮您处理。",
            store=store,
        )

        self.assertIsNotNone(lead)
        self.assertEqual("pending_human_confirmation", lead["status"])
        self.assertIn("人工", reply)
        self.assertIn("尚未预约成功", reply)
        self.assertNotIn("已帮您预约", reply)


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