"""Section 25 — Domains: 8-layer test suite. Covers musehub/services/musehub_domains.py and musehub/api/routes/musehub/domains.py. Layer map --------- 1. Unit — compute_manifest_hash, _to_response, dataclasses 2. Integration — service functions against real PostgreSQL DB 3. E2E — HTTP client against full app 4. Stress — many domains, concurrent queries 5. Data Integrity — sort order, deprecated exclusion, hash correctness 6. Security — auth enforcement, duplicate scoped_id 7. Performance — timing budgets 8. Admin — domain verification endpoint """ from __future__ import annotations import asyncio import json import secrets from collections.abc import Generator, Mapping import time import pytest import pytest_asyncio from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import blob_id from musehub.auth.request_signing import MSignContext, optional_signed_request, require_signed_request from musehub.db.musehub_domain_models import MusehubDomain, MusehubDomainInstall from datetime import datetime, timezone from musehub.core.genesis import compute_identity_id, compute_repo_id from musehub.db.musehub_identity_models import MusehubIdentity from musehub.db.musehub_repo_models import MusehubRepo from musehub.main import app from musehub.types.json_types import JSONObject, StrDict from musehub.services.musehub_domains import ( DomainListResponse, DomainReposResponse, DomainResponse, _to_response, compute_manifest_hash, count_public_repos_by_domain, create_domain, get_domain_by_id, get_domain_by_scoped_id, list_domains, list_repos_for_domain, record_domain_install, record_domain_uninstall, resolve_unambiguous_domain_id_by_category, ) # --------------------------------------------------------------------------- # DB helpers # --------------------------------------------------------------------------- def _uid() -> str: return secrets.token_hex(16) async def _db_domain( session: AsyncSession, *, author_slug: str = "alice", slug: str | None = None, display_name: str = "Test Domain", description: str = "A test domain", capabilities: JSONObject | None = None, viewer_type: str = "generic", version: str = "1.0.0", install_count: int = 0, is_verified: bool = False, is_deprecated: bool = False, ) -> MusehubDomain: from datetime import datetime, timezone slug = slug or f"domain-{_uid()[:8]}" caps = capabilities or {"dimensions": [], "merge_semantics": "three_way"} manifest_hash = compute_manifest_hash(caps) domain = MusehubDomain( domain_id=_uid(), author_user_id=author_slug, author_slug=author_slug, slug=slug, display_name=display_name, description=description, version=version, manifest_hash=manifest_hash, capabilities=caps, viewer_type=viewer_type, install_count=install_count, is_verified=is_verified, is_deprecated=is_deprecated, created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) session.add(domain) await session.flush() return domain async def _db_repo( session: AsyncSession, owner: str = "alice", *, domain_id: str | None = None, marketplace_domain_id: str | None = None, visibility: str = "public", deleted: bool = False, ) -> MusehubRepo: slug = f"repo-{_uid()[:8]}" owner_id = compute_identity_id(owner.encode()) created_at = datetime.now(tz=timezone.utc) repo = MusehubRepo( repo_id=compute_repo_id(owner_id, slug, "code", created_at.isoformat()), name=slug, slug=slug, owner=owner, owner_user_id=owner_id, visibility=visibility, domain_id=domain_id, marketplace_domain_id=marketplace_domain_id, created_at=created_at, updated_at=created_at, ) session.add(repo) await session.flush() if deleted: await session.delete(repo) await session.flush() return repo # =========================================================================== # Layer 1 — Unit # =========================================================================== class TestUnitComputeManifestHash: def test_returns_hex_string(self) -> None: h = compute_manifest_hash({"dimensions": []}) assert isinstance(h, str) assert h.startswith("sha256:") assert len(h) == 71 # sha256:<64-hex> def test_deterministic(self) -> None: caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"} assert compute_manifest_hash(caps) == compute_manifest_hash(caps) def test_sorted_keys_order_independent(self) -> None: caps_a = {"b": 1, "a": 2} caps_b = {"a": 2, "b": 1} assert compute_manifest_hash(caps_a) == compute_manifest_hash(caps_b) def test_different_capabilities_different_hash(self) -> None: h1 = compute_manifest_hash({"dimensions": []}) h2 = compute_manifest_hash({"dimensions": [{"name": "tempo"}]}) assert h1 != h2 def test_matches_manual_sha256(self) -> None: caps = {"x": 1} blob = json.dumps(caps, sort_keys=True, separators=(",", ":")).encode() expected = blob_id(blob) assert compute_manifest_hash(caps) == expected class TestUnitToResponse: """Unit tests for _to_response using an integration DB fixture to create real ORM rows.""" async def test_scoped_id_format(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, author_slug="gabriel", slug="midi") resp = _to_response(d) assert resp.scoped_id == "@gabriel/midi" async def test_install_count_preserved(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi-count", install_count=5) resp = _to_response(d) assert resp.install_count == 5 async def test_is_verified_preserved(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi-verified", is_verified=True) resp = _to_response(d) assert resp.is_verified is True async def test_capabilities_copied(self, db_session: AsyncSession) -> None: caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"} d = await _db_domain(db_session, slug="midi-caps", capabilities=caps) resp = _to_response(d) assert resp.capabilities == caps async def test_none_capabilities_become_empty_dict( self, db_session: AsyncSession ) -> None: d = await _db_domain(db_session, slug="midi-no-caps") d.capabilities = None resp = _to_response(d) assert resp.capabilities == {} class TestUnitDataclasses: def test_domain_response_fields(self) -> None: from datetime import datetime, timezone dr = DomainResponse( domain_id=_uid(), author_slug="alice", slug="midi", scoped_id="@alice/midi", display_name="MIDI", description="", version="1.0.0", manifest_hash="abc", capabilities={}, viewer_type="generic", install_count=0, is_verified=False, is_deprecated=False, created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc), ) assert dr.scoped_id == "@alice/midi" def test_domain_list_response_fields(self) -> None: dlr = DomainListResponse(domains=[], total=0) assert dlr.total == 0 def test_domain_repos_response_fields(self) -> None: drr = DomainReposResponse( domain_id=_uid(), scoped_id="@a/b", repos=[], total=0 ) assert drr.repos == [] # =========================================================================== # Layer 2 — Integration # =========================================================================== class TestIntegrationListDomains: async def test_returns_all_non_deprecated(self, db_session: AsyncSession) -> None: await _db_domain(db_session, slug="midi", display_name="MIDI") await _db_domain(db_session, slug="code", display_name="Code") await _db_domain(db_session, slug="old", is_deprecated=True) await db_session.flush() result = await list_domains(db_session) slugs = [d.slug for d in result.domains] assert "midi" in slugs assert "code" in slugs assert "old" not in slugs async def test_query_filters_by_display_name(self, db_session: AsyncSession) -> None: await _db_domain(db_session, slug="piano", display_name="Piano Roll Domain") await _db_domain(db_session, slug="genome", display_name="Genomics Domain") await db_session.flush() result = await list_domains(db_session, query="Piano") assert len(result.domains) == 1 assert result.domains[0].slug == "piano" async def test_verified_only_filter(self, db_session: AsyncSession) -> None: await _db_domain(db_session, slug="v", is_verified=True) await _db_domain(db_session, slug="u", is_verified=False) await db_session.flush() result = await list_domains(db_session, verified_only=True) slugs = [d.slug for d in result.domains] assert "v" in slugs assert "u" not in slugs async def test_pagination_page_size(self, db_session: AsyncSession) -> None: for i in range(5): await _db_domain(db_session, slug=f"d{i}") await db_session.flush() result = await list_domains(db_session, limit=2) assert len(result.domains) == 2 assert result.total == 5 assert result.next_cursor is not None class TestIntegrationGetDomain: async def test_get_by_scoped_id_found(self, db_session: AsyncSession) -> None: await _db_domain(db_session, author_slug="alice", slug="midi") await db_session.flush() result = await get_domain_by_scoped_id(db_session, "alice", "midi") assert result is not None assert result.scoped_id == "@alice/midi" async def test_get_by_scoped_id_not_found(self, db_session: AsyncSession) -> None: result = await get_domain_by_scoped_id(db_session, "nobody", "missing") assert result is None async def test_get_by_id_found(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="code") await db_session.flush() result = await get_domain_by_id(db_session, d.domain_id) assert result is not None assert result.domain_id == d.domain_id async def test_get_by_id_not_found(self, db_session: AsyncSession) -> None: result = await get_domain_by_id(db_session, "nonexistent-id") assert result is None class TestIntegrationListReposForDomain: async def test_returns_public_repos(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="public") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="public") await db_session.flush() result = await list_repos_for_domain(db_session, d.domain_id) assert result.total == 2 assert len(result.repos) == 2 async def test_private_repos_excluded(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="public") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="private") await db_session.flush() result = await list_repos_for_domain(db_session, d.domain_id) assert result.total == 1 async def test_deleted_repos_excluded(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="public") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="public", deleted=True) await db_session.flush() result = await list_repos_for_domain(db_session, d.domain_id) assert result.total == 1 async def test_nonexistent_domain_returns_empty(self, db_session: AsyncSession) -> None: result = await list_repos_for_domain(db_session, "nonexistent-id") assert result.total == 0 assert result.repos == [] async def test_repo_with_matching_plugin_category_but_no_marketplace_link_excluded( self, db_session: AsyncSession ) -> None: """Regression test for the column-mismatch bug this function once had. list_repos_for_domain used to compare MusehubRepo.domain_id (a plain VCS-plugin category string like "code") against the marketplace domain's genesis-addressed domain_id (a sha256 hash) — two different namespaces that could never match, so every domain always showed 0 repos. A repo whose *plugin category string* happens to equal the domain's slug must NOT be returned unless it's also explicitly linked via marketplace_domain_id. """ d = await _db_domain(db_session, slug="code") await _db_repo(db_session, domain_id="code", marketplace_domain_id=None, visibility="public") await db_session.flush() result = await list_repos_for_domain(db_session, d.domain_id) assert result.total == 0 assert result.repos == [] class TestIntegrationCountPublicReposByDomain: """musehub#120 follow-up — the /domains listing page showed domain.install_count mislabeled as a repo count. Fixes require a batch per-domain repo count (this function) rather than an N+1 call to list_repos_for_domain per row. """ async def test_counts_public_repos_per_domain(self, db_session: AsyncSession) -> None: code = await _db_domain(db_session, slug="code") mist = await _db_domain(db_session, slug="mist") await _db_repo(db_session, marketplace_domain_id=code.domain_id, visibility="public") await _db_repo(db_session, marketplace_domain_id=code.domain_id, visibility="public") await _db_repo(db_session, marketplace_domain_id=mist.domain_id, visibility="public") await db_session.flush() counts = await count_public_repos_by_domain( db_session, [code.domain_id, mist.domain_id] ) assert counts[code.domain_id] == 2 assert counts[mist.domain_id] == 1 async def test_private_repos_excluded(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="identity") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="private") await db_session.flush() counts = await count_public_repos_by_domain(db_session, [d.domain_id]) assert counts.get(d.domain_id, 0) == 0 async def test_domain_with_zero_repos_absent_from_result( self, db_session: AsyncSession ) -> None: d = await _db_domain(db_session, slug="lonely") await db_session.flush() counts = await count_public_repos_by_domain(db_session, [d.domain_id]) assert d.domain_id not in counts assert counts.get(d.domain_id, 0) == 0 async def test_empty_domain_ids_returns_empty_dict( self, db_session: AsyncSession ) -> None: assert await count_public_repos_by_domain(db_session, []) == {} async def test_does_not_count_other_domains_repos( self, db_session: AsyncSession ) -> None: target = await _db_domain(db_session, slug="target") other = await _db_domain(db_session, slug="other") await _db_repo(db_session, marketplace_domain_id=other.domain_id, visibility="public") await db_session.flush() counts = await count_public_repos_by_domain(db_session, [target.domain_id]) assert counts.get(target.domain_id, 0) == 0 class TestIntegrationResolveUnambiguousDomainByCategory: """musehub#120 Phase 1 (MDL_01-04) — the never-guess resolution helper. Consulted later by both create_repo (Phase 2) and the backfill script (Phase 3) to auto-link a repo's plain category string (e.g. "code") to a specific marketplace MusehubDomain, without ever guessing when the category is ambiguous or unmatched. """ async def test_mdl01_resolves_the_sole_matching_live_domain( self, db_session: AsyncSession ) -> None: d = await _db_domain(db_session, author_slug="gabriel", slug="code") await db_session.flush() result = await resolve_unambiguous_domain_id_by_category(db_session, "code") assert result == d.domain_id async def test_mdl02_returns_none_when_zero_domains_match( self, db_session: AsyncSession ) -> None: result = await resolve_unambiguous_domain_id_by_category(db_session, "no-such-category") assert result is None async def test_mdl03_returns_none_when_multiple_live_domains_share_the_category( self, db_session: AsyncSession ) -> None: await _db_domain(db_session, author_slug="gabriel", slug="code") await _db_domain(db_session, author_slug="alice", slug="code") await db_session.flush() result = await resolve_unambiguous_domain_id_by_category(db_session, "code") assert result is None async def test_mdl04_deprecated_sibling_does_not_create_ambiguity( self, db_session: AsyncSession ) -> None: live = await _db_domain(db_session, author_slug="gabriel", slug="code") await _db_domain(db_session, author_slug="alice", slug="code", is_deprecated=True) await db_session.flush() result = await resolve_unambiguous_domain_id_by_category(db_session, "code") assert result == live.domain_id class TestIntegrationCreateDomain: async def test_creates_domain_with_hash(self, db_session: AsyncSession) -> None: caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"} result = await create_domain( db_session, author_user_id="alice", author_slug="alice", slug="midi", display_name="MIDI", description="MIDI domain", capabilities=caps, ) assert result.domain_id != "" assert result.manifest_hash == compute_manifest_hash(caps) async def test_scoped_id_format(self, db_session: AsyncSession) -> None: result = await create_domain( db_session, author_user_id="bob", author_slug="bob", slug="code", display_name="Code", description="", capabilities={}, ) assert result.scoped_id == "@bob/code" async def test_not_verified_by_default(self, db_session: AsyncSession) -> None: result = await create_domain( db_session, author_user_id="alice", author_slug="alice", slug="genome", display_name="Genome", description="", capabilities={}, ) assert result.is_verified is False assert result.is_deprecated is False class TestIntegrationRecordDomainInstall: async def test_increments_install_count(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi", install_count=0) await db_session.flush() await record_domain_install(db_session, "user1", d.domain_id) await db_session.flush() # Verify via get_domain domain = await get_domain_by_id(db_session, d.domain_id) assert domain is not None assert domain.install_count == 1 async def test_idempotent_same_user(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi", install_count=0) await db_session.flush() await record_domain_install(db_session, "user1", d.domain_id) await record_domain_install(db_session, "user1", d.domain_id) # duplicate await db_session.flush() domain = await get_domain_by_id(db_session, d.domain_id) assert domain is not None assert domain.install_count == 1 # not 2 async def test_different_users_each_increment(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi", install_count=0) await db_session.flush() await record_domain_install(db_session, "user1", d.domain_id) await record_domain_install(db_session, "user2", d.domain_id) await db_session.flush() domain = await get_domain_by_id(db_session, d.domain_id) assert domain is not None assert domain.install_count == 2 class TestIntegrationRecordDomainUninstall: """musehub#117 DOM_13: record_domain_uninstall reverses install_count.""" async def test_decrements_install_count(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi-uninstall", install_count=0) await db_session.flush() await record_domain_install(db_session, "user1", d.domain_id) await record_domain_uninstall(db_session, "user1", d.domain_id) await db_session.flush() domain = await get_domain_by_id(db_session, d.domain_id) assert domain is not None assert domain.install_count == 0 async def test_idempotent_when_not_installed(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi-never-installed", install_count=0) await db_session.flush() await record_domain_uninstall(db_session, "user1", d.domain_id) await db_session.flush() domain = await get_domain_by_id(db_session, d.domain_id) assert domain is not None assert domain.install_count == 0 async def test_never_goes_negative(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi-floor", install_count=0) await db_session.flush() # Uninstall twice in a row without a matching install between them await record_domain_install(db_session, "user1", d.domain_id) await record_domain_uninstall(db_session, "user1", d.domain_id) await record_domain_uninstall(db_session, "user1", d.domain_id) await db_session.flush() domain = await get_domain_by_id(db_session, d.domain_id) assert domain is not None assert domain.install_count == 0 async def test_only_affects_the_uninstalling_user(self, db_session: AsyncSession) -> None: d = await _db_domain(db_session, slug="midi-per-user", install_count=0) await db_session.flush() await record_domain_install(db_session, "user1", d.domain_id) await record_domain_install(db_session, "user2", d.domain_id) await record_domain_uninstall(db_session, "user1", d.domain_id) await db_session.flush() domain = await get_domain_by_id(db_session, d.domain_id) assert domain is not None assert domain.install_count == 1 # =========================================================================== # Layer 3 — E2E # =========================================================================== class TestE2EListDomains: async def test_list_returns_200( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await _db_domain(db_session, slug="midi-api") await db_session.commit() r = await client.get("/api/domains") assert r.status_code == 200 body = r.json() assert "domains" in body assert "total" in body assert isinstance(body["domains"], list) async def test_list_no_auth_required( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await db_session.commit() r = await client.get("/api/domains") assert r.status_code == 200 async def test_list_query_param_filters( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await _db_domain(db_session, slug="piano-e2e", display_name="Piano Roll E2E") await _db_domain(db_session, slug="genome-e2e", display_name="Genome E2E") await db_session.commit() r = await client.get("/api/domains?q=Piano+Roll") assert r.status_code == 200 body = r.json() slugs = [d["slug"] for d in body["domains"]] assert "piano-e2e" in slugs assert "genome-e2e" not in slugs async def test_list_page_size_param( self, client: AsyncClient, db_session: AsyncSession, ) -> None: for i in range(5): await _db_domain(db_session, slug=f"e2e-page-{i}") await db_session.commit() r = await client.get("/api/domains?limit=2") assert r.status_code == 200 body = r.json() assert len(body["domains"]) <= 2 assert body["nextCursor"] is not None class TestE2ERegisterDomain: async def test_register_201( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await db_session.commit() body = { "author_slug": "testuser", "slug": "my-domain", "display_name": "My Domain", "description": "A domain for testing", "capabilities": {"dimensions": [], "merge_semantics": "three_way"}, "viewer_type": "generic", "version": "1.0.0", } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 201 resp = r.json() assert "domain_id" in resp assert "scoped_id" in resp assert "manifest_hash" in resp async def test_register_requires_auth( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await db_session.commit() body = { "author_slug": "testuser", "slug": "unauthed", "display_name": "Unauthed", "description": "", "capabilities": {}, } r = await client.post("/api/domains", json=body) assert r.status_code == 401 async def test_register_duplicate_409( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await db_session.commit() body = { "author_slug": "testuser", "slug": "dup-domain", "display_name": "Dup", "description": "", "capabilities": {}, } r1 = await client.post("/api/domains", json=body, headers=auth_headers) assert r1.status_code == 201 r2 = await client.post("/api/domains", json=body, headers=auth_headers) assert r2.status_code == 409 class TestE2EGetDomain: async def test_get_domain_200( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="alice", slug="midi-detail") await db_session.commit() r = await client.get("/api/domains/@alice/midi-detail") assert r.status_code == 200 body = r.json() assert body["scoped_id"] == "@alice/midi-detail" assert "capabilities" in body assert "manifest_hash" in body async def test_get_domain_404( self, client: AsyncClient, ) -> None: r = await client.get("/api/domains/@nobody/nonexistent") assert r.status_code == 404 async def test_get_domain_repos_200( self, client: AsyncClient, db_session: AsyncSession, ) -> None: d = await _db_domain(db_session, author_slug="alice", slug="midi-repos") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="public") await db_session.commit() r = await client.get("/api/domains/@alice/midi-repos/repos") assert r.status_code == 200 body = r.json() assert body["total"] == 1 assert len(body["repos"]) == 1 async def test_get_domain_repos_404_unknown_domain( self, client: AsyncClient, ) -> None: r = await client.get("/api/domains/@nobody/missing/repos") assert r.status_code == 404 async def test_domain_detail_page_json_carries_viewer_type_and_dimensions( self, client: AsyncClient, db_session: AsyncSession, ) -> None: """musehub#117 DOM_10: the UI page must thread viewer_type and dimensions into page_json so the client-side viewer (domain-viewers.ts) can actually render them — not just the API JSON response.""" await _db_domain( db_session, author_slug="alice", slug="piano-roll-detail", viewer_type="piano_roll", capabilities={ "dimensions": [{"name": "drums", "description": "Percussion track"}], }, ) await db_session.commit() r = await client.get("/domains/@alice/piano-roll-detail") assert r.status_code == 200 assert '"viewerType": "piano_roll"' in r.text assert "drums" in r.text assert 'id="dd-viewer-preview"' in r.text async def test_domains_listing_page_reports_real_repo_count_not_install_count( self, client: AsyncClient, db_session: AsyncSession, ) -> None: """Regression test: the /domains listing page once showed domain.install_count mislabeled as a repo count — two repos sharing one domain (one install, two repos) rendered "1 repo" instead of "2 repos". Uses ?format=json to assert on the exact repo_count field the template consumes.""" d = await _db_domain(db_session, author_slug="alice", slug="listing-count") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="public") await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="public") await db_session.commit() r = await client.get("/domains?format=json") assert r.status_code == 200 body = r.json() entry = next(dm for dm in body["domains"] if dm["scoped_id"] == "@alice/listing-count") assert entry["repo_count"] == 2 assert entry["install_count"] == 0 # distinct metric — nobody installed it # =========================================================================== # Layer 4 — Stress # =========================================================================== class TestStress: async def test_list_100_domains(self, db_session: AsyncSession) -> None: for i in range(100): await _db_domain(db_session, slug=f"domain-{i}") await db_session.flush() result = await list_domains(db_session, limit=100) assert result.total == 100 async def test_concurrent_list_domains(self, db_session: AsyncSession) -> None: for i in range(10): await _db_domain(db_session, slug=f"c{i}") await db_session.flush() results = await asyncio.gather( *[list_domains(db_session) for _ in range(5)] ) assert all(r.total == 10 for r in results) async def test_list_repos_for_domain_50_repos( self, db_session: AsyncSession ) -> None: d = await _db_domain(db_session, slug="big-domain") for i in range(50): await _db_repo(db_session, marketplace_domain_id=d.domain_id, visibility="public") await db_session.flush() result = await list_repos_for_domain(db_session, d.domain_id, limit=50) assert result.total == 50 # =========================================================================== # Layer 5 — Data Integrity # =========================================================================== class TestDataIntegrity: async def test_sorted_by_install_count_desc(self, db_session: AsyncSession) -> None: await _db_domain(db_session, slug="low", install_count=1) await _db_domain(db_session, slug="high", install_count=10) await _db_domain(db_session, slug="mid", install_count=5) await db_session.flush() result = await list_domains(db_session) counts = [d.install_count for d in result.domains] assert counts == sorted(counts, reverse=True) async def test_deprecated_excluded_from_list(self, db_session: AsyncSession) -> None: await _db_domain(db_session, slug="active") await _db_domain(db_session, slug="old", is_deprecated=True) await db_session.flush() result = await list_domains(db_session) slugs = [d.slug for d in result.domains] assert "active" in slugs assert "old" not in slugs async def test_manifest_hash_matches_capabilities( self, db_session: AsyncSession ) -> None: caps = {"dimensions": [{"name": "tempo"}], "merge_semantics": "ot"} result = await create_domain( db_session, author_user_id="alice", author_slug="alice", slug="verify-hash", display_name="Verify", description="", capabilities=caps, ) assert result.manifest_hash == compute_manifest_hash(caps) async def test_capabilities_json_not_lossy(self, db_session: AsyncSession) -> None: caps = { "dimensions": [{"name": "tempo", "unit": "bpm"}], "artifact_types": ["audio/midi"], "merge_semantics": "ot", } created = await create_domain( db_session, author_user_id="alice", author_slug="alice", slug="caps-test", display_name="Caps", description="", capabilities=caps, ) await db_session.flush() retrieved = await get_domain_by_id(db_session, created.domain_id) assert retrieved is not None assert retrieved.capabilities["dimensions"][0]["name"] == "tempo" async def test_total_count_reflects_query_filter( self, db_session: AsyncSession ) -> None: await _db_domain(db_session, slug="match-a", display_name="Match This") await _db_domain(db_session, slug="no-match", display_name="Something Else") await db_session.flush() result = await list_domains(db_session, query="Match This") assert result.total == 1 # =========================================================================== # Layer 6 — Security # =========================================================================== class TestSecurity: async def test_post_without_auth_returns_401( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await db_session.commit() r = await client.post( "/api/domains", json={ "author_slug": "hack", "slug": "hack-domain", "display_name": "Hack", "description": "", "capabilities": {}, }, ) assert r.status_code == 401 async def test_duplicate_scoped_id_returns_409( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await db_session.commit() body = { "author_slug": "testuser", "slug": "conflict-test", "display_name": "Conflict", "description": "", "capabilities": {}, } r1 = await client.post("/api/domains", json=body, headers=auth_headers) assert r1.status_code == 201 r2 = await client.post("/api/domains", json=body, headers=auth_headers) assert r2.status_code == 409 assert "already registered" in r2.json()["detail"] async def test_sql_injection_in_query_param_safe( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await db_session.commit() r = await client.get("/api/domains?q='; DROP TABLE musehub_domains; --") assert r.status_code == 200 # parameterized query — safe async def test_manifest_hash_tampering_detectable(self) -> None: """Different capabilities always produce different hashes.""" original_caps = {"dimensions": [{"name": "tempo"}]} tampered_caps = {"dimensions": [{"name": "tempo"}, {"name": "injected"}]} assert compute_manifest_hash(original_caps) != compute_manifest_hash(tampered_caps) async def test_author_slug_must_match_caller_handle( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """musehub#117 DOM_02: registering under someone else's author_slug is impersonation. auth_headers authenticates as 'testuser' (see conftest._TEST_HANDLE). Before this fix, author_slug was stored verbatim from the request body with no check against the caller's real authenticated handle — any authenticated user could register '@aaronrene/anything' or '@gabriel/anything' while being neither. """ await db_session.commit() body = { "author_slug": "someone-else", "slug": "impersonation-attempt", "display_name": "Not Mine", "description": "", "capabilities": {}, } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 403 assert "author_slug" in r.json()["detail"].lower() or "handle" in r.json()["detail"].lower() async def test_author_slug_matching_own_handle_still_succeeds( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Sanity check: the identity check does not break the legitimate case.""" await db_session.commit() body = { "author_slug": "testuser", "slug": "own-domain-after-fix", "display_name": "Mine", "description": "", "capabilities": {}, } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 201 # =========================================================================== # Layer 7 — Performance # =========================================================================== class TestPerformance: async def test_list_50_domains_under_200ms(self, db_session: AsyncSession) -> None: for i in range(50): await _db_domain(db_session, slug=f"perf-{i}") await db_session.flush() start = time.perf_counter() result = await list_domains(db_session, limit=50) elapsed = time.perf_counter() - start assert result.total == 50 assert elapsed < 0.2, f"list_domains took {elapsed:.3f}s" async def test_compute_manifest_hash_fast(self) -> None: caps = {"dimensions": [{"name": f"dim_{i}"} for i in range(100)]} start = time.perf_counter() for _ in range(1000): compute_manifest_hash(caps) elapsed = time.perf_counter() - start assert elapsed < 0.5, f"1000 hash computations took {elapsed:.3f}s" async def test_create_domain_under_100ms(self, db_session: AsyncSession) -> None: start = time.perf_counter() await create_domain( db_session, author_user_id="alice", author_slug="alice", slug="perf-create", display_name="Perf", description="", capabilities={"dimensions": [], "merge_semantics": "ot"}, ) elapsed = time.perf_counter() - start assert elapsed < 0.1, f"create_domain took {elapsed:.3f}s" # =========================================================================== # Layer 8 — Admin: domain verification # =========================================================================== _ADMIN_IDENTITY_ID = compute_identity_id(b"adminuser") _ADMIN_HANDLE = "adminuser" _ADMIN_CONTEXT = MSignContext( handle=_ADMIN_HANDLE, identity_id=_ADMIN_IDENTITY_ID, is_agent=False, is_admin=True, ) @pytest_asyncio.fixture async def admin_user(db_session: AsyncSession) -> MusehubIdentity: identity = MusehubIdentity( identity_id=_ADMIN_IDENTITY_ID, handle=_ADMIN_HANDLE, display_name="Admin User", identity_type="human", is_admin=True, ) db_session.add(identity) await db_session.commit() await db_session.refresh(identity) await db_session.commit() return identity @pytest.fixture def admin_headers(admin_user: MusehubIdentity) -> Generator[dict[str, str], None, None]: app.dependency_overrides[require_signed_request] = lambda: _ADMIN_CONTEXT app.dependency_overrides[optional_signed_request] = lambda: _ADMIN_CONTEXT yield {"Content-Type": "application/json"} app.dependency_overrides.pop(require_signed_request, None) app.dependency_overrides.pop(optional_signed_request, None) class TestAdminVerifyDomain: async def test_verify_sets_flag( self, client: AsyncClient, admin_headers: Mapping[str, str], db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="alice", slug="verifiable", is_verified=False) await db_session.commit() r = await client.post("/api/domains/@alice/verifiable/verify", headers=admin_headers) assert r.status_code == 200 assert r.json()["is_verified"] is True async def test_verify_idempotent( self, client: AsyncClient, admin_headers: Mapping[str, str], db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="alice", slug="already-verified", is_verified=True) await db_session.commit() r = await client.post("/api/domains/@alice/already-verified/verify", headers=admin_headers) assert r.status_code == 200 assert r.json()["is_verified"] is True async def test_unverify_clears_flag( self, client: AsyncClient, admin_headers: Mapping[str, str], db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="alice", slug="to-unverify", is_verified=True) await db_session.commit() r = await client.delete("/api/domains/@alice/to-unverify/verify", headers=admin_headers) assert r.status_code == 200 assert r.json()["is_verified"] is False async def test_verify_domain_not_found( self, client: AsyncClient, admin_headers: Mapping[str, str], db_session: AsyncSession, ) -> None: await db_session.commit() r = await client.post("/api/domains/@nobody/ghost/verify", headers=admin_headers) assert r.status_code == 404 async def test_verify_requires_admin( self, client: AsyncClient, auth_headers: Mapping[str, str], db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="alice", slug="non-admin-target") await db_session.commit() r = await client.post("/api/domains/@alice/non-admin-target/verify", headers=auth_headers) assert r.status_code == 403 async def test_verify_requires_auth( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="alice", slug="unauthed-verify") await db_session.commit() r = await client.post("/api/domains/@alice/unauthed-verify/verify") assert r.status_code == 401 # =========================================================================== # musehub#117 DOM_04 — one canonical URL, checked against the source of # truth (the actually-mounted FastAPI route), not a second hardcoded string. # Three places have historically drifted on this single feature: the muse # CLI targeted /api/v1/domains, the public docs claimed /api/musehub/domains, # and the real route was /api/domains the whole time. This guards the # musehub side (route + docs); the CLI side is guarded by # test_publish_targets_api_domains_not_v1 in the muse repo's # tests/test_domains_publish.py. # =========================================================================== class TestDomainRegistryURLConsistency: def test_register_domain_route_path(self) -> None: """The register-domain route must be mounted at /api/domains.""" matches = [r for r in app.routes if getattr(r, "name", "") == "register_domain"] assert len(matches) == 1, "expected exactly one register_domain route" assert matches[0].path == "/api/domains" def test_docs_page_references_the_real_route_path(self) -> None: """docs_muse_domains.html must reference the actual mounted path. Previously claimed GET /api/musehub/domains — a URL that never existed on the server. """ import pathlib matches = [r for r in app.routes if getattr(r, "name", "") == "register_domain"] real_path = matches[0].path docs_path = ( pathlib.Path(__file__).parent.parent / "musehub" / "templates" / "musehub" / "pages" / "docs_muse_domains.html" ) content = docs_path.read_text(encoding="utf-8") assert f"/api/domains" in content and real_path in content assert "/api/musehub/domains" not in content, ( "stale URL reintroduced — docs must reference the real mounted route" ) assert "/api/v1/domains" not in content, ( "stale URL reintroduced — docs must reference the real mounted route" ) # =========================================================================== # musehub#117 Phase 1 — schema hardening on the publish path (DOM_05..DOM_08) # =========================================================================== class TestDomainSchemaHardening: """DOM_05/DOM_06/DOM_07: viewer_type and capabilities are now real, schema-validated fields — malformed input gets a specific 422, not a silent accept or a generic 500 further down the pipeline.""" async def test_invalid_viewer_type_rejected_422( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """DOM_05: viewer_type is a fixed enum, not free text.""" await db_session.commit() body = { "author_slug": "testuser", "slug": "bad-viewer-type", "display_name": "Bad Viewer", "description": "", "capabilities": {}, "viewer_type": "not_a_real_viewer", } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 422 async def test_sequence_viewer_rejected_422( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """DOM_05: 'sequence_viewer' was documented but never implemented in the frontend — dropped from the enum rather than kept as a dead value (see musehub#117 Phase 2).""" await db_session.commit() body = { "author_slug": "testuser", "slug": "sequence-viewer-attempt", "display_name": "Sequence", "description": "", "capabilities": {}, "viewer_type": "sequence_viewer", } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 422 async def test_valid_viewer_types_accepted( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Sanity check: the real palette values still work.""" await db_session.commit() for viewer_type in ("symbol_graph", "piano_roll", "generic"): body = { "author_slug": "testuser", "slug": f"viewer-ok-{viewer_type}", "display_name": "OK", "description": "", "capabilities": {}, "viewer_type": viewer_type, } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 201, (viewer_type, r.text) async def test_invalid_merge_semantics_rejected_422( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """DOM_06: merge_semantics is a fixed enum.""" await db_session.commit() body = { "author_slug": "testuser", "slug": "bad-merge-semantics", "display_name": "Bad", "description": "", "capabilities": {"merge_semantics": "yolo"}, } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 422 async def test_malformed_dimensions_rejected_422( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """DOM_06: a dimension must be {name, description}, not an arbitrary shape.""" await db_session.commit() body = { "author_slug": "testuser", "slug": "bad-dimensions", "display_name": "Bad", "description": "", "capabilities": {"dimensions": ["just-a-string-not-an-object"]}, } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 422 async def test_empty_capabilities_still_valid( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Sanity check: an empty manifest (no capabilities declared yet) must remain valid — all fields default to safe empty values.""" await db_session.commit() body = { "author_slug": "testuser", "slug": "minimal-manifest", "display_name": "Minimal", "description": "", "capabilities": {}, } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 201 async def test_too_many_dimensions_rejected_422( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """DOM_07: dimension count is capped.""" await db_session.commit() body = { "author_slug": "testuser", "slug": "too-many-dimensions", "display_name": "Too Many", "description": "", "capabilities": { "dimensions": [{"name": f"dim{i}"} for i in range(51)], }, } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 422 async def test_oversized_capabilities_rejected_422( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """DOM_07: total serialized capabilities size is capped at 16 KB.""" await db_session.commit() body = { "author_slug": "testuser", "slug": "oversized-capabilities", "display_name": "Oversized", "description": "", "capabilities": { "dimensions": [ {"name": f"dim{i}", "description": "x" * 500} for i in range(50) ], }, } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 422 class TestDomainRegistrationRateLimit: """DOM_08: POST /api/domains is rate limited — protects the schema- validated but still-unbounded-frequency publish path from spam/DoS.""" async def test_register_domain_429_after_limit_exceeded( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: from musehub.rate_limits import DOMAIN_REGISTER_LIMIT limit_n = int(DOMAIN_REGISTER_LIMIT.split("/")[0]) await db_session.commit() for i in range(limit_n): body = { "author_slug": "testuser", "slug": f"rate-limit-domain-{i}", "display_name": "RL", "description": "", "capabilities": {}, } r = await client.post("/api/domains", json=body, headers=auth_headers) assert r.status_code == 201, r.text over = await client.post( "/api/domains", json={ "author_slug": "testuser", "slug": "rate-limit-domain-over", "display_name": "Over", "description": "", "capabilities": {}, }, headers=auth_headers, ) assert over.status_code == 429 # =========================================================================== # musehub#117 Phase 3 (DOM_12) — repo home page renders the marketplace # domain badge when a repo is linked, via repo_nav.html's dormant # repo_domain slot. # =========================================================================== class TestRepoHomeMarketplaceDomainBadge: async def test_linked_repo_shows_domain_badge( self, client: AsyncClient, db_session: AsyncSession, ) -> None: domain = await _db_domain( db_session, author_slug="alice", slug="badge-domain", viewer_type="piano_roll", ) repo = await _db_repo(db_session, owner="alice", visibility="public") repo.marketplace_domain_id = domain.domain_id await db_session.commit() resp = await client.get(f"/alice/{repo.slug}") assert resp.status_code == 200 assert "nav-domain-badge" in resp.text assert domain.scoped_id in resp.text async def test_unlinked_repo_shows_no_domain_badge( self, client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await _db_repo(db_session, owner="alice", visibility="public") await db_session.commit() resp = await client.get(f"/alice/{repo.slug}") assert resp.status_code == 200 assert "nav-domain-badge" not in resp.text # =========================================================================== # musehub#117 Phase 5 (DOM_16, DOM_17) — full CRUD parity: Update (PATCH) # and Delete (DELETE, soft via is_deprecated) for the domains entity. # auth_headers authenticates as handle "testuser" (see conftest.py); domains # built with author_slug="testuser" are "owned" by that fixture for the # owner-only guard tests. # =========================================================================== class TestIntegrationUpdateDomain: async def test_update_display_name_and_description( self, db_session: AsyncSession, ) -> None: from musehub.services.musehub_domains import update_domain d = await _db_domain(db_session, author_slug="alice", slug="to-update", display_name="Old", description="old desc") await db_session.commit() updated = await update_domain( db_session, d.domain_id, display_name="New", description="new desc", ) await db_session.commit() assert updated is not None assert updated.display_name == "New" assert updated.description == "new desc" # untouched fields stay as-is assert updated.viewer_type == "generic" async def test_update_capabilities_recomputes_manifest_hash( self, db_session: AsyncSession, ) -> None: from musehub.services.musehub_domains import compute_manifest_hash, update_domain d = await _db_domain(db_session, author_slug="alice", slug="hash-update") old_hash = d.manifest_hash await db_session.commit() new_caps: JSONObject = {"dimensions": [{"name": "bass", "description": ""}], "merge_semantics": "crdt"} updated = await update_domain(db_session, d.domain_id, capabilities=new_caps) await db_session.commit() assert updated is not None assert updated.manifest_hash != old_hash assert updated.manifest_hash == compute_manifest_hash(new_caps) async def test_update_partial_leaves_other_fields_unchanged( self, db_session: AsyncSession, ) -> None: from musehub.services.musehub_domains import update_domain d = await _db_domain(db_session, author_slug="alice", slug="partial-update", viewer_type="piano_roll", version="2.0.0") await db_session.commit() updated = await update_domain(db_session, d.domain_id, description="only this changed") await db_session.commit() assert updated is not None assert updated.description == "only this changed" assert updated.viewer_type == "piano_roll" assert updated.version == "2.0.0" async def test_update_nonexistent_domain_returns_none( self, db_session: AsyncSession, ) -> None: from musehub.services.musehub_domains import update_domain result = await update_domain(db_session, "sha256:doesnotexist", display_name="X") assert result is None class TestIntegrationSetDomainDeprecated: async def test_deprecate_sets_flag( self, db_session: AsyncSession, ) -> None: from musehub.services.musehub_domains import set_domain_deprecated d = await _db_domain(db_session, author_slug="alice", slug="to-deprecate", is_deprecated=False) await db_session.commit() updated = await set_domain_deprecated(db_session, d.domain_id, deprecated=True) await db_session.commit() assert updated is not None assert updated.is_deprecated is True async def test_undeprecate_clears_flag( self, db_session: AsyncSession, ) -> None: from musehub.services.musehub_domains import set_domain_deprecated d = await _db_domain(db_session, author_slug="alice", slug="to-undeprecate", is_deprecated=True) await db_session.commit() updated = await set_domain_deprecated(db_session, d.domain_id, deprecated=False) await db_session.commit() assert updated is not None assert updated.is_deprecated is False async def test_deprecated_domain_excluded_from_list( self, db_session: AsyncSession, ) -> None: from musehub.services.musehub_domains import list_domains, set_domain_deprecated d = await _db_domain(db_session, author_slug="alice", slug="hidden-after-deprecate") await db_session.commit() await set_domain_deprecated(db_session, d.domain_id, deprecated=True) await db_session.commit() result = await list_domains(db_session, query="hidden-after-deprecate") assert result.total == 0 async def test_deprecated_domain_still_fetchable_by_scoped_id( self, db_session: AsyncSession, ) -> None: from musehub.services.musehub_domains import get_domain_by_scoped_id, set_domain_deprecated d = await _db_domain(db_session, author_slug="alice", slug="deprecated-but-fetchable") await db_session.commit() await set_domain_deprecated(db_session, d.domain_id, deprecated=True) await db_session.commit() fetched = await get_domain_by_scoped_id(db_session, "alice", "deprecated-but-fetchable") assert fetched is not None assert fetched.is_deprecated is True class TestE2EUpdateDomain: async def test_update_200( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="testuser", slug="e2e-update", display_name="Before") await db_session.commit() r = await client.patch( "/api/domains/@testuser/e2e-update", json={"display_name": "After"}, headers=auth_headers, ) assert r.status_code == 200, r.text assert r.json()["display_name"] == "After" # Verify the change is really persisted, not just echoed back. get_r = await client.get("/api/domains/@testuser/e2e-update") assert get_r.json()["display_name"] == "After" async def test_update_requires_auth( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="testuser", slug="e2e-update-unauthed") await db_session.commit() r = await client.patch( "/api/domains/@testuser/e2e-update-unauthed", json={"display_name": "Nope"}, ) assert r.status_code == 401 async def test_update_requires_ownership( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="alice", slug="e2e-update-not-mine") await db_session.commit() r = await client.patch( "/api/domains/@alice/e2e-update-not-mine", json={"display_name": "Hijacked"}, headers=auth_headers, ) assert r.status_code == 403 async def test_update_404_unknown_domain( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await db_session.commit() r = await client.patch( "/api/domains/@testuser/does-not-exist", json={"display_name": "X"}, headers=auth_headers, ) assert r.status_code == 404 async def test_update_invalid_viewer_type_422( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="testuser", slug="e2e-update-bad-viewer") await db_session.commit() r = await client.patch( "/api/domains/@testuser/e2e-update-bad-viewer", json={"viewer_type": "not_a_real_viewer"}, headers=auth_headers, ) assert r.status_code == 422 async def test_update_capabilities_over_dimension_cap_422( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="testuser", slug="e2e-update-too-many-dims") await db_session.commit() r = await client.patch( "/api/domains/@testuser/e2e-update-too-many-dims", json={"capabilities": {"dimensions": [{"name": f"d{i}"} for i in range(51)]}}, headers=auth_headers, ) assert r.status_code == 422 class TestE2EDeleteDomain: async def test_delete_204( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="testuser", slug="e2e-delete") await db_session.commit() r = await client.delete("/api/domains/@testuser/e2e-delete", headers=auth_headers) assert r.status_code == 204 async def test_deleted_domain_gone_from_list_but_fetchable_directly( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="testuser", slug="e2e-delete-visibility") await db_session.commit() del_r = await client.delete("/api/domains/@testuser/e2e-delete-visibility", headers=auth_headers) assert del_r.status_code == 204 list_r = await client.get("/api/domains", params={"q": "e2e-delete-visibility"}) assert list_r.json()["total"] == 0 detail_r = await client.get("/api/domains/@testuser/e2e-delete-visibility") assert detail_r.status_code == 200 assert detail_r.json()["is_deprecated"] is True async def test_delete_requires_auth( self, client: AsyncClient, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="testuser", slug="e2e-delete-unauthed") await db_session.commit() r = await client.delete("/api/domains/@testuser/e2e-delete-unauthed") assert r.status_code == 401 async def test_delete_requires_ownership( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await _db_domain(db_session, author_slug="alice", slug="e2e-delete-not-mine") await db_session.commit() r = await client.delete("/api/domains/@alice/e2e-delete-not-mine", headers=auth_headers) assert r.status_code == 403 async def test_delete_404_unknown_domain( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: await db_session.commit() r = await client.delete("/api/domains/@testuser/does-not-exist", headers=auth_headers) assert r.status_code == 404 async def test_delete_idempotent_already_deprecated_404( self, client: AsyncClient, auth_headers: StrDict, db_session: AsyncSession, ) -> None: """Deleting an already-deprecated domain 404s, matching repos.py's delete_repo docstring convention ('already deleted' -> 404, not a silent no-op 204) — the caller can tell 'gone' from 'never existed' the same way either time, but a second delete on the same resource should not report fresh success.""" await _db_domain(db_session, author_slug="testuser", slug="e2e-delete-twice", is_deprecated=True) await db_session.commit() r = await client.delete("/api/domains/@testuser/e2e-delete-twice", headers=auth_headers) assert r.status_code == 404