interactive-story/backend/tests/test_local_only_surface.py

"""What M2 removed stays removed, and what it must not break stays working.

A subtractive milestone needs tests that fail if the surface grows back. These
are cheap, blunt, and deliberately not clever: they assert against the running
app's route table, the shipped configuration files, and the settings API.

    python -m pytest tests/test_local_only_surface.py -v
"""
import re
from pathlib import Path

import pytest
from fastapi import Depends
from fastapi.testclient import TestClient

from app import auth, models
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.providers.openai_compatible import (
    CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT, OpenAICompatibleProvider,
)

REPO = Path(__file__).resolve().parents[2]


def _paths() -> set[str]:
    """Every path the app serves, including those inside included routers."""
    found = set()

    def walk(routes):
        for r in routes:
            path = getattr(r, "path", None)
            if path:
                found.add(path)
            walk(getattr(r, "routes", []) or [])

    walk(app.routes)
    return found


@pytest.fixture()
def client():
    Base.metadata.create_all(bind=engine)
    setup = SessionLocal()
    user = models.User(is_guest=False)
    setup.add(user)
    setup.flush()
    setup.add(models.Settings(user_id=user.id, model="test-model"))
    setup.commit()
    user_id = user.id
    setup.close()

    def _current_user(db=Depends(get_db)):
        return db.get(models.User, user_id)

    app.dependency_overrides[auth.get_current_user] = _current_user
    c = TestClient(app)
    try:
        yield c
    finally:
        app.dependency_overrides.clear()
        Base.metadata.drop_all(bind=engine)


# --- the removed surfaces ------------------------------------------------

@pytest.mark.parametrize("prefix", ["/api/auth", "/api/analytics", "/api/scripts"])
def test_no_route_serves_a_removed_subsystem(prefix):
    """Accounts, the visitor dashboard, and campaign scripting are gone as
    routes, not merely hidden behind a flag."""
    assert not [p for p in _paths() if p.startswith(prefix)], prefix


@pytest.mark.parametrize("path", [
    "/api/auth/me", "/api/auth/login", "/api/auth/register", "/api/auth/logout",
    "/api/analytics/summary", "/api/analytics/collect", "/api/analytics/access",
    "/api/scripts", "/api/adventures/1/scripts", "/api/adventures/1/script-state",
])
def test_a_removed_endpoint_answers_404(client, path):
    assert client.get(path).status_code == 404, path


def test_the_application_has_no_scripting_engine():
    with pytest.raises(ImportError):
        __import__("app.scripting")


def test_no_module_imports_quickjs():
    """The dependency is gone from requirements; this catches an import that
    would put it back."""
    for py in (REPO / "backend" / "app").rglob("*.py"):
        assert "import quickjs" not in py.read_text(), py


def test_requirements_carry_no_hosted_dependencies():
    text = (REPO / "backend" / "requirements.txt").read_text()
    for gone in ("quickjs", "psycopg", "cryptography"):
        assert gone not in text, gone


def test_no_render_deployment_config():
    assert not (REPO / "render.yaml").exists()


# --- no cloud provider, no key ------------------------------------------

def test_settings_expose_no_api_key_field(client):
    body = client.get("/api/settings").json()
    assert "api_key" not in body
    assert "has_api_key" not in body


def test_an_api_key_cannot_be_set_through_the_api(client):
    """Pydantic ignores unknown fields, so this asserts the value does not
    land rather than that the request is refused."""
    client.put("/api/settings", json={"api_key": "sk-should-not-stick"})
    db = SessionLocal()
    try:
        assert db.query(models.Settings).first().api_key == ""
    finally:
        db.close()


def test_the_provider_sends_no_authorization_header():
    provider = OpenAICompatibleProvider("http://127.0.0.1:11434/v1", "m")
    assert "Authorization" not in provider._headers()


# --- the model timeout ---------------------------------------------------

def test_the_default_timeout_is_generous_but_finite():
    """M1 measured a cold model load exceeding the inherited hardcoded 120s on
    a CPU-only host. It must be longer than that, and it must be a number."""
    assert DEFAULT_READ_TIMEOUT > 120
    assert DEFAULT_READ_TIMEOUT <= 3600


def test_connect_stays_short_while_reading_stays_patient():
    """A wrong address should fail in seconds; a loading model should not."""
    provider = OpenAICompatibleProvider("http://127.0.0.1:11434/v1", "m")
    timeout = provider._timeout()
    assert timeout.connect == CONNECT_TIMEOUT <= 30
    assert timeout.read == DEFAULT_READ_TIMEOUT


