ducoterra
022da8e2bc
Foundation (phase 01, verified): - FastAPI app: /api/health, /api/suggestions, /api/chat (placeholder), static frontend served locally (no CDN) - Postgres 17 + pgvector via db/Containerfile + compose.yaml (podman compose up -d db), Alembic initial migration (documents, chunks with vector(768), query_log) - LLM client targeting https://aipi.reeseapps.com/v1 (turbo/embed); scripts/llm_probe.py verified models + 768-dim embeddings live - Conditional debugpy: imported only when DEBUGPY=1 (attach on demand, :5678); logging config for clean single-line logs - Frontend shell: mobile-first chat + Sources pages, tokens, a11y baselines - Tests: 24 unit+integration (99% coverage on app/), ruff + pyright clean, Playwright smoke E2E (3 tests) against a deterministic mock LLM - Planning: .agent/PLAN.md (architecture + LOCKED decisions), AGENTS.md, 6 user stories, 7 phase files (one story / one phase / one Playwright suite each)
127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
"""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 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
|
|
|
|
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"
|
|
|
|
|
|
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"
|
|
)
|
|
env.setdefault("BOR_DATABASE_URL", "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese")
|
|
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()
|