Ty
d9af8420bc
REVIEW-5 / PUBLISH-45: owner edits the stored photo lineup in Review — tap a card to keep/remove, drag to reorder, leftmost kept photo = primary (Woo cover). The kept set must contain a clean calibration + its dimension-overlay copy; publish is blocked until it does. - backend: role column on slab_photos (schema + idempotent _migrate ALTER), role in photo CRUD + new DELETE endpoint, Woo image order by role-rank (clean→overlay→batch) per PUBLISH-45 - frontend: Review photo strip (render from listPhotos, keep/remove, HTML5 drag-reorder, dynamic PRIMARY badge from .primary class), persistLineup() re-seq kept + drop removed before entering Publish - tests: lineup.ts unit, e2e review-photo-strip.spec, backend role + order Refs: #90, REVIEW-5, PUBLISH-45
136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
"""S3 photos BLOB storage tests: multipart upload, original/processed GET, 410 after purge."""
|
|
|
|
import base64
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import backend.db as db_mod
|
|
|
|
|
|
@pytest.fixture
|
|
def photos_env():
|
|
tmp = tempfile.mkdtemp(prefix="photos_")
|
|
os.environ["SLAB_DB_PATH"] = os.path.join(tmp, "p.db")
|
|
os.environ["SLAB_IMAGES_DIR"] = tmp
|
|
os.environ["WOO_BASE_URL"] = "https://example.com"
|
|
os.environ["SLAB_AES_KEY"] = base64.b64encode(b"0123456789abcdef0123456789abcdef").decode()
|
|
db_mod._reset_for_tests()
|
|
from backend.app import app
|
|
|
|
conn = db_mod.init_db()
|
|
# seed a slab
|
|
conn.execute("INSERT INTO slabs(id, sku, status, server_rev, created_at, updated_at) VALUES ('s1','SLAB-01','calibrated',1,'t','t')")
|
|
conn.commit()
|
|
yield TestClient(app), conn
|
|
db_mod._reset_for_tests()
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
PNG_ORIG = b"\x89PNG-orig-bytes"
|
|
PNG_PROC = b"\x89PNG-proc-bytes"
|
|
|
|
|
|
def _multipart(client, slab_id, photo_id, original=PNG_ORIG, processed=PNG_PROC, is_calibration=0, role="batch", seq=None):
|
|
files = {"original": ("orig.png", original, "image/png")}
|
|
if processed is not None:
|
|
files["processed"] = ("proc.png", processed, "image/png")
|
|
data = {"photo_id": photo_id, "is_calibration": str(is_calibration), "role": role}
|
|
if seq is not None:
|
|
data["seq"] = str(seq)
|
|
return client.post(
|
|
f"/api/v1/slabs/{slab_id}/photos",
|
|
data=data,
|
|
files=files,
|
|
)
|
|
|
|
|
|
def test_upload_and_get_original_processed(photos_env):
|
|
client, _ = photos_env
|
|
r = _multipart(client, "s1", "p1")
|
|
assert r.status_code == 201, r.text
|
|
body = r.json()
|
|
assert body["original_retained"] is True and body["processed_retained"] is True
|
|
|
|
r = client.get("/api/v1/slabs/s1/photos/p1/original")
|
|
assert r.status_code == 200
|
|
assert r.content == PNG_ORIG
|
|
assert r.headers["content-type"] == "image/png"
|
|
|
|
r = client.get("/api/v1/slabs/s1/photos/p1/processed")
|
|
assert r.status_code == 200
|
|
assert r.content == PNG_PROC
|
|
|
|
|
|
def test_processed_410_when_purged(photos_env):
|
|
client, conn = photos_env
|
|
_multipart(client, "s1", "p2")
|
|
# simulate purge (publish clears blobs)
|
|
conn.execute("UPDATE slab_photos SET original_blob=NULL, processed_blob=NULL WHERE id='p2'")
|
|
conn.commit()
|
|
assert client.get("/api/v1/slabs/s1/photos/p2/original").status_code == 410
|
|
assert client.get("/api/v1/slabs/s1/photos/p2/processed").status_code == 410
|
|
|
|
|
|
def test_upload_requires_slab(photos_env):
|
|
client, _ = photos_env
|
|
r = _multipart(client, "nope", "p1")
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_reupload_replaces(photos_env):
|
|
client, _ = photos_env
|
|
_multipart(client, "s1", "p3", original=PNG_ORIG, processed=PNG_PROC)
|
|
r = _multipart(client, "s1", "p3", original=b"NEW-ORIG", processed=None)
|
|
assert r.status_code == 201
|
|
assert r.json()["processed_retained"] is False
|
|
assert client.get("/api/v1/slabs/s1/photos/p3/original").content == b"NEW-ORIG"
|
|
assert client.get("/api/v1/slabs/s1/photos/p3/processed").status_code == 410
|
|
|
|
|
|
def test_list_photos(photos_env):
|
|
client, _ = photos_env
|
|
_multipart(client, "s1", "pa", is_calibration=1, role="calibration")
|
|
_multipart(client, "s1", "pb", role="overlay")
|
|
r = client.get("/api/v1/slabs/s1/photos")
|
|
assert r.status_code == 200
|
|
ids = [row["id"] for row in r.json()]
|
|
assert "pa" in ids and "pb" in ids
|
|
|
|
|
|
def test_upload_stores_and_returns_role(photos_env):
|
|
client, _ = photos_env
|
|
r = _multipart(client, "s1", "r1", role="overlay")
|
|
assert r.status_code == 201
|
|
assert r.json()["role"] == "overlay"
|
|
r = client.get("/api/v1/slabs/s1/photos")
|
|
row = {row_["id"]: row_ for row_ in r.json()}["r1"]
|
|
assert row["role"] == "overlay"
|
|
|
|
|
|
def test_upload_rejects_bad_role(photos_env):
|
|
client, _ = photos_env
|
|
r = _multipart(client, "s1", "r2", role="not-a-role")
|
|
assert r.status_code == 400
|
|
assert r.json()["error"]["code"] == "validation"
|
|
|
|
|
|
def test_delete_photo(photos_env):
|
|
client, _ = photos_env
|
|
_multipart(client, "s1", "d1")
|
|
assert client.get("/api/v1/slabs/s1/photos/d1/processed").status_code == 200
|
|
r = client.delete("/api/v1/slabs/s1/photos/d1")
|
|
assert r.status_code == 204
|
|
# Gone from the list and blob no longer served.
|
|
ids = [row["id"] for row in client.get("/api/v1/slabs/s1/photos").json()]
|
|
assert "d1" not in ids
|
|
assert client.get("/api/v1/slabs/s1/photos/d1/processed").status_code == 404
|
|
|
|
|
|
def test_delete_photo_not_found(photos_env):
|
|
client, _ = photos_env
|
|
assert client.delete("/api/v1/slabs/s1/photos/ghost").status_code == 404
|