pdf-quiz-generator/backend/tests/test_media_library.py

"""Image libraries: access is per library, and storage is behind one interface.

Run: DATABASE_URL=sqlite:///:memory: PYTHONPATH=backend python -m unittest discover -s backend/tests
"""
import os
os.environ["DATABASE_URL"] = "sqlite:///:memory:"

import io
import unittest
from unittest.mock import patch

from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from app.database import Base, get_db
from app.models.media import MediaAsset, MediaLibrary, MediaLibraryGrant
from app.models.user import User
from app.routers import media
from app.utils.auth import get_current_user

TAG_DDL = """
CREATE TABLE question_tags (id INTEGER PRIMARY KEY, name VARCHAR(200), type VARCHAR(50), exam_id INTEGER, parent_id INTEGER, sort_order INTEGER DEFAULT 100, created_at TIMESTAMP);
"""


class MediaLibraryTests(unittest.TestCase):
    def setUp(self):
        self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
        Base.metadata.create_all(self.engine)
        self.db = Session(self.engine)
        self.db.execute(text(TAG_DDL.strip()))

        self.mod = User(id=1, name="Mod", email="mod@example.test", hashed_password="unused", role="moderator")
        self.educator = User(id=2, name="Edu", email="edu@example.test", hashed_password="unused")
        self.outsider = User(id=3, name="Out", email="out@example.test", hashed_password="unused")
        self.db.add_all([self.mod, self.educator, self.outsider])
        self.db.add_all([
            MediaLibrary(id=1, name="Cardiology imaging"),
            MediaLibrary(id=2, name="Dermatology photos"),
        ])
        self.db.flush()
        self.db.add_all([
            MediaAsset(id=1, path="media/a.png", title="ECG", library_id=1, kind="image"),
            MediaAsset(id=2, path="media/b.png", title="Rash", library_id=2, kind="image"),
        ])
        self.db.add(MediaLibraryGrant(library_id=1, user_id=2, granted_by=1))
        self.db.commit()

        self.user = self.mod
        app = FastAPI()
        app.include_router(media.router, prefix="/media")
        app.dependency_overrides[get_db] = lambda: self.db
        app.dependency_overrides[get_current_user] = lambda: self.user
        self.client = TestClient(app)

    def tearDown(self):
        self.client.close()
        self.db.close()
        self.engine.dispose()

    def test_a_moderator_sees_every_library(self):
        names = {lib["name"] for lib in self.client.get("/media/libraries").json()}
        self.assertEqual(names, {"Cardiology imaging", "Dermatology photos"})

    def test_a_grant_limits_which_libraries_and_images_are_visible(self):
        self.user = self.educator
        libraries = self.client.get("/media/libraries").json()
        self.assertEqual([lib["name"] for lib in libraries], ["Cardiology imaging"])

        images = self.client.get("/media/").json()
        self.assertEqual([i["title"] for i in images["images"]], ["ECG"])

        # And an ungranted library cannot be browsed by asking for it directly.
        self.assertEqual(self.client.get("/media/", params={"library_id": 2}).status_code, 403)

    def test_someone_with_no_grant_sees_nothing(self):
        self.user = self.outsider
        self.assertEqual(self.client.get("/media/libraries").json(), [])
        self.assertEqual(self.client.get("/media/").json()["total"], 0)

    def test_editing_is_confined_to_granted_libraries(self):
        self.user = self.educator
        self.assertEqual(self.client.patch("/media/1", json={"caption": "Sinus rhythm"}).status_code, 200)
        self.assertEqual(self.db.get(MediaAsset, 1).caption, "Sinus rhythm")
        # Image 2 is in a library they were not granted.
        self.assertEqual(self.client.patch("/media/2", json={"caption": "No"}).status_code, 403)
        # Nor can an image be moved into one.
        self.assertEqual(self.client.patch("/media/1", json={"library_id": 2}).status_code, 403)

    def test_tags_reuse_the_shared_vocabulary(self):
        with patch.object(media.embedding_service, "embed_record", return_value=False):
            response = self.client.patch("/media/1", json={"tags": ["Arrhythmia", "ECG"]})
        self.assertEqual(response.status_code, 200, response.text)
        self.assertEqual(sorted(response.json()["tags"]), ["Arrhythmia", "ECG"])
        # A tag that already exists is reused rather than duplicated.
        existing = self.db.execute(text("SELECT COUNT(*) FROM question_tags")).scalar()
        with patch.object(media.embedding_service, "embed_record", return_value=False):
            self.client.patch("/media/2", json={"tags": ["ECG"]})
        self.assertEqual(self.db.execute(text("SELECT COUNT(*) FROM question_tags")).scalar(), existing)

    def test_upload_stores_through_the_storage_service(self):
        saved = {}

        def fake_save(key, data, content_type=None):
            saved["key"], saved["bytes"] = key, len(data)
            return key

        with patch.object(media.storage_service, "save", side_effect=fake_save), \
             patch.object(media.storage_service, "using_s3", return_value=True), \
             patch.object(media.embedding_service, "embed_record", return_value=False):
            response = self.client.post(
                "/media/upload",
                files={"file": ("scan.png", io.BytesIO(b"x" * 100), "image/png")},
                data={"library_id": "1"},
            )
        self.assertEqual(response.status_code, 201, response.text)
        body = response.json()
        # The row stores the key, never a URL, so the backend can change later.
        self.assertTrue(body["path"].startswith("media/"))
        self.assertNotIn("http", body["path"])
        self.assertEqual(body["storage"], "s3")
        self.assertEqual(saved["bytes"], 100)

    def test_upload_rejects_the_wrong_type_and_oversized_files(self):
        with patch.object(media.storage_service, "save", return_value="k"):
            bad_type = self.client.post(
                "/media/upload", files={"file": ("a.exe", io.BytesIO(b"x"), "application/x-msdownload")},
                data={"library_id": "1"})
            oversized = self.client.post(
                "/media/upload",
                files={"file": ("a.png", io.BytesIO(b"x" * (media.MAX_IMAGE_BYTES + 5)), "image/png")},
                data={"library_id": "1"})
        self.assertEqual(bad_type.status_code, 400)
        self.assertEqual(oversized.status_code, 413)

    def test_granting_and_revoking_a_library_is_moderator_only(self):
        self.assertEqual(self.client.post("/media/libraries/2/grants", json={"user_id": 3}).status_code, 201)
        self.assertEqual(self.client.post("/media/libraries/2/grants", json={"user_id": 3}).status_code, 409)
        self.assertEqual(self.client.delete("/media/libraries/2/grants/3").status_code, 204)
        self.user = self.educator
        self.assertEqual(self.client.post("/media/libraries/2/grants", json={"user_id": 3}).status_code, 403)


if __name__ == "__main__":
    unittest.main()