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