"""Section 9 — Compliance & Legal Minimums. Covers: Privacy policy : exists, linked from footer, covers agent-first model. Terms of Service : exists, implicit acceptance via key registration. Minimum data : no unnecessary PII fields in MusehubIdentity. GDPR / CCPA : GET /me/export and DELETE /me endpoints exist and work. DMCA : takedown process documented. OSS license audit : license-audit.md exists and covers all direct deps. DB migration : 0001 (consolidated) includes training_opt_out + tos_accepted_at/tos_version. Auth service : tos_accepted_at set at registration time. Training opt-out : MusehubRepo.training_opt_out field exists, defaults False. """ from __future__ import annotations import json from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest _ROOT = Path(__file__).resolve().parents[1] _MUSEHUB_PKG = _ROOT / "musehub" _DOCS_LEGAL = _ROOT / "docs" / "legal" _BASE_HTML = _MUSEHUB_PKG / "templates" / "musehub" / "base.html" _AUTH_SVC = _MUSEHUB_PKG / "services" / "musehub_auth.py" _USERS_ROUTES = _MUSEHUB_PKG / "api" / "routes" / "musehub" / "users.py" _MIGRATION = _ROOT / "alembic" / "versions" / "0001_consolidated_schema.py" _CHECKLIST = _ROOT / "docs" / "pre-launch-checklist.md" # ═══════════════════════════════════════════════════════════════════════════════ # Privacy Policy # ═══════════════════════════════════════════════════════════════════════════════ class TestPrivacyPolicy: _pp = _DOCS_LEGAL / "privacy-policy.md" def test_privacy_policy_exists(self) -> None: assert self._pp.exists(), "docs/legal/privacy-policy.md must exist" def test_privacy_policy_covers_pubkey_identity(self) -> None: text = self._pp.read_text() assert "public key" in text.lower() or "pubkey" in text.lower() def test_privacy_policy_covers_agents(self) -> None: text = self._pp.read_text() assert "agent" in text.lower() def test_privacy_policy_covers_training_data(self) -> None: text = self._pp.read_text() # Must mention training data policy assert "training" in text.lower() def test_privacy_policy_covers_training_opt_out(self) -> None: text = self._pp.read_text() assert "training_opt_out" in text or "opt-out" in text.lower() or "opt out" in text.lower() def test_privacy_policy_mentions_export_endpoint(self) -> None: text = self._pp.read_text() assert "/me/export" in text or "export" in text.lower() def test_privacy_policy_mentions_delete_endpoint(self) -> None: text = self._pp.read_text() assert "DELETE" in text or "deletion" in text.lower() def test_privacy_policy_covers_private_repos_exclusion(self) -> None: text = self._pp.read_text() # Private repos must never be used for training assert "private" in text.lower() def test_privacy_policy_has_effective_date(self) -> None: text = self._pp.read_text() assert "Effective date" in text or "effective date" in text.lower() # ═══════════════════════════════════════════════════════════════════════════════ # Terms of Service # ═══════════════════════════════════════════════════════════════════════════════ class TestTermsOfService: _tos = _DOCS_LEGAL / "terms-of-service.md" def test_tos_exists(self) -> None: assert self._tos.exists(), "docs/legal/terms-of-service.md must exist" def test_tos_implicit_acceptance_via_key_registration(self) -> None: text = self._tos.read_text() # Must explain that key registration = acceptance assert "key registration" in text.lower() or "registering a key" in text.lower() def test_tos_records_tos_accepted_at(self) -> None: text = self._tos.read_text() assert "tos_accepted_at" in text def test_tos_records_tos_version(self) -> None: text = self._tos.read_text() assert "tos_version" in text def test_tos_agent_operator_responsibility(self) -> None: text = self._tos.read_text() assert "operator" in text.lower() def test_tos_training_data_policy_section(self) -> None: text = self._tos.read_text() assert "training" in text.lower() def test_tos_private_repos_never_used_for_training(self) -> None: text = self._tos.read_text() # The word "private" must appear in context with training assert "private" in text.lower() def test_tos_training_opt_out_mentioned(self) -> None: text = self._tos.read_text() assert "training_opt_out" in text def test_tos_osi_license_condition(self) -> None: text = self._tos.read_text() # Must condition training use on OSI license assert "osi" in text.lower() or "open-source license" in text.lower() or "open source license" in text.lower() def test_tos_has_effective_date(self) -> None: text = self._tos.read_text() assert "Effective date" in text or "effective date" in text.lower() # ═══════════════════════════════════════════════════════════════════════════════ # DMCA # ═══════════════════════════════════════════════════════════════════════════════ class TestDmca: _dmca = _DOCS_LEGAL / "dmca.md" def test_dmca_exists(self) -> None: assert self._dmca.exists(), "docs/legal/dmca.md must exist" def test_dmca_has_contact_email(self) -> None: text = self._dmca.read_text() assert "dmca@" in text or "@musehub" in text def test_dmca_has_response_timeline(self) -> None: text = self._dmca.read_text() # Must commit to a response time assert "business day" in text.lower() or "days" in text.lower() def test_dmca_covers_counter_notice(self) -> None: text = self._dmca.read_text() assert "counter" in text.lower() def test_dmca_mentions_repeat_infringers(self) -> None: text = self._dmca.read_text() assert "repeat" in text.lower() def test_dmca_covers_agent_operators(self) -> None: text = self._dmca.read_text() assert "agent" in text.lower() or "operator" in text.lower() # ═══════════════════════════════════════════════════════════════════════════════ # OSS License Audit # ═══════════════════════════════════════════════════════════════════════════════ class TestLicenseAudit: _audit = _DOCS_LEGAL / "license-audit.md" def test_license_audit_exists(self) -> None: assert self._audit.exists(), "docs/legal/license-audit.md must exist" def test_license_audit_covers_fastapi(self) -> None: text = self._audit.read_text() assert "fastapi" in text.lower() def test_license_audit_covers_sqlalchemy(self) -> None: text = self._audit.read_text() assert "sqlalchemy" in text.lower() def test_license_audit_covers_cryptography(self) -> None: text = self._audit.read_text() assert "cryptography" in text.lower() def test_license_audit_covers_psycopg2(self) -> None: text = self._audit.read_text() assert "psycopg2" in text.lower() def test_license_audit_has_review_schedule(self) -> None: text = self._audit.read_text() assert "review" in text.lower() def test_all_direct_deps_are_osi_or_noted(self) -> None: """Every dep row must declare 'Yes' (OSI) or have an explanation.""" text = self._audit.read_text() # We just check that 'OSI approved' header is present and 'Yes' appears assert "OSI approved" in text or "osi" in text.lower() # ═══════════════════════════════════════════════════════════════════════════════ # Footer — legal links in base.html # ═══════════════════════════════════════════════════════════════════════════════ class TestLegalFooter: def test_footer_exists_in_base_html(self) -> None: text = _BASE_HTML.read_text() assert "site-footer" in text or " None: text = _BASE_HTML.read_text() assert "privacy" in text.lower() def test_footer_has_terms_link(self) -> None: text = _BASE_HTML.read_text() assert "terms" in text.lower() or "Terms" in text def test_footer_has_dmca_link(self) -> None: text = _BASE_HTML.read_text() assert "dmca" in text.lower() or "DMCA" in text # ═══════════════════════════════════════════════════════════════════════════════ # DB Model — compliance fields # ═══════════════════════════════════════════════════════════════════════════════ class TestDbComplianceFields: def test_musehub_repo_has_training_opt_out(self) -> None: from musehub.db.musehub_repo_models import MusehubRepo assert hasattr(MusehubRepo, "training_opt_out") def test_training_opt_out_defaults_false(self) -> None: from musehub.db.musehub_repo_models import MusehubRepo col = MusehubRepo.__table__.c["training_opt_out"] # default is False assert col.default is not None or col.server_default is not None or col.nullable is False def test_musehub_identity_has_tos_accepted_at(self) -> None: from musehub.db.musehub_identity_models import MusehubIdentity assert hasattr(MusehubIdentity, "tos_accepted_at") def test_musehub_identity_has_tos_version(self) -> None: from musehub.db.musehub_identity_models import MusehubIdentity assert hasattr(MusehubIdentity, "tos_version") # ═══════════════════════════════════════════════════════════════════════════════ # Alembic migration 0001 (consolidated) # ═══════════════════════════════════════════════════════════════════════════════ class TestMigration0023: def test_migration_file_exists(self) -> None: assert _MIGRATION.exists(), "alembic/versions/0001_consolidated_schema.py must exist" def test_revision_is_0023(self) -> None: src = _MIGRATION.read_text() assert 'revision = "0001"' in src def test_down_revision_is_0022(self) -> None: src = _MIGRATION.read_text() assert 'down_revision = None' in src def test_adds_training_opt_out_to_repos(self) -> None: src = _MIGRATION.read_text() assert "training_opt_out" in src assert "musehub_repos" in src def test_adds_tos_accepted_at_to_identities(self) -> None: src = _MIGRATION.read_text() assert "tos_accepted_at" in src assert "musehub_identities" in src def test_adds_tos_version_to_identities(self) -> None: src = _MIGRATION.read_text() assert "tos_version" in src def test_has_downgrade(self) -> None: src = _MIGRATION.read_text() assert "def downgrade" in src assert "drop_table" in src # ═══════════════════════════════════════════════════════════════════════════════ # Auth service — tos_accepted_at at registration # ═══════════════════════════════════════════════════════════════════════════════ class TestAuthTosRecording: def test_auth_sets_tos_accepted_at_on_registration(self) -> None: src = _AUTH_SVC.read_text() assert "tos_accepted_at" in src def test_auth_sets_tos_version_on_registration(self) -> None: src = _AUTH_SVC.read_text() assert "tos_version" in src def test_tos_version_is_1_0(self) -> None: src = _AUTH_SVC.read_text() assert '"1.0"' in src or "'1.0'" in src def test_tos_accepted_at_set_at_identity_creation(self) -> None: src = _AUTH_SVC.read_text() # tos_accepted_at should be passed into MusehubIdentity constructor assert "MusehubIdentity(" in src # After MusehubIdentity( the tos_accepted_at should appear before the next session.add idx_construct = src.index("MusehubIdentity(") idx_add = src.index("session.add(identity)", idx_construct) segment = src[idx_construct:idx_add] assert "tos_accepted_at" in segment # ═══════════════════════════════════════════════════════════════════════════════ # GDPR endpoints — source-level checks # ═══════════════════════════════════════════════════════════════════════════════ class TestGdprEndpointsExist: def test_export_endpoint_defined(self) -> None: src = _USERS_ROUTES.read_text() assert "/me/export" in src def test_delete_endpoint_defined(self) -> None: src = _USERS_ROUTES.read_text() assert '"/me"' in src assert "delete" in src.lower() def test_export_requires_auth(self) -> None: src = _USERS_ROUTES.read_text() # export endpoint must use require_valid_token # Find the block around /me/export idx = src.index("/me/export") segment = src[max(0, idx - 200):idx + 500] assert "require_valid_token" in segment def test_delete_requires_auth(self) -> None: src = _USERS_ROUTES.read_text() # Find the delete /me block — look for HTTP_204_NO_CONTENT (unique to delete) assert "HTTP_204_NO_CONTENT" in src idx = src.index("HTTP_204_NO_CONTENT") segment = src[max(0, idx - 500):idx + 1000] assert "require_valid_token" in segment def test_export_returns_identity_data(self) -> None: src = _USERS_ROUTES.read_text() # Export must include identity, keys, repos, commits assert '"identity"' in src or "\"identity\"" in src assert '"keys"' in src or "\"keys\"" in src assert '"repos"' in src or "\"repos\"" in src assert '"commits"' in src or "\"commits\"" in src def test_export_includes_tos_acceptance(self) -> None: src = _USERS_ROUTES.read_text() # Export response must include tos_accepted_at assert "tos_accepted_at" in src def test_delete_hard_deletes_auth_keys(self) -> None: src = _USERS_ROUTES.read_text() # Deletion of auth keys must happen in the delete endpoint assert "MusehubAuthKey" in src assert "delete(" in src or "Delete(" in src def test_export_schema_version_field(self) -> None: src = _USERS_ROUTES.read_text() assert "schema_version" in src def test_delete_returns_204(self) -> None: src = _USERS_ROUTES.read_text() assert "HTTP_204_NO_CONTENT" in src def test_gdpr_import_added_to_users(self) -> None: src = _USERS_ROUTES.read_text() assert "MusehubAuthKey" in src # ═══════════════════════════════════════════════════════════════════════════════ # GDPR endpoint integration — unit tests with mocked DB # ═══════════════════════════════════════════════════════════════════════════════ class TestGdprExportUnit: """Test GET /api/me/export response structure.""" @pytest.mark.asyncio async def test_export_response_structure(self) -> None: from datetime import datetime, timezone from musehub.api.routes.musehub.users import export_my_data from musehub.auth.dependencies import TokenClaims now = datetime.now(timezone.utc) mock_identity = MagicMock() mock_identity.identity_id = "id-123" mock_identity.handle = "gabriel" mock_identity.identity_type = "human" mock_identity.display_name = "Gabriel" mock_identity.bio = "Music maker" mock_identity.email = None mock_identity.website_url = None mock_identity.location = None mock_identity.created_at = now mock_identity.tos_accepted_at = now mock_identity.tos_version = "1.0" mock_key = MagicMock() mock_key.key_id = "key-1" mock_key.algorithm = "ed25519" mock_key.fingerprint = "abc123" mock_key.label = "main" mock_key.created_at = now mock_key.last_used_at = now mock_repo = MagicMock() mock_repo.repo_id = "repo-1" mock_repo.name = "test-repo" mock_repo.slug = "test-repo" mock_repo.visibility = "public" mock_repo.description = "" mock_repo.tags = [] mock_repo.training_opt_out = False mock_repo.created_at = now mock_commit = MagicMock() mock_commit.commit_id = "c-1" mock_commit.repo_id = "repo-1" mock_commit.branch = "main" mock_commit.message = "init" mock_commit.timestamp = now # Mock DB session db = AsyncMock() def make_result(obj_or_list: MagicMock | list[MagicMock]) -> None: r = MagicMock() if isinstance(obj_or_list, list): r.scalars.return_value.all.return_value = obj_or_list else: r.scalar_one_or_none.return_value = obj_or_list return r db.execute = AsyncMock(side_effect=[ make_result(mock_identity), # identity query make_result([mock_key]), # keys query make_result([mock_repo]), # repos query make_result([mock_commit]), # commits query ]) claims = MagicMock(spec=TokenClaims) claims.identity_id = "id-123" import json as _json response = await export_my_data(claims=claims, db=db) result = _json.loads(response.body) assert result["schema_version"] == "1.0" assert result["identity"]["handle"] == "gabriel" assert result["identity"]["tos_version"] == "1.0" assert len(result["keys"]) == 1 assert result["keys"][0]["algorithm"] == "ed25519" assert len(result["repos"]) == 1 assert result["repos"][0]["training_opt_out"] is False assert len(result["commits"]) == 1 @pytest.mark.asyncio async def test_export_404_when_identity_missing(self) -> None: from musehub.api.routes.musehub.users import export_my_data from musehub.auth.dependencies import TokenClaims from fastapi import HTTPException db = AsyncMock() result = MagicMock() result.scalar_one_or_none.return_value = None db.execute = AsyncMock(return_value=result) claims = MagicMock(spec=TokenClaims) claims.identity_id = "missing-id" with pytest.raises(HTTPException) as exc_info: await export_my_data(claims=claims, db=db) assert exc_info.value.status_code == 404 class TestGdprDeleteUnit: """Test DELETE /api/me endpoint.""" @pytest.mark.asyncio async def test_delete_calls_commit(self) -> None: from datetime import datetime, timezone from musehub.api.routes.musehub.users import delete_my_account from musehub.auth.dependencies import TokenClaims now = datetime.now(timezone.utc) mock_identity = MagicMock() mock_identity.identity_id = "id-123" mock_identity.handle = "gabriel" mock_identity.deleted_at = None db = AsyncMock() identity_result = MagicMock() identity_result.scalar_one_or_none.return_value = mock_identity db.execute = AsyncMock(return_value=MagicMock()) # First call returns identity, subsequent calls (delete keys, update repos) return MagicMock call_count = 0 async def execute_side_effect(stmt: MagicMock) -> None: nonlocal call_count call_count += 1 if call_count == 1: return identity_result return MagicMock() db.execute = execute_side_effect db.commit = AsyncMock() claims = MagicMock(spec=TokenClaims) claims.identity_id = "id-123" await delete_my_account(claims=claims, db=db) # commit must have been called db.commit.assert_awaited_once() @pytest.mark.asyncio async def test_delete_sets_deleted_at_on_identity(self) -> None: from datetime import datetime, timezone from musehub.api.routes.musehub.users import delete_my_account from musehub.auth.dependencies import TokenClaims mock_identity = MagicMock() mock_identity.identity_id = "id-123" mock_identity.handle = "gabriel" mock_identity.deleted_at = None db = AsyncMock() identity_result = MagicMock() identity_result.scalar_one_or_none.return_value = mock_identity call_count = 0 async def execute_side_effect(stmt: MagicMock) -> None: nonlocal call_count call_count += 1 if call_count == 1: return identity_result return MagicMock() db.execute = execute_side_effect db.commit = AsyncMock() claims = MagicMock(spec=TokenClaims) claims.identity_id = "id-123" await delete_my_account(claims=claims, db=db) # identity.deleted_at must have been set assert mock_identity.deleted_at is not None @pytest.mark.asyncio async def test_delete_404_when_identity_missing(self) -> None: from musehub.api.routes.musehub.users import delete_my_account from musehub.auth.dependencies import TokenClaims from fastapi import HTTPException db = AsyncMock() result = MagicMock() result.scalar_one_or_none.return_value = None db.execute = AsyncMock(return_value=result) claims = MagicMock(spec=TokenClaims) claims.identity_id = "missing-id" with pytest.raises(HTTPException) as exc_info: await delete_my_account(claims=claims, db=db) assert exc_info.value.status_code == 404 # ═══════════════════════════════════════════════════════════════════════════════ # Checklist updated # ═══════════════════════════════════════════════════════════════════════════════ class TestChecklistSection9: def test_checklist_section9_exists(self) -> None: text = _CHECKLIST.read_text() assert "## 9. Compliance" in text def test_checklist_has_six_items(self) -> None: text = _CHECKLIST.read_text() # Find the section 9 block start = text.index("## 9. Compliance") # End at next ## heading try: end = text.index("\n## ", start + 1) except ValueError: end = len(text) section = text[start:end] assert section.count("- [x]") >= 6, "All 6 section 9 items should be checked"