US01-07: Automate Phase A End-to-End Acceptance (#53)
This commit was merged in pull request #53.
This commit is contained in:
219
tests/e2e/test_phase_a_pipeline.py
Normal file
219
tests/e2e/test_phase_a_pipeline.py
Normal file
@@ -0,0 +1,219 @@
|
||||
"""Phase A end-to-end acceptance gate.
|
||||
|
||||
Drives the whole Phase A journey against a real API process over HTTP, plus one
|
||||
browser reload check, then restarts the process and asserts durable API + database
|
||||
state survived. Everything is offline: deterministic synthetic fixtures, no network.
|
||||
|
||||
Note: Phase A has no separate worker yet (durable jobs arrive in Phase B/US02); the
|
||||
harness launches the real API process, which is the production boundary for this
|
||||
phase. Scan and detection run synchronously via their endpoints.
|
||||
"""
|
||||
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
import numpy as np
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def _structured(path, seed, size=(320, 240)):
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
rng = np.random.default_rng(seed)
|
||||
w, h = size
|
||||
base = np.zeros((h, w, 3), dtype=np.uint8)
|
||||
for _ in range(6):
|
||||
x0, y0 = int(rng.integers(0, w - 60)), int(rng.integers(0, h - 60))
|
||||
base[y0 : y0 + 60, x0 : x0 + 60] = rng.integers(0, 256, 3)
|
||||
grad = np.linspace(0, 120, w, dtype=np.uint8)
|
||||
base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255)
|
||||
Image.fromarray(base).save(path, quality=95)
|
||||
return path
|
||||
|
||||
|
||||
def _resized(src, dst, scale=0.5):
|
||||
with Image.open(src) as image:
|
||||
image.resize(
|
||||
(int(image.width * scale), int(image.height * scale)), Image.LANCZOS
|
||||
).save(dst, quality=95)
|
||||
|
||||
|
||||
def _oriented(path, w, h, orientation=6, seed=7):
|
||||
arr = np.random.default_rng(seed).integers(0, 256, (h, w, 3), dtype=np.uint8)
|
||||
img = Image.fromarray(arr)
|
||||
exif = img.getexif()
|
||||
exif[274] = orientation
|
||||
img.save(path, exif=exif, quality=95)
|
||||
|
||||
|
||||
def _free_port():
|
||||
with socket.socket() as sock:
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
return sock.getsockname()[1]
|
||||
|
||||
|
||||
class ServerController:
|
||||
def __init__(self, data, lib):
|
||||
self.data = data
|
||||
self.lib = lib
|
||||
self.proc = None
|
||||
self.base = None
|
||||
self.client = None
|
||||
|
||||
def start(self):
|
||||
port = _free_port()
|
||||
env = {
|
||||
"PATH": __import__("os").environ.get("PATH", ""),
|
||||
"PHOTO_PIPELINE_DATA_DIR": str(self.data),
|
||||
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(self.lib),
|
||||
"PHOTO_PIPELINE_HOST": "127.0.0.1",
|
||||
"PHOTO_PIPELINE_PORT": str(port),
|
||||
}
|
||||
self.proc = subprocess.Popen(
|
||||
[sys.executable, "-m", "photo_pipeline", "serve"],
|
||||
cwd=str(REPO),
|
||||
env=env,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
self.base = f"http://127.0.0.1:{port}"
|
||||
deadline = time.monotonic() + 30
|
||||
while time.monotonic() < deadline:
|
||||
if self.proc.poll() is not None:
|
||||
_, err = self.proc.communicate()
|
||||
pytest.fail(f"server exited: {err.decode(errors='replace')}")
|
||||
try:
|
||||
if httpx.get(f"{self.base}/api/v1/health/ready", timeout=1).status_code == 200:
|
||||
self.client = httpx.Client(base_url=self.base, timeout=10)
|
||||
return
|
||||
except httpx.HTTPError:
|
||||
time.sleep(0.2)
|
||||
self.stop()
|
||||
pytest.fail("server never became ready")
|
||||
|
||||
def restart(self):
|
||||
self.stop()
|
||||
self.start()
|
||||
|
||||
def stop(self):
|
||||
if self.client:
|
||||
self.client.close()
|
||||
self.client = None
|
||||
if self.proc:
|
||||
self.proc.terminate()
|
||||
try:
|
||||
self.proc.wait(timeout=10)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.proc.kill()
|
||||
self.proc = None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pipeline(tmp_path):
|
||||
data = tmp_path / "data"
|
||||
data.mkdir()
|
||||
lib = tmp_path / "lib"
|
||||
lib.mkdir()
|
||||
|
||||
# Deterministic corpus exercising every Phase A journey.
|
||||
photo = _structured(lib / "album" / "photo.jpg", 1)
|
||||
shutil.copy2(photo, lib / "album" / "photo_copy.jpg") # exact duplicate
|
||||
scene = _structured(lib / "album" / "scene.jpg", 2)
|
||||
_resized(scene, lib / "album" / "scene_small.jpg") # perceptual near-duplicate
|
||||
_oriented(lib / "rotated.jpg", 800, 480, orientation=6) # orientation fixture
|
||||
_structured(lib / "movable.jpg", 5) # moved between scans
|
||||
_structured(lib / "_IGNORE" / "secret.jpg", 9) # exclusion sentinel
|
||||
|
||||
controller = ServerController(data, lib)
|
||||
controller.start()
|
||||
yield controller
|
||||
controller.stop()
|
||||
|
||||
|
||||
def _assets(client):
|
||||
return client.get("/api/v1/inventory/assets?limit=200").json()["items"]
|
||||
|
||||
|
||||
def _id_for(assets, suffix):
|
||||
return next(a["id"] for a in assets if (a["current_path"] or "").endswith(suffix))
|
||||
|
||||
|
||||
def test_phase_a_full_pipeline_and_restart(page, pipeline):
|
||||
client = pipeline.client
|
||||
|
||||
# 1. Scan + detect through the real API.
|
||||
assert client.post("/api/v1/inventory/scan").status_code == 200
|
||||
detect = client.post("/api/v1/duplicates/detect").json()
|
||||
assert detect["clusters"] >= 2
|
||||
|
||||
assets = _assets(client)
|
||||
paths = [a["current_path"] for a in assets]
|
||||
|
||||
# 2. Exclusion: the _IGNORE sentinel is never discovered.
|
||||
assert not any("_IGNORE" in p or p.endswith("secret.jpg") for p in paths)
|
||||
# 5 discovered files (photo, photo_copy, scene, scene_small, rotated, movable = 6)
|
||||
assert len(assets) == 6
|
||||
movable_id = _id_for(assets, "movable.jpg")
|
||||
rotated_id = _id_for(assets, "rotated.jpg")
|
||||
|
||||
# 3. Move reconciliation: identity survives a move across a rescan.
|
||||
shutil.move(str(pipeline.lib / "movable.jpg"), str(pipeline.lib / "album" / "moved.jpg"))
|
||||
assert client.post("/api/v1/inventory/scan").status_code == 200
|
||||
assert _id_for(_assets(client), "moved.jpg") == movable_id
|
||||
|
||||
# 4. Exact + fuzzy clusters exist with the right confidence handling.
|
||||
clusters = client.get("/api/v1/duplicates/clusters").json()["items"]
|
||||
exact = next(c for c in clusters if c["method"] == "exact")
|
||||
perceptual = next(c for c in clusters if c["method"] == "perceptual")
|
||||
assert exact["state"] == "decided" # exact auto-resolved
|
||||
assert perceptual["state"] == "open" # fuzzy left for review
|
||||
|
||||
# 5. Thumbnail orientation: a rotated source serves an upright preview.
|
||||
thumb = client.get(f"/api/v1/assets/{rotated_id}/thumbnail?size=256")
|
||||
assert thumb.status_code == 200 and thumb.headers["content-type"] == "image/webp"
|
||||
with Image.open(BytesIO(thumb.content)) as image:
|
||||
assert image.height > image.width # 800x480 landscape shown upright as portrait
|
||||
|
||||
# 6. Canonical selection on the fuzzy cluster persists with variant links.
|
||||
detail = client.get(f"/api/v1/duplicates/clusters/{perceptual['id']}").json()
|
||||
chosen = detail["members"][0]["asset_id"]
|
||||
decided = client.post(
|
||||
f"/api/v1/duplicates/clusters/{perceptual['id']}/decision",
|
||||
json={
|
||||
"decision": "canonical",
|
||||
"expected_version": detail["version"],
|
||||
"canonical_asset_id": chosen,
|
||||
},
|
||||
)
|
||||
assert decided.status_code == 200 and decided.json()["state"] == "decided"
|
||||
variant_ids = {m["asset_id"] for m in detail["members"]} - {chosen}
|
||||
after = {a["id"]: a for a in _assets(client)}
|
||||
assert all(after[v]["canonical_asset_id"] == chosen for v in variant_ids)
|
||||
|
||||
# 7. Reload in the browser: the decision is restored, not re-fetched fresh state.
|
||||
page.goto(f"{pipeline.base}/app/#/duplicates/{perceptual['id']}")
|
||||
page.get_by_test_id("cluster-state").wait_for()
|
||||
page.reload()
|
||||
page.get_by_test_id("cluster-state").wait_for()
|
||||
assert "decided" in page.get_by_test_id("cluster-state").inner_text()
|
||||
|
||||
# 8. Full process restart: durable API + DB state survives.
|
||||
pipeline.restart()
|
||||
client = pipeline.client
|
||||
assert client.get("/api/v1/health/ready").status_code == 200
|
||||
restored = client.get("/api/v1/duplicates/clusters").json()["items"]
|
||||
restored_perceptual = next(c for c in restored if c["id"] == perceptual["id"])
|
||||
assert restored_perceptual["state"] == "decided" # decision durable
|
||||
restored_assets = {a["id"]: a for a in _assets(client)}
|
||||
assert _id_for(list(restored_assets.values()), "moved.jpg") == movable_id # identity durable
|
||||
assert all(restored_assets[v]["canonical_asset_id"] == chosen for v in variant_ids)
|
||||
restored_paths = [a["current_path"] for a in restored_assets.values()]
|
||||
assert not any("_IGNORE" in p for p in restored_paths) # exclusion still holds
|
||||
Reference in New Issue
Block a user