test_http.py

"""HTTP gate tests. These use a mock transport -- the gates are pure logic and
must be testable without the network, even though the ADAPTERS do real HTTP."""

import re
from pathlib import Path

import httpx
import pytest

from promptgen.adapters.base import Signal
from promptgen.adapters.http import (
    HTML_TYPES,
    JSON_TYPES,
    SoftNotFound,
    TooLarge,
    WrongContentType,
    fetch_text,
    try_fetch,
)
from promptgen.store.repo import dedupe


def client_for(handler) -> httpx.AsyncClient:
    return httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=True)


async def test_soft_404_redirect_to_login_is_rejected():
    """vercel.com 307s any garbage path to /auth-redirect then /login, and the
    login page returns 200. Status alone would call that a hit."""
    def handler(request: httpx.Request) -> httpx.Response:
        if "login" in request.url.path:
            return httpx.Response(200, html="<html>sign in</html>")
        return httpx.Response(307, headers={"location": "https://x.test/login?next=%2Fnope"})

    async with client_for(handler) as http:
        with pytest.raises(SoftNotFound):
            await fetch_text(http, "https://x.test/nope", expect=HTML_TYPES)


async def test_content_type_gate_rejects_html_served_as_json():
    """/mint.json returns 200 with 2.4MB of text/html. The type gate rejects it
    from the headers, so the body is never downloaded."""
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, html="<html>marketing page</html>")

    async with client_for(handler) as http:
        with pytest.raises(WrongContentType):
            await fetch_text(http, "https://x.test/mint.json", expect=JSON_TYPES)


async def test_size_cap_counts_decompressed_bytes():
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, html="<html>" + "x" * 5000 + "</html>")

    async with client_for(handler) as http:
        with pytest.raises(TooLarge):
            await fetch_text(http, "https://x.test/", expect=HTML_TYPES, cap=1000)


async def test_try_fetch_returns_none_instead_of_raising():
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(404)

    async with client_for(handler) as http:
        assert await try_fetch(http, "https://x.test/", expect=HTML_TYPES) is None


async def test_happy_path_returns_final_url():
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, html="<html>ok</html>")

    async with client_for(handler) as http:
        got = await fetch_text(http, "https://x.test/pricing", expect=HTML_TYPES)
    assert got.status == 200 and got.final_url == "https://x.test/pricing"


# ------------------------------------------------------------------- store

def test_dedupe_keeps_highest_weight_per_ref():
    """Postgres raises if one ON CONFLICT statement carries a ref twice."""
    def sig(w):
        return Signal(text=f"w{w}", source="docs", kind="capability",
                      ref="docs:page:/a", weight=w)
    rows, collapsed = dedupe([sig(3), sig(9), sig(5)])
    assert len(rows) == 1 and rows[0].weight == 9 and collapsed == 2


# ---------------------------------------------------------------- tripwire

BANNED = re.compile(r"\b(DEMO|is_demo|use_fixtures|IS_FIXTURE|MOCK_MODE)\b")


def _code_only(path: Path) -> str:
    """Source with comments and string literals removed.

    Checking raw text would flag the docstrings that DESCRIBE this rule, and
    would equally let a real flag hide inside a string. Tokenizing tests the
    code itself.
    """
    import io
    import tokenize

    out = []
    with path.open("rb") as fh:
        for tok in tokenize.tokenize(io.BytesIO(fh.read()).readline):
            if tok.type not in (tokenize.COMMENT, tokenize.STRING):
                out.append(tok.string)
    return " ".join(out)


def test_no_mode_flags_outside_the_registry():
    """Fixture-vs-live is chosen at CONSTRUCTION time, in registry.py alone.
    `if DEMO:` and its synonyms must not appear in pipeline logic."""
    offenders = [
        p for p in Path("promptgen").rglob("*.py")
        if p.name != "registry.py" and BANNED.search(_code_only(p))
    ]
    assert not offenders, f"mode flag found in {offenders}"


def test_tripwire_would_actually_catch_a_flag(tmp_path):
    """A tripwire nobody has seen fail is not a tripwire."""
    bad = tmp_path / "bad.py"
    bad.write_text("if DEMO:\n    pass\n")
    assert BANNED.search(_code_only(bad))