Coverage for tests/test_mcp_search.py: 100%
260 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Offline regression tests for MCP search composition and validation."""
3from copy import deepcopy
4from datetime import datetime
5from itertools import product
7import pytest
8from conftest import make_course, set_courses
9from fastmcp import Client
10from pydantic import ValidationError
12from data_api.core.exceptions import DataNotAvailableException
13from data_api.data.manager import nthudata
14from data_api.domain.dining import services as dining_services
15from data_api.mcp.server import mcp
16from data_api.mcp.tools.courses import _search_courses
17from data_api.mcp.tools.dining import _find_dining
18from data_api.runtime import get_runtime
20FILTER_COMBINATIONS = list(product([False, True], repeat=3))
23@pytest.fixture(autouse=True)
24def fixed_course_refresh(monkeypatch):
25 set_courses(monkeypatch, [])
28@pytest.fixture
29def course_candidates(monkeypatch):
30 candidates = [
31 (
32 flags,
33 make_course(
34 {
35 "id": f"{'11510ISS 509900' if flags[2] else '11510OTHER'}-{index}",
36 "chinese_title": "服務科學導論" if flags[0] else "微積分",
37 "english_title": "Service Science" if flags[0] else "Calculus",
38 "lecturers": [{"label": "Teacher" if flags[1] else "Someone else"}],
39 }
40 ),
41 )
42 for index, flags in enumerate(FILTER_COMBINATIONS)
43 ]
44 set_courses(monkeypatch, [course for _, course in candidates])
45 return candidates
48@pytest.mark.parametrize("selected", FILTER_COMBINATIONS)
49@pytest.mark.parametrize("keyword", ["服務科學導論", "Service"])
50async def test_course_filters_all_contribute(course_candidates, selected, keyword):
51 filters = dict(
52 zip(("keyword", "teacher", "course_id"), (keyword, "Teacher", "11510ISS 509900"))
53 )
54 arguments = {
55 name: value for (name, value), enabled in zip(filters.items(), selected) if enabled
56 }
57 expected_ids = [
58 course.id
59 for flags, course in course_candidates
60 if all(not enabled or matches for enabled, matches in zip(selected, flags))
61 ]
63 result = await _search_courses(**arguments)
64 assert [course["id"] for course in result["courses"]] == expected_ids
65 assert result["count"] == len(expected_ids)
67 for name in arguments:
68 impossible = {**arguments, name: "NTHU_REVIEW_NONEXISTENT_20260928"}
69 assert await _search_courses(**impossible, limit=1) == {"count": 0, "courses": []}
72@pytest.mark.parametrize(
73 ("argument", "field"),
74 [
75 ("keyword", "chinese_title"),
76 ("keyword", "english_title"),
77 ("teacher", "teacher"),
78 ("course_id", "id"),
79 ],
80)
81@pytest.mark.parametrize(
82 ("literal", "decoy"),
83 [("C++", "CCC"), ("[AI]", "AI"), ("(", "ordinary"), (".", "ordinary"), ("\\", "ordinary")],
84)
85async def test_course_search_uses_literal_substrings(monkeypatch, argument, field, literal, decoy):
86 value, other_value = f"prefix {literal} suffix", decoy
87 if field == "teacher":
88 field, value, other_value = "lecturers", [{"label": value}], [{"label": other_value}]
89 if field == "id":
90 match = make_course(id=f"11510prefix {literal} suffix")
91 other = make_course(id=f"11510{decoy}")
92 else:
93 match = make_course({"id": "11510match", field: value})
94 other = make_course({"id": "11510decoy", field: other_value})
95 set_courses(monkeypatch, [other, match])
97 result = await _search_courses(**{argument: literal})
98 assert [course["id"] for course in result["courses"]] == [match.id]
101@pytest.mark.parametrize("limit", [-100, 0, 101, 10**9])
102async def test_course_limit_rejected_by_implementation(limit):
103 with pytest.raises(ValidationError):
104 await _search_courses(limit=limit)
107async def test_course_limit_schema_and_protocol_validation(monkeypatch):
108 set_courses(monkeypatch, [make_course(id=f"11510{index}") for index in range(105)])
109 async with Client(mcp) as client:
110 tools = await client.list_tools()
111 tool = next(tool for tool in tools if tool.name == "search_courses")
112 limit_schema = tool.input_schema["properties"]["limit"]
113 assert limit_schema["minimum"] == 1
114 assert limit_schema["maximum"] == 100
115 assert limit_schema["default"] == 20
117 for limit in [-100, 0, 101, 10**9]:
118 result = await client.call_tool(
119 "search_courses", {"limit": limit}, raise_on_error=False
120 )
121 assert result.is_error
123 for arguments, expected_count in [({}, 20), ({"limit": 1}, 1), ({"limit": 100}, 100)]:
124 result = await client.call_tool("search_courses", arguments)
125 assert result.data["count"] == expected_count
126 assert len(result.data["courses"]) == expected_count
129@pytest.fixture
130def dining_candidates(monkeypatch):
131 data = [
132 {
133 "building": building,
134 "restaurants": [
135 {
136 "name": f"{name}-{building_index}-{int(is_open)}",
137 "area": building,
138 "phone": "",
139 "schedule": "09:00-17:00",
140 "note": "" if is_open else "平日休息、週六休息、週日休息",
141 }
142 for name, is_open in product(["Pizza", "Sushi"], [False, True])
143 ],
144 }
145 for building_index, building in enumerate(["小吃部", "水木生活中心"])
146 ]
147 original = deepcopy(data)
148 calls = []
150 async def fake_get(endpoint):
151 assert endpoint == "dining.json"
152 calls.append(endpoint)
153 return "testhash", data
155 class Saturday(datetime):
156 @classmethod
157 def now(cls, tz=None):
158 return cls(2026, 9, 26, tzinfo=tz)
160 monkeypatch.setattr(nthudata, "get", fake_get)
161 monkeypatch.setattr(dining_services, "datetime", Saturday)
162 yield data, calls
163 assert data == original
166@pytest.mark.parametrize("selected", FILTER_COMBINATIONS)
167@pytest.mark.parametrize("schedule", ["today", "weekday", "saturday", "sunday"])
168async def test_dining_filters_all_contribute(dining_candidates, selected, schedule):
169 data, calls = dining_candidates
170 building, name, check_open = selected
171 arguments = {}
172 if building:
173 arguments["building"] = "水木"
174 if name:
175 arguments["restaurant_name"] = "Sushi"
176 if check_open:
177 arguments["check_open"] = schedule
179 result = await _find_dining(**arguments)
180 assert set(result) == {"buildings"}
181 restaurants = [r for b in result["buildings"] for r in b["restaurants"]]
183 expected_names = [
184 r["name"]
185 for b in data
186 if not building or b["building"] == "水木生活中心"
187 for r in b["restaurants"]
188 if (not name or r["name"].startswith("Sushi")) and (not check_open or not r["note"])
189 ]
190 assert [r["name"] for r in restaurants] == expected_names
191 assert calls == ["dining.json"]
194async def test_dining_filters_before_limiting(monkeypatch):
195 data = [
196 {"building": "小吃部", "restaurants": [{"name": "Sushi"} for _ in range(25)]},
197 {
198 "building": "水木生活中心",
199 "restaurants": (
200 [{"name": "Pizza"} for _ in range(25)]
201 + [{"name": "Sushi", "note": "平日休息、週六休息、週日休息"} for _ in range(25)]
202 + [{"name": "Sushi match"}]
203 ),
204 },
205 ]
207 async def fake_get(endpoint):
208 assert endpoint == "dining.json"
209 return "testhash", data
211 monkeypatch.setattr(nthudata, "get", fake_get)
212 result = await _find_dining(building="水木", restaurant_name="Sushi", check_open="today")
213 assert [b["building"] for b in result["buildings"]] == ["水木生活中心"]
214 assert [r["name"] for r in result["buildings"][0]["restaurants"]] == ["Sushi match"]
217@pytest.mark.parametrize("arguments", [{"building": "ZZZZZZ"}, {"restaurant_name": "ZZZZZZ"}])
218async def test_dining_impossible_filter_with_open_status(dining_candidates, arguments):
219 result = await _find_dining(**arguments, check_open="today")
220 assert result == {"buildings": []}
223@pytest.mark.parametrize("check_open", [None, "today", "weekday", "saturday", "sunday"])
224@pytest.mark.parametrize(
225 ("building", "expected_buildings"),
226 [
227 (None, ["小吃部", "水木生活中心", "水木餐廳"]),
228 ("", ["小吃部", "水木生活中心", "水木餐廳"]),
229 ("小吃部", ["小吃部"]),
230 ("水木", ["水木生活中心", "水木餐廳"]),
231 ("水木生活中心", ["水木生活中心", "水木餐廳"]),
232 ],
233)
234async def test_dining_grouped_limits_over_protocol(
235 monkeypatch, dining_candidates, check_open, building, expected_buildings
236):
237 data = [
238 {
239 "building": name,
240 "restaurants": (
241 [{"name": "Pizza", "note": ""} for _ in range(12)]
242 + [
243 {"name": "Sushi closed", "note": "平日休息、週六休息、週日休息"}
244 for _ in range(12)
245 ]
246 + [
247 {
248 "name": f"Sushi-{index}",
249 "phone": None,
250 "schedule": {"weekday": "09:00-17:00"},
251 "note": "",
252 }
253 for index in range(count)
254 ]
255 ),
256 }
257 for name, count in [("小吃部", 11), ("水木生活中心", 23), ("水木餐廳", 35)]
258 ]
259 data.append({"building": "Empty", "restaurants": []})
260 original = deepcopy(data)
261 calls = []
263 async def fake_get(endpoint):
264 assert endpoint == "dining.json"
265 calls.append(endpoint)
266 return "testhash", data
268 monkeypatch.setattr(nthudata, "get", fake_get)
269 async with Client(mcp) as client:
270 response = await client.call_tool(
271 "find_dining",
272 {"building": building, "restaurant_name": "Sushi", "check_open": check_open},
273 )
274 assert not response.is_error
275 assert set(response.data) == {"buildings"}
276 expected = []
277 for b in data:
278 if b["building"] not in expected_buildings:
279 continue
280 restaurants = [
281 r
282 for r in b["restaurants"]
283 if r["name"].startswith("Sushi") and (not check_open or not r["note"])
284 ]
285 if not building:
286 restaurants = restaurants[:10]
287 expected.append(
288 {
289 "building": b["building"],
290 "restaurants": [
291 {
292 "name": r["name"],
293 "phone": r.get("phone"),
294 "schedule": r.get("schedule"),
295 "note": r["note"] or None,
296 }
297 for r in restaurants
298 ],
299 }
300 )
301 assert response.data == {"buildings": expected}
302 assert data == original
303 assert calls == ["dining.json"]
306@pytest.mark.parametrize("check_open", [None, "today", "weekday", "saturday", "sunday"])
307async def test_dining_omits_empty_buildings(monkeypatch, dining_candidates, check_open):
308 async def fake_get(endpoint):
309 assert endpoint == "dining.json"
310 return "testhash", [
311 {"building": "Empty", "restaurants": []},
312 {
313 "building": "Closed",
314 "restaurants": [{"name": "Cafe", "note": "平日休息、週六休息、週日休息"}],
315 },
316 ]
318 monkeypatch.setattr(nthudata, "get", fake_get)
319 async with Client(mcp) as client:
320 response = await client.call_tool("find_dining", {"check_open": check_open})
321 assert not response.is_error
322 if check_open:
323 assert response.data == {"buildings": []}
324 else:
325 assert [b["building"] for b in response.data["buildings"]] == ["Closed"]
328@pytest.fixture
329def directory_and_locations(monkeypatch):
330 datasets = {
331 "directory.json": [
332 {
333 "index": str(index),
334 "name": "Office",
335 "details": {
336 "people": [
337 {"name": "Alice", "title": "Director"},
338 {"name": "Alice Smith", "title": None},
339 {"name": "Bob", "title": "Director"},
340 {"name": "Carol"},
341 {"name": "Dave", "title": ""},
342 ]
343 },
344 }
345 for index in range(8)
346 ],
347 "maps.json": {
348 "main": {
349 "Main gate": {"latitude": "24.79", "longitude": "120.99"},
350 **{
351 f"Main hall {index}": {"latitude": "24.79", "longitude": "120.99"}
352 for index in range(7)
353 },
354 },
355 },
356 }
357 original = deepcopy(datasets)
358 calls = []
360 async def get_dataset(path):
361 calls.append(path)
362 return "test-version", datasets[path]
364 monkeypatch.setattr(nthudata, "get", get_dataset)
365 yield datasets, calls
366 assert datasets == original
369@pytest.mark.parametrize("fuzzy", [None, True, False])
370@pytest.mark.parametrize(
371 "query,department_count,person_names",
372 [
373 ("Office", 5, []),
374 ("Alice", 0, ["Alice"] * 8),
375 ("Director", 0, ["Alice", "Bob"] * 5),
376 ("Off", 0, []),
377 ("Ali", 0, []),
378 ("Dir", 0, []),
379 ("office", 0, []),
380 ("alice", 0, []),
381 ("director", 0, []),
382 ("ZZZZZZ", 0, []),
383 ("", 0, []),
384 ],
385)
386async def test_registered_directory_search_modes(
387 directory_and_locations, query, department_count, person_names, fuzzy
388):
389 arguments = {"query": query}
390 if fuzzy is not None:
391 arguments["fuzzy"] = fuzzy
392 async with Client(mcp) as client:
393 response = await client.call_tool("search_directory", arguments)
394 assert not response.is_error
395 assert set(response.data) == {"departments", "people"}
396 assert directory_and_locations[1] == ["directory.json"]
397 if fuzzy is False or query in ("ZZZZZZ", ""):
398 assert len(response.data["departments"]) == department_count
399 assert [person["name"] for person in response.data["people"]] == person_names
400 else:
401 expected_departments = 5 if query.lower() in ("office", "off") else 0
402 assert len(response.data["departments"]) == expected_departments
403 if expected_departments:
404 assert response.data["departments"] == directory_and_locations[0]["directory.json"][:5]
405 elif query == "alice":
406 assert [person["name"] for person in response.data["people"]] == (
407 ["Alice"] * 8 + ["Alice Smith"] * 2
408 )
409 elif query.lower() in ("alice", "ali"):
410 assert [person["name"] for person in response.data["people"]] == [
411 "Alice",
412 "Alice Smith",
413 ] * 5
414 else:
415 assert [person["name"] for person in response.data["people"]] == ["Alice", "Bob"] * 5
418@pytest.mark.parametrize("fuzzy", [None, True, False])
419@pytest.mark.parametrize("query", ["Main gate", "Main", "main gate", "ZZZZZZ", ""])
420async def test_registered_locations_search_modes(directory_and_locations, query, fuzzy):
421 arguments = {"query": query}
422 if fuzzy is not None:
423 arguments["fuzzy"] = fuzzy
424 async with Client(mcp) as client:
425 response = await client.call_tool("search_locations", arguments)
426 assert not response.is_error
427 assert set(response.data) == {"locations"}
428 assert directory_and_locations[1] == ["maps.json"]
429 locations = response.data["locations"]
430 if fuzzy is False:
431 assert [location["name"] for location in locations] == (
432 ["Main gate"] if query == "Main gate" else []
433 )
434 elif query in ("ZZZZZZ", ""):
435 assert locations == []
436 else:
437 assert len(locations) == 5
438 assert locations[0]["name"] == "Main gate"
441@pytest.mark.parametrize("fuzzy", [True, False])
442@pytest.mark.parametrize(
443 "tool,path", [("search_directory", "directory.json"), ("search_locations", "maps.json")]
444)
445async def test_search_tools_unavailable_dataset(monkeypatch, tool, path, fuzzy):
446 async def unavailable(endpoint):
447 assert endpoint == path
448 raise DataNotAvailableException("private upstream")
450 monkeypatch.setattr(nthudata, "get", unavailable)
451 async with Client(mcp) as client:
452 response = await client.call_tool(
453 tool, {"query": "query", "fuzzy": fuzzy}, raise_on_error=False
454 )
455 assert response.is_error
456 assert "private upstream" not in str(response.content)
459@pytest.mark.parametrize("fuzzy", [True, False])
460async def test_empty_directory_search_unavailable_dataset(monkeypatch, fuzzy):
461 async def unavailable(endpoint):
462 assert endpoint == "directory.json"
463 raise DataNotAvailableException("private upstream")
465 monkeypatch.setattr(nthudata, "get", unavailable)
466 async with Client(mcp) as client:
467 response = await client.call_tool(
468 "search_directory", {"query": "", "fuzzy": fuzzy}, raise_on_error=False
469 )
470 assert response.is_error
471 assert "private upstream" not in str(response.content)
474@pytest.mark.parametrize("tool", ["search_directory", "search_locations"])
475async def test_search_tools_protocol_validation(tool):
476 async with Client(mcp) as client:
477 for arguments in ({}, {"query": "query", "fuzzy": "invalid"}):
478 response = await client.call_tool(tool, arguments, raise_on_error=False)
479 assert response.is_error
480 removed = await client.call_tool("search_campus", {"query": "query"}, raise_on_error=False)
481 assert removed.is_error
484@pytest.mark.parametrize("query", ["Office", "Alice", "Director", "Ali", "alice", "director"])
485async def test_directory_exact_rest_shape_and_case(directory_and_locations, query):
486 _, departments = await get_runtime().departments.get_directory(query, fuzzy=False)
487 if query == "Office":
488 assert departments == directory_and_locations[0]["directory.json"]
489 elif query in ("Alice", "Director"):
490 assert len(departments) == 8
491 for department in departments:
492 expected = ["Alice"] if query == "Alice" else ["Alice", "Bob"]
493 assert [person["name"] for person in department["details"]["people"]] == expected
494 else:
495 assert departments == []
498@pytest.mark.parametrize("fuzzy", [True, False])
499async def test_directory_filters_before_caps_and_preserves_exact_order(monkeypatch, fuzzy):
500 data = [
501 {"index": str(index), "name": "ZZZZZZ", "details": {"people": [{"name": "ZZZZZZ"}]}}
502 for index in range(20)
503 ] + [
504 {
505 "index": str(index),
506 "name": "Office",
507 "details": {"people": [{"name": "Alice"}]},
508 }
509 for index in range(20, 28)
510 ]
512 async def get_dataset(path):
513 assert path == "directory.json"
514 return "test-version", data
516 monkeypatch.setattr(nthudata, "get", get_dataset)
517 async with Client(mcp) as client:
518 departments = await client.call_tool(
519 "search_directory", {"query": "Office", "fuzzy": fuzzy}
520 )
521 people = await client.call_tool("search_directory", {"query": "Alice", "fuzzy": fuzzy})
522 assert [department["index"] for department in departments.data["departments"]] == [
523 str(index) for index in range(20, 25)
524 ]
525 assert people.data == {"departments": [], "people": [{"name": "Alice"}] * 8}