shinkan-jinkendo/backend/tests/test_media_assets_archive.py

"""
Medienarchiv: GET /api/media-assets und POST /api/exercises/{id}/media/from-asset (gemockte DB).
"""
from __future__ import annotations

import os
from unittest.mock import MagicMock, patch

import pytest
from fastapi.testclient import TestClient

os.environ.setdefault("SKIP_DB_MIGRATE", "1")

from auth import require_auth
from main import app
from tenant_context import TenantContext, get_tenant_context


@pytest.fixture
def client() -> TestClient:
    return TestClient(app)


@pytest.fixture(autouse=True)
def _clear_overrides() -> None:
    yield
    app.dependency_overrides.pop(require_auth, None)
    app.dependency_overrides.pop(get_tenant_context, None)


def _mock_db(mock_cur: MagicMock) -> MagicMock:
    mock_conn = MagicMock()
    mock_cm = MagicMock()
    mock_cm.__enter__.return_value = mock_conn
    mock_cm.__exit__.return_value = False
    return mock_cm


def test_list_media_assets_ok_mocked(client: TestClient) -> None:
    app.dependency_overrides[require_auth] = lambda: {"profile_id": 10, "role": "trainer"}
    app.dependency_overrides[get_tenant_context] = lambda: TenantContext(
        profile_id=10,
        global_role="trainer",
        effective_club_id=5,
        club_ids=frozenset({5}),
        memberships=[],
    )

    mock_cur = MagicMock()
    mock_cur.fetchall.return_value = [
        {
            "id": 1,
            "mime_type": "image/png",
            "byte_size": 100,
            "original_filename": "a.png",
            "visibility": "official",
            "club_id": None,
            "uploaded_by_profile_id": 2,
            "lifecycle_state": "active",
            "created_at": None,
            "sha256": "a" * 64,
        }
    ]
    mock_cm = _mock_db(mock_cur)

    with patch("routers.media_assets.get_db", return_value=mock_cm), patch(
        "routers.media_assets.get_cursor", return_value=mock_cur
    ):
        r = client.get("/api/media-assets?q=test", headers={"X-Auth-Token": "t"})

    assert r.status_code == 200
    body = r.json()
    assert body["limit"] == 30
    assert len(body["items"]) == 1
    assert body["items"][0]["original_filename"] == "a.png"


def test_attach_from_asset_duplicate_returns_400(client: TestClient) -> None:
    app.dependency_overrides[require_auth] = lambda: {"profile_id": 1, "role": "trainer"}
    app.dependency_overrides[get_tenant_context] = lambda: TenantContext(
        profile_id=1,
        global_role="trainer",
        effective_club_id=None,
        club_ids=frozenset(),
        memberships=[],
    )

    mock_cur = MagicMock()
    mock_cur.fetchone.side_effect = [
        {"created_by": 1, "visibility": "private", "club_id": None},
        {"c": 0},
        {
            "id": 5,
            "mime_type": "image/jpeg",
            "byte_size": 10,
            "original_filename": "x.jpg",
            "visibility": "official",
            "club_id": None,
            "uploaded_by_profile_id": 1,
            "lifecycle_state": "active",
            "storage_key": "exercises/x.jpg",
        },
        {"id": 1},
    ]
    mock_cm = _mock_db(mock_cur)

    with patch("routers.exercises.get_db", return_value=mock_cm), patch(
        "routers.exercises.get_cursor", return_value=mock_cur
    ):
        r = client.post(
            "/api/exercises/3/media/from-asset",
            headers={"X-Auth-Token": "t", "Content-Type": "application/json"},
            json={
                "media_asset_id": 5,
                "title": "",
                "description": "",
                "context": "ablauf",
                "is_primary": False,
            },
        )

    assert r.status_code == 400
    assert "bereits" in (r.json().get("detail") or "").lower()


