kirche-einfach/backend/scripts/smoke_e2e.py

"""Manual end-to-end smoke: real LFX + real API, frontend-shaped calls.

Usage (from backend/):
    uv run python scripts/smoke_e2e.py
Runs the API against a throwaway SQLite file and an ephemeral LFX pass.
Exits non-zero on any unexpected status/body.
"""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from tests._harness import (
    free_port,
    migrate_throwaway,
    plant_session_jwt,
    spawn_lfx,
    wait_healthy,
)

BACKEND = Path(__file__).resolve().parent.parent
# Local venv (uv run) or image venv (/opt/venv in containers): prefer PATH.
VENV_BIN = Path(shutil.which("lfx") or "").parent or (BACKEND / ".venv" / "bin")
FIXTURES = BACKEND / "tests" / "fixtures" / "flows"

# Ephemeral ports: the dev API/LFX may already occupy the defaults
# (e.g. when smoke runs inside the api container itself).
LFX_PORT = free_port()
API_PORT = free_port()
LFX_KEY = "lfx-759GsfJKjAeOBZ_44-cll9WdHp_yfYn_tIg7eHG38Rs"

# Stateless admin session JWT, minted in main() after planting the user.
_SMOKE_JWT = ""


def api(method: str, path: str, body: dict | None = None) -> tuple[int, dict | str]:
    import httpx

    with httpx.Client(base_url=f"http://127.0.0.1:{API_PORT}", timeout=120) as c:
        # Management + runs are login-gated; the smoke admin JWT is minted
        # in main() (stateless sessions, see app/auth/service.py).
        c.cookies.set("kirche_session", _SMOKE_JWT, domain="127.0.0.1")
        r = c.request(method, path, json=body)
        try:
            return r.status_code, r.json()
        except ValueError:
            return r.status_code, r.text


def check(name: str, cond: bool, detail: str = "") -> None:
    print(f"[{'OK' if cond else 'FAIL'}] {name} {detail}")
    if not cond:
        raise SystemExit(f"smoke failed: {name} {detail}")


def main() -> None:
    tmp = Path(tempfile.mkdtemp(prefix="kirche-smoke-"))
    flow_dir = Path(tempfile.mkdtemp(prefix="kirche-smoke-flows-"))
    (flow_dir / "seed.json").write_text((FIXTURES / "simple_rule.json").read_text())
    # Migrate a throwaway SQLite schema to head before starting the API.
    db_url = f"sqlite:///{tmp / 'smoke.db'}"
    migrate_throwaway(BACKEND, db_url)

    # Plant an admin user + mint its session JWT: management + runs
    # require login, and there is no SMTP in smoke to deliver a magic link.
    # Sessions are stateless (see app/auth/service.py) — the cookie value
    # below is the JWT sent as `kirche_session`.
    global _SMOKE_JWT
    from sqlmodel import Session, create_engine

    from app.settings import settings as app_settings

    if not app_settings.jwt_secret:
        app_settings.jwt_secret = os.environ.get(
            "JWT_SECRET", "smoke-jwt-secret-32-chars-minimum!!"
        )
    engine = create_engine(db_url)
    with Session(engine) as s:
        _, _SMOKE_JWT = plant_session_jwt(s, email="smoke@example.com", name="smoke")
        # Smoke predates the shared harness key; keep its long serve key.
        print("smoke admin JWT minted")
    engine.dispose()

    env_extra = {
        "DATABASE_URL": db_url,
        "FLOWS_WATCH_ENABLED": "false",
        "LFX_URL": f"http://127.0.0.1:{LFX_PORT}",
        "CALLBACK_BASE_URL": f"http://127.0.0.1:{API_PORT}",
        "CALLBACK_TOKEN": "smoke-callback-token",
    }
    lfx = spawn_lfx(flow_dir, LFX_PORT, env_extra=env_extra, api_key=LFX_KEY)
    uvicorn = subprocess.Popen(
        [str(VENV_BIN / "uvicorn"), "app.main:app", "--host", "127.0.0.1", "--port", str(API_PORT)],
        cwd=str(BACKEND),
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        env={
            **os.environ,
            "LANGFLOW_API_KEY": "",
            "LFX_API_KEY": LFX_KEY,
            **env_extra,
        },
    )
    try:
        wait_healthy(f"http://127.0.0.1:{LFX_PORT}", deadline_s=180)
        wait_healthy(f"http://127.0.0.1:{API_PORT}", deadline_s=180)

        s, flows = api("GET", "/api/flows")
        check("list flows", s == 200 and isinstance(flows, list))

        s, flow = api("POST", "/api/flows", {"name": "smoke-dach"})
        check("create flow", s == 201, json.dumps(flow)[:120])
        fid = flow["id"]

        fx = json.loads((FIXTURES / "simple_rule.json").read_text())
        s, ver = api(
            "POST",
            f"/api/flows/{fid}/versions",
            {"langflow_json": fx, "author": "smoke"},
        )
        check("create version", s == 201, f"v{ver.get('version_no')}")

        s, dep = api("POST", f"/api/flows/{fid}/versions/1/deploy")
        check("deploy", s == 200 and dep.get("lfx_flow_id"), json.dumps(dep)[:160])

        s, run = api(
            "POST",
            f"/api/flows/{fid}/run",
            {"input": {"betrag": 120000, "schwelle": 200000}},
        )
        check(
            "run standard",
            s == 200 and run.get("status") == "completed",
            json.dumps(run.get("output"))[:120],
        )

        s, run2 = api(
            "POST",
            f"/api/flows/{fid}/run",
            {"input": {"betrag": 999000, "schwelle": 200000}},
        )
        check(
            "run sonderfall",
            s == 200 and "Sonderfall" in (run2.get("output") or {}).get("result", ""),
        )

        s, tasks = api("GET", "/api/approvals")
        check("approvals empty (no callback flow)", s == 200 and tasks == [])

        print("smoke E2E: all green")
    finally:
        uvicorn.terminate()
        lfx.terminate()


if __name__ == "__main__":
    if "TEST_DATABASE_URL" not in os.environ:
        sys.exit("TEST_DATABASE_URL must be set")
    main()