gabriel / musehub public
test_domains.py python
914 lines 31.6 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 20 days ago
1 """Section 25 — Domains: 8-layer test suite.
2
3 Covers musehub/services/musehub_domains.py and
4 musehub/api/routes/musehub/domains.py.
5
6 Layer map
7 ---------
8 1. Unit — compute_manifest_hash, _to_response, dataclasses
9 2. Integration — service functions against real PostgreSQL DB
10 3. E2E — HTTP client against full app
11 4. Stress — many domains, concurrent queries
12 5. Data Integrity — sort order, deprecated exclusion, hash correctness
13 6. Security — auth enforcement, duplicate scoped_id
14 7. Performance — timing budgets
15 8. Admin — domain verification endpoint
16 """
17 from __future__ import annotations
18
19 import asyncio
20 import json
21 import secrets
22 from collections.abc import Generator, Mapping
23 import time
24
25 import pytest
26 import pytest_asyncio
27 from httpx import AsyncClient
28 from sqlalchemy.ext.asyncio import AsyncSession
29
30 from muse.core.types import blob_id
31 from musehub.auth.request_signing import MSignContext, optional_signed_request, require_signed_request
32 from musehub.db.musehub_domain_models import MusehubDomain, MusehubDomainInstall
33 from datetime import datetime, timezone
34
35 from musehub.core.genesis import compute_identity_id, compute_repo_id
36 from musehub.db.musehub_identity_models import MusehubIdentity
37 from musehub.db.musehub_repo_models import MusehubRepo
38 from musehub.main import app
39 from musehub.types.json_types import JSONObject, StrDict
40 from musehub.services.musehub_domains import (
41 DomainListResponse,
42 DomainReposResponse,
43 DomainResponse,
44 _to_response,
45 compute_manifest_hash,
46 create_domain,
47 get_domain_by_id,
48 get_domain_by_scoped_id,
49 list_domains,
50 list_repos_for_domain,
51 record_domain_install,
52 )
53
54
55 # ---------------------------------------------------------------------------
56 # DB helpers
57 # ---------------------------------------------------------------------------
58
59
60 def _uid() -> str:
61 return secrets.token_hex(16)
62
63
64 async def _db_domain(
65 session: AsyncSession,
66 *,
67 author_slug: str = "alice",
68 slug: str | None = None,
69 display_name: str = "Test Domain",
70 description: str = "A test domain",
71 capabilities: JSONObject | None = None,
72 viewer_type: str = "generic",
73 version: str = "1.0.0",
74 install_count: int = 0,
75 is_verified: bool = False,
76 is_deprecated: bool = False,
77 ) -> MusehubDomain:
78 from datetime import datetime, timezone
79
80 slug = slug or f"domain-{_uid()[:8]}"
81 caps = capabilities or {"dimensions": [], "merge_semantics": "three_way"}
82 manifest_hash = compute_manifest_hash(caps)
83 domain = MusehubDomain(
84 domain_id=_uid(),
85 author_user_id=author_slug,
86 author_slug=author_slug,
87 slug=slug,
88 display_name=display_name,
89 description=description,
90 version=version,
91 manifest_hash=manifest_hash,
92 capabilities=caps,
93 viewer_type=viewer_type,
94 install_count=install_count,
95 is_verified=is_verified,
96 is_deprecated=is_deprecated,
97 created_at=datetime.now(timezone.utc),
98 updated_at=datetime.now(timezone.utc),
99 )
100 session.add(domain)
101 await session.flush()
102 return domain
103
104
105 async def _db_repo(
106 session: AsyncSession,
107 owner: str = "alice",
108 *,
109 domain_id: str | None = None,
110 visibility: str = "public",
111 deleted: bool = False,
112 ) -> MusehubRepo:
113 slug = f"repo-{_uid()[:8]}"
114 owner_id = compute_identity_id(owner.encode())
115 created_at = datetime.now(tz=timezone.utc)
116 repo = MusehubRepo(
117 repo_id=compute_repo_id(owner_id, slug, "code", created_at.isoformat()),
118 name=slug,
119 slug=slug,
120 owner=owner,
121 owner_user_id=owner_id,
122 visibility=visibility,
123 domain_id=domain_id,
124 created_at=created_at,
125 updated_at=created_at,
126 )
127 session.add(repo)
128 await session.flush()
129 if deleted:
130 await session.delete(repo)
131 await session.flush()
132 return repo
133
134
135 # ===========================================================================
136 # Layer 1 — Unit
137 # ===========================================================================
138
139
140 class TestUnitComputeManifestHash:
141 def test_returns_hex_string(self) -> None:
142 h = compute_manifest_hash({"dimensions": []})
143 assert isinstance(h, str)
144 assert h.startswith("sha256:")
145 assert len(h) == 71 # sha256:<64-hex>
146
147 def test_deterministic(self) -> None:
148 caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"}
149 assert compute_manifest_hash(caps) == compute_manifest_hash(caps)
150
151 def test_sorted_keys_order_independent(self) -> None:
152 caps_a = {"b": 1, "a": 2}
153 caps_b = {"a": 2, "b": 1}
154 assert compute_manifest_hash(caps_a) == compute_manifest_hash(caps_b)
155
156 def test_different_capabilities_different_hash(self) -> None:
157 h1 = compute_manifest_hash({"dimensions": []})
158 h2 = compute_manifest_hash({"dimensions": [{"name": "tempo"}]})
159 assert h1 != h2
160
161 def test_matches_manual_sha256(self) -> None:
162 caps = {"x": 1}
163 blob = json.dumps(caps, sort_keys=True, separators=(",", ":")).encode()
164 expected = blob_id(blob)
165 assert compute_manifest_hash(caps) == expected
166
167
168 class TestUnitToResponse:
169 """Unit tests for _to_response using an integration DB fixture to create real ORM rows."""
170
171 async def test_scoped_id_format(self, db_session: AsyncSession) -> None:
172 d = await _db_domain(db_session, author_slug="gabriel", slug="midi")
173 resp = _to_response(d)
174 assert resp.scoped_id == "@gabriel/midi"
175
176 async def test_install_count_preserved(self, db_session: AsyncSession) -> None:
177 d = await _db_domain(db_session, slug="midi-count", install_count=5)
178 resp = _to_response(d)
179 assert resp.install_count == 5
180
181 async def test_is_verified_preserved(self, db_session: AsyncSession) -> None:
182 d = await _db_domain(db_session, slug="midi-verified", is_verified=True)
183 resp = _to_response(d)
184 assert resp.is_verified is True
185
186 async def test_capabilities_copied(self, db_session: AsyncSession) -> None:
187 caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"}
188 d = await _db_domain(db_session, slug="midi-caps", capabilities=caps)
189 resp = _to_response(d)
190 assert resp.capabilities == caps
191
192 async def test_none_capabilities_become_empty_dict(
193 self, db_session: AsyncSession
194 ) -> None:
195 d = await _db_domain(db_session, slug="midi-no-caps")
196 d.capabilities = None
197 resp = _to_response(d)
198 assert resp.capabilities == {}
199
200
201 class TestUnitDataclasses:
202 def test_domain_response_fields(self) -> None:
203 from datetime import datetime, timezone
204
205 dr = DomainResponse(
206 domain_id=_uid(),
207 author_slug="alice",
208 slug="midi",
209 scoped_id="@alice/midi",
210 display_name="MIDI",
211 description="",
212 version="1.0.0",
213 manifest_hash="abc",
214 capabilities={},
215 viewer_type="generic",
216 install_count=0,
217 is_verified=False,
218 is_deprecated=False,
219 created_at=datetime.now(timezone.utc),
220 updated_at=datetime.now(timezone.utc),
221 )
222 assert dr.scoped_id == "@alice/midi"
223
224 def test_domain_list_response_fields(self) -> None:
225 dlr = DomainListResponse(domains=[], total=0)
226 assert dlr.total == 0
227
228 def test_domain_repos_response_fields(self) -> None:
229 drr = DomainReposResponse(
230 domain_id=_uid(), scoped_id="@a/b", repos=[], total=0
231 )
232 assert drr.repos == []
233
234
235 # ===========================================================================
236 # Layer 2 — Integration
237 # ===========================================================================
238
239
240 class TestIntegrationListDomains:
241 async def test_returns_all_non_deprecated(self, db_session: AsyncSession) -> None:
242 await _db_domain(db_session, slug="midi", display_name="MIDI")
243 await _db_domain(db_session, slug="code", display_name="Code")
244 await _db_domain(db_session, slug="old", is_deprecated=True)
245 await db_session.flush()
246
247 result = await list_domains(db_session)
248 slugs = [d.slug for d in result.domains]
249 assert "midi" in slugs
250 assert "code" in slugs
251 assert "old" not in slugs
252
253 async def test_query_filters_by_display_name(self, db_session: AsyncSession) -> None:
254 await _db_domain(db_session, slug="piano", display_name="Piano Roll Domain")
255 await _db_domain(db_session, slug="genome", display_name="Genomics Domain")
256 await db_session.flush()
257
258 result = await list_domains(db_session, query="Piano")
259 assert len(result.domains) == 1
260 assert result.domains[0].slug == "piano"
261
262 async def test_verified_only_filter(self, db_session: AsyncSession) -> None:
263 await _db_domain(db_session, slug="v", is_verified=True)
264 await _db_domain(db_session, slug="u", is_verified=False)
265 await db_session.flush()
266
267 result = await list_domains(db_session, verified_only=True)
268 slugs = [d.slug for d in result.domains]
269 assert "v" in slugs
270 assert "u" not in slugs
271
272 async def test_pagination_page_size(self, db_session: AsyncSession) -> None:
273 for i in range(5):
274 await _db_domain(db_session, slug=f"d{i}")
275 await db_session.flush()
276
277 result = await list_domains(db_session, limit=2)
278 assert len(result.domains) == 2
279 assert result.total == 5
280 assert result.next_cursor is not None
281
282
283 class TestIntegrationGetDomain:
284 async def test_get_by_scoped_id_found(self, db_session: AsyncSession) -> None:
285 await _db_domain(db_session, author_slug="alice", slug="midi")
286 await db_session.flush()
287
288 result = await get_domain_by_scoped_id(db_session, "alice", "midi")
289 assert result is not None
290 assert result.scoped_id == "@alice/midi"
291
292 async def test_get_by_scoped_id_not_found(self, db_session: AsyncSession) -> None:
293 result = await get_domain_by_scoped_id(db_session, "nobody", "missing")
294 assert result is None
295
296 async def test_get_by_id_found(self, db_session: AsyncSession) -> None:
297 d = await _db_domain(db_session, slug="code")
298 await db_session.flush()
299
300 result = await get_domain_by_id(db_session, d.domain_id)
301 assert result is not None
302 assert result.domain_id == d.domain_id
303
304 async def test_get_by_id_not_found(self, db_session: AsyncSession) -> None:
305 result = await get_domain_by_id(db_session, "nonexistent-id")
306 assert result is None
307
308
309 class TestIntegrationListReposForDomain:
310 async def test_returns_public_repos(self, db_session: AsyncSession) -> None:
311 d = await _db_domain(db_session, slug="midi")
312 await _db_repo(db_session, domain_id=d.domain_id, visibility="public")
313 await _db_repo(db_session, domain_id=d.domain_id, visibility="public")
314 await db_session.flush()
315
316 result = await list_repos_for_domain(db_session, d.domain_id)
317 assert result.total == 2
318 assert len(result.repos) == 2
319
320 async def test_private_repos_excluded(self, db_session: AsyncSession) -> None:
321 d = await _db_domain(db_session, slug="midi")
322 await _db_repo(db_session, domain_id=d.domain_id, visibility="public")
323 await _db_repo(db_session, domain_id=d.domain_id, visibility="private")
324 await db_session.flush()
325
326 result = await list_repos_for_domain(db_session, d.domain_id)
327 assert result.total == 1
328
329 async def test_deleted_repos_excluded(self, db_session: AsyncSession) -> None:
330 d = await _db_domain(db_session, slug="midi")
331 await _db_repo(db_session, domain_id=d.domain_id, visibility="public")
332 await _db_repo(db_session, domain_id=d.domain_id, visibility="public", deleted=True)
333 await db_session.flush()
334
335 result = await list_repos_for_domain(db_session, d.domain_id)
336 assert result.total == 1
337
338 async def test_nonexistent_domain_returns_empty(self, db_session: AsyncSession) -> None:
339 result = await list_repos_for_domain(db_session, "nonexistent-id")
340 assert result.total == 0
341 assert result.repos == []
342
343
344 class TestIntegrationCreateDomain:
345 async def test_creates_domain_with_hash(self, db_session: AsyncSession) -> None:
346 caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"}
347 result = await create_domain(
348 db_session,
349 author_user_id="alice",
350 author_slug="alice",
351 slug="midi",
352 display_name="MIDI",
353 description="MIDI domain",
354 capabilities=caps,
355 )
356 assert result.domain_id != ""
357 assert result.manifest_hash == compute_manifest_hash(caps)
358
359 async def test_scoped_id_format(self, db_session: AsyncSession) -> None:
360 result = await create_domain(
361 db_session,
362 author_user_id="bob",
363 author_slug="bob",
364 slug="code",
365 display_name="Code",
366 description="",
367 capabilities={},
368 )
369 assert result.scoped_id == "@bob/code"
370
371 async def test_not_verified_by_default(self, db_session: AsyncSession) -> None:
372 result = await create_domain(
373 db_session,
374 author_user_id="alice",
375 author_slug="alice",
376 slug="genome",
377 display_name="Genome",
378 description="",
379 capabilities={},
380 )
381 assert result.is_verified is False
382 assert result.is_deprecated is False
383
384
385 class TestIntegrationRecordDomainInstall:
386 async def test_increments_install_count(self, db_session: AsyncSession) -> None:
387 d = await _db_domain(db_session, slug="midi", install_count=0)
388 await db_session.flush()
389
390 await record_domain_install(db_session, "user1", d.domain_id)
391 await db_session.flush()
392
393 # Verify via get_domain
394 domain = await get_domain_by_id(db_session, d.domain_id)
395 assert domain is not None
396 assert domain.install_count == 1
397
398 async def test_idempotent_same_user(self, db_session: AsyncSession) -> None:
399 d = await _db_domain(db_session, slug="midi", install_count=0)
400 await db_session.flush()
401
402 await record_domain_install(db_session, "user1", d.domain_id)
403 await record_domain_install(db_session, "user1", d.domain_id) # duplicate
404 await db_session.flush()
405
406 domain = await get_domain_by_id(db_session, d.domain_id)
407 assert domain is not None
408 assert domain.install_count == 1 # not 2
409
410 async def test_different_users_each_increment(self, db_session: AsyncSession) -> None:
411 d = await _db_domain(db_session, slug="midi", install_count=0)
412 await db_session.flush()
413
414 await record_domain_install(db_session, "user1", d.domain_id)
415 await record_domain_install(db_session, "user2", d.domain_id)
416 await db_session.flush()
417
418 domain = await get_domain_by_id(db_session, d.domain_id)
419 assert domain is not None
420 assert domain.install_count == 2
421
422
423 # ===========================================================================
424 # Layer 3 — E2E
425 # ===========================================================================
426
427
428 class TestE2EListDomains:
429 async def test_list_returns_200(
430 self,
431 client: AsyncClient,
432 db_session: AsyncSession,
433 ) -> None:
434 await _db_domain(db_session, slug="midi-api")
435 await db_session.commit()
436
437 r = await client.get("/api/domains")
438 assert r.status_code == 200
439 body = r.json()
440 assert "domains" in body
441 assert "total" in body
442 assert isinstance(body["domains"], list)
443
444 async def test_list_no_auth_required(
445 self,
446 client: AsyncClient,
447 db_session: AsyncSession,
448 ) -> None:
449 await db_session.commit()
450 r = await client.get("/api/domains")
451 assert r.status_code == 200
452
453 async def test_list_query_param_filters(
454 self,
455 client: AsyncClient,
456 db_session: AsyncSession,
457 ) -> None:
458 await _db_domain(db_session, slug="piano-e2e", display_name="Piano Roll E2E")
459 await _db_domain(db_session, slug="genome-e2e", display_name="Genome E2E")
460 await db_session.commit()
461
462 r = await client.get("/api/domains?q=Piano+Roll")
463 assert r.status_code == 200
464 body = r.json()
465 slugs = [d["slug"] for d in body["domains"]]
466 assert "piano-e2e" in slugs
467 assert "genome-e2e" not in slugs
468
469 async def test_list_page_size_param(
470 self,
471 client: AsyncClient,
472 db_session: AsyncSession,
473 ) -> None:
474 for i in range(5):
475 await _db_domain(db_session, slug=f"e2e-page-{i}")
476 await db_session.commit()
477
478 r = await client.get("/api/domains?limit=2")
479 assert r.status_code == 200
480 body = r.json()
481 assert len(body["domains"]) <= 2
482 assert body["nextCursor"] is not None
483
484
485 class TestE2ERegisterDomain:
486 async def test_register_201(
487 self,
488 client: AsyncClient,
489 auth_headers: StrDict,
490 db_session: AsyncSession,
491 ) -> None:
492 await db_session.commit()
493 body = {
494 "author_slug": "testuser",
495 "slug": "my-domain",
496 "display_name": "My Domain",
497 "description": "A domain for testing",
498 "capabilities": {"dimensions": [], "merge_semantics": "three_way"},
499 "viewer_type": "generic",
500 "version": "1.0.0",
501 }
502 r = await client.post("/api/domains", json=body, headers=auth_headers)
503 assert r.status_code == 201
504 resp = r.json()
505 assert "domain_id" in resp
506 assert "scoped_id" in resp
507 assert "manifest_hash" in resp
508
509 async def test_register_requires_auth(
510 self,
511 client: AsyncClient,
512 db_session: AsyncSession,
513 ) -> None:
514 await db_session.commit()
515 body = {
516 "author_slug": "testuser",
517 "slug": "unauthed",
518 "display_name": "Unauthed",
519 "description": "",
520 "capabilities": {},
521 }
522 r = await client.post("/api/domains", json=body)
523 assert r.status_code == 401
524
525 async def test_register_duplicate_409(
526 self,
527 client: AsyncClient,
528 auth_headers: StrDict,
529 db_session: AsyncSession,
530 ) -> None:
531 await db_session.commit()
532 body = {
533 "author_slug": "testuser",
534 "slug": "dup-domain",
535 "display_name": "Dup",
536 "description": "",
537 "capabilities": {},
538 }
539 r1 = await client.post("/api/domains", json=body, headers=auth_headers)
540 assert r1.status_code == 201
541
542 r2 = await client.post("/api/domains", json=body, headers=auth_headers)
543 assert r2.status_code == 409
544
545
546 class TestE2EGetDomain:
547 async def test_get_domain_200(
548 self,
549 client: AsyncClient,
550 db_session: AsyncSession,
551 ) -> None:
552 await _db_domain(db_session, author_slug="alice", slug="midi-detail")
553 await db_session.commit()
554
555 r = await client.get("/api/domains/@alice/midi-detail")
556 assert r.status_code == 200
557 body = r.json()
558 assert body["scoped_id"] == "@alice/midi-detail"
559 assert "capabilities" in body
560 assert "manifest_hash" in body
561
562 async def test_get_domain_404(
563 self,
564 client: AsyncClient,
565 ) -> None:
566 r = await client.get("/api/domains/@nobody/nonexistent")
567 assert r.status_code == 404
568
569 async def test_get_domain_repos_200(
570 self,
571 client: AsyncClient,
572 db_session: AsyncSession,
573 ) -> None:
574 d = await _db_domain(db_session, author_slug="alice", slug="midi-repos")
575 await _db_repo(db_session, domain_id=d.domain_id, visibility="public")
576 await db_session.commit()
577
578 r = await client.get("/api/domains/@alice/midi-repos/repos")
579 assert r.status_code == 200
580 body = r.json()
581 assert body["total"] == 1
582 assert len(body["repos"]) == 1
583
584 async def test_get_domain_repos_404_unknown_domain(
585 self,
586 client: AsyncClient,
587 ) -> None:
588 r = await client.get("/api/domains/@nobody/missing/repos")
589 assert r.status_code == 404
590
591
592 # ===========================================================================
593 # Layer 4 — Stress
594 # ===========================================================================
595
596
597 class TestStress:
598 async def test_list_100_domains(self, db_session: AsyncSession) -> None:
599 for i in range(100):
600 await _db_domain(db_session, slug=f"domain-{i}")
601 await db_session.flush()
602
603 result = await list_domains(db_session, limit=100)
604 assert result.total == 100
605
606 async def test_concurrent_list_domains(self, db_session: AsyncSession) -> None:
607 for i in range(10):
608 await _db_domain(db_session, slug=f"c{i}")
609 await db_session.flush()
610
611 results = await asyncio.gather(
612 *[list_domains(db_session) for _ in range(5)]
613 )
614 assert all(r.total == 10 for r in results)
615
616 async def test_list_repos_for_domain_50_repos(
617 self, db_session: AsyncSession
618 ) -> None:
619 d = await _db_domain(db_session, slug="big-domain")
620 for i in range(50):
621 await _db_repo(db_session, domain_id=d.domain_id, visibility="public")
622 await db_session.flush()
623
624 result = await list_repos_for_domain(db_session, d.domain_id, limit=50)
625 assert result.total == 50
626
627
628 # ===========================================================================
629 # Layer 5 — Data Integrity
630 # ===========================================================================
631
632
633 class TestDataIntegrity:
634 async def test_sorted_by_install_count_desc(self, db_session: AsyncSession) -> None:
635 await _db_domain(db_session, slug="low", install_count=1)
636 await _db_domain(db_session, slug="high", install_count=10)
637 await _db_domain(db_session, slug="mid", install_count=5)
638 await db_session.flush()
639
640 result = await list_domains(db_session)
641 counts = [d.install_count for d in result.domains]
642 assert counts == sorted(counts, reverse=True)
643
644 async def test_deprecated_excluded_from_list(self, db_session: AsyncSession) -> None:
645 await _db_domain(db_session, slug="active")
646 await _db_domain(db_session, slug="old", is_deprecated=True)
647 await db_session.flush()
648
649 result = await list_domains(db_session)
650 slugs = [d.slug for d in result.domains]
651 assert "active" in slugs
652 assert "old" not in slugs
653
654 async def test_manifest_hash_matches_capabilities(
655 self, db_session: AsyncSession
656 ) -> None:
657 caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"}
658 result = await create_domain(
659 db_session,
660 author_user_id="alice",
661 author_slug="alice",
662 slug="verify-hash",
663 display_name="Verify",
664 description="",
665 capabilities=caps,
666 )
667 assert result.manifest_hash == compute_manifest_hash(caps)
668
669 async def test_capabilities_json_not_lossy(self, db_session: AsyncSession) -> None:
670 caps = {
671 "dimensions": [{"name": "tempo", "unit": "bpm"}],
672 "artifact_types": ["audio/midi"],
673 "merge_semantics": "ot",
674 }
675 created = await create_domain(
676 db_session,
677 author_user_id="alice",
678 author_slug="alice",
679 slug="caps-test",
680 display_name="Caps",
681 description="",
682 capabilities=caps,
683 )
684 await db_session.flush()
685
686 retrieved = await get_domain_by_id(db_session, created.domain_id)
687 assert retrieved is not None
688 assert retrieved.capabilities["dimensions"][0]["name"] == "tempo"
689
690 async def test_total_count_reflects_query_filter(
691 self, db_session: AsyncSession
692 ) -> None:
693 await _db_domain(db_session, slug="match-a", display_name="Match This")
694 await _db_domain(db_session, slug="no-match", display_name="Something Else")
695 await db_session.flush()
696
697 result = await list_domains(db_session, query="Match This")
698 assert result.total == 1
699
700
701 # ===========================================================================
702 # Layer 6 — Security
703 # ===========================================================================
704
705
706 class TestSecurity:
707 async def test_post_without_auth_returns_401(
708 self,
709 client: AsyncClient,
710 db_session: AsyncSession,
711 ) -> None:
712 await db_session.commit()
713 r = await client.post(
714 "/api/domains",
715 json={
716 "author_slug": "hack",
717 "slug": "hack-domain",
718 "display_name": "Hack",
719 "description": "",
720 "capabilities": {},
721 },
722 )
723 assert r.status_code == 401
724
725 async def test_duplicate_scoped_id_returns_409(
726 self,
727 client: AsyncClient,
728 auth_headers: StrDict,
729 db_session: AsyncSession,
730 ) -> None:
731 await db_session.commit()
732 body = {
733 "author_slug": "testuser",
734 "slug": "conflict-test",
735 "display_name": "Conflict",
736 "description": "",
737 "capabilities": {},
738 }
739 r1 = await client.post("/api/domains", json=body, headers=auth_headers)
740 assert r1.status_code == 201
741
742 r2 = await client.post("/api/domains", json=body, headers=auth_headers)
743 assert r2.status_code == 409
744 assert "already registered" in r2.json()["detail"]
745
746 async def test_sql_injection_in_query_param_safe(
747 self,
748 client: AsyncClient,
749 db_session: AsyncSession,
750 ) -> None:
751 await db_session.commit()
752 r = await client.get("/api/domains?q='; DROP TABLE musehub_domains; --")
753 assert r.status_code == 200 # parameterized query — safe
754
755 async def test_manifest_hash_tampering_detectable(self) -> None:
756 """Different capabilities always produce different hashes."""
757 original_caps = {"dimensions": [{"name": "tempo"}]}
758 tampered_caps = {"dimensions": [{"name": "tempo"}, {"name": "injected"}]}
759 assert compute_manifest_hash(original_caps) != compute_manifest_hash(tampered_caps)
760
761
762 # ===========================================================================
763 # Layer 7 — Performance
764 # ===========================================================================
765
766
767 class TestPerformance:
768 async def test_list_50_domains_under_200ms(self, db_session: AsyncSession) -> None:
769 for i in range(50):
770 await _db_domain(db_session, slug=f"perf-{i}")
771 await db_session.flush()
772
773 start = time.perf_counter()
774 result = await list_domains(db_session, limit=50)
775 elapsed = time.perf_counter() - start
776
777 assert result.total == 50
778 assert elapsed < 0.2, f"list_domains took {elapsed:.3f}s"
779
780 async def test_compute_manifest_hash_fast(self) -> None:
781 caps = {"dimensions": [{"name": f"dim_{i}"} for i in range(100)]}
782 start = time.perf_counter()
783 for _ in range(1000):
784 compute_manifest_hash(caps)
785 elapsed = time.perf_counter() - start
786 assert elapsed < 0.5, f"1000 hash computations took {elapsed:.3f}s"
787
788 async def test_create_domain_under_100ms(self, db_session: AsyncSession) -> None:
789 start = time.perf_counter()
790 await create_domain(
791 db_session,
792 author_user_id="alice",
793 author_slug="alice",
794 slug="perf-create",
795 display_name="Perf",
796 description="",
797 capabilities={"dimensions": [], "merge_semantics": "ot"},
798 )
799 elapsed = time.perf_counter() - start
800 assert elapsed < 0.1, f"create_domain took {elapsed:.3f}s"
801
802
803 # ===========================================================================
804 # Layer 8 — Admin: domain verification
805 # ===========================================================================
806
807 _ADMIN_IDENTITY_ID = compute_identity_id(b"adminuser")
808 _ADMIN_HANDLE = "adminuser"
809
810 _ADMIN_CONTEXT = MSignContext(
811 handle=_ADMIN_HANDLE,
812 identity_id=_ADMIN_IDENTITY_ID,
813 is_agent=False,
814 is_admin=True,
815 )
816
817
818 @pytest_asyncio.fixture
819 async def admin_user(db_session: AsyncSession) -> MusehubIdentity:
820 identity = MusehubIdentity(
821 identity_id=_ADMIN_IDENTITY_ID,
822 handle=_ADMIN_HANDLE,
823 display_name="Admin User",
824 identity_type="human",
825 is_admin=True,
826 )
827 db_session.add(identity)
828 await db_session.commit()
829 await db_session.refresh(identity)
830 await db_session.commit()
831 return identity
832
833
834 @pytest.fixture
835 def admin_headers(admin_user: MusehubIdentity) -> Generator[dict[str, str], None, None]:
836 app.dependency_overrides[require_signed_request] = lambda: _ADMIN_CONTEXT
837 app.dependency_overrides[optional_signed_request] = lambda: _ADMIN_CONTEXT
838 yield {"Content-Type": "application/json"}
839 app.dependency_overrides.pop(require_signed_request, None)
840 app.dependency_overrides.pop(optional_signed_request, None)
841
842
843 class TestAdminVerifyDomain:
844 async def test_verify_sets_flag(
845 self,
846 client: AsyncClient,
847 admin_headers: Mapping[str, str],
848 db_session: AsyncSession,
849 ) -> None:
850 await _db_domain(db_session, author_slug="alice", slug="verifiable", is_verified=False)
851 await db_session.commit()
852
853 r = await client.post("/api/domains/@alice/verifiable/verify", headers=admin_headers)
854 assert r.status_code == 200
855 assert r.json()["is_verified"] is True
856
857 async def test_verify_idempotent(
858 self,
859 client: AsyncClient,
860 admin_headers: Mapping[str, str],
861 db_session: AsyncSession,
862 ) -> None:
863 await _db_domain(db_session, author_slug="alice", slug="already-verified", is_verified=True)
864 await db_session.commit()
865
866 r = await client.post("/api/domains/@alice/already-verified/verify", headers=admin_headers)
867 assert r.status_code == 200
868 assert r.json()["is_verified"] is True
869
870 async def test_unverify_clears_flag(
871 self,
872 client: AsyncClient,
873 admin_headers: Mapping[str, str],
874 db_session: AsyncSession,
875 ) -> None:
876 await _db_domain(db_session, author_slug="alice", slug="to-unverify", is_verified=True)
877 await db_session.commit()
878
879 r = await client.delete("/api/domains/@alice/to-unverify/verify", headers=admin_headers)
880 assert r.status_code == 200
881 assert r.json()["is_verified"] is False
882
883 async def test_verify_domain_not_found(
884 self,
885 client: AsyncClient,
886 admin_headers: Mapping[str, str],
887 db_session: AsyncSession,
888 ) -> None:
889 await db_session.commit()
890 r = await client.post("/api/domains/@nobody/ghost/verify", headers=admin_headers)
891 assert r.status_code == 404
892
893 async def test_verify_requires_admin(
894 self,
895 client: AsyncClient,
896 auth_headers: Mapping[str, str],
897 db_session: AsyncSession,
898 ) -> None:
899 await _db_domain(db_session, author_slug="alice", slug="non-admin-target")
900 await db_session.commit()
901
902 r = await client.post("/api/domains/@alice/non-admin-target/verify", headers=auth_headers)
903 assert r.status_code == 403
904
905 async def test_verify_requires_auth(
906 self,
907 client: AsyncClient,
908 db_session: AsyncSession,
909 ) -> None:
910 await _db_domain(db_session, author_slug="alice", slug="unauthed-verify")
911 await db_session.commit()
912
913 r = await client.post("/api/domains/@alice/unauthed-verify/verify")
914 assert r.status_code == 401
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 20 days ago