kirche-einfach/backend/tests/conftest.py

"""Shared pytest fixtures.

Requires TEST_DATABASE_URL (a dedicated Postgres database, e.g. the
`kirche_einfach_test` database from the compose stack). The URL is mapped
onto DATABASE_URL *before* any app module is imported, so settings and the
alembic env both target the test database. Never point this at production.
"""

import os
from pathlib import Path

import pytest

TEST_DB_URL = os.environ.get("TEST_DATABASE_URL")
if not TEST_DB_URL:
    raise RuntimeError(
        "TEST_DATABASE_URL is not set. Refusing to run tests without "
        "an explicit test database (see .env.example)."
    )

# Must precede all `app.*` imports: Settings is instantiated at import time.
os.environ["DATABASE_URL"] = TEST_DB_URL

BACKEND_DIR = Path(__file__).resolve().parent.parent


@pytest.fixture(scope="session")
def db_engine():
    """Engine on the test DB, migrated to head once per session."""
    from alembic import command as alembic_command
    from alembic.config import Config

    from app import db as app_db

    app_db.reset_engine()
    engine = app_db.get_engine()

    cfg = Config()
    cfg.set_main_option("script_location", str(BACKEND_DIR / "migrations"))
    alembic_command.upgrade(cfg, "head")

    yield engine

    engine.dispose()
    app_db.reset_engine()


@pytest.fixture()
def session(db_engine):
    """Function-scoped session for direct DB access in tests."""
    from sqlmodel import Session

    with Session(db_engine) as s:
        yield s


@pytest.fixture(autouse=True)
def _clean_tables(db_engine):
    """Truncate all domain tables before and after every test (isolation)."""
    from sqlmodel import SQLModel

    def _truncate() -> None:
        with db_engine.begin() as conn:
            for table in reversed(SQLModel.metadata.sorted_tables):
                conn.execute(table.delete())

    _truncate()
    yield
    _truncate()


@pytest.fixture()
async def client():
    """ASGI test client for the FastAPI app."""
    from httpx import ASGITransport, AsyncClient

    from app.main import app

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


@pytest.fixture(scope="session")
def lfx_server(tmp_path_factory):
    """Real `lfx serve` subprocess on a free port; points settings at it.

    Serves an (initially seeded) flow dir; tests upload fixture flows through
    the same deploy path production uses.
    """
    import shutil
    import socket
    import subprocess
    import sys
    import time
    import urllib.request
    from pathlib import Path

    from app.settings import settings

    bindir = Path(sys.executable).parent
    lfx_bin = shutil.which("lfx", path=str(bindir)) or "lfx"

    sock = socket.socket()
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    sock.close()

    flow_dir = tmp_path_factory.mktemp("lfx-flows")
    seed = (
        Path(__file__).parent / "fixtures" / "flows" / "simple_rule.json"
    ).read_text()
    (flow_dir / "seed.json").write_text(seed)

    env = {
        **os.environ,
        # lfx serve refuses to start without a serve key; requests need it too.
        "LFX_API_KEY": "test-lfx-key",
    }
    proc = subprocess.Popen(
        [
            lfx_bin,
            "serve",
            "--host",
            "127.0.0.1",
            "--port",
            str(port),
            "--flow-dir",
            str(flow_dir),
            str(flow_dir),
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        env=env,
    )
    url = f"http://127.0.0.1:{port}"
    deadline = time.time() + 240
    healthy = False
    while time.time() < deadline:
        if proc.poll() is not None:
            out = proc.stdout.read() if proc.stdout else ""
            raise RuntimeError(f"lfx serve exited early:\n{out[-4000:]}")
        try:
            with urllib.request.urlopen(url + "/health", timeout=3) as response:
                if response.status == 200:
                    healthy = True
                    break
        except OSError:
            time.sleep(1)
    if not healthy:
        proc.terminate()
        raise RuntimeError("lfx serve did not become healthy in time")

    old_url = settings.lfx_url
    old_key = settings.lfx_api_key
    settings.lfx_url = url
    settings.lfx_api_key = "test-lfx-key"
    yield url
    settings.lfx_url = old_url
    settings.lfx_api_key = old_key
    proc.terminate()
    try:
        proc.wait(timeout=20)
    except subprocess.TimeoutExpired:
        proc.kill()


@pytest.fixture(scope="session")
def live_server():
    """Real uvicorn server (ephemeral port) so LFX flows can call back
    into our API over HTTP, exactly like production."""
    import threading
    import time

    import uvicorn

    from app.main import app

    config = uvicorn.Config(app, host="127.0.0.1", port=0, log_level="warning")
    server = uvicorn.Server(config)
    thread = threading.Thread(target=server.run, daemon=True)
    thread.start()
    deadline = time.time() + 30
    while not server.started and time.time() < deadline:
        time.sleep(0.05)
    assert server.started, "uvicorn test server did not start"
    port = server.servers[0].sockets[0].getsockname()[1]
    yield f"http://127.0.0.1:{port}"
    server.should_exit = True
    thread.join(timeout=15)