feat: unify image storage onto files backend + Apple-style collections UI

- Image model: new file OneToOneField → FileUpload; get_url/get_thumbnail_url
  serve via /ui/files/ (Pillow thumbs) with legacy R2 fallback until backfill
- upload_images API now persists through _save_uploaded_file (files backend);
  deletes clean up FileUpload disk+record; collection destroy cascades files
- R2Storage.delete_file implemented (was silently missing → legacy R2 delete no-op)
- manage.py migrate_image_storage: idempotent R2→files backfill (dry-run default,
  --commit, optional --delete-r2); nothing removed from R2 without explicit flag
- Collections list: Apple design system, cover cards, hover actions, mobile-first
- Collection detail: native drag&drop upload (CDN Dropzone removed), lightbox with
  keyboard/swipe nav + description edit + delete, valid JSON image data via json_script
- Slideshow untouched (automatically uses new backend)
- tests: 10 new (upload/delete/URLs/backfill), 148 total green
This commit is contained in:
OpenClaw Sub-agent
2026-08-03 16:08:34 +10:00
parent 84ec8c2c24
commit 366cc35ff5
11 changed files with 1444 additions and 580 deletions
+206
View File
@@ -0,0 +1,206 @@
"""
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"),
)
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)
fake_storage = mock.Mock()
fake_storage.bucket = "bucket"
fake_storage.client.get_object.return_value = {"Body": io.BytesIO(raw)}
with mock.patch(
"links.management.commands.migrate_image_storage.R2Storage",
return_value=fake_storage,
):
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 mock.patch(
"links.management.commands.migrate_image_storage.R2Storage",
return_value=fake_storage,
):
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")
fake_storage = mock.Mock()
fake_storage.bucket = "bucket"
fake_storage.client.get_object.side_effect = Exception("network down")
with mock.patch(
"links.management.commands.migrate_image_storage.R2Storage",
return_value=fake_storage,
):
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