SmartBizTrip/tests/test_review_fixes.py

"""业务流程回归测试：使用模拟模型、代理和数据库，不访问外部服务。"""

import asyncio
import json
import sys
import unittest
from pathlib import Path
from unittest.mock import AsyncMock, Mock

from langchain_core.messages import AIMessage
from langchain_core.runnables import RunnableLambda

from chat_service import ChatService
from memory import ConversationMemory


class FlowTests(unittest.IsolatedAsyncioTestCase):
    def service(self):
        service = ChatService.__new__(ChatService)
        service.memory = ConversationMemory()
        service.messages = []
        service.conversation_history = ""
        service.llm = RunnableLambda(lambda _: AIMessage(content="汇总结果"))
        return service

    async def test_city_stream_supplies_weather_even_when_unavailable(self):
        for weather in ("成都小雨", ""):
            service = self.service()
            prompts = []

            def model(prompt):
                prompts.append(prompt.to_string())
                return AIMessage(content="接待建议")

            service.llm = RunnableLambda(model)
            service._fetch_weather_for_city_service = AsyncMock(return_value=weather)
            service.intent_agent = Mock(
                return_value=(["city_service"], {"city_service": "成都接待"}, "")
            )
            reply = "".join([x async for x in service.chat_stream("成都接待")])
            self.assertEqual(reply, "接待建议")
            self.assertIn("目的地天气：" + weather, prompts[0])
            service._fetch_weather_for_city_service.assert_awaited_once_with("成都接待")

    async def test_multi_intent_stream_matches_saved_reply(self):
        service = self.service()
        service.intent_agent = Mock(
            return_value=(
                ["weather", "train"],
                {"weather": "天气", "train": "高铁"},
                "",
            )
        )

        async def call(intent, query):
            yield query

        service._call_agent_intent_stream = call
        reply = "".join([x async for x in service.chat_stream("天气和高铁")])
        self.assertEqual(reply, "天气\n\n高铁")
        self.assertEqual(service.messages[-1]["content"], reply)

    async def test_dependencies_reach_later_agent_in_both_modes(self):
        for streaming in (False, True):
            service = self.service()
            calls = []

            async def call(intent, query):
                calls.append((intent, query))
                return "航班MU123，800元" if intent == "flight" else "模拟预订"

            service._call_agent_intent = call
            # 刻意传入逆序计划，验证实际执行仍满足依赖。
            steps = [
                {
                    "step": 2,
                    "intent": "order",
                    "action": "订最便宜的航班",
                    "depends_on": 1,
                },
                {"step": 1, "intent": "flight", "action": "查询航班", "depends_on": 0},
            ]
            queries = {"flight": "查询航班", "order": "预订查询结果"}
            if streaming:
                _ = [x async for x in service._react_loop_stream(steps, queries)]
            else:
                await service.react_loop(steps, queries)
            self.assertEqual([x[0] for x in calls], ["flight", "order"])
            payload = json.loads(calls[1][1])
            self.assertEqual(
                payload["previous_results"][0]["result"], "航班MU123，800元"
            )
            self.assertEqual(payload["action"], "订最便宜的航班")

    async def test_invalid_plans_execute_nothing(self):
        for steps in (
            [{"step": 1, "depends_on": 2}],
            [{"step": 1, "depends_on": 2}, {"step": 2, "depends_on": 1}],
            [{"step": 1}, {"step": 1}],
        ):
            service = self.service()
            service.execute_step = AsyncMock()
            with self.assertRaises(ValueError):
                await service._execute_plan(steps, {})
            service.execute_step.assert_not_called()

    async def test_independent_steps_still_run_concurrently(self):
        service = self.service()
        started = set()
        ready = asyncio.Event()

        async def execute(step, queries, previous):
            started.add(step["step"])
            if len(started) == 2:
                ready.set()
            await asyncio.wait_for(ready.wait(), timeout=1)
            return "ok"

        service.execute_step = execute
        result = await service._execute_plan([{"step": 1}, {"step": 2}], {})
        self.assertEqual(len(result), 2)

    def test_complex_goes_to_planning(self):
        self.assertFalse(self.service()._should_skip_planning(["complex"]))
        self.assertTrue(self.service()._should_skip_planning(["weather"]))


class ProfileTests(unittest.TestCase):
    def test_replace_deletes_and_merge_keeps_other_keys(self):
        memory = ConversationMemory()
        memory.update_profile({"座位": "二等座", "舱位": "经济舱"})
        memory.update_profile({"座位": "一等座"})
        self.assertIn("舱位", memory.user_profile)
        memory.update_profile({"座位": "一等座"}, replace=True)
        self.assertEqual(memory.user_profile, {"座位": "一等座"})
        memory.update_profile({}, replace=True)
        self.assertEqual(memory.user_profile, {})

    def test_database_replace_deletes_stale_rows_and_commits(self):
        memory = ConversationMemory()
        db = Mock()
        memory.set_db_connection(db)
        memory.update_profile({"座位": "二等座"}, replace=True)
        calls = db.cursor.return_value.execute.call_args_list
        self.assertEqual(calls[0].args, ("DELETE FROM user_profiles",))
        self.assertEqual(calls[1].args[1], ("座位", "二等座", "二等座"))
        db.commit.assert_called_once()
        memory.update_profile({}, replace=True)
        self.assertEqual(
            db.cursor.return_value.execute.call_args.args,
            ("DELETE FROM user_profiles",),
        )

    def test_failed_replace_rolls_back_memory_and_database(self):
        memory = ConversationMemory()
        memory.user_profile = {"座位": "二等座"}
        db = Mock()
        db.cursor.return_value.execute.side_effect = RuntimeError("模拟数据库失败")
        memory.set_db_connection(db)
        with self.assertRaises(RuntimeError):
            memory.update_profile({}, replace=True)
        self.assertEqual(memory.user_profile, {"座位": "二等座"})
        db.rollback.assert_called_once()
        db.commit.assert_not_called()


class SimulationTests(unittest.TestCase):
    def test_all_booking_tools_explicitly_report_simulation_without_database(self):
        sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
        from mcp_server.mcp_ticket_server import TicketService
        from mcp_server.mcp_trip_server import TripService

        ticket = TicketService.__new__(TicketService)
        trip = TripService.__new__(TripService)
        replies = [
            ticket.order_train("2026-10-01", "G1", "二等座", 1),
            ticket.order_flight("2026-10-01", "MU123", "经济舱", 1),
            ticket.order_event("2026-10-01", "活动", "场馆", "普通票", 1),
            trip.order_car_rental("2026-10-01", "SUV", 1),
            trip.order_meeting_package("2026-10-01", "套餐", 1),
            trip.order_insurance("综合型", "2026-10-01", 1),
        ]
        for reply in replies:
            self.assertIn("模拟", reply)
            self.assertIn("未实际下单", reply)


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