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