Coverage for tests/test_announcements.py: 100%
106 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""Tests for announcements endpoints."""
3from copy import deepcopy
4from itertools import product
6import pytest
7from httpx import ASGITransport, AsyncClient
9from data_api.api.api import app
10from data_api.data.manager import nthudata
11from data_api.domain.announcements.services import announcements_service
13pytestmark = pytest.mark.usefixtures("dataset_runtime")
16@pytest.fixture
17def announcement_sources(monkeypatch):
18 sources = [
19 {
20 "title": "Notice",
21 "link": "https://academic.site.nthu.edu.tw/p/403-1007-1504-1.php",
22 "department": "Academic",
23 "language": "en",
24 "articles": [
25 {
26 "title": "Notice",
27 "link": "https://example.com/article",
28 "date": "2026-10-05",
29 }
30 ],
31 },
32 {
33 "title": "Notice",
34 "link": "hTtP://Academic.Site.Nthu.Edu.Tw/p/403-1007-1505-1.php",
35 "department": "Academic",
36 "language": "zh-tw",
37 "articles": [],
38 },
39 {
40 "title": "Notice",
41 "link": "https://student.site.nthu.edu.tw/",
42 "department": "Student",
43 "language": "en",
44 "articles": [
45 {
46 "title": "Notice",
47 "link": "https://academic.site.nthu.edu.tw/article",
48 "date": "2026-10-05",
49 }
50 ],
51 },
52 {
53 "title": "Notice",
54 "link": "HtTpS://Events.Site.Nthu.Edu.Tw/P/Case?Token=AbC#Frag",
55 "department": "Events",
56 "language": "en",
57 "articles": [],
58 },
59 ]
60 original = deepcopy(sources)
62 async def get_announcements(endpoint):
63 assert endpoint == "announcements.json"
64 return "test-commit", sources
66 monkeypatch.setattr(nthudata, "get", get_announcements)
67 yield sources
68 assert sources == original
71class TestAnnouncementsEndpoints:
72 """Tests for announcements endpoints."""
74 @pytest.fixture
75 async def client(self):
76 """Create async test client."""
77 async with AsyncClient(
78 transport=ASGITransport(app=app), base_url="http://test", follow_redirects=True
79 ) as client:
80 yield client
82 async def test_get_all_announcements(self, client: AsyncClient):
83 """Test getting all announcements."""
84 response = await client.get("/announcements")
85 assert response.status_code == 200
87 @pytest.mark.parametrize(
88 "department",
89 ["清華公佈欄", "國立清華大學學生會"],
90 )
91 async def test_get_announcements_by_department(self, client: AsyncClient, department: str):
92 """Test getting announcements filtered by department."""
93 params = {"department": department}
94 response = await client.get("/announcements", params=params)
95 assert response.status_code == 200
97 @pytest.mark.parametrize("path", ["/announcements"])
98 @pytest.mark.parametrize("fuzzy", [False, True])
99 @pytest.mark.parametrize(
100 "url, indices",
101 [
102 ("academic.site.nthu.edu.tw", [0, 1]),
103 ("academic.site.nthu.edu.tw/p/403-1007-1504-1.php", [0]),
104 ("403-1007-1504-1.php", [0]),
105 ("403-1007-1504-1.PHP", []),
106 ("https://academic.site.nthu.edu.tw", [0, 1]),
107 ("http://academic.site.nthu.edu.tw", [0, 1]),
108 ("HTTPS://Academic.Site.Nthu.Edu.Tw", [0, 1]),
109 ("hTtPs://aCaDeMiC.sItE.nThU.eDu.Tw/p/403-1007-1504-1.php", [0]),
110 ("HTTPS://events.site.nthu.edu.tw/P/Case?Token=AbC#Frag", [3]),
111 ("events.site.nthu.edu.tw/P/Case", [3]),
112 ("P/Case", [3]),
113 ("p/Case", []),
114 ("EVENTS.SITE.NTHU.EDU.TW/p/Case", []),
115 ("HTTPS://events.site.nthu.edu.tw/P/case", []),
116 ("events.site.nthu.edu.tw/P/Case?token=AbC", []),
117 ("events.site.nthu.edu.tw/P/Case?Token=abc", []),
118 ("events.site.nthu.edu.tw/P/Case?Token=AbC#frag", []),
119 ("/P/Case?Token=AbC#Frag", [3]),
120 ("/p/Case?Token=AbC#Frag", []),
121 ("nonexistent.site.nthu.edu.tw", []),
122 ("academic.site.nthu.edu.tx", []),
123 ("", [0, 1, 2, 3]),
124 ],
125 )
126 async def test_filter_by_source_url(
127 self, client: AsyncClient, announcement_sources, path, fuzzy, url, indices
128 ):
129 response = await client.get(path, params={"url": url, "fuzzy": fuzzy})
130 assert response.status_code == 200
131 expected = [deepcopy(announcement_sources[index]) for index in indices]
132 for source in expected:
133 if source["department"] == "Events":
134 source["link"] = "https://events.site.nthu.edu.tw/P/Case?Token=AbC#Frag"
135 elif source["language"] == "zh-tw":
136 source["link"] = "https://academic.site.nthu.edu.tw/p/403-1007-1505-1.php"
137 assert response.json() == expected
138 assert response.headers["X-Data-Commit-Hash"] == "test-commit"
140 @pytest.mark.parametrize("fuzzy", [False, True])
141 async def test_url_combines_with_other_filters(
142 self, client: AsyncClient, announcement_sources, fuzzy
143 ):
144 response = await client.get(
145 "/announcements",
146 params={
147 "url": "academic.site.nthu.edu.tw",
148 "department": "Academic",
149 "title": "Notice",
150 "language": "en",
151 "fuzzy": fuzzy,
152 },
153 )
154 assert response.status_code == 200
155 assert response.json() == announcement_sources[:1]
157 @pytest.mark.parametrize("department", [None, "教務處", "不存在的部門"])
158 async def test_get_announcements_sources(self, client: AsyncClient, monkeypatch, department):
159 """Test getting announcement sources."""
160 sources = [
161 {
162 "title": "教務處公告",
163 "link": "https://example.com/academic",
164 "department": "教務處",
165 "language": "zh-tw",
166 },
167 {
168 "title": "學生事務處公告",
169 "link": "https://example.com/student",
170 "department": "學生事務處",
171 "language": "zh-tw",
172 },
173 ]
175 async def get_sources(endpoint):
176 assert endpoint == "announcements_list.json"
177 return "test-commit", sources
179 monkeypatch.setattr(nthudata, "get", get_sources)
180 params = {"department": department} if department else {}
181 response = await client.get("/announcements/sources", params=params)
182 assert response.status_code == 200
183 expected = [
184 source for source in sources if not department or source["department"] == department
185 ]
186 assert response.json() == expected
187 assert response.headers["X-Data-Commit-Hash"] == "test-commit"
190def test_announcements_openapi_contract():
191 schema = app.openapi()
192 paths = schema["paths"]
193 announcement_paths = {path for path in paths if path.startswith("/announcements")}
194 assert announcement_paths == {
195 "/announcements",
196 "/announcements/sources",
197 }
198 assert paths["/announcements"]["get"]["operationId"] == "getAnnouncements"
199 assert paths["/announcements/sources"]["get"]["operationId"] == "getAnnouncementsList"
200 assert not paths["/announcements"]["get"].get("deprecated", False)
201 assert not paths["/announcements/sources"]["get"].get("deprecated", False)
202 parameters = {
203 parameter["name"]: parameter for parameter in paths["/announcements"]["get"]["parameters"]
204 }
205 assert {"department", "title", "language", "fuzzy", "url"} <= parameters.keys()
206 assert parameters["url"]["required"] is False
207 assert parameters["url"]["schema"]["type"] == "string"
208 description = parameters["department"]["description"]
209 assert "/announcements/sources" in description
210 assert "/announcements/lists/departments" not in description
211 response_schema = paths["/announcements/sources"]["get"]["responses"]["200"]["content"][
212 "application/json"
213 ]["schema"]
214 assert response_schema["type"] == "array"
215 assert response_schema["items"] == {"$ref": "#/components/schemas/AnnouncementSource"}
216 source_schema = schema["components"]["schemas"]["AnnouncementSource"]
217 fields = {"title", "link", "language", "department"}
218 assert set(source_schema["required"]) == fields
219 assert set(source_schema["properties"]) == fields
220 assert all(field["type"] == "string" for field in source_schema["properties"].values())
221 assert source_schema["properties"]["link"]["format"] == "uri"
224@pytest.mark.parametrize("selected", list(product([False, True], repeat=3)))
225async def test_fuzzy_filters_preserve_sources_and_cached_articles(monkeypatch, selected):
226 flags = list(product([False, True], repeat=3))
227 data = [
228 {
229 "department": "Alpha" if department else "ZZZZZZ",
230 "language": "en" if language else "zh-tw",
231 "articles": [{"title": "Notice" if title else "XXXXXX"}],
232 }
233 for department, language, title in flags
234 ]
235 empty_source = {"department": "Alpha", "language": "en", "articles": []}
236 data.append(empty_source)
237 original = deepcopy(data)
239 async def get_announcements(endpoint):
240 return "test", data
242 monkeypatch.setattr(nthudata, "get", get_announcements)
243 department, language, title = selected
244 commit, result = await announcements_service.fuzzy_search_announcements(
245 department="Alpha" if department else None,
246 language="en" if language else None,
247 title="Notice" if title else None,
248 )
249 expected = [
250 source
251 for source, matches in zip(data, flags)
252 if all(not enabled or matched for enabled, matched in zip(selected, matches))
253 ]
254 if not title:
255 expected.append(empty_source)
256 assert commit == "test"
257 assert result == expected
258 assert data == original