tests/test_http.py

"""HTTP-layer tests (app/main.py) via FastAPI TestClient.

Covers health/UI, upload validation errors, the honest missing-secret path,
the happy path with a mocked LLM, artifact serving, rate limiting and
concurrency.
"""

from __future__ import annotations

import asyncio
import dataclasses
import json
import time
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from app import main as main_module
from app.config import Settings
from app.gallery import GalleryStore
from app.pipeline import JobRegistry
from app.ratelimit import RateLimiter
from tests.conftest import MockLLM

FIXTURES = Path(__file__).resolve().parent / "fixtures"

ALL_LLM_VARS = (
    "LLM_API_KEY", "LLM_MODEL", "LLM_BASE_URL", "LLM_API_STYLE",
    "ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_BASE_URL",
    "ANTHROPIC_MODEL",
)


@pytest.fixture()
def env(tmp_path, monkeypatch, canned_spec_text):
    """Isolate the app: scrubbed env, tmp registry, mock LLM factory."""
    for var in ALL_LLM_VARS:
        monkeypatch.delenv(var, raising=False)
    monkeypatch.setenv("LLM_API_KEY", "test-key-not-real")
    monkeypatch.setenv("LLM_MODEL", "test-model")
    monkeypatch.setenv("LLM_BASE_URL", "https://llm.test/api")
    monkeypatch.setenv("RUNS_DIR", str(tmp_path / "runs"))
    monkeypatch.setenv("GALLERY_DIR", str(tmp_path / "gallery"))

    configured = Settings()
    monkeypatch.delenv("LLM_API_KEY")
    monkeypatch.delenv("LLM_MODEL")
    unconfigured = Settings()

    registry = JobRegistry(tmp_path / "runs")
    monkeypatch.setattr(main_module, "registry", registry)
    monkeypatch.setattr(
        main_module, "gallery_store", GalleryStore(tmp_path / "gallery")
    )
    monkeypatch.setattr(main_module, "limiter", RateLimiter(100))
    monkeypatch.setattr(main_module, "settings", configured)
    main_module.app.state.llm_factory = lambda s: MockLLM([canned_spec_text])

    with TestClient(main_module.app) as client:
        yield client, registry, configured, unconfigured

    main_module.app.state.llm_factory = lambda s: main_module.LLMClient(s)


def wait_terminal(client: TestClient, job_id: str, timeout: float = 90.0) -> dict:
    deadline = time.time() + timeout
    while time.time() < deadline:
        payload = client.get(f"/api/jobs/{job_id}").json()
        if payload["status"] in {"done", "error"}:
            return payload
        time.sleep(0.4)
    raise AssertionError(f"job {job_id} did not finish in {timeout}s")


def post_png(client: TestClient, data: bytes | None = None, **kwargs):
    payload = data if data is not None else (FIXTURES / "ref.png").read_bytes()
    resp = client.post("/api/jobs",
                       files={"file": ("ref.png", payload, "image/png")}, **kwargs)
    if resp.status_code not in (202, 429):
        print("\nPOST /api/jobs ->", resp.status_code, resp.text[:400])
    return resp


class TestBasics:
    def test_health(self, env):
        client, *_ = env
        payload = client.get("/health").json()
        assert payload["status"] == "ok"
        assert "llm_configured" in payload

    def test_index_served(self, env):
        client, *_ = env
        resp = client.get("/")
        assert resp.status_code == 200
        assert "img2threejs" in resp.text
        assert "viewer-frame" in resp.text

    def test_static_assets(self, env):
        client, *_ = env
        for asset in ("app.js", "styles.css", "viewer.html", "viewer-core.js", "logo.svg"):
            assert client.get(f"/static/{asset}").status_code == 200, asset

    def test_config_never_leaks_secrets(self, env):
        client, *_ = env
        resp = client.get("/api/config")
        assert resp.status_code == 200
        assert resp.json()["llm_configured"] is True
        assert "test-key-not-real" not in resp.text

    def test_security_headers(self, env):
        client, *_ = env
        resp = client.get("/")
        assert "content-security-policy" in resp.headers
        assert resp.headers["x-content-type-options"] == "nosniff"


