test_app.py

"""Checks for the properties that make link-only sharing actually hold."""

import json
import os
import tempfile

os.environ["TRACE_DIR"] = tempfile.mkdtemp()
os.environ["UPLOAD_TOKEN"] = "s3cret"  # exercised below; unset in deployment

import pytest  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402

import app as app_module  # noqa: E402
import render  # noqa: E402

client = TestClient(app_module.app, raise_server_exceptions=False)
AUTH = {"x-upload-token": "s3cret"}


@pytest.fixture(autouse=True)
def fresh_state():
    """Every test looks like a new client; otherwise the shared bucket from the
    TestClient's single IP starts returning 429 partway through the suite.

    UPLOAD_TOKEN is pinned here rather than inherited from import order — the
    app module is shared with test_auth.py, which turns the token off.
    """
    app_module._hits.clear()
    app_module.UPLOAD_TOKEN = "s3cret"
    client.cookies.clear()
    yield
    app_module._hits.clear()


def create(trace, title="t"):
    response = client.post("/api/traces", json={"title": title, "trace": trace}, headers=AUTH)
    assert response.status_code == 201, response.text
    return response.json()


def test_ids_are_128_bit_and_unique():
    ids = {app_module.new_id() for _ in range(2000)}
    assert len(ids) == 2000
    assert all(app_module.ID_RE.match(i) for i in ids)
    assert app_module.ID_BYTES * 8 == 128
    # and the id the API hands back has the same shape
    assert app_module.ID_RE.match(create("hello")["id"])


def test_roundtrip_and_raw():
    messages = [
        {"role": "user", "content": "what's 2+2"},
        {"role": "assistant", "content": [{"type": "text", "text": "4"}]},
    ]
    created = create({"messages": messages}, title="math")
    page = client.get(f"/t/{created['id']}")
    assert page.status_code == 200
    assert "what's 2+2" in page.text or "what's 2+2" in page.text
    assert "math" in page.text

    raw = client.get(f"/t/{created['id']}/raw")
    assert raw.json()["messages"] == messages


def test_no_listing_or_index_leak():
    created = create("secret trace")
    home = client.get("/")
    assert created["id"] not in home.text
    # No route enumerates traces.
    for path in ("/traces", "/api/traces/list", "/t/", "/t"):
        assert client.get(path).status_code in (404, 405)


def test_unknown_and_malformed_ids_are_indistinguishable():
    unknown = client.get("/t/" + "A" * 22)          # well-formed, absent
    malformed = client.get("/t/short")               # wrong shape
    assert unknown.status_code == malformed.status_code == 404
    assert unknown.json() == malformed.json()


def test_security_headers_present():
    created = create("x")
    headers = client.get(f"/t/{created['id']}").headers
    assert headers["referrer-policy"] == "no-referrer"
    assert "noindex" in headers["x-robots-tag"]
    assert "default-src 'none'" in headers["content-security-policy"]
    assert headers["cache-control"] == "private, no-store"


def test_robots_disallows_everything():
    assert "Disallow: /" in client.get("/robots.txt").text


def test_uploads_are_open_when_no_token_configured(monkeypatch):
    """The deployed configuration: anyone may upload."""
    monkeypatch.setattr(app_module, "UPLOAD_TOKEN", "")
    assert client.post("/api/traces", json={"trace": "anyone"}).status_code == 201
    assert client.post("/", data={"content": "anyone"}).status_code == 201


def test_upload_requires_token():
    assert client.post("/api/traces", json={"trace": "x"}).status_code == 401
    assert client.post("/api/traces", json={"trace": "x"},
                       headers={"x-upload-token": "wrong"}).status_code == 401
    assert client.post("/api/traces", json={"trace": "x"},
                       headers={"authorization": "Bearer s3cret"}).status_code == 201


def test_oversized_trace_rejected():
    original = app_module.MAX_BYTES
    app_module.MAX_BYTES = 500
    try:
        response = client.post("/api/traces", json={"trace": "x" * 2000}, headers=AUTH)
        assert response.status_code == 413
    finally:
        app_module.MAX_BYTES = original


def test_html_is_escaped():
    created = create([{"role": "user", "content": "<script>alert(1)</script>"}])
    page = client.get(f"/t/{created['id']}")
    assert "<script>alert(1)</script>" not in page.text
    assert "&lt;script&gt;" in page.text


def test_read_rate_limit_blocks_scanning():
    app_module._hits.clear()
    limit, _ = app_module.LIMITS["read"]
    codes = {client.get("/t/" + "B" * 22).status_code for _ in range(limit + 10)}
    assert 429 in codes
    app_module._hits.clear()


def test_render_handles_dialects():
    anthropic = render.normalize({"messages": [
        {"role": "assistant", "content": [
            {"type": "thinking", "thinking": "hmm"},
            {"type": "tool_use", "name": "bash", "input": {"cmd": "ls"}},
        ]},
        {"role": "user", "content": [
            {"type": "tool_result", "tool_use_id": "1", "content": "a.txt", "is_error": False}
        ]},
    ]})
    assert anthropic["kind"] == "messages"
    kinds = [b["type"] for m in anthropic["messages"] for b in m["blocks"]]
    assert kinds == ["thinking", "tool_use", "tool_result"]
    # a user turn carrying only tool results is displayed as the tool speaking
    assert anthropic["messages"][1]["role"] == "tool"

    openai = render.normalize([{
        "role": "assistant",
        "content": None,
        "tool_calls": [{"function": {"name": "search", "arguments": '{"q": "x"}'}}],
    }])
    assert openai["messages"][0]["blocks"][0]["name"] == "search"

    langchain = render.normalize({"messages": [{"type": "human", "content": "hi"}]})
    assert langchain["messages"][0]["role"] == "user"

    assert render.normalize("just text")["kind"] == "text"
    assert render.normalize({"nested": {"a": 1}})["kind"] == "json"


def test_remote_images_are_not_rendered():
    view = render.normalize([{
        "role": "user",
        "content": [{"type": "image_url", "image_url": {"url": "https://evil.test/p.png"}}],
    }])
    block = view["messages"][0]["blocks"][0]
    assert block["type"] == "note"
    assert "evil.test" not in json.dumps(view)


def test_form_post_creates_trace():
    app_module._hits.clear()
    response = client.post("/", data={"content": "hello", "title": "via form", "token": "s3cret"})
    assert response.status_code == 201
    assert "/t/" in response.text