links/tests/test_collections.py

"""
Tests for Image Collections: unified files-backend storage, upload/delete
API flows, URL resolution and the R2→files backfill command.
"""

import io
import os
from unittest import mock

import pytest
from django.core.files.uploadedfile import SimpleUploadedFile
from django.core.management import call_command
from PIL import Image as PILImage

from links.models import ImageCollection, Image, FileUpload


def _real_png(name="photo.png"):
    """A fully decodable 8x8 PNG (Pillow can open it → thumbnails work)."""
    buf = io.BytesIO()
    PILImage.new("RGB", (8, 8), (200, 60, 60)).save(buf, format="PNG")
    return SimpleUploadedFile(name, buf.getvalue(), content_type="image/png")


def _text_file():
    return SimpleUploadedFile("note.txt", b"hello", content_type="text/plain")


@pytest.mark.django_db
class TestImageFilesBackend:
    def _collection(self):
        return ImageCollection.objects.create(name="Test Collection")

    def test_upload_images_uses_files_backend(self, api_client):
        coll = self._collection()
        r = api_client.post(
            f"/api/collections/{coll.pk}/upload_images",
            {"file": _real_png()},
            format="multipart",
        )
        assert r.status_code == 201
        assert r.json()["status"] == "success"

        image = coll.images.get()
        assert image.file is not None
        # The image is served through the files backend URL, not R2
        assert image.file_key == ""
        assert image.file.mime_type == "image/png"
        assert image.file.size > 0
        assert os.path.exists(image.file.file_path)
        assert image.get_url() == image.file.download_url
        assert image.get_url().startswith("/ui/files/")
        assert image.get_thumbnail_url() == image.file.thumbnail_url

    def test_upload_rejects_non_image(self, api_client):
        coll = self._collection()
        r = api_client.post(
            f"/api/collections/{coll.pk}/upload_images",
            {"file": _text_file()},
            format="multipart",
        )
        assert r.status_code == 400
        assert coll.images.count() == 0
        assert FileUpload.objects.count() == 0

    def test_upload_multiple_images(self, api_client):
        coll = self._collection()
        r = api_client.post(
            f"/api/collections/{coll.pk}/upload_images",
            {"file": [_real_png("a.png"), _real_png("b.png")]},
            format="multipart",
        )
        assert r.status_code == 201
        assert coll.images.count() == 2
        assert FileUpload.objects.count() == 2

    def test_serializer_exposes_url_and_thumbnail(self, api_client):
        coll = self._collection()
        api_client.post(
            f"/api/collections/{coll.pk}/upload_images",
            {"file": _real_png()},
            format="multipart",
        )
        data = api_client.get(f"/api/collections/{coll.pk}").json()
        assert data["image_count"] == 1
        r = api_client.get(f"/api/images/{coll.images.get().pk}")
        assert r.status_code == 200
        body = r.json()
        assert body["url"].startswith("/ui/files/")
        assert body["thumbnail_url"] == "/ui/files/" + str(coll.images.get().file.pk) + "/thumb/"

    def test_delete_image_removes_file_and_record(self, api_client):
        coll = self._collection()
        api_client.post(
            f"/api/collections/{coll.pk}/upload_images",
            {"file": _real_png()},
            format="multipart",
        )
        image = coll.images.get()
        path = image.file.file_path
        assert os.path.exists(path)

        r = api_client.delete(f"/api/images/{image.pk}")
        assert r.status_code == 204 or r.status_code == 200
        assert not Image.objects.filter(pk=image.pk).exists()
        assert not FileUpload.objects.filter(pk=image.file.pk).exists()
        assert not os.path.exists(path)

    def test_delete_collection_cascades(self, api_client):
        coll = self._collection()
        api_client.post(
            f"/api/collections/{coll.pk}/upload_images",
            {"file": [_real_png("a.png"), _real_png("b.png")]},
            format="multipart",
        )
        paths = [img.file.file_path for img in coll.images.all()]
        assert all(os.path.exists(p) for p in paths)

        r = api_client.delete(f"/api/collections/{coll.pk}")
        assert r.status_code == 204 or r.status_code == 200
        assert not ImageCollection.objects.filter(pk=coll.pk).exists()
        assert not Image.objects.filter(collection=coll).exists()
        assert FileUpload.objects.count() == 0
        assert not any(os.path.exists(p) for p in paths)

    def test_legacy_r2_image_falls_back_to_r2_url(self):
        """Images not yet backfilled (file_key set, no file) still resolve via R2."""
        coll = self._collection()
        image = Image.objects.create(
            collection=coll, title="legacy", file_key="images/legacy/x.jpg",
            content_type="image/jpeg", size=123,
        )
        with mock.patch("links.models.R2Storage") as fake:
            fake.return_value.get_url.return_value = "https://r2.example/x.jpg?sig=1"
            assert image.get_url().startswith("https://r2.example")
            assert fake.return_value.get_url.call_count == 1
        # And thumbnails request the cdn-cgi transformation variant
        with mock.patch("links.models.R2Storage") as fake:
            fake.return_value.get_url.return_value = "https://cdn.example/x.jpg?w=200"
            image.get_thumbnail_url()
            kwargs = fake.return_value.get_url.call_args.kwargs
            assert kwargs.get("width") == 200


@pytest.mark.django_db
class TestMigrateImageStorageCommand:
    def _legacy_image(self, coll, key="images/old/cat.jpg", raw=None):
        return Image.objects.create(
            collection=coll, title="old.png", file_key=key,
            content_type="image/png", size=len(raw or b"data"),
        )

    @staticmethod
    def _patch_download(raw=None, exc=None):
        return mock.patch(
            "links.management.commands.migrate_image_storage._download_r2_object",
            return_value=raw if raw is not None else b"\x89PNGdata",
            side_effect=exc,
        )

    def test_dry_run_does_nothing(self, tmp_path):
        coll = ImageCollection.objects.create(name="C")
        self._legacy_image(coll, raw=b"\x89PNGdata")
        call_command("migrate_image_storage")
        img = Image.objects.get()
        assert img.file is None
        assert img.file_key == "images/old/cat.jpg"

    def test_backfill_creates_fileupload_and_links(self, tmp_path):
        coll = ImageCollection.objects.create(name="C")
        raw = io.BytesIO()
        PILImage.new("RGB", (8, 8)).save(raw, format="PNG")
        raw = raw.getvalue()
        self._legacy_image(coll, raw=raw)

        with self._patch_download(raw=raw):
            call_command("migrate_image_storage", "--commit")

        img = Image.objects.get()
        assert img.file is not None
        assert img.file_key == ""
        assert os.path.exists(img.file.file_path)
        assert img.file.size == len(raw)
        # Idempotent: second run has nothing to do
        with self._patch_download(raw=raw):
            call_command("migrate_image_storage", "--commit")
        assert Image.objects.get().file is not None

    def test_backfill_failure_leaves_record_for_retry(self, tmp_path):
        coll = ImageCollection.objects.create(name="C")
        self._legacy_image(coll, raw=b"\x89PNGdata")

        with self._patch_download(exc=Exception("network down")):
            call_command("migrate_image_storage", "--commit")

        img = Image.objects.get()
        assert img.file is None
        assert img.file_key == "images/old/cat.jpg"
        assert FileUpload.objects.count() == 0