Coverage for tests/test_mcp_tools.py: 97%
250 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Tests for MCP tools."""
3from types import SimpleNamespace
4from unittest.mock import AsyncMock
6import pytest
7from conftest import course_payload
8from fastmcp import Client
9from httpx import Response
10from pydantic import TypeAdapter
12from data_api.domain.announcements import services as announcements_services
13from data_api.domain.buses import services as buses_services
14from data_api.domain.departments import services as departments_services
15from data_api.domain.dining import services as dining_services
16from data_api.domain.energy import services as energy_services
17from data_api.domain.locations import services as locations_services
18from data_api.domain.newsletters import services as newsletters_services
19from data_api.mcp.server import mcp
20from data_api.mcp.tools.announcements import _get_announcements
21from data_api.mcp.tools.buses import _get_bus_schedule
22from data_api.mcp.tools.courses import _search_courses
23from data_api.mcp.tools.dining import _find_dining
24from data_api.mcp.tools.directory import _search_directory
25from data_api.mcp.tools.energy import _get_energy_usage
26from data_api.mcp.tools.library import _get_library_info
27from data_api.mcp.tools.locations import _search_locations
28from data_api.mcp.tools.newsletters import _get_newsletters
29from data_api.mcp.tools.result_types import (
30 AnnouncementsResult,
31 BusResult,
32 CoursesResult,
33 DiningBuildingsResult,
34 DirectoryResult,
35 EnergyResult,
36 LibraryItemsResult,
37 LibrarySpacesResult,
38 LocationsResult,
39 NewslettersResult,
40)
41from data_api.models.courses import CourseData
42from data_api.runtime import get_runtime
44pytestmark = pytest.mark.usefixtures("dataset_runtime")
47class TestMCPTools:
48 """Tests for MCP tools functionality."""
50 async def test_search_directory(self):
51 """Test directory search tool."""
52 result = await _search_directory(query="教務處")
53 assert set(result) == {"departments", "people"}
54 assert "departments" in result
55 assert "people" in result
56 assert isinstance(result["departments"], list)
57 assert isinstance(result["people"], list)
59 async def test_search_locations(self):
60 """Test location search."""
61 result = await _search_locations(query="校門")
62 assert set(result) == {"locations"}
63 assert isinstance(result["locations"], list)
65 async def test_get_bus_schedule_default(self):
66 """Test bus schedules with default parameters."""
67 result = await _get_bus_schedule()
68 assert "current_time" in result
69 assert "day_type" in result
70 assert "buses" in result
71 assert "route_info" in result
72 assert isinstance(result["buses"], list)
74 async def test_get_bus_schedule_main_route(self):
75 """Test schedules for the main campus route."""
76 result = await _get_bus_schedule(route="main", direction="up", limit=3)
77 assert "buses" in result
78 assert isinstance(result["buses"], list)
80 async def test_get_bus_schedule_nanda_route(self):
81 """Test schedules for the Nanda route."""
82 result = await _get_bus_schedule(route="nanda", direction="down", limit=3)
83 assert "buses" in result
84 assert isinstance(result["buses"], list)
86 async def test_search_courses_by_keyword(self):
87 """Test course search by keyword."""
88 result = await _search_courses(keyword="微積分", limit=5)
89 assert "count" in result
90 assert "courses" in result
91 assert isinstance(result["courses"], list)
93 async def test_search_courses_by_teacher(self):
94 """Test course search by teacher."""
95 result = await _search_courses(teacher="王", limit=5)
96 assert "count" in result
97 assert "courses" in result
98 assert isinstance(result["courses"], list)
100 async def test_search_courses_no_filter(self):
101 """Test course search without filter returns all courses."""
102 result = await _search_courses(limit=10)
103 assert "count" in result
104 assert "courses" in result
105 assert isinstance(result["courses"], list)
107 async def test_get_announcements(self):
108 """Test get announcements."""
109 result = await _get_announcements(limit=5)
110 assert "count" in result
111 assert "sources" in result
112 assert isinstance(result["sources"], list)
114 async def test_get_announcements_with_department(self):
115 """Test get announcements with department filter."""
116 result = await _get_announcements(department="學生", limit=5)
117 assert "count" in result
118 assert "sources" in result
119 assert isinstance(result["sources"], list)
121 async def test_find_dining(self):
122 """Test find dining without filters."""
123 result = await _find_dining()
124 assert "buildings" in result
125 assert isinstance(result["buildings"], list)
127 async def test_find_dining_by_building(self):
128 """Test find dining by building."""
129 result = await _find_dining(building="小吃部")
130 assert "buildings" in result
131 assert isinstance(result["buildings"], list)
133 async def test_find_dining_check_open(self):
134 """Test find dining with open check."""
135 result = await _find_dining(check_open="today")
136 assert set(result) == {"buildings"}
137 assert isinstance(result["buildings"], list)
139 @pytest.mark.parametrize("empty", [False, True])
140 async def test_get_library_info_space(self, mock_upstream, library_space_payload, empty):
141 if empty:
142 library_space_payload["rows"] = []
143 mock_upstream(lambda request: Response(200, json=library_space_payload))
144 result = await _get_library_info(info_type="space")
145 assert result == {
146 "spaces": (
147 [] if empty else [{"zone": "Main library", "type": "Study room", "available": "12"}]
148 )
149 }
151 @pytest.mark.parametrize("count", [0, 12])
152 async def test_get_library_info_lost_and_found(self, mock_upstream, lost_items_html, count):
153 mock_upstream(lambda request: Response(200, text=lost_items_html(count)))
154 result = await _get_library_info(info_type="lost_and_found")
155 assert result == {
156 "items": [
157 {
158 "序號": str(index),
159 "拾獲時間": "2026-09-30",
160 "拾獲地點": "Main library",
161 "描述": "Book title",
162 }
163 for index in range(min(count, 10))
164 ]
165 }
167 async def test_get_newsletters(self):
168 """Test get newsletters."""
169 result = await _get_newsletters()
170 assert "count" in result
171 assert "newsletters" in result
172 assert isinstance(result["newsletters"], list)
174 async def test_get_newsletters_with_search(self):
175 """Test get newsletters with search."""
176 result = await _get_newsletters(search="教務處")
177 assert "count" in result
178 assert "newsletters" in result
179 assert isinstance(result["newsletters"], list)
181 async def test_get_energy_usage(self, mock_upstream):
182 mock_upstream(lambda request: Response(200, text='<img alt="kW: 0">'))
183 result = await _get_energy_usage()
184 assert len(result["zones"]) == 3
185 for zone in result["zones"]:
186 assert set(zone) == {
187 "name",
188 "usage_kw",
189 "capacity_kw",
190 "usage_percent",
191 "last_updated",
192 }
193 assert zone["usage_kw"] == 0
194 assert zone["usage_percent"] == 0
196 async def test_get_bus_schedule_with_stop(self):
197 """Test schedule queries include the selected stop's metadata."""
198 from data_api.domain.buses.enums import BusStopsName
200 result = await _get_bus_schedule(stop=BusStopsName.M1)
201 assert "stop_info" in result
202 assert "stop_name" in result
203 assert result["stop_name"] == "北校門口"
204 assert "current_time" in result
205 assert "day_type" in result
206 assert "buses" in result
207 assert isinstance(result["buses"], list)
209 async def test_get_bus_schedule_with_stop_name_string(self):
210 """Test MCP converts a stop name string to its enum."""
211 from fastmcp import Client
213 from data_api.mcp.server import mcp
215 async with Client(mcp) as client:
216 response = await client.call_tool("get_bus_schedule", {"stop": "台積館"})
217 result = response.data
218 assert "stop_info" in result
219 assert "stop_name" in result
220 assert "buses" in result
221 assert isinstance(result["buses"], list)
224class TestMCPToolIntegration:
225 """Integration tests for MCP tools."""
227 async def test_search_directory_returns_limited_results(self):
228 """Test that directory search limits results appropriately."""
229 result = await _search_directory(query="系")
230 assert len(result["departments"]) <= 5
231 assert len(result["people"]) <= 10
233 async def test_search_locations_returns_limited_results(self):
234 result = await _search_locations(query="館")
235 assert len(result["locations"]) <= 5
237 async def test_get_bus_schedule_respects_limit(self):
238 """Test that bus results respect limit parameter."""
239 result = await _get_bus_schedule(limit=3)
240 assert len(result["buses"]) <= 3
242 async def test_search_courses_respects_limit(self):
243 """Test that course search respects limit parameter."""
244 result = await _search_courses(limit=5)
245 assert len(result["courses"]) <= 5
247 async def test_course_search_result_format(self):
248 """Test that course search returns expected fields."""
249 result = await _search_courses(limit=1)
250 if result["courses"]:
251 course = result["courses"][0]
252 assert set(course) == set(CourseData.model_fields)
253 CourseData.model_validate(course)
255 async def test_announcements_result_format(self):
256 """Test that announcements return expected structure."""
257 result = await _get_announcements(limit=1)
258 if result["sources"]:
259 source = result["sources"][0]
260 assert "department" in source
261 assert "articles" in source
262 if source["articles"]:
263 article = source["articles"][0]
264 assert "title" in article
265 assert "link" in article
267 async def test_get_bus_schedule_with_stop_respects_limit(self):
268 """Test that stop-filtered schedules respect the limit."""
269 from data_api.domain.buses.enums import BusStopsName
271 result = await _get_bus_schedule(stop=BusStopsName.M1, limit=2)
272 assert len(result["buses"]) <= 2
275class TestRegisteredPayloads:
276 """Exercise the private payload shapes through the registered MCP transport."""
278 @pytest.mark.parametrize(
279 ("name", "arguments", "result_type", "expected"),
280 [
281 (
282 "search_directory",
283 {"query": "not-found"},
284 DirectoryResult,
285 {
286 "departments": [],
287 "people": [],
288 },
289 ),
290 ("search_locations", {"query": "not-found"}, LocationsResult, {"locations": []}),
291 (
292 "get_bus_schedule",
293 {"day": "weekday"},
294 BusResult,
295 {
296 "buses": [],
297 "route_info": [],
298 },
299 ),
300 (
301 "search_courses",
302 {"keyword": "not-found"},
303 CoursesResult,
304 {
305 "count": 0,
306 "courses": [],
307 },
308 ),
309 (
310 "get_announcements",
311 {},
312 AnnouncementsResult,
313 {
314 "count": 0,
315 "sources": [],
316 },
317 ),
318 ("find_dining", {}, DiningBuildingsResult, {"buildings": []}),
319 (
320 "find_dining",
321 {"check_open": "weekday"},
322 DiningBuildingsResult,
323 {"buildings": []},
324 ),
325 (
326 "get_newsletters",
327 {"search": "not-found"},
328 NewslettersResult,
329 {
330 "count": 0,
331 "newsletters": [],
332 },
333 ),
334 ],
335 )
336 async def test_empty_dataset_payloads(self, name, arguments, result_type, expected):
337 async with Client(mcp) as client:
338 response = await client.call_tool(name, arguments)
339 assert not response.is_error
340 TypeAdapter(result_type).validate_python(response.data)
341 assert expected.items() <= response.data.items()
343 async def test_nested_dataset_payloads_and_nulls(self, monkeypatch):
344 monkeypatch.setattr(
345 locations_services.locations_service,
346 "fuzzy_search_locations",
347 AsyncMock(
348 return_value=(
349 "fixture",
350 [{"name": "Main gate", "latitude": "24.79", "longitude": "120.99"}],
351 )
352 ),
353 )
354 monkeypatch.setattr(
355 departments_services.departments_service,
356 "search_departments_and_people",
357 AsyncMock(
358 return_value=(
359 "fixture",
360 {
361 "departments": [
362 {
363 "index": "01",
364 "name": "Office",
365 "details": {"people": [{"name": "Alice", "title": None}]},
366 }
367 ],
368 "people": [{"name": "Alice", "title": None}],
369 },
370 )
371 ),
372 )
373 monkeypatch.setattr(
374 announcements_services.announcements_service,
375 "fuzzy_search_announcements",
376 AsyncMock(
377 return_value=(
378 "fixture",
379 [
380 {
381 "department": "Office",
382 "language": None,
383 "articles": [
384 {"title": "Notice", "link": "https://example.com", "date": None}
385 ],
386 }
387 ],
388 )
389 ),
390 )
391 monkeypatch.setattr(
392 dining_services.dining_service,
393 "fuzzy_search_dining_data",
394 AsyncMock(
395 return_value=(
396 "fixture",
397 [
398 {
399 "building": "Hall",
400 "restaurants": [
401 {
402 "name": "Cafe",
403 "phone": None,
404 "schedule": {"weekday": "09:00-17:00"},
405 "note": "",
406 }
407 ],
408 }
409 ],
410 )
411 ),
412 )
413 monkeypatch.setattr(
414 newsletters_services.newsletters_service,
415 "get_all_newsletters",
416 AsyncMock(
417 return_value=(
418 "fixture",
419 [
420 {
421 "name": "Newsletter",
422 "link": None,
423 "articles": [{"title": "Issue", "date": None}],
424 }
425 ],
426 )
427 ),
428 )
429 cases = [
430 (
431 "search_directory",
432 {"query": "Alice"},
433 DirectoryResult,
434 {
435 "departments": [
436 {
437 "index": "01",
438 "name": "Office",
439 "details": {"people": [{"name": "Alice", "title": None}]},
440 }
441 ],
442 "people": [{"name": "Alice", "title": None}],
443 },
444 ),
445 (
446 "search_locations",
447 {"query": "Main"},
448 LocationsResult,
449 {"locations": [{"name": "Main gate", "latitude": "24.79", "longitude": "120.99"}]},
450 ),
451 (
452 "get_announcements",
453 {},
454 AnnouncementsResult,
455 {
456 "count": 1,
457 "sources": [
458 {
459 "department": "Office",
460 "language": None,
461 "articles": [
462 {"title": "Notice", "link": "https://example.com", "date": None}
463 ],
464 }
465 ],
466 },
467 ),
468 (
469 "find_dining",
470 {},
471 DiningBuildingsResult,
472 {
473 "buildings": [
474 {
475 "building": "Hall",
476 "restaurants": [
477 {
478 "name": "Cafe",
479 "phone": None,
480 "schedule": {"weekday": "09:00-17:00"},
481 "note": None,
482 }
483 ],
484 }
485 ],
486 },
487 ),
488 (
489 "find_dining",
490 {"check_open": "weekday"},
491 DiningBuildingsResult,
492 {
493 "buildings": [
494 {
495 "building": "Hall",
496 "restaurants": [
497 {
498 "name": "Cafe",
499 "phone": None,
500 "schedule": {"weekday": "09:00-17:00"},
501 "note": None,
502 }
503 ],
504 }
505 ],
506 },
507 ),
508 (
509 "get_newsletters",
510 {},
511 NewslettersResult,
512 {
513 "count": 1,
514 "newsletters": [
515 {
516 "name": "Newsletter",
517 "link": None,
518 "latest_articles": [{"title": "Issue", "date": None}],
519 }
520 ],
521 },
522 ),
523 ]
524 async with Client(mcp) as client:
525 for name, arguments, result_type, expected in cases:
526 response = await client.call_tool(name, arguments)
527 assert not response.is_error
528 assert response.data == expected
529 TypeAdapter(result_type).validate_python(response.data)
531 async def test_bus_simple_detailed_and_stop_payloads(self, monkeypatch):
532 from data_api.domain.buses.enums import BusStopsName
534 prepared = buses_services.BusesService.prepare(
535 {
536 "towardTSMCBuildingInfo": {
537 "direction": "up",
538 "duration": "2026",
539 "route": "main-up",
540 "routeEN": "Main up",
541 },
542 "weekdayBusScheduleTowardTSMCBuilding": [
543 {"time": "08:00", "description": "", "dep_stop": "校門", "line": "red"}
544 ],
545 }
546 )
547 monkeypatch.setattr(
548 get_runtime(),
549 "buses",
550 SimpleNamespace(
551 update_data=AsyncMock(),
552 query_schedule=prepared.query_schedule,
553 get_route_info=prepared.get_route_info,
554 gen_bus_stops_info=prepared.gen_bus_stops_info,
555 ),
556 )
557 async with Client(mcp) as client:
558 for details in (False, True):
559 response = await client.call_tool(
560 "get_bus_schedule",
561 {
562 "route": "main",
563 "direction": "up",
564 "day": "weekday",
565 "details": details,
566 "stop": BusStopsName.M1.value,
567 },
568 )
569 assert not response.is_error
570 result = response.data
571 TypeAdapter(BusResult).validate_python(result)
572 assert result["route_info"] == [
573 {
574 "direction": "up",
575 "duration": "2026",
576 "route": "main-up",
577 "routeEN": "Main up",
578 }
579 ]
580 assert result["stop_name"] == BusStopsName.M1.value
581 assert result["stop_info"][0]["name"] == BusStopsName.M1.value
582 assert {"name_en", "latitude", "longitude"} <= result["stop_info"][0].keys()
583 assert result["buses"]
584 departure = result["buses"][0]["dep_info"] if details else result["buses"][0]
585 assert departure["time"] == "08:00"
586 assert departure["line"] == "main_red"
587 if details:
588 assert {"stop", "arrive_time"} <= result["buses"][0]["stops_time"][0].keys()
589 else:
590 assert "stops_time" not in result["buses"][0]
592 async def test_course_payload_preserves_normalized_values(self):
593 async with Client(mcp) as client:
594 response = await client.call_tool("search_courses", {"limit": 1})
595 assert not response.is_error
596 TypeAdapter(CoursesResult).validate_python(response.data)
597 assert response.data["count"] == 1
598 assert response.data["courses"][0] == course_payload()
600 async def test_live_payloads_and_empty_results(
601 self,
602 monkeypatch,
603 mock_upstream,
604 library_space_payload,
605 lost_items_html,
606 lost_items_empty_page,
607 ):
608 async with Client(mcp) as client:
609 mock_upstream(lambda request: Response(200, json=library_space_payload))
610 space = await client.call_tool("get_library_info", {"info_type": "space"})
611 assert space.data == {
612 "spaces": [
613 {
614 "zone": "Main library",
615 "type": "Study room",
616 "available": "12",
617 }
618 ]
619 }
620 TypeAdapter(LibrarySpacesResult).validate_python(space.data)
622 library_space_payload["rows"] = []
623 empty_space = await client.call_tool("get_library_info", {"info_type": "space"})
624 assert empty_space.data == {"spaces": []}
626 mock_upstream(lambda request: Response(200, text=lost_items_html(1)))
627 items = await client.call_tool("get_library_info", {"info_type": "lost_and_found"})
628 assert items.data == {
629 "items": [
630 {
631 "序號": "0",
632 "拾獲時間": "2026-09-30",
633 "拾獲地點": "Main library",
634 "描述": "Book title",
635 }
636 ]
637 }
638 TypeAdapter(LibraryItemsResult).validate_python(items.data)
640 mock_upstream(lambda request: Response(200, text=lost_items_empty_page))
641 empty_items = await client.call_tool(
642 "get_library_info", {"info_type": "lost_and_found"}
643 )
644 assert empty_items.data == {"items": []}
646 mock_upstream(lambda request: Response(200, text='<img alt="kW: 0">'))
647 energy = await client.call_tool("get_energy_usage", {})
648 TypeAdapter(EnergyResult).validate_python(energy.data)
649 assert len(energy.data["zones"]) == 3
650 assert all(
651 zone["usage_kw"] == 0 and zone["usage_percent"] == 0
652 for zone in energy.data["zones"]
653 )
655 monkeypatch.setattr(
656 energy_services.energy_service,
657 "get_realtime_electricity_usage",
658 AsyncMock(return_value=[]),
659 )
660 empty_energy = await client.call_tool("get_energy_usage", {})
661 assert empty_energy.data == {"zones": []}
662 TypeAdapter(EnergyResult).validate_python(empty_energy.data)