class TestHonestFailures:
    def test_missing_secret_503_no_fabrication(self, env, monkeypatch):
        client, _, _, unconfigured = env
        monkeypatch.setattr(main_module, "settings", unconfigured)
        resp = post_png(client)
        assert resp.status_code == 503
        payload = resp.json()
        assert payload["error"] == "llm_not_configured"
        assert "LLM_API_KEY" in payload["detail"]
        assert "LLM_MODEL" in payload["detail"]
        # Never fabricate: no artifact references whatsoever.
        assert "sculptRuntime" not in resp.text
        assert "bundle" not in resp.text

    def test_no_file_422(self, env):
        client, *_ = env
        assert client.post("/api/jobs").status_code == 422

    def test_non_image_content_type_415(self, env):
        client, *_ = env
        resp = client.post("/api/jobs",
                           files={"file": ("x.html", b"<html></html>", "text/html")})
        assert resp.status_code == 415

    def test_oversize_413(self, env, monkeypatch):
        client, *_ = env
        big = b"\x89PNG\r\n\x1a\n" + b"0" * (10 * 1024 * 1024 + 2)
        resp = client.post("/api/jobs",
                           files={"file": ("big.png", big, "image/png")})
        assert resp.status_code == 413

    def test_asgi_content_length_cap_rejects_before_multipart(self, env, monkeypatch):
        client, _, configured, _ = env
        bounded = dataclasses.replace(configured, max_upload_bytes=64 * 1024)
        monkeypatch.setattr(main_module, "settings", bounded)
        body = b"x" * (
            bounded.max_upload_bytes + main_module.MULTIPART_OVERHEAD_BYTES + 1
        )
        resp = client.post(
            "/api/jobs", content=body,
            headers={"content-type": "multipart/form-data; boundary=x"},
        )
        assert resp.status_code == 413
        assert resp.json()["error"] == "request_too_large"
        # Body-limit responses still pass through the outer security policy.
        assert "content-security-policy" in resp.headers

    def test_asgi_receive_cap_without_content_length(self, env, monkeypatch):
        _, _, configured, _ = env
        bounded = dataclasses.replace(configured, max_upload_bytes=64 * 1024)
        monkeypatch.setattr(main_module, "settings", bounded)
        limit = bounded.max_upload_bytes + main_module.MULTIPART_OVERHEAD_BYTES
        chunks = [
            {"type": "http.request", "body": b"a" * (limit // 2), "more_body": True},
            {"type": "http.request", "body": b"b" * (limit // 2 + 1), "more_body": False},
        ]
        sent: list[dict] = []

        async def drain_app(scope, receive, send):
            while True:
                message = await receive()
                if not message.get("more_body"):
                    return

        async def receive():
            return chunks.pop(0)

        async def send(message):
            sent.append(message)

        scope = {
            "type": "http", "asgi": {"version": "3.0"},
            "http_version": "1.1", "method": "POST", "scheme": "http",
            "path": "/api/jobs", "raw_path": b"/api/jobs", "query_string": b"",
            "headers": [], "client": ("test-peer", 1), "server": ("test", 80),
        }
        middleware = main_module.UploadBodyLimitMiddleware(drain_app)
        asyncio.run(middleware(scope, receive, send))
        start = next(message for message in sent if message["type"] == "http.response.start")
        assert start["status"] == 413

    def test_corrupt_image_fails_honestly(self, env):
        client, *_ = env
        resp = client.post(
            "/api/jobs",
            files={"file": ("bad.png", b"\x89PNG\r\n\x1a\ntruncated", "image/png")})
        assert resp.status_code == 202
        terminal = wait_terminal(client, resp.json()["job_id"])
        assert terminal["status"] == "error"
        assert terminal["error"]["stage"] == "intake"

    def test_unknown_job_404(self, env):
        client, *_ = env
        assert client.get("/api/jobs/deadbeefdeadbeef").status_code == 404
        assert client.get("/api/jobs/deadbeefdeadbeef/events").status_code == 404
        assert client.get("/api/jobs/deadbeefdeadbeef/artifacts/spec.json").status_code == 404
        assert client.get("/api/jobs/../../etc/passwd").status_code in {404, 422}

    def test_artifact_name_whitelist(self, env):
        client, registry, *_ = env
        job = registry.create()
        (job.dir / "evil.sh").write_text("echo nope")
        assert client.get(f"/api/jobs/{job.id}/artifacts/evil.sh").status_code == 404

    def test_failed_job_cannot_serve_late_model_artifacts(self, env):
        client, registry, *_ = env
        job = registry.create()
        (job.dir / "spec.json").write_text(
            '{"targetName":"Diagnostic"}', encoding="utf-8"
        )
        (job.dir / "factory.ts").write_text(
            "export const late = true;", encoding="utf-8"
        )
        (job.dir / "model.bundle.js").write_text(
            "export const late = true;", encoding="utf-8"
        )
        job.status = "error"

        assert client.get(
            f"/api/jobs/{job.id}/artifacts/spec.json"
        ).status_code == 200
        assert client.get(
            f"/api/jobs/{job.id}/artifacts/factory.ts"
        ).status_code == 404
        assert client.get(
            f"/api/jobs/{job.id}/artifacts/model.bundle.js"
        ).status_code == 404


class TestHappyPath:
    def test_full_job_and_artifacts(self, env):
        client, *_ = env
        resp = post_png(client, data=(FIXTURES / "ref.png").read_bytes())
        assert resp.status_code == 202
        job_id = resp.json()["job_id"]

        terminal = wait_terminal(client, job_id)
        assert terminal["status"] == "done", terminal.get("error")
        status_response = client.get(f"/api/jobs/{job_id}")
        assert status_response.headers["cache-control"] == "no-store"
        result = terminal["result"]
        assert result["targetName"] == "Mug"
        assert result["shareRequested"] is True
        assert result["shared"] is True
        assert result["galleryItem"]["id"] == job_id
        gallery_detail = client.get(result["galleryItem"]["detailUrl"])
        assert gallery_detail.status_code == 200
        assert gallery_detail.json()["targetName"] == "Mug"

        factory = client.get(f"/api/jobs/{job_id}/artifacts/factory.ts")
        assert factory.status_code == 200
        assert factory.headers["cache-control"] == "no-store"
        assert "createMugModel" in factory.text
        assert "sculptRuntime" in factory.text

        spec = client.get(f"/api/jobs/{job_id}/artifacts/spec.json")
        assert spec.status_code == 200
        assert spec.json()["targetName"] == "Mug"

        bundle = client.get(f"/api/jobs/{job_id}/artifacts/model.bundle.js")
        assert bundle.status_code == 200
        assert len(bundle.content) > 100_000

        standalone = client.get(f"/api/jobs/{job_id}/artifacts/standalone.html")
        assert standalone.status_code == 200
        assert "mountViewer" in standalone.text

        probe = client.get(f"/api/jobs/{job_id}/artifacts/probe.json")
        assert probe.json()["width"] == 64

        reference = client.get(f"/api/jobs/{job_id}/artifacts/reference.png")
        assert reference.status_code == 200
        assert reference.content[:8] == b"\x89PNG\r\n\x1a\n"

    def test_explicit_share_false_never_publishes(self, env):
        client, *_ = env
        response = client.post(
            "/api/jobs",
            files={
                "file": (
                    "ref.png",
                    (FIXTURES / "ref.png").read_bytes(),
                    "image/png",
                )
            },
            data={"share": "false"},
        )
        assert response.status_code == 202
        assert response.json()["share_requested"] is False
        terminal = wait_terminal(client, response.json()["job_id"])
        assert terminal["status"] == "done", terminal.get("error")
        assert terminal["result"]["shareRequested"] is False
        assert terminal["result"]["shared"] is False
        assert terminal["result"]["galleryItem"] is None
        assert client.get("/api/gallery").json()["items"] == []
        assert any(
            event["stage"] == "publishing"
            and event["status"] == "done"
            and event["data"]["shared"] is False
            for event in main_module.registry.get(response.json()["job_id"]).events
        )

    def test_sse_replay_terminates(self, env):
        client, *_ = env
        job_id = post_png(client).json()["job_id"]
        terminal = wait_terminal(client, job_id)
        assert terminal["status"] == "done"
        resp = client.get(f"/api/jobs/{job_id}/events")
        assert resp.status_code == 200
        assert "text/event-stream" in resp.headers["content-type"]
        assert resp.headers["cache-control"] == "no-store"
        lines = [l[5:] for l in resp.text.splitlines() if l.startswith("data:")]
        events = [json.loads(l) for l in lines]
        assert events[0]["stage"] == "queued"
        assert events[1]["stage"] == "queued" and events[1]["status"] == "done"
        assert "id: 1" in resp.text
        assert events[-1]["stage"] == "done" and events[-1]["status"] == "done"
        assert events[-1]["data"]["result"]["targetName"] == "Mug"

        resumed = client.get(
            f"/api/jobs/{job_id}/events",
            headers={"Last-Event-ID": str(events[-2]["seq"])},
        )
        resumed_events = [
            json.loads(line[5:])
            for line in resumed.text.splitlines()
            if line.startswith("data:")
        ]
        assert [event["seq"] for event in resumed_events] == [events[-1]["seq"]]

    def test_concurrent_http_jobs(self, env):
        client, *_ = env
        ids = [post_png(client).json()["job_id"] for _ in range(3)]
        terminals = [wait_terminal(client, jid) for jid in ids]
        assert {t["status"] for t in terminals} == {"done"}
        assert len(set(ids)) == 3

    def test_queue_wait_reports_elapsed_then_honestly_times_out(
        self, env, monkeypatch
    ):
        client, registry, configured, _ = env
        monkeypatch.setattr(main_module, "job_semaphore", asyncio.Semaphore(0))
        monkeypatch.setattr(main_module, "QUEUE_PROGRESS_INTERVAL_S", 0.02)
        monkeypatch.setattr(
            main_module,
            "settings",
            dataclasses.replace(configured, job_timeout_s=0.08),
        )

        response = post_png(client)
        assert response.status_code == 202
        terminal = wait_terminal(client, response.json()["job_id"], timeout=5)
        assert terminal["status"] == "error"
        assert terminal["error"]["code"] == "job_timeout"
        assert terminal["error"]["stage"] == "queued"
        events = registry.get(response.json()["job_id"]).events
        assert any(
            event["stage"] == "queued"
            and event["status"] == "progress"
            and event["data"]["elapsedSeconds"] >= 1
            for event in events
        )


class TestRateLimit:
    def test_limit_enforced(self, env, monkeypatch):
        client, *_ = env
        monkeypatch.setattr(main_module, "limiter", RateLimiter(2))
        assert post_png(client).status_code == 202
        assert post_png(client).status_code == 202
        resp = post_png(client)
        assert resp.status_code == 429
        assert "retry-after" in resp.headers