gabriel / musehub public
test_musehub_pagination.py python
388 lines 12.7 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Tests for RFC 8288 Link header pagination on MuseHub list endpoints.
2
3 Covers acceptance criteria:
4 - PaginationParams dependency parses page/per_page and cursor/limit query params
5 - build_link_header emits correct RFC 8288 rel links for first/last/prev/next
6 - build_cursor_link_header emits a rel="next" link with cursor and limit
7 - paginate_list slices correctly and returns accurate total
8 - GET /repos/{repo_id}/issues returns Link header and total field
9 - GET /repos/{repo_id}/proposals returns Link header and total field
10 - GET /repos/{repo_id}/commits returns Link header when per_page > 0
11 - GET /musehub/repos returns rel="next" Link header when next_cursor is present
12
13 All tests use fixtures from conftest.py. No live external APIs are called.
14 """
15 from __future__ import annotations
16
17 import pytest
18 from httpx import AsyncClient
19 from sqlalchemy.ext.asyncio import AsyncSession
20 from starlette.requests import Request as StarletteRequest
21
22 from musehub.muse_contracts.json_types import StrDict
23 from musehub.api.routes.musehub.pagination import (
24 PaginationParams,
25 build_cursor_link_header,
26 build_link_header,
27 paginate_list,
28 )
29
30
31 # ---------------------------------------------------------------------------
32 # Unit tests — pagination helpers
33 # ---------------------------------------------------------------------------
34
35
36 def _make_request(url: str) -> StarletteRequest:
37 """Build a minimal Starlette Request for testing URL construction."""
38 scope = {
39 "type": "http",
40 "method": "GET",
41 "path": url.split("?")[0],
42 "query_string": url.split("?")[1].encode() if "?" in url else b"",
43 "headers": [],
44 }
45 return StarletteRequest(scope)
46
47
48 def test_paginate_list_first_page() -> None:
49 """paginate_list returns the first page slice and correct total."""
50 items = list(range(55))
51 page, total = paginate_list(items, page=1, per_page=20)
52 assert total == 55
53 assert page == list(range(20))
54
55
56 def test_paginate_list_middle_page() -> None:
57 """paginate_list returns the correct middle page slice."""
58 items = list(range(55))
59 page, total = paginate_list(items, page=2, per_page=20)
60 assert total == 55
61 assert page == list(range(20, 40))
62
63
64 def test_paginate_list_last_partial_page() -> None:
65 """paginate_list returns a partial slice on the final page."""
66 items = list(range(55))
67 page, total = paginate_list(items, page=3, per_page=20)
68 assert total == 55
69 assert page == list(range(40, 55))
70
71
72 def test_paginate_list_beyond_last_page_returns_empty() -> None:
73 """paginate_list returns an empty slice when page exceeds total."""
74 items = list(range(10))
75 page, total = paginate_list(items, page=5, per_page=10)
76 assert total == 10
77 assert page == []
78
79
80 def test_paginate_list_empty_input() -> None:
81 """paginate_list handles empty input gracefully."""
82 page_items: list[int]
83 page_items, total = paginate_list([], page=1, per_page=20)
84 assert total == 0
85 assert page_items == []
86
87
88 def test_build_link_header_single_page() -> None:
89 """build_link_header emits only first and last when there is exactly one page."""
90 req = _make_request("http://test/api/repos/r1/issues?page=1&per_page=20")
91 header = build_link_header(req, total=5, page=1, per_page=20)
92 assert 'rel="first"' in header
93 assert 'rel="last"' in header
94 assert 'rel="next"' not in header
95 assert 'rel="prev"' not in header
96
97
98 def test_build_link_header_first_of_many() -> None:
99 """build_link_header emits first, last, and next (but not prev) on page 1 of N."""
100 req = _make_request("http://test/api/repos/r1/issues?page=1&per_page=10")
101 header = build_link_header(req, total=55, page=1, per_page=10)
102 assert 'rel="first"' in header
103 assert 'rel="last"' in header
104 assert 'rel="next"' in header
105 assert 'rel="prev"' not in header
106 assert "page=2" in header
107 assert "page=6" in header # last page for 55 items at 10/page
108
109
110 def test_build_link_header_middle_page() -> None:
111 """build_link_header emits all four rels on an interior page."""
112 req = _make_request("http://test/api/repos/r1/issues?page=3&per_page=10")
113 header = build_link_header(req, total=55, page=3, per_page=10)
114 assert 'rel="first"' in header
115 assert 'rel="last"' in header
116 assert 'rel="next"' in header
117 assert 'rel="prev"' in header
118 assert "page=4" in header
119 assert "page=2" in header
120
121
122 def test_build_link_header_last_page() -> None:
123 """build_link_header emits prev (but not next) on the last page."""
124 req = _make_request("http://test/api/repos/r1/issues?page=6&per_page=10")
125 header = build_link_header(req, total=55, page=6, per_page=10)
126 assert 'rel="first"' in header
127 assert 'rel="last"' in header
128 assert 'rel="prev"' in header
129 assert 'rel="next"' not in header
130
131
132 def test_build_link_header_preserves_existing_query_params() -> None:
133 """build_link_header keeps non-pagination query params on generated URLs."""
134 req = _make_request("http://test/api/repos/r1/issues?state=open&page=1&per_page=10")
135 header = build_link_header(req, total=30, page=1, per_page=10)
136 assert "state=open" in header
137
138
139 def test_build_cursor_link_header_emits_next_only() -> None:
140 """build_cursor_link_header emits exactly one rel="next" with cursor and limit encoded."""
141 req = _make_request("http://test/api/repos?limit=20")
142 header = build_cursor_link_header(req, next_cursor="abc123", limit=20)
143 assert 'rel="next"' in header
144 assert "cursor=abc123" in header
145 assert "limit=20" in header
146 assert 'rel="prev"' not in header
147 assert 'rel="first"' not in header
148
149
150 # ---------------------------------------------------------------------------
151 # Integration tests — issues list endpoint
152 # ---------------------------------------------------------------------------
153
154
155 async def _create_repo(client: AsyncClient, auth_headers: StrDict, name: str) -> str:
156 r = await client.post(
157 "/api/repos",
158 json={"name": name, "owner": "testuser"},
159 headers=auth_headers,
160 )
161 assert r.status_code == 201
162 repo_id: str = r.json()["repoId"]
163 return repo_id
164
165
166 async def _create_issue(
167 client: AsyncClient,
168 auth_headers: StrDict,
169 repo_id: str,
170 title: str,
171 ) -> None:
172 r = await client.post(
173 f"/api/repos/{repo_id}/issues",
174 json={"title": title, "body": ""},
175 headers=auth_headers,
176 )
177 assert r.status_code == 201
178
179
180 @pytest.mark.anyio
181 async def test_list_issues_link_header_present(
182 client: AsyncClient,
183 auth_headers: StrDict,
184 ) -> None:
185 """GET /issues includes a Link header when pagination is active."""
186 repo_id = await _create_repo(client, auth_headers, "pagination-issues-link")
187 for i in range(5):
188 await _create_issue(client, auth_headers, repo_id, f"Issue {i}")
189
190 r = await client.get(
191 f"/api/repos/{repo_id}/issues?page=1&per_page=2",
192 headers=auth_headers,
193 )
194 assert r.status_code == 200
195 assert "Link" in r.headers
196 link = r.headers["Link"]
197 assert 'rel="first"' in link
198 assert 'rel="last"' in link
199 assert 'rel="next"' in link
200
201
202 @pytest.mark.anyio
203 async def test_list_issues_total_field_returned(
204 client: AsyncClient,
205 auth_headers: StrDict,
206 ) -> None:
207 """GET /issues response body includes ``total`` with the count across all pages."""
208 repo_id = await _create_repo(client, auth_headers, "pagination-issues-total")
209 for i in range(7):
210 await _create_issue(client, auth_headers, repo_id, f"Track issue {i}")
211
212 r = await client.get(
213 f"/api/repos/{repo_id}/issues?page=1&per_page=3",
214 headers=auth_headers,
215 )
216 assert r.status_code == 200
217 body = r.json()
218 assert body["total"] == 7
219 assert len(body["issues"]) == 3
220
221
222 @pytest.mark.anyio
223 async def test_list_issues_last_page_no_next(
224 client: AsyncClient,
225 auth_headers: StrDict,
226 ) -> None:
227 """GET /issues Link header on the last page has no rel=\"next\"."""
228 repo_id = await _create_repo(client, auth_headers, "pagination-issues-last")
229 for i in range(4):
230 await _create_issue(client, auth_headers, repo_id, f"Issue {i}")
231
232 r = await client.get(
233 f"/api/repos/{repo_id}/issues?page=2&per_page=3",
234 headers=auth_headers,
235 )
236 assert r.status_code == 200
237 link = r.headers["Link"]
238 assert 'rel="next"' not in link
239 assert 'rel="prev"' in link
240
241
242 @pytest.mark.anyio
243 async def test_list_issues_default_page_returns_all_when_small(
244 client: AsyncClient,
245 auth_headers: StrDict,
246 ) -> None:
247 """GET /issues with no pagination params returns results on page 1 (default)."""
248 repo_id = await _create_repo(client, auth_headers, "pagination-issues-default")
249 for i in range(3):
250 await _create_issue(client, auth_headers, repo_id, f"Default page {i}")
251
252 r = await client.get(
253 f"/api/repos/{repo_id}/issues",
254 headers=auth_headers,
255 )
256 assert r.status_code == 200
257 body = r.json()
258 assert len(body["issues"]) == 3
259 assert body["total"] == 3
260
261
262 # ---------------------------------------------------------------------------
263 # Integration tests — Proposals list endpoint
264 # ---------------------------------------------------------------------------
265
266
267
268 @pytest.mark.anyio
269 async def test_list_proposals_link_header_present(
270 client: AsyncClient,
271 auth_headers: StrDict,
272 db_session: AsyncSession,
273 ) -> None:
274 """GET /proposals includes a Link header when pagination is active."""
275 from musehub.db.musehub_models import MusehubProposal
276
277 repo_id = await _create_repo(client, auth_headers, "pagination-proposals-link")
278
279 # Insert proposals directly to avoid branch validation complexity in tests
280 for i in range(3):
281 db_session.add(MusehubProposal(
282 repo_id=repo_id,
283 proposal_number=i + 1,
284 title=f"Proposal {i}",
285 from_branch=f"feat/{i}",
286 to_branch="main",
287 author="testuser",
288 ))
289 await db_session.commit()
290
291 r = await client.get(
292 f"/api/repos/{repo_id}/proposals?page=1&per_page=2",
293 headers=auth_headers,
294 )
295 assert r.status_code == 200
296 assert "Link" in r.headers
297 link = r.headers["Link"]
298 assert 'rel="first"' in link
299 assert 'rel="next"' in link
300
301
302 @pytest.mark.anyio
303 async def test_list_proposals_total_field_returned(
304 client: AsyncClient,
305 auth_headers: StrDict,
306 db_session: AsyncSession,
307 ) -> None:
308 """GET /proposals response body includes ``total`` field."""
309 from musehub.db.musehub_models import MusehubProposal
310
311 repo_id = await _create_repo(client, auth_headers, "pagination-proposals-total")
312
313 for i in range(4):
314 db_session.add(MusehubProposal(
315 repo_id=repo_id,
316 proposal_number=i + 1,
317 title=f"Proposal {i}",
318 from_branch=f"feat/{i}",
319 to_branch="main",
320 author="testuser",
321 ))
322 await db_session.commit()
323
324 r = await client.get(
325 f"/api/repos/{repo_id}/proposals",
326 headers=auth_headers,
327 )
328 assert r.status_code == 200
329 body = r.json()
330 assert "total" in body
331 assert body["total"] == 4
332
333
334 # ---------------------------------------------------------------------------
335 # Integration tests — commits list endpoint
336 # ---------------------------------------------------------------------------
337
338
339 @pytest.mark.anyio
340 async def test_list_commits_link_header_with_per_page(
341 client: AsyncClient,
342 auth_headers: StrDict,
343 db_session: AsyncSession,
344 ) -> None:
345 """GET /commits with per_page > 0 includes an RFC 8288 Link header."""
346 from datetime import datetime, timezone, timedelta
347 from musehub.db.musehub_models import MusehubCommit
348
349 repo_id = await _create_repo(client, auth_headers, "pagination-commits-link")
350 now = datetime.now(tz=timezone.utc)
351
352 for i in range(5):
353 db_session.add(MusehubCommit(
354 commit_id=f"sha-{i:04d}",
355 repo_id=repo_id,
356 branch="main",
357 parent_ids=[],
358 message=f"Commit {i}",
359 author="testuser",
360 timestamp=now + timedelta(seconds=i),
361 ))
362 await db_session.commit()
363
364 r = await client.get(
365 f"/api/repos/{repo_id}/commits?page=1&per_page=2",
366 headers=auth_headers,
367 )
368 assert r.status_code == 200
369 assert "Link" in r.headers
370 link = r.headers["Link"]
371 assert 'rel="first"' in link
372 assert 'rel="next"' in link
373
374
375 @pytest.mark.anyio
376 async def test_list_commits_no_link_header_without_per_page(
377 client: AsyncClient,
378 auth_headers: StrDict,
379 ) -> None:
380 """GET /commits without per_page does NOT add a Link header (legacy mode)."""
381 repo_id = await _create_repo(client, auth_headers, "pagination-commits-no-link")
382
383 r = await client.get(
384 f"/api/repos/{repo_id}/commits",
385 headers=auth_headers,
386 )
387 assert r.status_code == 200
388 assert "Link" not in r.headers
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago