gabriel / musehub public
test_domains_section25.py python
832 lines 28.1 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 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.muse_contracts.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 from datetime import datetime, timezone
105
106 slug = f"repo-{_uid()[:8]}"
107 repo = MusehubRepo(
108 repo_id=_uid(),
109 name=slug,
110 slug=slug,
111 owner=owner,
112 owner_user_id=owner,
113 visibility=visibility,
114 domain_id=domain_id,
115 deleted_at=datetime.now(timezone.utc) if deleted else None,
116 )
117 session.add(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 @pytest.mark.anyio
158 async def test_scoped_id_format(self, db_session: AsyncSession) -> None:
159 d = await _db_domain(db_session, author_slug="gabriel", slug="midi")
160 resp = _to_response(d)
161 assert resp.scoped_id == "@gabriel/midi"
162
163 @pytest.mark.anyio
164 async def test_install_count_preserved(self, db_session: AsyncSession) -> None:
165 d = await _db_domain(db_session, slug="midi-count", install_count=5)
166 resp = _to_response(d)
167 assert resp.install_count == 5
168
169 @pytest.mark.anyio
170 async def test_is_verified_preserved(self, db_session: AsyncSession) -> None:
171 d = await _db_domain(db_session, slug="midi-verified", is_verified=True)
172 resp = _to_response(d)
173 assert resp.is_verified is True
174
175 @pytest.mark.anyio
176 async def test_capabilities_copied(self, db_session: AsyncSession) -> None:
177 caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"}
178 d = await _db_domain(db_session, slug="midi-caps", capabilities=caps)
179 resp = _to_response(d)
180 assert resp.capabilities == caps
181
182 @pytest.mark.anyio
183 async def test_none_capabilities_become_empty_dict(
184 self, db_session: AsyncSession
185 ) -> None:
186 d = await _db_domain(db_session, slug="midi-no-caps")
187 d.capabilities = None
188 resp = _to_response(d)
189 assert resp.capabilities == {}
190
191
192 class TestUnitDataclasses:
193 def test_domain_response_fields(self) -> None:
194 from datetime import datetime, timezone
195
196 dr = DomainResponse(
197 domain_id=_uid(),
198 author_slug="alice",
199 slug="midi",
200 scoped_id="@alice/midi",
201 display_name="MIDI",
202 description="",
203 version="1.0.0",
204 manifest_hash="abc",
205 capabilities={},
206 viewer_type="generic",
207 install_count=0,
208 is_verified=False,
209 is_deprecated=False,
210 created_at=datetime.now(timezone.utc),
211 updated_at=datetime.now(timezone.utc),
212 )
213 assert dr.scoped_id == "@alice/midi"
214
215 def test_domain_list_response_fields(self) -> None:
216 dlr = DomainListResponse(domains=[], total=0)
217 assert dlr.total == 0
218
219 def test_domain_repos_response_fields(self) -> None:
220 drr = DomainReposResponse(
221 domain_id=_uid(), scoped_id="@a/b", repos=[], total=0
222 )
223 assert drr.repos == []
224
225
226 # ===========================================================================
227 # Layer 2 — Integration
228 # ===========================================================================
229
230
231 class TestIntegrationListDomains:
232 @pytest.mark.anyio
233 async def test_returns_all_non_deprecated(self, db_session: AsyncSession) -> None:
234 await _db_domain(db_session, slug="midi", display_name="MIDI")
235 await _db_domain(db_session, slug="code", display_name="Code")
236 await _db_domain(db_session, slug="old", is_deprecated=True)
237 await db_session.flush()
238
239 result = await list_domains(db_session)
240 slugs = [d.slug for d in result.domains]
241 assert "midi" in slugs
242 assert "code" in slugs
243 assert "old" not in slugs
244
245 @pytest.mark.anyio
246 async def test_query_filters_by_display_name(self, db_session: AsyncSession) -> None:
247 await _db_domain(db_session, slug="piano", display_name="Piano Roll Domain")
248 await _db_domain(db_session, slug="genome", display_name="Genomics Domain")
249 await db_session.flush()
250
251 result = await list_domains(db_session, query="Piano")
252 assert len(result.domains) == 1
253 assert result.domains[0].slug == "piano"
254
255 @pytest.mark.anyio
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 @pytest.mark.anyio
267 async def test_pagination_page_size(self, db_session: AsyncSession) -> None:
268 for i in range(5):
269 await _db_domain(db_session, slug=f"d{i}")
270 await db_session.flush()
271
272 result = await list_domains(db_session, page=1, page_size=2)
273 assert len(result.domains) == 2
274 assert result.total == 5
275
276
277 class TestIntegrationGetDomain:
278 @pytest.mark.anyio
279 async def test_get_by_scoped_id_found(self, db_session: AsyncSession) -> None:
280 await _db_domain(db_session, author_slug="alice", slug="midi")
281 await db_session.flush()
282
283 result = await get_domain_by_scoped_id(db_session, "alice", "midi")
284 assert result is not None
285 assert result.scoped_id == "@alice/midi"
286
287 @pytest.mark.anyio
288 async def test_get_by_scoped_id_not_found(self, db_session: AsyncSession) -> None:
289 result = await get_domain_by_scoped_id(db_session, "nobody", "missing")
290 assert result is None
291
292 @pytest.mark.anyio
293 async def test_get_by_id_found(self, db_session: AsyncSession) -> None:
294 d = await _db_domain(db_session, slug="code")
295 await db_session.flush()
296
297 result = await get_domain_by_id(db_session, d.domain_id)
298 assert result is not None
299 assert result.domain_id == d.domain_id
300
301 @pytest.mark.anyio
302 async def test_get_by_id_not_found(self, db_session: AsyncSession) -> None:
303 result = await get_domain_by_id(db_session, "nonexistent-id")
304 assert result is None
305
306
307 class TestIntegrationListReposForDomain:
308 @pytest.mark.anyio
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 @pytest.mark.anyio
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 @pytest.mark.anyio
330 async def test_deleted_repos_excluded(self, db_session: AsyncSession) -> None:
331 d = await _db_domain(db_session, slug="midi")
332 await _db_repo(db_session, domain_id=d.domain_id, visibility="public")
333 await _db_repo(db_session, domain_id=d.domain_id, visibility="public", deleted=True)
334 await db_session.flush()
335
336 result = await list_repos_for_domain(db_session, d.domain_id)
337 assert result.total == 1
338
339 @pytest.mark.anyio
340 async def test_nonexistent_domain_returns_empty(self, db_session: AsyncSession) -> None:
341 result = await list_repos_for_domain(db_session, "nonexistent-id")
342 assert result.total == 0
343 assert result.repos == []
344
345
346 class TestIntegrationCreateDomain:
347 @pytest.mark.anyio
348 async def test_creates_domain_with_hash(self, db_session: AsyncSession) -> None:
349 caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"}
350 result = await create_domain(
351 db_session,
352 author_user_id="alice",
353 author_slug="alice",
354 slug="midi",
355 display_name="MIDI",
356 description="MIDI domain",
357 capabilities=caps,
358 )
359 assert result.domain_id != ""
360 assert result.manifest_hash == compute_manifest_hash(caps)
361
362 @pytest.mark.anyio
363 async def test_scoped_id_format(self, db_session: AsyncSession) -> None:
364 result = await create_domain(
365 db_session,
366 author_user_id="bob",
367 author_slug="bob",
368 slug="code",
369 display_name="Code",
370 description="",
371 capabilities={},
372 )
373 assert result.scoped_id == "@bob/code"
374
375 @pytest.mark.anyio
376 async def test_not_verified_by_default(self, db_session: AsyncSession) -> None:
377 result = await create_domain(
378 db_session,
379 author_user_id="alice",
380 author_slug="alice",
381 slug="genome",
382 display_name="Genome",
383 description="",
384 capabilities={},
385 )
386 assert result.is_verified is False
387 assert result.is_deprecated is False
388
389
390 class TestIntegrationRecordDomainInstall:
391 @pytest.mark.anyio
392 async def test_increments_install_count(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 db_session.flush()
398
399 # Verify via get_domain
400 domain = await get_domain_by_id(db_session, d.domain_id)
401 assert domain is not None
402 assert domain.install_count == 1
403
404 @pytest.mark.anyio
405 async def test_idempotent_same_user(self, db_session: AsyncSession) -> None:
406 d = await _db_domain(db_session, slug="midi", install_count=0)
407 await db_session.flush()
408
409 await record_domain_install(db_session, "user1", d.domain_id)
410 await record_domain_install(db_session, "user1", d.domain_id) # duplicate
411 await db_session.flush()
412
413 domain = await get_domain_by_id(db_session, d.domain_id)
414 assert domain is not None
415 assert domain.install_count == 1 # not 2
416
417 @pytest.mark.anyio
418 async def test_different_users_each_increment(self, db_session: AsyncSession) -> None:
419 d = await _db_domain(db_session, slug="midi", install_count=0)
420 await db_session.flush()
421
422 await record_domain_install(db_session, "user1", d.domain_id)
423 await record_domain_install(db_session, "user2", d.domain_id)
424 await db_session.flush()
425
426 domain = await get_domain_by_id(db_session, d.domain_id)
427 assert domain is not None
428 assert domain.install_count == 2
429
430
431 # ===========================================================================
432 # Layer 3 — E2E
433 # ===========================================================================
434
435
436 class TestE2EListDomains:
437 @pytest.mark.anyio
438 async def test_list_returns_200(
439 self,
440 client: AsyncClient,
441 db_session: AsyncSession,
442 ) -> None:
443 await _db_domain(db_session, slug="midi-api")
444 await db_session.commit()
445
446 r = await client.get("/api/domains")
447 assert r.status_code == 200
448 body = r.json()
449 assert "domains" in body
450 assert "total" in body
451 assert isinstance(body["domains"], list)
452
453 @pytest.mark.anyio
454 async def test_list_no_auth_required(
455 self,
456 client: AsyncClient,
457 db_session: AsyncSession,
458 ) -> None:
459 await db_session.commit()
460 r = await client.get("/api/domains")
461 assert r.status_code == 200
462
463 @pytest.mark.anyio
464 async def test_list_query_param_filters(
465 self,
466 client: AsyncClient,
467 db_session: AsyncSession,
468 ) -> None:
469 await _db_domain(db_session, slug="piano-e2e", display_name="Piano Roll E2E")
470 await _db_domain(db_session, slug="genome-e2e", display_name="Genome E2E")
471 await db_session.commit()
472
473 r = await client.get("/api/domains?q=Piano+Roll")
474 assert r.status_code == 200
475 body = r.json()
476 slugs = [d["slug"] for d in body["domains"]]
477 assert "piano-e2e" in slugs
478 assert "genome-e2e" not in slugs
479
480 @pytest.mark.anyio
481 async def test_list_page_size_param(
482 self,
483 client: AsyncClient,
484 db_session: AsyncSession,
485 ) -> None:
486 for i in range(5):
487 await _db_domain(db_session, slug=f"e2e-page-{i}")
488 await db_session.commit()
489
490 r = await client.get("/api/domains?page_size=2")
491 assert r.status_code == 200
492 assert len(r.json()["domains"]) <= 2
493
494
495 class TestE2ERegisterDomain:
496 @pytest.mark.anyio
497 async def test_register_201(
498 self,
499 client: AsyncClient,
500 auth_headers: StrDict,
501 db_session: AsyncSession,
502 ) -> None:
503 await db_session.commit()
504 body = {
505 "author_slug": "testuser",
506 "slug": "my-domain",
507 "display_name": "My Domain",
508 "description": "A domain for testing",
509 "capabilities": {"dimensions": [], "merge_semantics": "three_way"},
510 "viewer_type": "generic",
511 "version": "1.0.0",
512 }
513 r = await client.post("/api/domains", json=body, headers=auth_headers)
514 assert r.status_code == 201
515 resp = r.json()
516 assert "domain_id" in resp
517 assert "scoped_id" in resp
518 assert "manifest_hash" in resp
519
520 @pytest.mark.anyio
521 async def test_register_requires_auth(
522 self,
523 client: AsyncClient,
524 db_session: AsyncSession,
525 ) -> None:
526 await db_session.commit()
527 body = {
528 "author_slug": "testuser",
529 "slug": "unauthed",
530 "display_name": "Unauthed",
531 "description": "",
532 "capabilities": {},
533 }
534 r = await client.post("/api/domains", json=body)
535 assert r.status_code == 401
536
537 @pytest.mark.anyio
538 async def test_register_duplicate_409(
539 self,
540 client: AsyncClient,
541 auth_headers: StrDict,
542 db_session: AsyncSession,
543 ) -> None:
544 await db_session.commit()
545 body = {
546 "author_slug": "testuser",
547 "slug": "dup-domain",
548 "display_name": "Dup",
549 "description": "",
550 "capabilities": {},
551 }
552 r1 = await client.post("/api/domains", json=body, headers=auth_headers)
553 assert r1.status_code == 201
554
555 r2 = await client.post("/api/domains", json=body, headers=auth_headers)
556 assert r2.status_code == 409
557
558
559 class TestE2EGetDomain:
560 @pytest.mark.anyio
561 async def test_get_domain_200(
562 self,
563 client: AsyncClient,
564 db_session: AsyncSession,
565 ) -> None:
566 await _db_domain(db_session, author_slug="alice", slug="midi-detail")
567 await db_session.commit()
568
569 r = await client.get("/api/domains/@alice/midi-detail")
570 assert r.status_code == 200
571 body = r.json()
572 assert body["scoped_id"] == "@alice/midi-detail"
573 assert "capabilities" in body
574 assert "manifest_hash" in body
575
576 @pytest.mark.anyio
577 async def test_get_domain_404(
578 self,
579 client: AsyncClient,
580 ) -> None:
581 r = await client.get("/api/domains/@nobody/nonexistent")
582 assert r.status_code == 404
583
584 @pytest.mark.anyio
585 async def test_get_domain_repos_200(
586 self,
587 client: AsyncClient,
588 db_session: AsyncSession,
589 ) -> None:
590 d = await _db_domain(db_session, author_slug="alice", slug="midi-repos")
591 await _db_repo(db_session, domain_id=d.domain_id, visibility="public")
592 await db_session.commit()
593
594 r = await client.get("/api/domains/@alice/midi-repos/repos")
595 assert r.status_code == 200
596 body = r.json()
597 assert body["total"] == 1
598 assert len(body["repos"]) == 1
599
600 @pytest.mark.anyio
601 async def test_get_domain_repos_404_unknown_domain(
602 self,
603 client: AsyncClient,
604 ) -> None:
605 r = await client.get("/api/domains/@nobody/missing/repos")
606 assert r.status_code == 404
607
608
609 # ===========================================================================
610 # Layer 4 — Stress
611 # ===========================================================================
612
613
614 class TestStress:
615 @pytest.mark.anyio
616 async def test_list_100_domains(self, db_session: AsyncSession) -> None:
617 for i in range(100):
618 await _db_domain(db_session, slug=f"domain-{i}")
619 await db_session.flush()
620
621 result = await list_domains(db_session, page=1, page_size=100)
622 assert result.total == 100
623
624 @pytest.mark.anyio
625 async def test_concurrent_list_domains(self, db_session: AsyncSession) -> None:
626 for i in range(10):
627 await _db_domain(db_session, slug=f"c{i}")
628 await db_session.flush()
629
630 results = await asyncio.gather(
631 *[list_domains(db_session) for _ in range(5)]
632 )
633 assert all(r.total == 10 for r in results)
634
635 @pytest.mark.anyio
636 async def test_list_repos_for_domain_50_repos(
637 self, db_session: AsyncSession
638 ) -> None:
639 d = await _db_domain(db_session, slug="big-domain")
640 for i in range(50):
641 await _db_repo(db_session, domain_id=d.domain_id, visibility="public")
642 await db_session.flush()
643
644 result = await list_repos_for_domain(db_session, d.domain_id, page_size=50)
645 assert result.total == 50
646
647
648 # ===========================================================================
649 # Layer 5 — Data Integrity
650 # ===========================================================================
651
652
653 class TestDataIntegrity:
654 @pytest.mark.anyio
655 async def test_sorted_by_install_count_desc(self, db_session: AsyncSession) -> None:
656 await _db_domain(db_session, slug="low", install_count=1)
657 await _db_domain(db_session, slug="high", install_count=10)
658 await _db_domain(db_session, slug="mid", install_count=5)
659 await db_session.flush()
660
661 result = await list_domains(db_session)
662 counts = [d.install_count for d in result.domains]
663 assert counts == sorted(counts, reverse=True)
664
665 @pytest.mark.anyio
666 async def test_deprecated_excluded_from_list(self, db_session: AsyncSession) -> None:
667 await _db_domain(db_session, slug="active")
668 await _db_domain(db_session, slug="old", is_deprecated=True)
669 await db_session.flush()
670
671 result = await list_domains(db_session)
672 slugs = [d.slug for d in result.domains]
673 assert "active" in slugs
674 assert "old" not in slugs
675
676 @pytest.mark.anyio
677 async def test_manifest_hash_matches_capabilities(
678 self, db_session: AsyncSession
679 ) -> None:
680 caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"}
681 result = await create_domain(
682 db_session,
683 author_user_id="alice",
684 author_slug="alice",
685 slug="verify-hash",
686 display_name="Verify",
687 description="",
688 capabilities=caps,
689 )
690 assert result.manifest_hash == compute_manifest_hash(caps)
691
692 @pytest.mark.anyio
693 async def test_capabilities_json_not_lossy(self, db_session: AsyncSession) -> None:
694 caps = {
695 "dimensions": [{"name": "tempo", "unit": "bpm"}],
696 "artifact_types": ["audio/midi"],
697 "merge_semantics": "ot",
698 }
699 created = await create_domain(
700 db_session,
701 author_user_id="alice",
702 author_slug="alice",
703 slug="caps-test",
704 display_name="Caps",
705 description="",
706 capabilities=caps,
707 )
708 await db_session.flush()
709
710 retrieved = await get_domain_by_id(db_session, created.domain_id)
711 assert retrieved is not None
712 assert retrieved.capabilities["dimensions"][0]["name"] == "tempo"
713
714 @pytest.mark.anyio
715 async def test_total_count_reflects_query_filter(
716 self, db_session: AsyncSession
717 ) -> None:
718 await _db_domain(db_session, slug="match-a", display_name="Match This")
719 await _db_domain(db_session, slug="no-match", display_name="Something Else")
720 await db_session.flush()
721
722 result = await list_domains(db_session, query="Match This")
723 assert result.total == 1
724
725
726 # ===========================================================================
727 # Layer 6 — Security
728 # ===========================================================================
729
730
731 class TestSecurity:
732 @pytest.mark.anyio
733 async def test_post_without_auth_returns_401(
734 self,
735 client: AsyncClient,
736 db_session: AsyncSession,
737 ) -> None:
738 await db_session.commit()
739 r = await client.post(
740 "/api/domains",
741 json={
742 "author_slug": "hack",
743 "slug": "hack-domain",
744 "display_name": "Hack",
745 "description": "",
746 "capabilities": {},
747 },
748 )
749 assert r.status_code == 401
750
751 @pytest.mark.anyio
752 async def test_duplicate_scoped_id_returns_409(
753 self,
754 client: AsyncClient,
755 auth_headers: StrDict,
756 db_session: AsyncSession,
757 ) -> None:
758 await db_session.commit()
759 body = {
760 "author_slug": "testuser",
761 "slug": "conflict-test",
762 "display_name": "Conflict",
763 "description": "",
764 "capabilities": {},
765 }
766 r1 = await client.post("/api/domains", json=body, headers=auth_headers)
767 assert r1.status_code == 201
768
769 r2 = await client.post("/api/domains", json=body, headers=auth_headers)
770 assert r2.status_code == 409
771 assert "already registered" in r2.json()["detail"]
772
773 @pytest.mark.anyio
774 async def test_sql_injection_in_query_param_safe(
775 self,
776 client: AsyncClient,
777 db_session: AsyncSession,
778 ) -> None:
779 await db_session.commit()
780 r = await client.get("/api/domains?q='; DROP TABLE musehub_domains; --")
781 assert r.status_code == 200 # parameterized query — safe
782
783 @pytest.mark.anyio
784 async def test_manifest_hash_tampering_detectable(self) -> None:
785 """Different capabilities always produce different hashes."""
786 original_caps = {"dimensions": [{"name": "tempo"}]}
787 tampered_caps = {"dimensions": [{"name": "tempo"}, {"name": "injected"}]}
788 assert compute_manifest_hash(original_caps) != compute_manifest_hash(tampered_caps)
789
790
791 # ===========================================================================
792 # Layer 7 — Performance
793 # ===========================================================================
794
795
796 class TestPerformance:
797 @pytest.mark.anyio
798 async def test_list_50_domains_under_200ms(self, db_session: AsyncSession) -> None:
799 for i in range(50):
800 await _db_domain(db_session, slug=f"perf-{i}")
801 await db_session.flush()
802
803 start = time.perf_counter()
804 result = await list_domains(db_session, page=1, page_size=50)
805 elapsed = time.perf_counter() - start
806
807 assert result.total == 50
808 assert elapsed < 0.2, f"list_domains took {elapsed:.3f}s"
809
810 @pytest.mark.anyio
811 async def test_compute_manifest_hash_fast(self) -> None:
812 caps = {"dimensions": [{"name": f"dim_{i}"} for i in range(100)]}
813 start = time.perf_counter()
814 for _ in range(1000):
815 compute_manifest_hash(caps)
816 elapsed = time.perf_counter() - start
817 assert elapsed < 0.5, f"1000 hash computations took {elapsed:.3f}s"
818
819 @pytest.mark.anyio
820 async def test_create_domain_under_100ms(self, db_session: AsyncSession) -> None:
821 start = time.perf_counter()
822 await create_domain(
823 db_session,
824 author_user_id="alice",
825 author_slug="alice",
826 slug="perf-create",
827 display_name="Perf",
828 description="",
829 capabilities={"dimensions": [], "merge_semantics": "ot"},
830 )
831 elapsed = time.perf_counter() - start
832 assert elapsed < 0.1, f"create_domain took {elapsed:.3f}s"
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago