stroblmeandClaude Opus 5
961a8f881d
One process owns this database — the image has run a single uvicorn worker for that reason since the four-engines bug — so a file beside the flows is the honest shape for it, and it is what lets `fluksio serve` need no infrastructure at all. Live values, node execution and the work queue never came here anyway; what does is a rollup a minute at a time, a row per cascade and the run history, and WAL keeps the readers going while that one writer works. DATA_DIR is now the one setting that moves everything an installation keeps; the rest derive from it and the images still spell theirs out. The schema is prepared in-process at startup, so the prestart service is gone, and the ten Postgres-only revisions collapse into one portable baseline. Three things only worked because psycopg was casting for us: a token's subject arriving as a string where the column is a UUID, `greatest`, and `date_bin`. The timestamps needed a column type of their own — SQLite stores no offset, and a naive datetime read back either raises against an aware `now` or serialises as local time. Postgres stays in the stack only for Umami, behind the analytics profile. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
118 lines
4.0 KiB
Python
118 lines
4.0 KiB
Python
import uuid
|
|
from collections.abc import Generator
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.engine import make_url
|
|
from sqlmodel import Session, select
|
|
|
|
from fluksio.cloud import config as cloud_config
|
|
from fluksio.core.config import settings
|
|
from fluksio.core.db import engine, prepare
|
|
from fluksio.main import app
|
|
from fluksio.models import User
|
|
from tests.utils.portal import INSTALLATION_ID, ISSUER, jwks
|
|
from tests.utils.user import authentication_token_from_email
|
|
from tests.utils.utils import get_superuser_token_headers
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def flow_data(tmp_path_factory: pytest.TempPathFactory) -> Generator[None, None, None]:
|
|
"""Keep flows and secrets written by tests out of the real store."""
|
|
root = tmp_path_factory.mktemp("flow-data")
|
|
settings.FLOWS_DIR = root / "flows"
|
|
settings.SECRETS_FILE = root / "secrets.enc"
|
|
settings.PANELS_FILE = root / "panels.json"
|
|
yield
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def db() -> Generator[Session, None, None]:
|
|
"""Create the throwaway database `tests/__init__.py` points at, drop it after."""
|
|
url = make_url(settings.SQLALCHEMY_DATABASE_URI)
|
|
# The teardown deletes this database, so refuse to run against anything but
|
|
# the dedicated test one.
|
|
assert url.get_backend_name() == "sqlite", url.get_backend_name()
|
|
assert url.database and url.database.endswith("_test.db"), url.database
|
|
|
|
path = Path(url.database)
|
|
path.unlink(missing_ok=True)
|
|
# The real path, so the suite runs against a database built the way a
|
|
# deployment's is — including the alembic stamp the app's own startup
|
|
# would otherwise trip over.
|
|
prepare(engine)
|
|
with Session(engine) as session:
|
|
yield session
|
|
|
|
engine.dispose()
|
|
# The write-ahead log and its index are part of the database.
|
|
for suffix in ("", "-wal", "-shm"):
|
|
path.with_name(path.name + suffix).unlink(missing_ok=True)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client() -> Generator[TestClient, None, None]:
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def superuser_token_headers(client: TestClient) -> dict[str, str]:
|
|
return get_superuser_token_headers(client)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def normal_user_token_headers(client: TestClient, db: Session) -> dict[str, str]:
|
|
return authentication_token_from_email(
|
|
client=client, email=settings.EMAIL_TEST_USER, db=db
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def portal_key() -> rsa.RSAPrivateKey:
|
|
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
|
|
|
|
@pytest.fixture
|
|
def enrolled(
|
|
tmp_path_factory: pytest.TempPathFactory,
|
|
portal_key: rsa.RSAPrivateKey,
|
|
db: Session,
|
|
) -> Any:
|
|
"""Enrol this installation with a fake portal, then undo it."""
|
|
local_user = db.exec(
|
|
select(User).where(User.email == settings.FIRST_SUPERUSER)
|
|
).one()
|
|
original = settings.CLOUD_CONFIG_FILE
|
|
settings.CLOUD_CONFIG_FILE = (
|
|
tmp_path_factory.mktemp(f"cloud-{uuid.uuid4().hex[:6]}") / "cloud.json"
|
|
)
|
|
cloud_config.save(
|
|
cloud_config.CloudConfig(
|
|
portal_url=ISSUER,
|
|
ws_url=f"{ISSUER}/api/v1/tunnel/attach",
|
|
installation_id=INSTALLATION_ID,
|
|
token="installation-token",
|
|
issuer=ISSUER,
|
|
jwks=jwks(portal_key),
|
|
local_user_id=str(local_user.id),
|
|
enrolled_at=datetime.now(UTC).isoformat(),
|
|
portal_account=settings.FIRST_SUPERUSER,
|
|
)
|
|
)
|
|
# Enrolment also maps the enrolling account to the portal identity that
|
|
# owns the installation; without it a portal session resolves to nobody.
|
|
local_user.portal_sub = "portal-user-1"
|
|
db.add(local_user)
|
|
db.commit()
|
|
yield local_user
|
|
local_user.portal_sub = None
|
|
db.add(local_user)
|
|
db.commit()
|
|
cloud_config.delete()
|
|
settings.CLOUD_CONFIG_FILE = original
|