tests/test_gallery.py

"""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",
        )