Lars
56e952f084
All checks were successful
Deploy Development / deploy (push) Successful in 39s
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 58s
- check_rights_coverage: rights_status='declared' gibt immer 'ok' zurück (P-06-Erklärung gilt inhaltlich, nicht sichtbarkeitsabhängig) - assert_rights_for_promotion: 'insufficient'-Pfad entfernt - Tests: test_declared_private_insufficient_for_club → test_declared_covers_any_visibility version: 0.8.81 module: media_rights Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
431 lines
18 KiB
Python
431 lines
18 KiB
Python
"""
|
||
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"]
|