""" P-06: Rechte-Erklaerung – Backend-Tests. Abgedeckt: 1. validate_rights_declaration – alle Pflichtfelder 2. check_rights_coverage – ok / insufficient / legacy / blocked 3. assert_rights_for_promotion – richtiges Fehlermuster 4. PATCH /api/media-assets/{id} – Promotion mit und ohne P-06 5. POST /api/media-assets/{id}/rights-declarations – Re-Deklaration 6. POST /api/media-assets/bulk-patch – P-06-Pfad im Bulk """ from __future__ import annotations import os from contextlib import ExitStack from unittest.mock import MagicMock, patch, call import pytest from fastapi.testclient import TestClient from fastapi import HTTPException os.environ.setdefault("SKIP_DB_MIGRATE", "1") from main import app from media_rights import ( validate_rights_declaration, check_rights_coverage, assert_rights_for_promotion, rights_covers_target, visibility_level, ) from tenant_context import TenantContext, get_tenant_context # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _SUPERADMIN_TENANT = TenantContext( profile_id=1, global_role="superadmin", effective_club_id=None, club_ids=frozenset(), memberships=[], ) _FULL_DECL = { "rights_holder_confirmed": True, "contains_identifiable_persons": False, "person_consent_confirmed": None, "contains_minors": False, "parental_consent_confirmed": None, "contains_music": False, "music_rights_confirmed": None, "contains_third_party_content": False, "third_party_rights_confirmed": None, } _PRIVATE_ASSET = { "id": 42, "visibility": "private", "club_id": 7, "uploaded_by_profile_id": 1, "lifecycle_state": "active", "copyright_notice": "Rechteinhaber 2026", "original_filename": "foto.jpg", "sha256": "a" * 64, "storage_key": f"library/verein-7/image/{'a' * 64}.jpg", "storage_backend": "local", "mime_type": "image/jpeg", "byte_size": 1024, "created_at": None, "tags": [], "rights_status": "legacy_unreviewed", "rights_declared_for_visibility": None, } _DECLARED_ASSET = {**_PRIVATE_ASSET, "rights_status": "declared", "rights_declared_for_visibility": "private"} _BLOCKED_ASSET = {**_PRIVATE_ASSET, "rights_status": "blocked", "rights_declared_for_visibility": None} @pytest.fixture def client() -> TestClient: return TestClient(app) @pytest.fixture(autouse=True) def _clear_overrides(): yield app.dependency_overrides.pop(get_tenant_context, None) def _make_db_mocks(asset: dict) -> tuple[MagicMock, MagicMock]: mock_cur = MagicMock() mock_cur.fetchone.return_value = asset mock_conn = MagicMock() mock_cm = MagicMock() mock_cm.__enter__.return_value = mock_conn mock_cm.__exit__.return_value = False return mock_cm, mock_cur _PERMISSION_PATCHES = [ ("routers.media_assets.assert_can_edit_media_asset_metadata", {}), ("routers.media_assets.assert_valid_governance_visibility", {}), ("routers.media_assets._media_assets_tags_column_present", {"return_value": False}), ("routers.media_assets.get_effective_media_root", {"return_value": "/tmp/media"}), ("routers.media_assets._relocate_asset_file_if_governance_changed", {"return_value": None}), ] def _enter_permission_patches(stack: ExitStack) -> None: for target, kwargs in _PERMISSION_PATCHES: stack.enter_context(patch(target, **kwargs)) # =========================================================================== # 1. validate_rights_declaration – Unit-Tests (kein HTTP) # =========================================================================== class TestValidateRightsDeclaration: def test_missing_rights_holder_raises(self): decl = {**_FULL_DECL, "rights_holder_confirmed": False} with pytest.raises(HTTPException) as exc: validate_rights_declaration(decl, "private") assert exc.value.status_code == 400 assert exc.value.detail["code"] == "RIGHTS_DECLARATION_REQUIRED" def test_identifiable_persons_none_raises(self): decl = {**_FULL_DECL, "contains_identifiable_persons": None} with pytest.raises(HTTPException) as exc: validate_rights_declaration(decl, "private") assert exc.value.status_code == 400 assert exc.value.detail["code"] == "RIGHTS_DECLARATION_REQUIRED" def test_person_consent_required_when_persons_present(self): decl = {**_FULL_DECL, "contains_identifiable_persons": True, "person_consent_confirmed": None} with pytest.raises(HTTPException) as exc: validate_rights_declaration(decl, "club") assert exc.value.detail["code"] == "PERSON_CONSENT_REQUIRED" def test_minors_none_raises(self): decl = {**_FULL_DECL, "contains_minors": None} with pytest.raises(HTTPException) as exc: validate_rights_declaration(decl, "private") assert exc.value.detail["code"] == "RIGHTS_DECLARATION_REQUIRED" def test_parental_consent_required_when_minors_present(self): decl = {**_FULL_DECL, "contains_minors": True, "parental_consent_confirmed": False} with pytest.raises(HTTPException) as exc: validate_rights_declaration(decl, "official") assert exc.value.detail["code"] == "PARENTAL_CONSENT_REQUIRED" def test_music_none_raises(self): decl = {**_FULL_DECL, "contains_music": None} with pytest.raises(HTTPException) as exc: validate_rights_declaration(decl, "private") assert exc.value.detail["code"] == "RIGHTS_DECLARATION_REQUIRED" def test_music_rights_required_when_music_present(self): decl = {**_FULL_DECL, "contains_music": True, "music_rights_confirmed": False} with pytest.raises(HTTPException) as exc: validate_rights_declaration(decl, "private") assert exc.value.detail["code"] == "MUSIC_RIGHTS_REQUIRED" def test_third_party_none_raises(self): decl = {**_FULL_DECL, "contains_third_party_content": None} with pytest.raises(HTTPException) as exc: validate_rights_declaration(decl, "private") assert exc.value.detail["code"] == "RIGHTS_DECLARATION_REQUIRED" def test_third_party_rights_required_when_content_present(self): decl = {**_FULL_DECL, "contains_third_party_content": True, "third_party_rights_confirmed": None} with pytest.raises(HTTPException) as exc: validate_rights_declaration(decl, "official") assert exc.value.detail["code"] == "THIRD_PARTY_RIGHTS_REQUIRED" def test_full_clean_decl_passes(self): validate_rights_declaration(_FULL_DECL, "official") def test_full_decl_with_all_true_passes(self): decl = { "rights_holder_confirmed": True, "contains_identifiable_persons": True, "person_consent_confirmed": True, "contains_minors": True, "parental_consent_confirmed": True, "contains_music": True, "music_rights_confirmed": True, "contains_third_party_content": True, "third_party_rights_confirmed": True, } validate_rights_declaration(decl, "official") def test_private_also_requires_full_declaration(self): """Konservative Erstannahme: private erfordert dieselbe Erklaerung wie official.""" decl = {**_FULL_DECL, "contains_identifiable_persons": None} with pytest.raises(HTTPException): validate_rights_declaration(decl, "private") # =========================================================================== # 2. check_rights_coverage – Unit-Tests mit Mock-Cursor # =========================================================================== class TestCheckRightsCoverage: def _cur(self, row): cur = MagicMock() cur.fetchone.return_value = row return cur def test_no_asset_returns_no_declaration(self): cur = self._cur(None) assert check_rights_coverage(cur, 1, "private") == "no_declaration" def test_blocked_returns_blocked(self): cur = self._cur({"rights_status": "blocked"}) assert check_rights_coverage(cur, 1, "private") == "blocked" def test_legacy_returns_legacy(self): cur = self._cur({"rights_status": "legacy_unreviewed"}) assert check_rights_coverage(cur, 1, "club") == "legacy" def test_declared_covers_any_visibility(self): # Eine bestehende Erklaerung gilt sichtbarkeitsunabhaengig for target in ("private", "club", "official"): cur = self._cur({"rights_status": "declared"}) assert check_rights_coverage(cur, 1, target) == "ok", f"failed for target={target}" # =========================================================================== # 3. assert_rights_for_promotion – Fehlermuster # =========================================================================== class TestAssertRightsForPromotion: def _cur(self, row): cur = MagicMock() cur.fetchone.return_value = row return cur def test_ok_passes_for_any_target(self): for target in ("private", "club", "official"): cur = self._cur({"rights_status": "declared"}) assert_rights_for_promotion(cur, 1, target) # no raise def test_legacy_raises_legacy_code(self): cur = self._cur({"rights_status": "legacy_unreviewed"}) with pytest.raises(HTTPException) as exc: assert_rights_for_promotion(cur, 1, "club") assert exc.value.status_code == 400 assert exc.value.detail["code"] == "LEGACY_REDECLARATION_REQUIRED" def test_blocked_raises_403(self): cur = self._cur({"rights_status": "blocked"}) with pytest.raises(HTTPException) as exc: assert_rights_for_promotion(cur, 1, "official") assert exc.value.status_code == 403 assert exc.value.detail["code"] == "RIGHTS_BLOCKED" # =========================================================================== # 4. PATCH /api/media-assets/{id} – P-06-Promotion via HTTP # =========================================================================== class TestPatchP06Promotion: def test_promote_legacy_without_decl_returns_400(self, client): """PATCH private->club ohne P-06-Felder muss LEGACY_REDECLARATION_REQUIRED liefern.""" app.dependency_overrides[get_tenant_context] = lambda: _SUPERADMIN_TENANT mock_cm, mock_cur = _make_db_mocks(_PRIVATE_ASSET) mock_cur.fetchone.side_effect = [_PRIVATE_ASSET, {"rights_status": "legacy_unreviewed", "rights_declared_for_visibility": None}] with ExitStack() as stack: stack.enter_context(patch("routers.media_assets.get_db", return_value=mock_cm)) stack.enter_context(patch("routers.media_assets.get_cursor", return_value=mock_cur)) _enter_permission_patches(stack) r = client.patch( "/api/media-assets/42", json={"visibility": "club", "club_id": 7, "copyright_notice": "Rechteinhaber 2026"}, headers={"X-Auth-Token": "t"}, ) assert r.status_code == 400 d = r.json()["detail"] assert d["code"] == "LEGACY_REDECLARATION_REQUIRED" def test_promote_legacy_with_full_decl_calls_write_declaration(self, client): """PATCH private->club mit vollstaendiger P-06-Erklaerung schreibt Declaration-Log.""" app.dependency_overrides[get_tenant_context] = lambda: _SUPERADMIN_TENANT mock_cm, mock_cur = _make_db_mocks(_PRIVATE_ASSET) updated_asset = {**_PRIVATE_ASSET, "visibility": "club"} mock_cur.fetchone.side_effect = [ _PRIVATE_ASSET, {"rights_status": "legacy_unreviewed", "rights_declared_for_visibility": None}, updated_asset, ] with ExitStack() as stack: stack.enter_context(patch("routers.media_assets.get_db", return_value=mock_cm)) stack.enter_context(patch("routers.media_assets.get_cursor", return_value=mock_cur)) _enter_permission_patches(stack) wr = stack.enter_context(patch("routers.media_assets.write_rights_declaration", return_value=1)) uq = stack.enter_context(patch("routers.media_assets.update_rights_quick_fields")) r = client.patch( "/api/media-assets/42", json={ "visibility": "club", "club_id": 7, "copyright_notice": "Rechteinhaber 2026", **_FULL_DECL, }, headers={"X-Auth-Token": "t"}, ) assert r.status_code == 200 assert wr.called call_args = wr.call_args assert call_args.args[3] == "legacy_re_declaration" assert call_args.args[4] == "club" assert uq.called # =========================================================================== # 5. POST /api/media-assets/{id}/rights-declarations – Re-Deklaration # =========================================================================== class TestPostRightsDeclaration: def test_redeclaration_for_legacy_asset_succeeds(self, client): """Nachdeklaration fuer Altmedium setzt action_type='legacy_re_declaration'.""" app.dependency_overrides[get_tenant_context] = lambda: _SUPERADMIN_TENANT mock_cm, mock_cur = _make_db_mocks(_PRIVATE_ASSET) mock_cur.fetchone.return_value = _PRIVATE_ASSET with ExitStack() as stack: stack.enter_context(patch("routers.media_assets.get_db", return_value=mock_cm)) stack.enter_context(patch("routers.media_assets.get_cursor", return_value=mock_cur)) stack.enter_context(patch("routers.media_assets.assert_can_edit_media_asset_metadata")) wr = stack.enter_context(patch("routers.media_assets.write_rights_declaration", return_value=99)) stack.enter_context(patch("routers.media_assets.update_rights_quick_fields")) r = client.post( "/api/media-assets/42/rights-declarations", json={"target_visibility": "club", **_FULL_DECL}, headers={"X-Auth-Token": "t"}, ) assert r.status_code == 200 data = r.json() assert data["action_type"] == "legacy_re_declaration" assert data["declaration_id"] == 99 assert wr.called def test_redeclaration_incomplete_decl_returns_400_or_422(self, client): """Fehlende Erklaerungsfelder fuehren zu 400 (business logic) oder 422 (Pydantic).""" app.dependency_overrides[get_tenant_context] = lambda: _SUPERADMIN_TENANT mock_cm, mock_cur = _make_db_mocks(_PRIVATE_ASSET) with ExitStack() as stack: stack.enter_context(patch("routers.media_assets.get_db", return_value=mock_cm)) stack.enter_context(patch("routers.media_assets.get_cursor", return_value=mock_cur)) stack.enter_context(patch("routers.media_assets.assert_can_edit_media_asset_metadata")) r = client.post( "/api/media-assets/42/rights-declarations", # Fehlende Pflichtfelder (contains_identifiable_persons etc.) -> Pydantic 422 ODER # validate_rights_declaration 400 je nach welche Felder fehlen json={"target_visibility": "private", "rights_holder_confirmed": True}, headers={"X-Auth-Token": "t"}, ) assert r.status_code in (400, 422) def test_redeclaration_asset_not_found_returns_404(self, client): app.dependency_overrides[get_tenant_context] = lambda: _SUPERADMIN_TENANT mock_cm, mock_cur = _make_db_mocks(None) with ExitStack() as stack: stack.enter_context(patch("routers.media_assets.get_db", return_value=mock_cm)) stack.enter_context(patch("routers.media_assets.get_cursor", return_value=mock_cur)) stack.enter_context(patch("routers.media_assets.assert_can_edit_media_asset_metadata")) r = client.post( "/api/media-assets/999/rights-declarations", json={"target_visibility": "private", **_FULL_DECL}, headers={"X-Auth-Token": "t"}, ) assert r.status_code == 404 # =========================================================================== # 6. Bulk-Patch – P-06-Promotion im Batch # =========================================================================== class TestBulkPatchP06: def test_bulk_promote_legacy_without_decl_reports_failure(self, client): """Bulk-Patch: Legacy-Asset ohne P-06-Felder landet in 'failed', nicht 422.""" app.dependency_overrides[get_tenant_context] = lambda: _SUPERADMIN_TENANT mock_cm, mock_cur = _make_db_mocks(_PRIVATE_ASSET) mock_cur.fetchone.side_effect = [ _PRIVATE_ASSET, {"rights_status": "legacy_unreviewed", "rights_declared_for_visibility": None}, ] with ExitStack() as stack: stack.enter_context(patch("routers.media_assets.get_db", return_value=mock_cm)) stack.enter_context(patch("routers.media_assets.get_cursor", return_value=mock_cur)) stack.enter_context(patch("routers.media_assets.assert_can_edit_media_asset_metadata")) stack.enter_context(patch("routers.media_assets.assert_valid_governance_visibility")) stack.enter_context(patch("routers.media_assets._media_assets_tags_column_present", return_value=False)) stack.enter_context(patch("routers.media_assets.get_effective_media_root", return_value="/tmp")) stack.enter_context(patch("routers.media_assets._relocate_asset_file_if_governance_changed", return_value=None)) r = client.post( "/api/media-assets/bulk-patch", json={ "media_asset_ids": [42], "visibility": "club", "club_id": 7, "copyright_notice": "Rechteinhaber 2026", }, headers={"X-Auth-Token": "t"}, ) assert r.status_code == 200 data = r.json() assert data["failed_count"] == 1 assert data["updated_count"] == 0 assert "LEGACY_REDECLARATION_REQUIRED" in data["failed"][0]["detail"] or \ "RIGHTS_SCOPE_INSUFFICIENT" in data["failed"][0]["detail"]