282 lines
12 KiB
Python
282 lines
12 KiB
Python
|
"""
|
|||
|
|
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"
|
|||
|
|
)
|