Coverage for tests/test_directory.py: 100%
116 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Tests for the unified directory endpoint."""
3from copy import deepcopy
5import pytest
6from httpx import ASGITransport, AsyncClient
8from data_api.api.api import app
9from data_api.api.schemas.departments import Department
10from data_api.core.exceptions import DataNotAvailableException
11from data_api.data.manager import nthudata
12from data_api.domain.departments import services
14pytestmark = pytest.mark.usefixtures("dataset_runtime")
17@pytest.fixture
18def directory_data(monkeypatch):
19 data = [
20 {
21 "index": "admin",
22 "name": "校務處",
23 "parent_name": "清華大學",
24 "url": "https://example.com/admin",
25 "details": {
26 "contact": {"phone": "123"},
27 "departments": [{"name": "行政組"}],
28 "people": [
29 {"name": "王小明", "title": "主任", "extension": "100"},
30 {"name": "陳大華", "title": None},
31 {"name": "林美玲"},
32 ],
33 },
34 },
35 {
36 "index": "computing",
37 "name": "計算中心",
38 "details": {
39 "contact": {"email": "computing@example.com"},
40 "people": [
41 {"name": "王小明", "title": ""},
42 {"name": "李大同", "title": "主任"},
43 ],
44 },
45 },
46 {"index": "library", "name": "圖書館"},
47 ]
49 async def get_directory(endpoint):
50 assert endpoint == services.JSON_PATH
51 return "directory-commit", data
53 monkeypatch.setattr(nthudata, "get", get_directory)
54 return data
57@pytest.mark.parametrize("fuzzy", [None, True, False])
58@pytest.mark.parametrize(
59 "query,indices,people",
60 [
61 (None, ["admin", "computing", "library"], None),
62 ("", ["admin", "computing", "library"], None),
63 ("校務處", ["admin"], None),
64 ("圖書館", ["library"], None),
65 ("王小明", ["admin", "computing"], [["王小明"], ["王小明"]]),
66 ("主任", ["admin", "computing"], [["王小明"], ["李大同"]]),
67 ("ZZZ", [], None),
68 ],
69)
70async def test_directory_query(directory_data, query, indices, people, fuzzy):
71 original = deepcopy(directory_data)
72 expected = [deepcopy(item) for item in directory_data if item["index"] in indices]
73 if people is not None:
74 for department, names in zip(expected, people, strict=True):
75 department["details"]["people"] = [
76 person for person in department["details"]["people"] if person["name"] in names
77 ]
78 async with AsyncClient(
79 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=False
80 ) as client:
81 params = {} if query is None else {"query": query}
82 if fuzzy is not None:
83 params["fuzzy"] = fuzzy
84 response = await client.get("/directory", params=params)
85 unfiltered = await client.get("/directory")
87 assert response.status_code == 200
88 assert "location" not in response.headers
89 assert response.headers["X-Data-Commit-Hash"] == "directory-commit"
90 assert response.json() == [
91 Department.model_validate(item).model_dump(mode="json") for item in expected
92 ]
93 assert unfiltered.json() == [
94 Department.model_validate(item).model_dump(mode="json") for item in original
95 ]
96 assert directory_data == original
99async def test_directory_department_match_keeps_nonmatching_people(directory_data, monkeypatch):
100 scores = {"校務處": 80, "計算中心": 79, "王小明": 80, "主任": 89}
101 monkeypatch.setattr(services.fuzz, "partial_ratio", lambda query, value: scores.get(value, 0))
102 commit, results = await services.departments_service.get_directory("query")
103 assert commit == "directory-commit"
104 assert results[0] == directory_data[0]
105 assert results[1]["details"]["people"] == [directory_data[1]["details"]["people"][0]]
108@pytest.mark.parametrize("score,expected_count", [(89, 0), (90, 2)])
109async def test_directory_title_match_threshold(directory_data, monkeypatch, score, expected_count):
110 monkeypatch.setattr(
111 services.fuzz, "partial_ratio", lambda query, value: score if value == "主任" else 0
112 )
113 _, results = await services.departments_service.get_directory("query")
114 assert len(results) == expected_count
115 for department in results:
116 assert len(department["details"]["people"]) == 1
117 assert department["details"]["people"][0]["title"] == "主任"
120async def test_directory_sorts_by_match_score(directory_data, monkeypatch):
121 scores = {"校務處": 80, "計算中心": 100}
122 monkeypatch.setattr(services.fuzz, "partial_ratio", lambda query, value: scores.get(value, 0))
123 _, results = await services.departments_service.get_directory("query")
124 assert [department["index"] for department in results] == ["computing", "admin"]
127async def test_directory_search_keeps_separate_results(directory_data):
128 commit, result = await services.departments_service.search_departments_and_people("王小明")
129 assert commit == "directory-commit"
130 assert result["departments"] == []
131 assert [person["name"] for person in result["people"]] == ["王小明", "王小明"]
134async def test_directory_without_commit_hash(monkeypatch):
135 async def get_directory(endpoint):
136 return None, []
138 monkeypatch.setattr(nthudata, "get", get_directory)
139 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
140 response = await client.get("/directory")
141 assert response.status_code == 200
142 assert response.json() == []
143 assert "X-Data-Commit-Hash" not in response.headers
146def test_directory_openapi():
147 schema = app.openapi()
148 paths = schema["paths"]
149 assert "/directory/search" not in paths
150 operation = paths["/directory"]["get"]
151 assert operation["tags"] == ["Directory"]
152 assert operation["operationId"] == "getDirectory"
153 assert not operation.get("deprecated", False)
154 assert operation["parameters"][0]["name"] == "query"
155 assert operation["parameters"][0]["required"] is False
156 parameters = {parameter["name"]: parameter for parameter in operation["parameters"]}
157 assert parameters.keys() == {"query", "fuzzy"}
158 assert parameters["fuzzy"]["schema"]["type"] == "boolean"
159 assert parameters["fuzzy"]["schema"]["default"] is True
160 assert operation["responses"]["200"]["content"]["application/json"]["schema"]["items"] == {
161 "$ref": "#/components/schemas/Department"
162 }
165async def test_directory_search_path_is_not_registered():
166 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
167 response = await client.get("/directory/search", params={"query": "test"})
168 assert response.status_code == 404
171@pytest.mark.parametrize("fuzzy", [True, False])
172@pytest.mark.parametrize("query", [None, "王小明"])
173async def test_directory_unavailable_dataset(monkeypatch, query, fuzzy):
174 async def unavailable(endpoint):
175 raise DataNotAvailableException("Dataset temporarily unavailable")
177 monkeypatch.setattr(nthudata, "get", unavailable)
178 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
179 params = {"fuzzy": fuzzy}
180 if query is not None:
181 params["query"] = query
182 response = await client.get("/directory", params=params)
183 assert response.status_code == 503
184 assert response.json() == {"detail": "Service temporarily unavailable"}
187@pytest.mark.parametrize("query", ["校務", "小明", "主"])
188async def test_directory_exact_rejects_partial_matches(directory_data, query):
189 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
190 fuzzy = await client.get("/directory", params={"query": query})
191 exact = await client.get("/directory", params={"query": query, "fuzzy": False})
192 assert fuzzy.status_code == exact.status_code == 200
193 assert fuzzy.json()
194 assert exact.json() == []
197async def test_directory_invalid_fuzzy(directory_data):
198 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
199 response = await client.get("/directory", params={"fuzzy": "invalid"})
200 assert response.status_code == 422
201 assert response.json()["detail"][0]["loc"] == ["query", "fuzzy"]