mirror of
https://github.com/bleak-ai/gcontext.git
synced 2026-08-11 13:19:23 +02:00
bernatsamperaandClaude Fable 5
aca7523754
FastAPI + Postgres service that stores workflow templates as file bundles with indexed manifest fields, submit-for-review publishing, and moderation endpoints behind an admin token. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
74 lines
1.9 KiB
Python
74 lines
1.9 KiB
Python
import os
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
|
|
import pytest
|
|
|
|
CONTAINER = "gcontext-api-test-pg"
|
|
PORT = 55433
|
|
|
|
|
|
def _wait_for_port(port: int, timeout: float = 30.0) -> None:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
with socket.socket() as sock:
|
|
sock.settimeout(1)
|
|
if sock.connect_ex(("127.0.0.1", port)) == 0:
|
|
return
|
|
time.sleep(0.5)
|
|
raise RuntimeError("postgres did not come up")
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def postgres():
|
|
subprocess.run(["docker", "rm", "-f", CONTAINER], capture_output=True)
|
|
subprocess.run(
|
|
[
|
|
"docker", "run", "-d", "--name", CONTAINER,
|
|
"-e", "POSTGRES_PASSWORD=test",
|
|
"-e", "POSTGRES_DB=workflows",
|
|
"-p", f"{PORT}:5432",
|
|
"postgres:16-alpine",
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
_wait_for_port(PORT)
|
|
os.environ["DATABASE_URL"] = (
|
|
f"postgresql+psycopg://postgres:test@127.0.0.1:{PORT}/workflows"
|
|
)
|
|
os.environ["ADMIN_TOKEN"] = "test-admin-token"
|
|
# The port answers before postgres accepts connections; retry the first connect.
|
|
from app.db import init_db
|
|
|
|
for _ in range(30):
|
|
try:
|
|
init_db()
|
|
break
|
|
except Exception:
|
|
time.sleep(1)
|
|
else:
|
|
raise RuntimeError("could not initialize the database")
|
|
yield
|
|
subprocess.run(["docker", "rm", "-f", CONTAINER], capture_output=True)
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(postgres):
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.db import engine
|
|
from app.main import app
|
|
from app.models import Base
|
|
|
|
Base.metadata.drop_all(engine())
|
|
Base.metadata.create_all(engine())
|
|
with TestClient(app) as test_client:
|
|
yield test_client
|
|
|
|
|
|
@pytest.fixture()
|
|
def admin():
|
|
return {"Authorization": "Bearer test-admin-token"}
|