"""业务流程回归测试：使用模拟模型、代理和数据库，不访问外部服务。""" 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()