""" P-11: Legal-Hold – Backend-Tests. Abgedeckt (15 Faelle): 1. assert_superadmin_for_legal_hold – Superadmin darf 2. assert_superadmin_for_legal_hold – Nicht-Superadmin wird geblockt 3. is_media_available_for_normal_use – ohne Hold True 4. is_media_available_for_normal_use – mit Hold False 5. assert_not_under_legal_hold – ohne Hold passiert nichts 6. assert_not_under_legal_hold – mit Hold wirft 403 LEGAL_HOLD_ACTIVE 7. set_legal_hold – ungueltige reason_code → 400 8. set_legal_hold – leere reason_note → 400 9. set_legal_hold – Asset bereits unter Hold → 409 10. set_legal_hold – Erfolgspfad: DB-Update + Audit-Log 11. release_legal_hold – leere release_note → 400 12. release_legal_hold – Asset nicht unter Hold → 409 13. release_legal_hold – Erfolgspfad ohne Deklaration → legacy_unreviewed 14. release_legal_hold – Erfolgspfad mit Deklaration → declared 15. Retention-Job ueberspringt Legal-Hold-Assets (run_retention_pass query) """ from __future__ import annotations import os from unittest.mock import MagicMock, patch, call import pytest from fastapi import HTTPException os.environ.setdefault("SKIP_DB_MIGRATE", "1") from media_legal_hold import ( LEGAL_HOLD_REASON_CODES, assert_not_under_legal_hold, assert_superadmin_for_legal_hold, is_media_available_for_normal_use, release_legal_hold, set_legal_hold, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _asset(legal_hold_active=False, rights_status="legacy_unreviewed", **kw) -> dict: base = { "id": 99, "visibility": "private", "lifecycle_state": "active", "rights_status": rights_status, "legal_hold_active": legal_hold_active, "legal_hold_reason_code": None, "legal_hold_reason_note": None, } base.update(kw) return base def _make_cur(fetchone_val, fetchone_seq=None): """Erstellt einen Mock-Cursor. fetchone_seq: wenn angegeben, wird fetchone() sequenziell diese Werte liefern. """ cur = MagicMock() if fetchone_seq is not None: cur.fetchone.side_effect = fetchone_seq else: cur.fetchone.return_value = fetchone_val return cur def _dict_row(d: dict): """Simuliert ein psycopg2-DictRow-Objekt (keys() + Subscript).""" m = MagicMock() m.keys.return_value = list(d.keys()) m.__getitem__ = lambda self, k: d[k] m.__iter__ = lambda self: iter(d.values()) return m # --------------------------------------------------------------------------- # 1–2: assert_superadmin_for_legal_hold # --------------------------------------------------------------------------- class TestAssertSuperadmin: def test_superadmin_passes(self): # kein Fehler assert_superadmin_for_legal_hold("superadmin") def test_non_superadmin_raises_403(self): for role in ("admin", "user", "club_admin", None, ""): with pytest.raises(HTTPException) as exc: assert_superadmin_for_legal_hold(role) assert exc.value.status_code == 403 # --------------------------------------------------------------------------- # 3–4: is_media_available_for_normal_use # --------------------------------------------------------------------------- class TestIsMediaAvailable: def test_no_hold_is_available(self): assert is_media_available_for_normal_use(_asset(legal_hold_active=False)) is True def test_hold_active_is_not_available(self): assert is_media_available_for_normal_use(_asset(legal_hold_active=True)) is False # --------------------------------------------------------------------------- # 5–6: assert_not_under_legal_hold # --------------------------------------------------------------------------- class TestAssertNotUnderLegalHold: def test_no_hold_does_not_raise(self): assert_not_under_legal_hold(_asset(legal_hold_active=False)) def test_hold_active_raises_403_with_code(self): with pytest.raises(HTTPException) as exc: assert_not_under_legal_hold(_asset(legal_hold_active=True, id=55)) assert exc.value.status_code == 403 detail = exc.value.detail assert isinstance(detail, dict) assert detail["code"] == "LEGAL_HOLD_ACTIVE" assert detail["asset_id"] == 55 # --------------------------------------------------------------------------- # 7–10: set_legal_hold # --------------------------------------------------------------------------- class TestSetLegalHold: def test_invalid_reason_code_raises_400(self): cur = MagicMock() conn = MagicMock() with pytest.raises(HTTPException) as exc: set_legal_hold(cur, conn, asset_id=1, acting_profile_id=1, reason_code="totally_made_up", reason_note="some reason") assert exc.value.status_code == 400 def test_empty_reason_note_raises_400(self): cur = MagicMock() conn = MagicMock() with pytest.raises(HTTPException) as exc: set_legal_hold(cur, conn, asset_id=1, acting_profile_id=1, reason_code="rights_dispute", reason_note="") assert exc.value.status_code == 400 def test_already_under_hold_raises_409(self): asset = _asset(legal_hold_active=True) cur = _make_cur(_dict_row(asset)) conn = MagicMock() with patch("db.r2d", return_value=asset), \ patch("media_legal_hold.write_audit_log_entry"): with pytest.raises(HTTPException) as exc: set_legal_hold(cur, conn, asset_id=99, acting_profile_id=1, reason_code="rights_dispute", reason_note="Verletzung bestätigt") assert exc.value.status_code == 409 def test_success_sets_hold_and_writes_audit(self): asset_before = _asset(legal_hold_active=False, rights_status="declared") asset_after = { **asset_before, "legal_hold_active": True, "legal_hold_reason_code": "copyright_complaint", "legal_hold_reason_note": "Urheberrechtsverletzung gemeldet", "legal_hold_set_by_profile_id": 1, "legal_hold_set_at": "2026-05-11T12:00:00Z", "rights_status": "blocked", } cur = _make_cur(None) cur.fetchone.side_effect = [_dict_row(asset_before), _dict_row(asset_after)] conn = MagicMock() with patch("db.r2d", side_effect=[asset_before, asset_after]), \ patch("media_legal_hold.write_audit_log_entry") as mock_audit: result = set_legal_hold( cur, conn, asset_id=99, acting_profile_id=1, reason_code="copyright_complaint", reason_note="Urheberrechtsverletzung gemeldet", ) assert result["legal_hold_active"] is True assert result["rights_status"] == "blocked" conn.commit.assert_called_once() mock_audit.assert_called_once() audit_call_kwargs = mock_audit.call_args assert audit_call_kwargs[1]["event_type"] == "legal_hold_set" assert audit_call_kwargs[1]["new_values"]["legal_hold_active"] is True assert audit_call_kwargs[1]["new_values"]["reason_code"] == "copyright_complaint" # --------------------------------------------------------------------------- # 11–14: release_legal_hold # --------------------------------------------------------------------------- class TestReleaseLegalHold: def test_empty_release_note_raises_400(self): cur = MagicMock() conn = MagicMock() with pytest.raises(HTTPException) as exc: release_legal_hold(cur, conn, asset_id=99, acting_profile_id=1, release_note="") assert exc.value.status_code == 400 def test_not_under_hold_raises_409(self): asset = _asset(legal_hold_active=False) cur = _make_cur(_dict_row(asset)) conn = MagicMock() with patch("db.r2d", return_value=asset): with pytest.raises(HTTPException) as exc: release_legal_hold(cur, conn, asset_id=99, acting_profile_id=1, release_note="Falschmeldung") assert exc.value.status_code == 409 def test_release_without_declaration_restores_legacy_unreviewed(self): asset = _asset(legal_hold_active=True, rights_status="blocked") asset_after = {**asset, "legal_hold_active": False, "rights_status": "legacy_unreviewed", "legal_hold_released_by_profile_id": 1, "legal_hold_release_note": "Klaerung abgeschlossen"} cur = _make_cur(None) # fetchone-Sequenz: 1) Asset (fuer r2d), 2) decl count row, 3) updated asset row (fuer r2d) decl_cnt_row = type("Row", (), {"__getitem__": staticmethod(lambda k: 0), "__iter__": staticmethod(lambda: iter([0]))})() cur.fetchone.side_effect = [_dict_row(asset), decl_cnt_row, _dict_row(asset_after)] conn = MagicMock() with patch("db.r2d", side_effect=[asset, asset_after]), \ patch("media_legal_hold.write_audit_log_entry") as mock_audit: result = release_legal_hold( cur, conn, asset_id=99, acting_profile_id=1, release_note="Klaerung abgeschlossen", ) assert result["legal_hold_active"] is False assert result["rights_status"] == "legacy_unreviewed" conn.commit.assert_called_once() mock_audit.assert_called_once() assert mock_audit.call_args[1]["event_type"] == "legal_hold_released" def test_release_with_declaration_restores_declared(self): asset = _asset(legal_hold_active=True, rights_status="blocked") asset_after = {**asset, "legal_hold_active": False, "rights_status": "declared", "legal_hold_released_by_profile_id": 1, "legal_hold_release_note": "Einigung erzielt"} cur = _make_cur(None) decl_cnt_row = type("Row", (), {"__getitem__": staticmethod(lambda k: 1), "__iter__": staticmethod(lambda: iter([1]))})() cur.fetchone.side_effect = [_dict_row(asset), decl_cnt_row, _dict_row(asset_after)] conn = MagicMock() with patch("db.r2d", side_effect=[asset, asset_after]), \ patch("media_legal_hold.write_audit_log_entry") as mock_audit: result = release_legal_hold( cur, conn, asset_id=99, acting_profile_id=1, release_note="Einigung erzielt", ) assert result["legal_hold_active"] is False assert result["rights_status"] == "declared" mock_audit.assert_called_once() assert mock_audit.call_args[1]["new_values"]["rights_status"] == "declared" # --------------------------------------------------------------------------- # 15: Retention-Job ueberspringt Legal-Hold-Assets # --------------------------------------------------------------------------- class TestRetentionJobSkipsLegalHold: def test_retention_queries_exclude_legal_hold(self): """Prueft dass run_retention_pass keine Legal-Hold-Assets anfasst.""" import inspect import media_lifecycle source = inspect.getsource(media_lifecycle.run_retention_pass) # Beide Queries (trash_soft→trash_hidden und trash_hidden→purge) # muessen den Legal-Hold-Filter enthalten assert "legal_hold_active" in source, ( "run_retention_pass muss legal_hold_active in den Retention-Queries filtern" ) # Sicherstellen, dass der Filter korrekt lautet (FALSE oder IS NULL) assert "legal_hold_active = FALSE" in source or "legal_hold_active IS NULL" in source, ( "Filter muss 'legal_hold_active = FALSE' oder 'IS NULL' enthalten" )