"""Focused persistence, safety, API, and progress tests for the gallery.""" from __future__ import annotations import asyncio from concurrent.futures import ThreadPoolExecutor from pathlib import Path import pytest from fastapi.testclient import TestClient from app import main as main_module from app.gallery import ( GALLERY_ARTIFACT_NAMES, GalleryError, GalleryStore, REQUIRED_GALLERY_ARTIFACTS, ) from app.pipeline import JobRegistry, _complete_vision_with_feedback def _fake_job(root: Path, item_id: str, marker: str = "one") -> Path: job_dir = root / item_id job_dir.mkdir(parents=True) for name in GALLERY_ARTIFACT_NAMES: (job_dir / name).write_bytes(f"{marker}:{name}".encode()) return job_dir def _result(name: str = "Test Object") -> dict: return { "targetName": name, "components": 7, "materials": 3, "elapsedSeconds": 12.5, "generationMode": "hosted-unreviewed-preview", "generatedPass": "hosted-preview", "reviewStatus": "unreviewed", } def test_atomic_publish_is_restart_readable_and_independent_of_job(tmp_path): item_id = "a" * 32 source = _fake_job(tmp_path / "runs", item_id) store = GalleryStore(tmp_path / "gallery") item = store.publish( item_id=item_id, job_dir=source, result=_result("Persistent Mug"), created_at=100.25, ) assert item["id"] == item_id assert item["targetName"] == "Persistent Mug" assert item["thumbnailUrl"].endswith("/reference.png") assert item["stats"] == { "components": 7, "materials": 3, "elapsedSeconds": 12.5, } assert set(item["artifacts"]) == set(GALLERY_ARTIFACT_NAMES) # Gallery copies do not depend on temporary run retention. for child in source.iterdir(): child.unlink() source.rmdir() restarted = GalleryStore(tmp_path / "gallery") detail = restarted.get(item_id) assert detail == item assert restarted.artifact_path(item_id, "factory.ts").read_bytes().startswith(b"one:") assert restarted.list_items()["items"] == [item] def test_newest_first_pagination_and_concurrent_commits(tmp_path): store = GalleryStore(tmp_path / "gallery") ids = [f"{number:032x}" for number in range(1, 7)] sources = { item_id: _fake_job(tmp_path / "runs", item_id, item_id) for item_id in ids } def publish(index: int): item_id = ids[index] return store.publish( item_id=item_id, job_dir=sources[item_id], result=_result(f"Object {index}"), created_at=100.0 + index, ) with ThreadPoolExecutor(max_workers=6) as pool: list(pool.map(publish, range(len(ids)))) page = store.list_items(offset=1, limit=2) assert page["total"] == 6 assert page["hasMore"] is True assert [item["id"] for item in page["items"]] == [ids[4], ids[3]] # Hidden staging debris and malformed complete-looking directories never # leak into a listing. (tmp_path / "gallery" / ".publishing-debris").mkdir() corrupt = tmp_path / "gallery" / ("f" * 32) corrupt.mkdir() (corrupt / "item.json").write_text("{}") assert store.list_items()["total"] == 6 def test_failed_or_unsafe_publish_leaves_no_visible_item(tmp_path): item_id = "b" * 32 source = _fake_job(tmp_path / "runs", item_id) (source / "standalone.html").unlink() store = GalleryStore(tmp_path / "gallery") with pytest.raises(GalleryError, match="missing required"): store.publish(item_id=item_id, job_dir=source, result=_result()) assert store.get(item_id) is None assert store.list_items()["items"] == [] assert not (tmp_path / "gallery" / item_id).exists() assert not list((tmp_path / "gallery").glob(".publishing-*")) # Strict ids and artifact names/path containment are enforced at reads. assert store.get("../" + item_id) is None assert store.artifact_path(item_id, "../../etc/passwd") is None def test_symlinked_required_artifact_outside_job_is_refused(tmp_path): item_id = "c" * 32 source = _fake_job(tmp_path / "runs", item_id) (source / "factory.ts").unlink() outside = tmp_path / "outside.ts" outside.write_text("not a job artifact") (source / "factory.ts").symlink_to(outside) store = GalleryStore(tmp_path / "gallery") with pytest.raises(GalleryError): store.publish(item_id=item_id, job_dir=source, result=_result()) assert store.get(item_id) is None def test_gallery_http_api_and_strict_pagination(tmp_path, monkeypatch): item_id = "d" * 32 store = GalleryStore(tmp_path / "gallery") store.publish( item_id=item_id, job_dir=_fake_job(tmp_path / "runs", item_id), result=_result("API Mug"), created_at=200, ) monkeypatch.setattr(main_module, "gallery_store", store) with TestClient(main_module.app) as client: listing = client.get("/api/gallery?offset=0&limit=1") assert listing.status_code == 200 assert listing.json()["items"][0]["id"] == item_id detail = client.get(f"/api/gallery/{item_id}") assert detail.status_code == 200 assert detail.json()["targetName"] == "API Mug" artifact = client.get(f"/api/gallery/{item_id}/artifacts/model.bundle.js") assert artifact.status_code == 200 assert artifact.headers["cache-control"].endswith("immutable") assert client.get("/api/gallery?offset=-1").status_code == 422 assert client.get("/api/gallery?limit=101").status_code == 422 assert client.get("/api/gallery/not-an-id").status_code == 404 assert client.get( f"/api/gallery/{item_id}/artifacts/item.json" ).status_code == 404 assert client.get("/gallery").status_code == 200 assert client.get(f"/gallery/{item_id}").status_code == 200 assert client.get("/gallery/not-an-id").status_code == 404 @pytest.mark.parametrize( ("value", "expected"), [ (None, True), ("true", True), ("ON", True), ("1", True), ("false", False), ("Off", False), ("0", False), ("", None), ("maybe", None), ], ) def test_share_multipart_parser(value, expected): assert main_module._parse_share_preference(value) is expected @pytest.mark.asyncio async def test_slow_llm_emits_truthful_bounded_progress(tmp_path, monkeypatch): class DelayedLLM: async def complete_vision(self, *, system, messages): await asyncio.sleep(0.055) return "finished" monkeypatch.setattr("app.pipeline.LLM_PROGRESS_INTERVAL_S", 0.01) job = JobRegistry(tmp_path / "runs").create() result = await _complete_vision_with_feedback( job, llm=DelayedLLM(), system="system", messages=[], attempt=2, max_attempts=4, ) assert result == "finished" progress = [ event for event in job.events if event["stage"] == "spec-authoring" and event["status"] == "progress" ] assert len(progress) >= 2 assert all(event["data"]["attempt"] == 2 for event in progress) assert all(event["data"]["maxAttempts"] == 4 for event in progress) assert all("elapsedSeconds" in event["data"] for event in progress) assert "%" not in " ".join(event["message"] for event in progress) def test_job_event_waiter_stays_set_until_a_stream_consumes_it(tmp_path): job = JobRegistry(tmp_path / "runs").create() job.waiter.clear() job.emit("generation", "progress", "Factory still running.") assert job.waiter.is_set() def test_required_gallery_artifact_set_is_served_allowlist_subset(): assert REQUIRED_GALLERY_ARTIFACTS <= set(GALLERY_ARTIFACT_NAMES) def test_publish_rejects_untrusted_staging_token(tmp_path): store = GalleryStore(tmp_path / "gallery") job_dir = tmp_path / "job" job_dir.mkdir() with pytest.raises(GalleryError, match="staging token"): store.publish( item_id="a" * 32, job_dir=job_dir, result={"targetName": "Object"}, staging_token="../../escape", )