def test_the_timeout_is_configurable(client):
    r = client.put("/api/settings", json={"model_timeout_seconds": 900})
    assert r.status_code == 200, r.text
    assert client.get("/api/settings").json()["model_timeout_seconds"] == 900


@pytest.mark.parametrize("value", [0, 29, 3601, -1])
def test_an_unusable_timeout_is_refused(client, value):
    """Not zero, not negative, and not "wait forever" spelled as a big number."""
    assert client.put(
        "/api/settings", json={"model_timeout_seconds": value}
    ).status_code == 422


def test_the_provider_honours_the_configured_timeout():
    provider = OpenAICompatibleProvider("http://127.0.0.1:11434/v1", "m", read_timeout=45)
    assert provider._timeout().read == 45


# --- the storyteller stays on loopback -----------------------------------

def test_the_native_start_scripts_bind_loopback():
    for script in ("start.sh", "start.ps1"):
        text = (REPO / script).read_text(errors="ignore")
        assert "--host 127.0.0.1" in text, script
        assert "--host 0.0.0.0" not in text, script


def test_compose_publishes_to_loopback_only():
    """The container listens on 0.0.0.0 because a published port cannot reach
    anything else. What must stay loopback is the *host* side of the mapping."""
    text = (REPO / "docker-compose.yml").read_text()
    published = re.findall(r'^\s*-\s*"([^"]+)"', text, re.M)
    assert published, "no published ports found — has the file moved?"
    for mapping in published:
        assert mapping.startswith("127.0.0.1:"), mapping


# --- every provider factory must actually build ---------------------------

def test_every_provider_factory_builds_from_a_real_settings_row(client):
    """M2 shipped with a defect this test would have caught.

    `Settings.api_key_plain` was removed with the API key, but `memorybank`'s
    two provider factories still read it. Nothing failed at import, and no test
    noticed, because every memory test stubs those factories out — so the break
    only appeared at runtime, in a background task, as a swallowed
    `AttributeError` that silently stopped summaries and embeddings.

    Constructing each factory from a real row is the cheapest thing that would
    have caught it, and it catches the same shape of mistake next time a
    Settings column moves.
    """
    from app import memorybank

    db = SessionLocal()
    try:
        settings = db.query(models.Settings).first()
        settings.embedding_model = "nomic-embed-text:latest"
        settings.summary_model = ""
        db.commit()

        summary = memorybank.summary_provider(settings)
        assert summary.model == settings.model  # falls back to the narrator
        assert summary.base_url == settings.endpoint_url.rstrip("/")

        embed = memorybank.embedding_provider(settings)
        assert embed.model == "nomic-embed-text:latest"

        turn = OpenAICompatibleProvider(
            settings.endpoint_url, settings.model, settings.api_mode,
            settings.model_timeout_seconds,
        )
        assert turn.read_timeout == settings.model_timeout_seconds
    finally:
        db.close()


def test_the_configured_timeout_reaches_every_generating_client(client, monkeypatch):
    """The second defect this review caught. The setting was stored, validated
    and exposed, and then not passed to the provider — so the turn engine kept
    using the module default and "configurable" was a claim rather than a fact.

    Each generating path is driven for real and the constructed provider is
    recorded. Embeddings are deliberately excluded: they are short, never
    cold-load a large model, and keep their own shorter constant.
    """
    from app import memorybank
    from app.routers import chat as chat_router
    from app.routers.adventures import turns as turns_router

    seen = []

    class Recorder:
        last_usage = None

        def __init__(self, endpoint_url, model, api_mode="chat", read_timeout=None):
            seen.append(read_timeout)

        async def generate(self, *a, **k):
            yield ("text", "narration")

        async def chat(self, *a, **k):
            yield ("text", "reply")

    monkeypatch.setattr(turns_router, "OpenAICompatibleProvider", Recorder)
    monkeypatch.setattr(chat_router, "OpenAICompatibleProvider", Recorder)
    monkeypatch.setattr(memorybank, "OpenAICompatibleProvider", Recorder)

    assert client.put("/api/settings", json={"model_timeout_seconds": 777}).status_code == 200

    adv = client.post("/api/adventures", json={"title": "T"}).json()["id"]
    client.post(f"/api/adventures/{adv}/actions", json={"type": "story", "text": "hello"})
    assert seen and seen[-1] == 777, f"turn engine used {seen[-1]!r}"

    client.post("/api/chat/stream", json={"messages": [{"role": "user", "content": "hi"}]})
    assert seen[-1] == 777, f"chat used {seen[-1]!r}"

    db = SessionLocal()
    try:
        memorybank.summary_provider(db.query(models.Settings).first())
    finally:
        db.close()
    assert seen[-1] == 777, f"summarizer used {seen[-1]!r}"