def test_attach_from_asset_ok_mocked(client: TestClient) -> None:
    app.dependency_overrides[require_auth] = lambda: {"profile_id": 1, "role": "trainer"}
    app.dependency_overrides[get_tenant_context] = lambda: TenantContext(
        profile_id=1,
        global_role="trainer",
        effective_club_id=None,
        club_ids=frozenset(),
        memberships=[],
    )

    inserted = {
        "id": 99,
        "exercise_id": 3,
        "media_type": "image",
        "file_path": "/media/exercises/h.jpg",
        "file_size": 10,
        "mime_type": "image/jpeg",
        "original_filename": "h.jpg",
        "embed_url": None,
        "embed_platform": None,
        "title": "h.jpg",
        "description": None,
        "sort_order": 1,
        "is_primary": False,
        "context": "ablauf",
        "created_at": None,
        "media_asset_id": 5,
    }
    mock_cur = MagicMock()
    mock_cur.fetchone.side_effect = [
        {"created_by": 1, "visibility": "private", "club_id": None},
        {"c": 0},
        {
            "id": 5,
            "mime_type": "image/jpeg",
            "byte_size": 10,
            "original_filename": "h.jpg",
            "visibility": "official",
            "club_id": None,
            "uploaded_by_profile_id": 1,
            "lifecycle_state": "active",
            "storage_key": "exercises/h.jpg",
        },
        None,
        inserted,
    ]
    mock_cm = _mock_db(mock_cur)

    with patch("routers.exercises.get_db", return_value=mock_cm), patch(
        "routers.exercises.get_cursor", return_value=mock_cur
    ):
        r = client.post(
            "/api/exercises/3/media/from-asset",
            headers={"X-Auth-Token": "t", "Content-Type": "application/json"},
            json={
                "media_asset_id": 5,
                "context": "detail",
                "is_primary": False,
            },
        )

    assert r.status_code == 201
    body = r.json()
    assert body["id"] == 99
    assert body["media_asset_id"] == 5
    assert body["asset_lifecycle_state"] == "active"


def test_delete_exercise_media_returns_orphan_when_last_ref(client: TestClient) -> None:
    app.dependency_overrides[require_auth] = lambda: {"profile_id": 1, "role": "trainer"}
    app.dependency_overrides[get_tenant_context] = lambda: TenantContext(
        profile_id=1,
        global_role="trainer",
        effective_club_id=None,
        club_ids=frozenset(),
        memberships=[],
    )

    mock_cur = MagicMock()
    mock_cur.fetchone.side_effect = [
        {"created_by": 1, "visibility": "private", "club_id": None},
        {"media_asset_id": 88},
        {"c": 1},
    ]
    mock_cm = _mock_db(mock_cur)

    with patch("routers.exercises.get_db", return_value=mock_cm), patch(
        "routers.exercises.get_cursor", return_value=mock_cur
    ):
        r = client.delete("/api/exercises/10/media/20", headers={"X-Auth-Token": "t"})

    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is True
    assert body["orphan_media_asset_id"] == 88


def test_delete_exercise_media_no_orphan_when_shared(client: TestClient) -> None:
    app.dependency_overrides[require_auth] = lambda: {"profile_id": 1, "role": "trainer"}
    app.dependency_overrides[get_tenant_context] = lambda: TenantContext(
        profile_id=1,
        global_role="trainer",
        effective_club_id=None,
        club_ids=frozenset(),
        memberships=[],
    )

    mock_cur = MagicMock()
    mock_cur.fetchone.side_effect = [
        {"created_by": 1, "visibility": "private", "club_id": None},
        {"media_asset_id": 88},
        {"c": 2},
    ]
    mock_cm = _mock_db(mock_cur)

    with patch("routers.exercises.get_db", return_value=mock_cm), patch(
        "routers.exercises.get_cursor", return_value=mock_cur
    ):
        r = client.delete("/api/exercises/10/media/20", headers={"X-Auth-Token": "t"})

    assert r.status_code == 200
    assert r.json()["orphan_media_asset_id"] is None


def test_delete_exercise_media_embed_row_no_orphan(client: TestClient) -> None:
    app.dependency_overrides[require_auth] = lambda: {"profile_id": 1, "role": "trainer"}
    app.dependency_overrides[get_tenant_context] = lambda: TenantContext(
        profile_id=1,
        global_role="trainer",
        effective_club_id=None,
        club_ids=frozenset(),
        memberships=[],
    )

    mock_cur = MagicMock()
    mock_cur.fetchone.side_effect = [
        {"created_by": 1, "visibility": "private", "club_id": None},
        {"media_asset_id": None},
    ]
    mock_cm = _mock_db(mock_cur)

    with patch("routers.exercises.get_db", return_value=mock_cm), patch(
        "routers.exercises.get_cursor", return_value=mock_cur
    ):
        r = client.delete("/api/exercises/10/media/21", headers={"X-Auth-Token": "t"})

    assert r.status_code == 200
    assert r.json() == {"ok": True, "orphan_media_asset_id": None}