"""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()