Lars
1ce6d929ce
All checks were successful
Deploy Development / deploy (push) Successful in 40s
Test Suite / pytest-backend (push) Successful in 34s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 10s
Test Suite / playwright-tests (push) Successful in 54s
- Added backend support for Legal Hold with new endpoints to set and release holds on media assets. - Introduced new database columns for managing Legal Hold status and reasons. - Updated frontend to include UI elements for setting and releasing Legal Holds, including a confirmation dialog. - Enhanced Media Library page to display Legal Hold status and actions for superadmins. - Implemented comprehensive backend tests covering all aspects of Legal Hold functionality. - Updated documentation to reflect changes in the upload rights specification and interface models. - Bumped version to 0.8.84 and updated MediaLibraryPage version to 1.6.0.
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"
|
||
)
|