brain-of-reese/tests/e2e/conftest.py

"""Playwright E2E fixtures (shared by every story's test file).

Each user story in ``.agent/user_stories/`` gets its own isolated E2E test
file; this conftest provides the shared environment:

* ``mock_llm``   — deterministic OpenAI-compatible server (see mock_llm.py).
                    Set ``E2E_REAL_LLM=1`` to point at the real aipi endpoint
                    instead (requires an imported knowledge base).
* ``app_server`` — the real FastAPI app under test (uvicorn subprocess).
* ``browser``/``page`` — headless Chromium pointed at the app.

Prerequisite for story tests that touch the database:
    podman compose up -d db
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import time
from collections.abc import Iterator
from pathlib import Path

import httpx
import pytest
from playwright.sync_api import Browser, Page, sync_playwright

from app.config import Settings as _Settings

REPO = Path(__file__).resolve().parents[2]
APP_PORT = int(os.environ.get("E2E_APP_PORT", "8123"))
MOCK_PORT = int(os.environ.get("E2E_MOCK_PORT", "8901"))
APP_URL = f"http://127.0.0.1:{APP_PORT}"
USE_REAL_LLM = os.environ.get("E2E_REAL_LLM") == "1"

# Phase 16: the app under test boots with single-admin auth configured
# (fail-loud otherwise). Known E2E values — the shared form-login helper
# (tests/e2e/auth_helpers.py) uses ADMIN_PASSWORD; the secret is fixed so
# session cookies stay valid across a session-scoped app restart.
ADMIN_PASSWORD = "e2e-admin-password"
SESSION_SECRET = "e2e-session-secret-0123456789abcdef0123456789abcdef"


def _wait_http(url: str, timeout: float = 40.0) -> None:
    deadline = time.monotonic() + timeout
    last_err = "unknown"
    while time.monotonic() < deadline:
        try:
            httpx.get(url, timeout=2.0)
            return
        except Exception as e:  # noqa: BLE001 — retry until deadline
            last_err = str(e)
            time.sleep(0.5)
    raise RuntimeError(f"server at {url} did not come up: {last_err}")


@pytest.fixture(scope="session")
def mock_llm() -> Iterator[int]:
    """Deterministic OpenAI-compatible LLM (chat + embeddings)."""
    if USE_REAL_LLM:
        yield 0
        return
    env = dict(os.environ)
    env.pop("DEBUGPY", None)
    proc = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "tests.e2e.mock_llm:app",
         "--host", "127.0.0.1", "--port", str(MOCK_PORT), "--log-level", "warning"],
        cwd=REPO,
        env=env,
    )
    try:
        _wait_http(f"http://127.0.0.1:{MOCK_PORT}/v1/models")
        yield MOCK_PORT
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()


@pytest.fixture(scope="session")
def app_server(mock_llm: int) -> Iterator[str]:
    """The real app under test."""
    env = dict(os.environ)
    env.pop("DEBUGPY", None)
    env["BOR_ENVIRONMENT"] = "e2e"
    env["BOR_STATIC_DIR"] = str(REPO / "frontend")
    env["BOR_LLM_BASE_URL"] = (
        "https://aipi.reeseapps.com/v1"
        if USE_REAL_LLM
        else f"http://127.0.0.1:{mock_llm}/v1"
    )
    # The E2E mock's token-overlap embeddings have their own score
    # distribution (phase 09) — the app under test gets the mock-calibrated
    # threshold so every story suite keeps its deterministic gate behavior.
    # The production default stays 0.62 (re-tuned against the real
    # `embed` model's 0.41–0.84 cosine range, PLAN A8).
    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
    # Phase 67 (LLM retry): the e2e pins the retry MECHANISM with instant
    # waits (BOR_LLM_RETRY_DELAY=0 — the 5 s default is unit-pinned via
    # tests/unit/test_config.py). BOR_LLM_RETRIES is forced to the code
    # default (derived from the class field, never drifts from
    # app/config.py) so the exhaustion test relies on the REAL budget and
    # an operator's local (gitignored) .env cannot leak a different one
    # into the app under test (the phase-61 leak-guard pattern).
    env["BOR_LLM_RETRY_DELAY"] = "0"
    env["BOR_LLM_RETRIES"] = str(_Settings.model_fields["llm_retries"].default)
    env.setdefault("BOR_DATABASE_URL", "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese")
    # Phase 16: admin auth must be set or create_app() refuses to boot.
    env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
    env["BOR_SESSION_SECRET"] = SESSION_SECRET
    # Phase 61 (defect fix): force the code DEFAULTS so an operator's local
    # (gitignored) ``.env`` — loaded by pydantic-settings from ``cwd=REPO``
    # — cannot leak corpus-specific chips / a docs repo into the app under
    # test: process env ranks above the ``.env`` file. The suggestions
    # default is derived from the class field (never drifts from
    # ``app/config.py``); docs-push stays inert (empty repo).
    env["BOR_DOCS_REPO"] = ""
    env["BOR_SUGGESTIONS"] = json.dumps(
        _Settings.model_fields["suggestions"].default
    )
    # Phase 62: the same leak class for the new UI customization
    # settings — an operator's local (gitignored) ``.env`` may
    # legitimately carry ``BOR_INPUT_PLACEHOLDER`` / ``BOR_FOOTER_TEXT``
    # / ``BOR_THEME``, and the byte-identical default contract (task
    # 05's ``test_default_server_is_byte_identical``) must see the code
    # defaults (derived from the class fields, never drifts from
    # ``app/config.py``).
    env["BOR_INPUT_PLACEHOLDER"] = _Settings.model_fields["input_placeholder"].default
    env["BOR_FOOTER_TEXT"] = _Settings.model_fields["footer_text"].default
    env["BOR_THEME"] = _Settings.model_fields["theme"].default
    proc = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "app.main:app",
         "--host", "127.0.0.1", "--port", str(APP_PORT), "--log-level", "warning"],
        cwd=REPO,
        env=env,
    )
    try:
        _wait_http(f"{APP_URL}/api/health")
        yield APP_URL
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()


@pytest.fixture(scope="session")
def app_url(app_server: str) -> str:
    return app_server


@pytest.fixture()
def db_ready(app_url: str) -> None:
    """Skip a test with clear instructions when Postgres is not running."""
    body = httpx.get(f"{app_url}/api/health", timeout=5).json()
    if body["db"] != "up":
        pytest.skip("Postgres not reachable — run `podman compose up -d db` first")


@pytest.fixture(scope="session")
def browser() -> Iterator[Browser]:
    with sync_playwright() as p:
        yield p.chromium.launch(headless=True)


@pytest.fixture()
def page(browser: Browser) -> Iterator[Page]:
    pg = browser.new_page(viewport={"width": 1280, "height": 800})
    yield pg
    pg.close()