app/backend/tests/conftest.py

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 import delete, inspect
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 (
    EngineEvent,
    FlowRun,
    MetricBucket,
    Run,
    RunArtifact,
    RunMetric,
    RunNode,
    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)


#: Tables the engine writes to on its own, from the collector and the run
#: service the module-scoped app starts.
_ENGINE_TABLES = (
    RunArtifact,
    RunMetric,
    RunNode,
    Run,
    FlowRun,
    EngineEvent,
    MetricBucket,
)


@pytest.fixture(scope="module", autouse=True)
def _clean_engine_rows() -> Generator[None, None, None]:
    """Clear what a module wrote before the next one queries it.

    The `db` session is session-scoped and every commit in it is permanent, so
    a `FlowRun` one module left marked `running` turned up in another module's
    query. Per-*test* rollback is not available here: the module-scoped
    `client` runs the real lifespan, and the collector and run service write
    through sessions of their own that no test transaction wraps. Clearing
    between modules bounds it to where the writes actually come from.
    """
    yield
    # `tests/flow` overrides `db` with a no-op — the engine holds no database
    # state, so those tests build no schema — and there is nothing to clear
    # there. Running that directory on its own is how that shows.
    if not inspect(engine).has_table("run"):
        return
    with Session(engine) as session:
        for model in _ENGINE_TABLES:
            session.execute(delete(model))
        session.commit()


@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