gabriel / musehub public
test_ui_ssr_section38.py python
936 lines 38.5 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Section 38 — UI / SSR Routes: 7-layer test suite.
2
3 Existing coverage (347 tests across 27 files) focuses on E2E page rendering for
4 issues, commits, labels, milestones, notifications, forks, topics, user profiles,
5 and settings. This file adds the missing layers and fills coverage gaps:
6
7 1. Unit — pure utility functions: is_htmx, htmx_trigger, _infer_sym_kind,
8 _fmt_relative, licenses_for_viewer_type (no DB, no HTTP)
9 2. Integration — repo home page, proposals list, agents swarm, blob/blame pages
10 with real DB state
11 3. E2E — pages with no existing test files: repo home, symbols, proposals,
12 agents, intel, blob/raw; plus HTMX fragment routing
13 4. Stress — sequential loads across multiple pages; empty-state fallbacks
14 5. Data Integrity — nav counts correct, filter accuracy, XSS content escaped
15 6. Security — auth-required pages return 401 without token; injected content
16 escaped; debug info not leaked
17 7. Performance — key pages respond under 500ms
18 """
19 from __future__ import annotations
20
21 import time
22 import uuid
23 from datetime import datetime, timezone
24
25 import pytest
26 from httpx import AsyncClient
27 from sqlalchemy.ext.asyncio import AsyncSession
28
29 from musehub.muse_contracts.json_types import StrDict
30 from musehub.db.musehub_models import (
31 MusehubIdentity,
32 MusehubIssue,
33 MusehubProposal,
34 MusehubRepo,
35 )
36
37 # ─────────────────────────────────────────────────────────────────────────────
38 # Helpers shared across layers
39 # ─────────────────────────────────────────────────────────────────────────────
40
41 _OWNER = "ssr-tester"
42 _SLUG = "ssr-test-repo"
43
44
45 async def _seed_identity(db: AsyncSession, handle: str = _OWNER) -> MusehubIdentity:
46 identity = MusehubIdentity(handle=handle, identity_type="human", display_name=handle)
47 db.add(identity)
48 await db.commit()
49 await db.refresh(identity)
50 return identity
51
52
53 async def _seed_repo(
54 db: AsyncSession,
55 owner: str = _OWNER,
56 slug: str = _SLUG,
57 visibility: str = "public",
58 ) -> MusehubRepo:
59 repo = MusehubRepo(
60 name=slug,
61 owner=owner,
62 slug=slug,
63 visibility=visibility,
64 owner_user_id=f"uid-{owner}",
65 )
66 db.add(repo)
67 await db.commit()
68 await db.refresh(repo)
69 return repo
70
71
72 async def _seed_issue(
73 db: AsyncSession,
74 repo_id: str,
75 *,
76 number: int = 1,
77 title: str = "Test issue",
78 state: str = "open",
79 ) -> MusehubIssue:
80 issue = MusehubIssue(
81 repo_id=repo_id,
82 number=number,
83 title=title,
84 body="Body text.",
85 state=state,
86 labels=[],
87 author=_OWNER,
88 )
89 db.add(issue)
90 await db.commit()
91 await db.refresh(issue)
92 return issue
93
94
95 async def _seed_proposal(
96 db: AsyncSession,
97 repo_id: str,
98 *,
99 proposal_number: int = 1,
100 title: str = "Add harmony voice",
101 state: str = "open",
102 ) -> MusehubProposal:
103 proposal = MusehubProposal(
104 repo_id=repo_id,
105 proposal_number=proposal_number,
106 title=title,
107 body="Proposal body.",
108 state=state,
109 from_branch="feat/harmony",
110 to_branch="main",
111 author=_OWNER,
112 )
113 db.add(proposal)
114 await db.commit()
115 await db.refresh(proposal)
116 return proposal
117
118
119 # ─────────────────────────────────────────────────────────────────────────────
120 # LAYER 1 — UNIT
121 # ─────────────────────────────────────────────────────────────────────────────
122
123
124 class TestIsHtmx:
125 """Unit tests for is_htmx / is_htmx_boosted helpers."""
126
127 def _req(self, headers: StrDict):
128 from starlette.testclient import TestClient
129 from starlette.applications import Starlette
130 from starlette.routing import Route
131 from starlette.requests import Request as StarletteRequest
132 from starlette.responses import PlainTextResponse
133
134 captured: list[StarletteRequest] = []
135
136 def handler(req: StarletteRequest):
137 captured.append(req)
138 return PlainTextResponse("ok")
139
140 app = Starlette(routes=[Route("/", handler)])
141 client = TestClient(app, raise_server_exceptions=False)
142 client.get("/", headers=headers)
143 return captured[0]
144
145 def test_no_header_is_not_htmx(self):
146 from musehub.api.routes.musehub.htmx_helpers import is_htmx
147 req = self._req({})
148 assert is_htmx(req) is False
149
150 def test_hx_request_true_is_htmx(self):
151 from musehub.api.routes.musehub.htmx_helpers import is_htmx
152 req = self._req({"HX-Request": "true"})
153 assert is_htmx(req) is True
154
155 def test_hx_request_false_is_not_htmx(self):
156 from musehub.api.routes.musehub.htmx_helpers import is_htmx
157 req = self._req({"HX-Request": "false"})
158 assert is_htmx(req) is False
159
160 def test_hx_boosted_false_when_absent(self):
161 from musehub.api.routes.musehub.htmx_helpers import is_htmx_boosted
162 req = self._req({})
163 assert is_htmx_boosted(req) is False
164
165 def test_hx_boosted_true_when_present(self):
166 from musehub.api.routes.musehub.htmx_helpers import is_htmx_boosted
167 req = self._req({"HX-Boosted": "true"})
168 assert is_htmx_boosted(req) is True
169
170
171 class TestHtmxTrigger:
172 """Unit tests for htmx_trigger helper."""
173
174 def test_sets_hx_trigger_header(self):
175 from starlette.responses import Response
176 from musehub.api.routes.musehub.htmx_helpers import htmx_trigger
177 import json
178
179 r = Response()
180 htmx_trigger(r, "toast", {"message": "saved"})
181 payload = json.loads(r.headers["HX-Trigger"])
182 assert payload == {"toast": {"message": "saved"}}
183
184 def test_sets_hx_trigger_header_no_detail(self):
185 from starlette.responses import Response
186 from musehub.api.routes.musehub.htmx_helpers import htmx_trigger
187 import json
188
189 r = Response()
190 htmx_trigger(r, "refresh")
191 payload = json.loads(r.headers["HX-Trigger"])
192 assert payload == {"refresh": True}
193
194 def test_multiple_events_possible(self):
195 """Calling twice overwrites but last write wins — not a bug, just a spec."""
196 from starlette.responses import Response
197 from musehub.api.routes.musehub.htmx_helpers import htmx_trigger
198
199 r = Response()
200 htmx_trigger(r, "reload")
201 htmx_trigger(r, "scroll")
202 # Only one value in header (last write wins)
203 assert "HX-Trigger" in r.headers
204
205
206 class TestHtmxRedirect:
207 """Unit tests for htmx_redirect helper."""
208
209 def test_returns_200(self):
210 from musehub.api.routes.musehub.htmx_helpers import htmx_redirect
211
212 r = htmx_redirect("/some/url")
213 assert r.status_code == 200
214
215 def test_sets_hx_redirect_header(self):
216 from musehub.api.routes.musehub.htmx_helpers import htmx_redirect
217
218 r = htmx_redirect("/dashboard")
219 assert r.headers["HX-Redirect"] == "/dashboard"
220
221 def test_url_preserved_exactly(self):
222 from musehub.api.routes.musehub.htmx_helpers import htmx_redirect
223
224 url = "/gabriel/my-repo?welcome=1"
225 r = htmx_redirect(url)
226 assert r.headers["HX-Redirect"] == url
227
228
229 class TestInferSymKind:
230 """Unit tests for _infer_sym_kind."""
231
232 def _infer(self, addr: str) -> str:
233 from musehub.api.routes.musehub.ui_symbols import _infer_sym_kind
234 return _infer_sym_kind(addr)
235
236 def test_camel_case_is_class(self):
237 assert self._infer("file.py::MyClass") == "class"
238
239 def test_snake_case_is_fn(self):
240 assert self._infer("file.py::my_function") == "fn"
241
242 def test_all_caps_is_sym(self):
243 assert self._infer("file.py::MAX_RETRIES") == "sym"
244
245 def test_no_namespace_camel(self):
246 assert self._infer("CamelCase") == "class"
247
248 def test_no_namespace_snake(self):
249 assert self._infer("my_func") == "fn"
250
251 def test_private_fn_underscore_prefix(self):
252 assert self._infer("_private_func") == "fn"
253
254 def test_dunder_is_sym_or_fn(self):
255 # __init__ after stripping underscores: "init" → fn
256 assert self._infer("__init__") == "fn"
257
258 def test_empty_addr_is_sym(self):
259 assert self._infer("") == "sym"
260
261 def test_private_class(self):
262 assert self._infer("_PrivateClass") == "class"
263
264
265 class TestLicensesForViewerType:
266 """Unit tests for licenses_for_viewer_type."""
267
268 def test_symbol_graph_returns_code_licenses(self):
269 from musehub.api.routes.musehub.ui_new_repo import licenses_for_viewer_type
270 result = licenses_for_viewer_type("symbol_graph")
271 assert isinstance(result, list)
272 assert len(result) > 0
273 assert all(isinstance(item, tuple) and len(item) == 2 for item in result)
274
275 def test_default_returns_generic_licenses(self):
276 from musehub.api.routes.musehub.ui_new_repo import licenses_for_viewer_type
277 default = licenses_for_viewer_type("audio")
278 code = licenses_for_viewer_type("symbol_graph")
279 # They may overlap but should be distinct lists
280 assert isinstance(default, list)
281
282 def test_unknown_type_returns_list(self):
283 from musehub.api.routes.musehub.ui_new_repo import licenses_for_viewer_type
284 result = licenses_for_viewer_type("totally_unknown")
285 assert isinstance(result, list)
286
287
288 # ─────────────────────────────────────────────────────────────────────────────
289 # LAYER 2 — INTEGRATION
290 # ─────────────────────────────────────────────────────────────────────────────
291
292
293 class TestRepoPageIntegration:
294 """Integration: repo home page with real DB (no HTTP)."""
295
296 async def test_repo_page_200_with_empty_repo(
297 self, client: AsyncClient, db_session: AsyncSession
298 ) -> None:
299 repo = await _seed_repo(db_session, owner="int-owner", slug="int-repo")
300 resp = await client.get(f"/int-owner/int-repo")
301 assert resp.status_code == 200
302
303 async def test_repo_page_404_for_unknown_slug(
304 self, client: AsyncClient, db_session: AsyncSession
305 ) -> None:
306 resp = await client.get("/nobody/totally-unknown-repo-xyz")
307 assert resp.status_code == 404
308
309 async def test_repo_page_shows_owner_in_html(
310 self, client: AsyncClient, db_session: AsyncSession
311 ) -> None:
312 await _seed_repo(db_session, owner="int-owner2", slug="int-repo2")
313 resp = await client.get("/int-owner2/int-repo2")
314 assert resp.status_code == 200
315 assert "int-owner2" in resp.text
316
317
318 class TestProposalListIntegration:
319 """Integration: proposals list page with real DB."""
320
321 async def test_proposal_list_200_empty(
322 self, client: AsyncClient, db_session: AsyncSession
323 ) -> None:
324 await _seed_repo(db_session, owner="prop-owner", slug="prop-repo")
325 resp = await client.get("/prop-owner/prop-repo/proposals")
326 assert resp.status_code == 200
327
328 async def test_proposal_list_shows_pr_title(
329 self, client: AsyncClient, db_session: AsyncSession
330 ) -> None:
331 repo = await _seed_repo(db_session, owner="prop-owner2", slug="prop-repo2")
332 await _seed_proposal(db_session, str(repo.repo_id), title="Reverb on chorus")
333 resp = await client.get("/prop-owner2/prop-repo2/proposals")
334 assert resp.status_code == 200
335 assert "Reverb on chorus" in resp.text
336
337 async def test_proposal_list_404_for_unknown_repo(
338 self, client: AsyncClient, db_session: AsyncSession
339 ) -> None:
340 resp = await client.get("/nobody/unknown-proposals-repo/proposals")
341 assert resp.status_code == 404
342
343
344 class TestAgentsPageIntegration:
345 """Integration: agents swarm / coord pages with real DB."""
346
347 async def test_agents_swarm_page_200(
348 self, client: AsyncClient, db_session: AsyncSession
349 ) -> None:
350 await _seed_repo(db_session, owner="agent-owner", slug="agent-repo")
351 resp = await client.get("/agent-owner/agent-repo/agents/swarm")
352 assert resp.status_code in (200, 404) # 404 if page not registered at /agents/swarm
353
354 async def test_agents_coord_page_200(
355 self, client: AsyncClient, db_session: AsyncSession
356 ) -> None:
357 await _seed_repo(db_session, owner="agent-owner2", slug="agent-repo2")
358 resp = await client.get("/agent-owner2/agent-repo2/agents/coord")
359 assert resp.status_code in (200, 404)
360
361
362 # ─────────────────────────────────────────────────────────────────────────────
363 # LAYER 3 — E2E
364 # ─────────────────────────────────────────────────────────────────────────────
365
366
367 class TestRepoHomeE2E:
368 """E2E: repo home page (/{owner}/{slug}) — no existing test file."""
369
370 async def test_get_repo_home_returns_200(
371 self, client: AsyncClient, db_session: AsyncSession
372 ) -> None:
373 await _seed_repo(db_session, owner="e2e-home", slug="home-repo")
374 resp = await client.get("/e2e-home/home-repo")
375 assert resp.status_code == 200
376
377 async def test_repo_home_returns_html(
378 self, client: AsyncClient, db_session: AsyncSession
379 ) -> None:
380 await _seed_repo(db_session, owner="e2e-html", slug="html-repo")
381 resp = await client.get("/e2e-html/html-repo")
382 assert "text/html" in resp.headers.get("content-type", "")
383
384 async def test_repo_home_no_auth_required(
385 self, client: AsyncClient, db_session: AsyncSession
386 ) -> None:
387 """Public repos should be accessible without authentication."""
388 await _seed_repo(db_session, owner="e2e-pub", slug="pub-repo", visibility="public")
389 resp = await client.get("/e2e-pub/pub-repo")
390 assert resp.status_code != 401
391
392 async def test_repo_home_json_format_returns_json(
393 self, client: AsyncClient, db_session: AsyncSession
394 ) -> None:
395 """?format=json should return JSON instead of HTML."""
396 await _seed_repo(db_session, owner="e2e-json", slug="json-repo")
397 resp = await client.get("/e2e-json/json-repo", params={"format": "json"})
398 assert resp.status_code == 200
399 assert "application/json" in resp.headers.get("content-type", "")
400
401 async def test_repo_home_accept_json_returns_json(
402 self, client: AsyncClient, db_session: AsyncSession
403 ) -> None:
404 """Accept: application/json triggers the JSON shortcut."""
405 await _seed_repo(db_session, owner="e2e-acc", slug="acc-repo")
406 resp = await client.get(
407 "/e2e-acc/acc-repo",
408 headers={"Accept": "application/json"},
409 )
410 assert resp.status_code == 200
411 data = resp.json()
412 assert "slug" in data or "repoId" in data
413
414 async def test_repo_home_htmx_returns_fragment(
415 self, client: AsyncClient, db_session: AsyncSession
416 ) -> None:
417 """HX-Request returns file_tree fragment, not full page."""
418 await _seed_repo(db_session, owner="e2e-htmx", slug="htmx-repo")
419 resp = await client.get(
420 "/e2e-htmx/htmx-repo",
421 headers={"HX-Request": "true"},
422 )
423 # Fragment should NOT contain the full <html> wrapper
424 assert resp.status_code == 200
425 body = resp.text
426 assert "<html" not in body.lower()
427
428 async def test_repo_home_unknown_repo_returns_404(
429 self, client: AsyncClient, db_session: AsyncSession
430 ) -> None:
431 resp = await client.get("/nobody/repo-that-does-not-exist-xyz-abc")
432 assert resp.status_code == 404
433
434
435 class TestSymbolsPageE2E:
436 """E2E: symbols page (/{owner}/{slug}/symbols) — no existing test file."""
437
438 async def test_symbols_list_page_returns_200(
439 self, client: AsyncClient, db_session: AsyncSession
440 ) -> None:
441 await _seed_repo(db_session, owner="sym-owner", slug="sym-repo")
442 resp = await client.get("/sym-owner/sym-repo/symbols")
443 assert resp.status_code == 200
444
445 async def test_symbols_list_no_auth_required(
446 self, client: AsyncClient, db_session: AsyncSession
447 ) -> None:
448 await _seed_repo(db_session, owner="sym-noauth", slug="sym-noauth-repo")
449 resp = await client.get("/sym-noauth/sym-noauth-repo/symbols")
450 assert resp.status_code != 401
451
452 async def test_symbols_list_unknown_repo_404(
453 self, client: AsyncClient, db_session: AsyncSession
454 ) -> None:
455 resp = await client.get("/nobody/unknown-sym-repo-xyz/symbols")
456 assert resp.status_code == 404
457
458 async def test_symbol_detail_unknown_repo_404(
459 self, client: AsyncClient, db_session: AsyncSession
460 ) -> None:
461 resp = await client.get("/nobody/unknown-sym-repo-xyz/symbol/file.py::MyFn")
462 assert resp.status_code == 404
463
464
465 class TestProposalsPageE2E:
466 """E2E: proposals pages — supplementing the minimal ssr file."""
467
468 async def test_proposal_list_returns_200(
469 self, client: AsyncClient, db_session: AsyncSession
470 ) -> None:
471 await _seed_repo(db_session, owner="proposal-e2e", slug="proposal-e2e-repo")
472 resp = await client.get("/proposal-e2e/proposal-e2e-repo/proposals")
473 assert resp.status_code == 200
474
475 async def test_proposal_list_html(
476 self, client: AsyncClient, db_session: AsyncSession
477 ) -> None:
478 await _seed_repo(db_session, owner="proposal-e2e2", slug="proposal-e2e-repo2")
479 resp = await client.get("/proposal-e2e2/proposal-e2e-repo2/proposals")
480 assert "text/html" in resp.headers.get("content-type", "")
481
482 async def test_proposal_detail_404_for_unknown_pr_id(
483 self, client: AsyncClient, db_session: AsyncSession
484 ) -> None:
485 await _seed_repo(db_session, owner="proposal-e2e3", slug="proposal-e2e-repo3")
486 fake_uuid = str(uuid.uuid4())
487 resp = await client.get(f"/proposal-e2e3/proposal-e2e-repo3/proposals/{fake_uuid}")
488 assert resp.status_code == 404
489
490 async def test_proposal_detail_renders_pr_title(
491 self, client: AsyncClient, db_session: AsyncSession
492 ) -> None:
493 repo = await _seed_repo(db_session, owner="proposal-detail", slug="proposal-detail-repo")
494 proposal = await _seed_proposal(
495 db_session, str(repo.repo_id), title="Unique proposal XYZ"
496 )
497 # URL uses proposal_id (UUID), not proposal_number
498 resp = await client.get(
499 f"/proposal-detail/proposal-detail-repo/proposals/{proposal.proposal_id}"
500 )
501 assert resp.status_code == 200
502 assert "Unique proposal XYZ" in resp.text
503
504
505 class TestIntelPageE2E:
506 """E2E: intel pages — no existing test file."""
507
508 async def test_intel_page_returns_200_or_empty(
509 self, client: AsyncClient, db_session: AsyncSession
510 ) -> None:
511 await _seed_repo(db_session, owner="intel-owner", slug="intel-repo")
512 resp = await client.get("/intel-owner/intel-repo/intel")
513 assert resp.status_code in (200, 404)
514
515 async def test_intel_dead_page_no_500(
516 self, client: AsyncClient, db_session: AsyncSession
517 ) -> None:
518 await _seed_repo(db_session, owner="intel-dead", slug="intel-dead-repo")
519 resp = await client.get("/intel-dead/intel-dead-repo/intel/dead")
520 assert resp.status_code != 500
521
522
523 class TestBlobAndRawE2E:
524 """E2E: blob and raw file pages — no existing test file."""
525
526 async def test_blob_page_404_for_unknown_repo(
527 self, client: AsyncClient, db_session: AsyncSession
528 ) -> None:
529 resp = await client.get("/nobody/unknown-blob-repo/blob/HEAD/README.md")
530 assert resp.status_code == 404
531
532 async def test_raw_file_404_for_unknown_repo(
533 self, client: AsyncClient, db_session: AsyncSession
534 ) -> None:
535 resp = await client.get("/nobody/unknown-raw-repo/raw/HEAD/README.md")
536 assert resp.status_code == 404
537
538 async def test_blob_page_404_for_unknown_file(
539 self, client: AsyncClient, db_session: AsyncSession
540 ) -> None:
541 await _seed_repo(db_session, owner="blob-owner", slug="blob-repo")
542 resp = await client.get("/blob-owner/blob-repo/blob/HEAD/nonexistent.md")
543 assert resp.status_code in (200, 404) # empty repo may return 404 or empty page
544
545
546 class TestHTMXFragmentRoutingE2E:
547 """E2E: HTMX fragment routing for pages that support it."""
548
549 async def test_issue_list_htmx_returns_fragment(
550 self, client: AsyncClient, db_session: AsyncSession
551 ) -> None:
552 repo = await _seed_repo(db_session, owner="htmx-frag", slug="htmx-frag-repo")
553 await _seed_issue(db_session, str(repo.repo_id), title="HTMX test issue")
554 resp = await client.get(
555 "/htmx-frag/htmx-frag-repo/issues",
556 headers={"HX-Request": "true"},
557 )
558 assert resp.status_code == 200
559 assert "<html" not in resp.text.lower()
560
561 async def test_issue_list_htmx_boosted_returns_full_page(
562 self, client: AsyncClient, db_session: AsyncSession
563 ) -> None:
564 await _seed_repo(db_session, owner="htmx-boost", slug="htmx-boost-repo")
565 resp = await client.get(
566 "/htmx-boost/htmx-boost-repo/issues",
567 headers={"HX-Request": "true", "HX-Boosted": "true"},
568 )
569 # Boosted requests must get the full page
570 assert resp.status_code == 200
571
572 async def test_search_page_returns_200(
573 self, client: AsyncClient, db_session: AsyncSession
574 ) -> None:
575 resp = await client.get("/search", params={"q": "test"})
576 assert resp.status_code == 200
577
578 async def test_explore_page_returns_200(
579 self, client: AsyncClient, db_session: AsyncSession
580 ) -> None:
581 resp = await client.get("/explore")
582 assert resp.status_code == 200
583
584
585 # ─────────────────────────────────────────────────────────────────────────────
586 # LAYER 4 — STRESS
587 # ─────────────────────────────────────────────────────────────────────────────
588
589
590 class TestUISSRStress:
591 """Stress: sequential page loads and empty-state robustness."""
592
593 async def test_multiple_repo_pages_sequential(
594 self, client: AsyncClient, db_session: AsyncSession
595 ) -> None:
596 """Load 5 different repos' pages in sequence — none should 500."""
597 for i in range(5):
598 owner = f"stress-owner-{i}"
599 slug = f"stress-repo-{i}"
600 await _seed_repo(db_session, owner=owner, slug=slug)
601 resp = await client.get(f"/{owner}/{slug}")
602 assert resp.status_code == 200, f"Repo {i} returned {resp.status_code}"
603
604 async def test_issue_list_with_many_issues(
605 self, client: AsyncClient, db_session: AsyncSession
606 ) -> None:
607 """Issue list renders correctly with 20 seeded issues."""
608 repo = await _seed_repo(db_session, owner="stress-issues", slug="stress-iss-repo")
609 for i in range(20):
610 await _seed_issue(
611 db_session, str(repo.repo_id), number=i + 1, title=f"Issue {i}"
612 )
613 resp = await client.get("/stress-issues/stress-iss-repo/issues")
614 assert resp.status_code == 200
615
616 async def test_empty_state_repo_home(
617 self, client: AsyncClient, db_session: AsyncSession
618 ) -> None:
619 """Repo with no commits/files renders without 500."""
620 await _seed_repo(db_session, owner="empty-owner", slug="empty-repo")
621 resp = await client.get("/empty-owner/empty-repo")
622 assert resp.status_code == 200
623
624 async def test_empty_state_proposals_page(
625 self, client: AsyncClient, db_session: AsyncSession
626 ) -> None:
627 """Proposals page with no proposals renders without 500."""
628 await _seed_repo(db_session, owner="empty-proposal", slug="empty-proposal-repo")
629 resp = await client.get("/empty-proposal/empty-proposal-repo/proposals")
630 assert resp.status_code == 200
631
632 async def test_empty_state_symbols_page(
633 self, client: AsyncClient, db_session: AsyncSession
634 ) -> None:
635 """Symbols page with no symbol index renders without 500."""
636 await _seed_repo(db_session, owner="empty-sym", slug="empty-sym-repo")
637 resp = await client.get("/empty-sym/empty-sym-repo/symbols")
638 assert resp.status_code == 200
639
640 async def test_topics_page_returns_200(
641 self, client: AsyncClient, db_session: AsyncSession
642 ) -> None:
643 resp = await client.get("/topics")
644 assert resp.status_code == 200
645
646 async def test_search_page_empty_query_200(
647 self, client: AsyncClient, db_session: AsyncSession
648 ) -> None:
649 resp = await client.get("/search")
650 assert resp.status_code == 200
651
652 async def test_concurrent_issue_list_requests(
653 self, client: AsyncClient, db_session: AsyncSession
654 ) -> None:
655 """3 sequential requests to the same issues page succeed."""
656 repo = await _seed_repo(db_session, owner="conc-owner", slug="conc-repo")
657 await _seed_issue(db_session, str(repo.repo_id), title="Concurrent test")
658 for _ in range(3):
659 resp = await client.get("/conc-owner/conc-repo/issues")
660 assert resp.status_code == 200
661
662
663 # ─────────────────────────────────────────────────────────────────────────────
664 # LAYER 5 — DATA INTEGRITY
665 # ─────────────────────────────────────────────────────────────────────────────
666
667
668 class TestUISSRDataIntegrity:
669 """Data integrity: nav counts correct, filter accuracy, XSS escaping."""
670
671 async def test_issue_open_count_matches_db(
672 self, client: AsyncClient, db_session: AsyncSession
673 ) -> None:
674 """Nav tab shows the correct open issue count seeded in DB."""
675 repo = await _seed_repo(db_session, owner="count-owner", slug="count-repo")
676 for i in range(3):
677 await _seed_issue(
678 db_session, str(repo.repo_id), number=i + 1, state="open"
679 )
680 resp = await client.get("/count-owner/count-repo/issues")
681 assert resp.status_code == 200
682 # The number 3 should appear in the open tab count
683 assert "3" in resp.text
684
685 async def test_closed_issues_not_shown_on_open_tab(
686 self, client: AsyncClient, db_session: AsyncSession
687 ) -> None:
688 repo = await _seed_repo(db_session, owner="filter-owner", slug="filter-repo")
689 await _seed_issue(
690 db_session, str(repo.repo_id), number=1, title="Open issue", state="open"
691 )
692 await _seed_issue(
693 db_session,
694 str(repo.repo_id),
695 number=2,
696 title="Closed issue xyz",
697 state="closed",
698 )
699 resp = await client.get("/filter-owner/filter-repo/issues", params={"state": "open"})
700 assert resp.status_code == 200
701 assert "Open issue" in resp.text
702 assert "Closed issue xyz" not in resp.text
703
704 async def test_open_issues_not_shown_on_closed_tab(
705 self, client: AsyncClient, db_session: AsyncSession
706 ) -> None:
707 repo = await _seed_repo(db_session, owner="filter-owner2", slug="filter-repo2")
708 await _seed_issue(
709 db_session, str(repo.repo_id), number=1, title="Open issue abc", state="open"
710 )
711 await _seed_issue(
712 db_session,
713 str(repo.repo_id),
714 number=2,
715 title="Closed issue only",
716 state="closed",
717 )
718 resp = await client.get(
719 "/filter-owner2/filter-repo2/issues", params={"state": "closed"}
720 )
721 assert resp.status_code == 200
722 assert "Closed issue only" in resp.text
723 assert "Open issue abc" not in resp.text
724
725 async def test_proposal_title_rendered_in_list(
726 self, client: AsyncClient, db_session: AsyncSession
727 ) -> None:
728 repo = await _seed_repo(db_session, owner="proposal-render", slug="proposal-render-repo")
729 await _seed_proposal(
730 db_session, str(repo.repo_id), title="Piano roll refactor"
731 )
732 resp = await client.get("/proposal-render/proposal-render-repo/proposals")
733 assert "Piano roll refactor" in resp.text
734
735 async def test_xss_title_escaped_in_issue_list(
736 self, client: AsyncClient, db_session: AsyncSession
737 ) -> None:
738 """Issue titles with HTML special chars must be escaped."""
739 repo = await _seed_repo(db_session, owner="xss-owner", slug="xss-repo")
740 xss_title = '<script>alert("xss")</script>'
741 await _seed_issue(db_session, str(repo.repo_id), title=xss_title)
742 resp = await client.get("/xss-owner/xss-repo/issues")
743 assert resp.status_code == 200
744 # The raw script tag must not appear unescaped
745 assert "<script>alert" not in resp.text
746 # The escaped version must be present instead
747 assert "&lt;script&gt;" in resp.text or "alert" not in resp.text
748
749 async def test_xss_in_proposal_title_escaped(
750 self, client: AsyncClient, db_session: AsyncSession
751 ) -> None:
752 repo = await _seed_repo(db_session, owner="xss-proposal", slug="xss-proposal-repo")
753 xss_title = "<img src=x onerror=alert(1)>"
754 await _seed_proposal(db_session, str(repo.repo_id), title=xss_title)
755 resp = await client.get("/xss-proposal/xss-proposal-repo/proposals")
756 assert resp.status_code == 200
757 assert "<img src=x onerror" not in resp.text
758
759 async def test_owner_name_in_repo_home_html(
760 self, client: AsyncClient, db_session: AsyncSession
761 ) -> None:
762 await _seed_repo(db_session, owner="render-owner-z", slug="render-repo-z")
763 resp = await client.get("/render-owner-z/render-repo-z")
764 assert "render-owner-z" in resp.text
765
766 async def test_repo_slug_in_repo_home_html(
767 self, client: AsyncClient, db_session: AsyncSession
768 ) -> None:
769 await _seed_repo(db_session, owner="slug-owner", slug="visible-slug-abc")
770 resp = await client.get("/slug-owner/visible-slug-abc")
771 assert "visible-slug-abc" in resp.text
772
773
774 # ─────────────────────────────────────────────────────────────────────────────
775 # LAYER 6 — SECURITY
776 # ─────────────────────────────────────────────────────────────────────────────
777
778
779 class TestUISSRSecurity:
780 """Security: auth enforcement, XSS, no debug leaks."""
781
782 async def test_create_repo_post_401_without_auth(
783 self, client: AsyncClient, db_session: AsyncSession
784 ) -> None:
785 """POST /new (create repo wizard) requires a valid token — no token → 401."""
786 resp = await client.post(
787 "/new",
788 json={"name": "test-repo", "owner": "sec-owner", "visibility": "public"},
789 )
790 assert resp.status_code == 401
791
792 async def test_labels_post_401_without_auth(
793 self, client: AsyncClient, db_session: AsyncSession
794 ) -> None:
795 """POST to label mutations requires auth — no token → 401."""
796 await _seed_repo(db_session, owner="sec-lbl", slug="sec-lbl-repo")
797 resp = await client.post(
798 "/sec-lbl/sec-lbl-repo/labels",
799 json={"name": "bug", "color": "#ff0000"},
800 )
801 assert resp.status_code == 401
802
803 async def test_new_repo_get_returns_200_or_redirect(
804 self, client: AsyncClient, db_session: AsyncSession
805 ) -> None:
806 """GET /new may require auth — should not 500."""
807 resp = await client.get("/new")
808 assert resp.status_code in (200, 401, 302, 303)
809
810 async def test_no_traceback_on_404(
811 self, client: AsyncClient, db_session: AsyncSession
812 ) -> None:
813 resp = await client.get("/nobody/no-such-repo-xyzabc")
814 assert "Traceback" not in resp.text
815
816 async def test_no_traceback_on_repo_home(
817 self, client: AsyncClient, db_session: AsyncSession
818 ) -> None:
819 await _seed_repo(db_session, owner="notrace-owner", slug="notrace-repo")
820 resp = await client.get("/notrace-owner/notrace-repo")
821 assert "Traceback" not in resp.text
822
823 async def test_no_stack_trace_on_issue_list(
824 self, client: AsyncClient, db_session: AsyncSession
825 ) -> None:
826 await _seed_repo(db_session, owner="notrace2", slug="notrace-repo2")
827 resp = await client.get("/notrace2/notrace-repo2/issues")
828 assert "Traceback" not in resp.text
829
830 async def test_private_repo_home_does_not_500(
831 self, client: AsyncClient, db_session: AsyncSession
832 ) -> None:
833 """Private repo home renders without error (visibility enforced at write APIs)."""
834 await _seed_repo(
835 db_session, owner="priv-owner", slug="priv-repo", visibility="private"
836 )
837 resp = await client.get("/priv-owner/priv-repo")
838 # Read layer is publicly accessible; write mutations require auth
839 assert resp.status_code != 500
840
841 async def test_xss_in_query_param_not_reflected_raw(
842 self, client: AsyncClient, db_session: AsyncSession
843 ) -> None:
844 """Search query containing XSS payload must not be reflected unescaped."""
845 resp = await client.get("/search", params={"q": '<script>alert(1)</script>'})
846 assert "<script>alert(1)</script>" not in resp.text
847
848 async def test_no_debug_info_in_production_errors(
849 self, client: AsyncClient, db_session: AsyncSession
850 ) -> None:
851 """404 responses must not contain Python module paths."""
852 resp = await client.get("/nobody/sec-unknown-repo-xyz")
853 assert "/Users/" not in resp.text
854 assert "musehub/" not in resp.text or resp.status_code != 500
855
856 async def test_sql_injection_in_owner_does_not_500(
857 self, client: AsyncClient, db_session: AsyncSession
858 ) -> None:
859 """URL path parameters passed as SQL injection should result in 404, not 500."""
860 resp = await client.get("/'; DROP TABLE musehub_repos; --/repo")
861 assert resp.status_code != 500
862
863
864 # ─────────────────────────────────────────────────────────────────────────────
865 # LAYER 7 — PERFORMANCE
866 # ─────────────────────────────────────────────────────────────────────────────
867
868
869 class TestUISSRPerformance:
870 """Performance: page responses under time budgets."""
871
872 async def test_repo_home_under_500ms(
873 self, client: AsyncClient, db_session: AsyncSession
874 ) -> None:
875 await _seed_repo(db_session, owner="perf-owner", slug="perf-repo")
876 # Warm-up
877 await client.get("/perf-owner/perf-repo")
878 start = time.perf_counter()
879 resp = await client.get("/perf-owner/perf-repo")
880 elapsed = time.perf_counter() - start
881 assert resp.status_code == 200
882 assert elapsed < 0.500, f"Repo home took {elapsed*1000:.1f}ms (limit 500ms)"
883
884 async def test_issue_list_under_500ms(
885 self, client: AsyncClient, db_session: AsyncSession
886 ) -> None:
887 repo = await _seed_repo(db_session, owner="perf-iss", slug="perf-iss-repo")
888 for i in range(5):
889 await _seed_issue(
890 db_session, str(repo.repo_id), number=i + 1, title=f"Issue {i}"
891 )
892 await client.get("/perf-iss/perf-iss-repo/issues")
893 start = time.perf_counter()
894 resp = await client.get("/perf-iss/perf-iss-repo/issues")
895 elapsed = time.perf_counter() - start
896 assert resp.status_code == 200
897 assert elapsed < 0.500, f"Issue list took {elapsed*1000:.1f}ms (limit 500ms)"
898
899 async def test_proposals_page_under_500ms(
900 self, client: AsyncClient, db_session: AsyncSession
901 ) -> None:
902 repo = await _seed_repo(db_session, owner="perf-proposal", slug="perf-proposal-repo")
903 await _seed_proposal(db_session, str(repo.repo_id), title="Perf test proposal")
904 await client.get("/perf-proposal/perf-proposal-repo/proposals")
905 start = time.perf_counter()
906 resp = await client.get("/perf-proposal/perf-proposal-repo/proposals")
907 elapsed = time.perf_counter() - start
908 assert resp.status_code == 200
909 assert elapsed < 0.500, f"Proposals page took {elapsed*1000:.1f}ms (limit 500ms)"
910
911 async def test_explore_page_under_500ms(
912 self, client: AsyncClient, db_session: AsyncSession
913 ) -> None:
914 await client.get("/explore") # warm-up
915 start = time.perf_counter()
916 resp = await client.get("/explore")
917 elapsed = time.perf_counter() - start
918 assert resp.status_code == 200
919 assert elapsed < 0.500, f"Explore page took {elapsed*1000:.1f}ms (limit 500ms)"
920
921 async def test_search_page_under_500ms(
922 self, client: AsyncClient, db_session: AsyncSession
923 ) -> None:
924 await client.get("/search")
925 start = time.perf_counter()
926 resp = await client.get("/search", params={"q": "test"})
927 elapsed = time.perf_counter() - start
928 assert resp.status_code == 200
929 assert elapsed < 0.500, f"Search page took {elapsed*1000:.1f}ms (limit 500ms)"
930
931 def test_to_camel_via_htmx_helpers_import_fast(self) -> None:
932 """Importing htmx_helpers is fast (module already loaded)."""
933 start = time.perf_counter()
934 from musehub.api.routes.musehub.htmx_helpers import is_htmx, htmx_trigger # noqa: F401
935 elapsed = time.perf_counter() - start
936 assert elapsed < 0.050, f"Import took {elapsed*1000:.1f}ms (limit 50ms)"
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago