116 lines
3.5 KiB
Python
116 lines
3.5 KiB
Python
"""Shared e2e harness: free ports, health waits, lfx spawn, migrate, sessions.
|
|
|
|
Single place for logic shared between tests/conftest.py and
|
|
scripts/smoke_e2e.py. Full real-LFX runs are kept.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
# Single constant: lfx prefers LANGFLOW_API_KEY over LFX_API_KEY, so the
|
|
# repo .env value must be cleared wherever a test key is used.
|
|
TEST_LFX_KEY = "test-lfx-key"
|
|
|
|
|
|
def free_port() -> int:
|
|
sock = socket.socket()
|
|
sock.bind(("127.0.0.1", 0))
|
|
port = sock.getsockname()[1]
|
|
sock.close()
|
|
return port
|
|
|
|
|
|
def wait_healthy(url: str, deadline_s: float = 240.0) -> None:
|
|
deadline = time.time() + deadline_s
|
|
while time.time() < deadline:
|
|
try:
|
|
with urllib.request.urlopen(url + "/health", timeout=3) as response:
|
|
if response.status == 200:
|
|
return
|
|
except OSError:
|
|
time.sleep(1)
|
|
raise RuntimeError(f"server did not become healthy: {url}")
|
|
|
|
|
|
def spawn_lfx(
|
|
flow_dir: Path, port: int, env_extra: dict | None = None, api_key: str = TEST_LFX_KEY
|
|
) -> subprocess.Popen:
|
|
bindir = Path(sys.executable).parent
|
|
lfx_bin = shutil.which("lfx", path=str(bindir)) or "lfx"
|
|
env = {**os.environ, "LANGFLOW_API_KEY": "", "LFX_API_KEY": api_key}
|
|
if env_extra:
|
|
env.update(env_extra)
|
|
return 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,
|
|
)
|
|
|
|
|
|
def migrate_throwaway(backend_dir: Path, db_url: str | None = None) -> None:
|
|
from alembic import command as alembic_command
|
|
from alembic.config import Config
|
|
|
|
cfg = Config()
|
|
cfg.set_main_option("script_location", str(backend_dir / "migrations"))
|
|
if db_url is not None:
|
|
cfg.set_main_option("sqlalchemy.url", db_url)
|
|
# migrations/env.py takes the URL from settings (not the ini value).
|
|
from app.settings import settings as app_settings
|
|
|
|
app_settings.database_url = db_url
|
|
alembic_command.upgrade(cfg, "head")
|
|
|
|
|
|
def spawn_uvicorn(app, host: str = "127.0.0.1", port: int = 0): # type: ignore[no-untyped-def]
|
|
"""Real uvicorn server in a daemon thread. Returns (server, thread)."""
|
|
import threading
|
|
import time
|
|
|
|
import uvicorn
|
|
|
|
config = uvicorn.Config(app, host=host, port=port, 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"
|
|
return server, thread
|
|
|
|
|
|
def plant_session_jwt(session, email: str, role: str = "admin", name: str = "Test") -> tuple:
|
|
"""Idempotent test user + session JWT (shared with smoke pattern)."""
|
|
from sqlmodel import select
|
|
|
|
from app.auth import service
|
|
from app.models import User
|
|
|
|
user = session.exec(select(User).where(User.email == email)).first()
|
|
if user is None:
|
|
user = User(name=name, email=email, role=role)
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
return user, service.mint_session_token(user)
|