Ty
49c22c9385
Clean re-implementation of the full product (not a port of the old M3 tree): - P0 foundation: 13-table schema + per-call sqlite factory (thread-safe), FastAPI app (health, JSON logging + secret redaction, CORS, startup guard). - P1 M1 Capture: mask/sheetKeep, geometry/rect (min-area rect), measure, png crop/inventory/overlay, EXIF decodeOrientedBitmap. - P2 M2 Photos: stateful capture session (batch, pick-cal, apply mask/crop to batch) + photos-ready lineup. - P3 M3 Create: real backend (crypto, woo client, taxonomy sync, pricing, inference Call1/Call2, review, publish, settings, slabs, photos) + real FE review/publish/settings screens consuming the API (no hardcoded Woo ids). - P4 gate exit: check_requirements + check_coverage both exit 0 (137/155 IDs, 0 gaps); backend 130/130, frontend 150/150, build green. Cite: docs/REQUIREMENTS.md (155 rows), docs/MILESTONES.md M1-M3.
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(monkeypatch):
|
|
tmp = tempfile.mkdtemp(prefix="photos_")
|
|
monkeypatch.setenv("SLAB_DB_PATH", os.path.join(tmp, "p.db"))
|
|
monkeypatch.setenv("SLAB_IMAGES_DIR", tmp)
|
|
monkeypatch.setenv("WOO_BASE_URL", "https://example.com")
|
|
monkeypatch.setenv("SLAB_AES_KEY", base64.b64encode(b"0123456789abcdef0123456789abcdef").decode())
|
|
from backend.app import app
|
|
|
|
db_mod.init_db()
|
|
|
|
conn = db_mod.get_connection()
|
|
# 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
|
|
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
|