mirror of
https://github.com/wahyd4/links.git
synced 2026-08-09 05:06:16 +10:00
OpenClaw Sub-agent
030e0dc5af
Direct boto3 calls against the R2 S3 endpoint fail with SignatureDoesNotMatch for this account, while presigned URLs served through the R2 custom domain are the path the app already uses. Switch migrate_image_storage download step to urlopen on the presigned URL.
201 lines
7.7 KiB
Python
201 lines
7.7 KiB
Python
"""
|
|
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)
|
|
|
|
with mock.patch(
|
|
"links.management.commands.migrate_image_storage._download_via_presigned_url",
|
|
return_value=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 mock.patch(
|
|
"links.management.commands.migrate_image_storage._download_via_presigned_url",
|
|
return_value=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 mock.patch(
|
|
"links.management.commands.migrate_image_storage._download_via_presigned_url",
|
|
side_effect=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
|