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