SmartBizTrip/tests/test_student_release.py

"""学生版的小修正回归测试：使用模拟对象，不调用真实模型、天气或数据库。"""
import contextlib
import importlib
import io
import json
import os
import runpy
import unittest
from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import Mock, patch

ROOT = Path(__file__).resolve().parents[1]


class StudentReleaseTests(unittest.TestCase):
    def test_course_config_uses_matching_environment_names(self):
        with patch.dict(os.environ, {
            "DEEPSEEK_API_KEY": "offline-chat-placeholder",
            "DASHSCOPE_API_KEY": "offline-embedding-placeholder",
            "QWEATHER_API_KEY": "offline-weather-placeholder",
            "QWEATHER_API_HOST": "weather.example.invalid",
        }):
            namespace = runpy.run_path(str(ROOT / "课件示例/项目实现/02-02.py"))
            config = namespace["Config"]()
        self.assertEqual(config.api_key, "offline-chat-placeholder")
        self.assertEqual(config.embedding_api_key, "offline-embedding-placeholder")
        self.assertEqual(config.weather_api_key, "offline-weather-placeholder")
        self.assertEqual(config.weather_api_host, "weather.example.invalid")

    def test_intent_prompt_receives_current_date(self):
        from main_prompts import SmartBizTripPrompts
        text = SmartBizTripPrompts.intent_prompt().format(
            query="明天天气", current_date="2026-10-09", user_profile="",
            task_context="{}", conversation_history="",
        )
        self.assertIn("当前日期：2026-10-09", text)

    def test_weather_timezone_and_update_boundary(self):
        from utils import spider_weather as spider
        now = spider.TZ.localize(datetime(2026, 10, 9, 12))
        with patch.object(spider, "datetime") as clock:
            clock.now.return_value = now
            self.assertFalse(spider.should_update_data((now-timedelta(hours=23, minutes=59)).replace(tzinfo=None)))
            self.assertTrue(spider.should_update_data((now-timedelta(hours=24)).replace(tzinfo=None)))
            self.assertTrue(spider.should_update_data(None))
            self.assertTrue(spider.should_update_data(now, force_update=True))

    def test_config_demo_runs_without_missing_method(self):
        with contextlib.redirect_stdout(io.StringIO()):
            runpy.run_path(str(ROOT / "config.py"), run_name="__main__")

    def test_ticket_task_passes_text_to_executor(self):
        from a2a_server import ticket_server as ticket
        from python_a2a import Task, Message, TextContent, MessageRole
        executor = Mock()
        executor.invoke.return_value = {"output": "未找到符合条件的车次"}
        message = Message(content=TextContent(text="北京到成都高铁"), role=MessageRole.USER)
        task = Task(id="offline-ticket-test", message=message.to_dict())
        with patch.object(ticket, "to_langchain_tool", return_value=[]), \
             patch.object(ticket, "create_tool_calling_agent"), \
             patch.object(ticket, "AgentExecutor", return_value=executor):
            result = ticket.TicketServer().handle_task(task)
        self.assertEqual(executor.invoke.call_args.args[0]["input"], "北京到成都高铁")
        self.assertEqual(result.artifacts[0]["parts"][0]["text"], "未找到符合条件的车次")

    def test_weather_api_start_does_not_connect_mysql(self):
        from mcp_server import mcp_weather_server as weather
        with patch.object(weather.conf, "weather_source", "api"), \
             patch.object(weather, "WeatherService") as db, \
             patch.object(weather, "create_fastapi_app", return_value="mock-app"), \
             patch.object(weather.uvicorn, "run") as serve, \
             contextlib.redirect_stdout(io.StringIO()):
            weather.create_weather_mcp_server()
        db.assert_not_called()
        serve.assert_called_once_with("mock-app", host="0.0.0.0", port=8002)

    def test_weather_http_calls_have_timeout_and_status_check(self):
        from mcp_server import mcp_weather_server as weather
        geo, forecast = Mock(), Mock()
        geo.json.return_value = {"location": [{"id": "101010100"}]}
        forecast.json.return_value = {"daily": []}
        with patch.object(weather.requests, "get", side_effect=[geo, forecast]) as get:
            result = json.loads(weather.fetch_weather_from_api("北京", "2026-10-09", "2026-10-09"))
        self.assertEqual(result["status"], "no_data")
        self.assertTrue(all(c.kwargs["timeout"] == 15 for c in get.call_args_list))
        geo.raise_for_status.assert_called_once()
        forecast.raise_for_status.assert_called_once()

    def test_city_filter_keeps_quotes_inside_literal(self):
        from mcp_server import mcp_trip_server as trip
        collection = Mock()
        collection.search.return_value = []
        city = '成都" or city != "北京'
        with patch("pymilvus.connections.connect"), \
             patch("pymilvus.Collection", return_value=collection), \
             patch.object(trip, "get_embedding", return_value=[0.0]):
            trip.search_meeting_packages_in_milvus("会务", city)
        self.assertEqual(collection.search.call_args.kwargs["expr"],
                         "city == " + json.dumps(city, ensure_ascii=False))

    def test_manual_weather_script_does_not_request_on_import(self):
        with patch("requests.get") as get:
            runpy.run_path(str(ROOT / "test.py"), run_name="manual_weather_import_test")
        get.assert_not_called()

    def test_imports_stay_in_this_copy(self):
        for name in ("config", "chat_service", "mcp_server.mcp_weather_server",
                     "mcp_server.mcp_ticket_server", "mcp_server.mcp_trip_server"):
            module = importlib.import_module(name)
            self.assertTrue(Path(module.__file__).resolve().is_relative_to(ROOT))