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 subprocess
import sys
import tempfile
import urllib.request
from pathlib import Path

BACKEND = Path(__file__).resolve().parent.parent
VENV_BIN = BACKEND / ".venv" / "bin"
FIXTURES = BACKEND / "tests" / "fixtures" / "flows"

LFX_PORT = 6004
API_PORT = 8000
LFX_KEY = "lfx-759GsfJKjAeOBZ_44-cll9WdHp_yfYn_tIg7eHG38Rs"


def wait_healthy(url: str, timeout: int = 180) -> None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with urllib.request.urlopen(url, timeout=3) as r:
                if r.status == 200:
                    return
        except OSError:
            time.sleep(1)
    raise RuntimeError(f"never healthy: {url}")


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:
        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("/tmp/smoke-e2e-flows")
    flow_dir.mkdir(exist_ok=True)
    (flow_dir / "seed.json").write_text(
        (FIXTURES / "simple_rule.json").read_text()
    )
    # Migrate a throwaway SQLite schema to head before starting the API.
    from alembic import command as alembic_command
    from alembic.config import Config

    db_url = f"sqlite:///{tmp / 'smoke.db'}"
    cfg = Config()
    cfg.set_main_option("script_location", str(BACKEND / "migrations"))
    cfg.set_main_option("sqlalchemy.url", db_url)
    alembic_command.upgrade(cfg, "head")

    env = {
        **os.environ,
        # lfx prefers LANGFLOW_API_KEY over LFX_API_KEY; the repo .env may set
        # it. Clear it so the serve key below is the one we send.
        "LANGFLOW_API_KEY": "",
        "LFX_API_KEY": LFX_KEY,
        "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}",
    }
    lfx = subprocess.Popen(
        [str(VENV_BIN / "lfx"), "serve", "--host", "127.0.0.1",
         "--port", str(LFX_PORT), "--flow-dir", str(flow_dir), str(flow_dir)],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env,
    )
    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=env,
    )
    try:
        wait_healthy(f"http://127.0.0.1:{LFX_PORT}/health")
        wait_healthy(f"http://127.0.0.1:{API_PORT}/health")

        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()