mqtt-broker-stack/backend/tests/conftest.py

"""Shared pytest fixtures: sqlite-backed DB + FastAPI TestClient."""

from __future__ import annotations

import os

import pytest
from fastapi.testclient import TestClient

# Ensure test settings before any app import resolves them.
os.environ.setdefault("SESSION_SECRET", "test-session-secret-value-123456")
os.environ.setdefault("SECRET_ENCRYPTION_KEY", "test-encryption-key-value-1234")
os.environ.setdefault("ADMIN_INITIAL_PASSWORD", "test-admin-password-123")
os.environ.setdefault("MOSQUITTO_DYNSEC_PASSWORD", "test-dynsec-password")
os.environ.setdefault("DATABASE_URL", "sqlite+pysqlite:///:memory:")

from sqlalchemy import create_engine  # noqa: E402
from sqlalchemy.orm import Session, sessionmaker  # noqa: E402
from sqlalchemy.pool import StaticPool  # noqa: E402

from app import db as db_module  # noqa: E402
from app import main as main_module  # noqa: E402
from app.models import ManagementRole, ManagementUser  # noqa: E402
from app.security import hash_password  # noqa: E402


@pytest.fixture()
def db_session():
    """Fresh in-memory database per test with schema + seeded users.

    StaticPool + check_same_thread=False: the TestClient serves requests
    on a different thread than the fixture, but all access must share ONE
    sqlite connection (in-memory DBs are per-connection).
    """
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        future=True,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    db_module.Base.metadata.create_all(engine)
    factory = sessionmaker(bind=engine, future=True)
    session = factory()
    # Seed one super admin and one auditor for RBAC tests.
    session.add_all(
        [
            ManagementUser(
                username="admin",
                password_hash=hash_password("admin-password-123"),
                role=ManagementRole.SUPER_ADMIN,
                must_change_password=False,
            ),
            ManagementUser(
                username="auditor",
                password_hash=hash_password("auditor-password-123"),
                role=ManagementRole.AUDITOR,
                must_change_password=False,
            ),
        ]
    )
    session.commit()
    yield session
    session.close()

@pytest.fixture()
def client(db_session, monkeypatch):
    """TestClient wired to the fixture DB via dependency_overrides."""

    def _override_get_db():
        # Yield the SAME session the test asserts against; the request
        # handler shares its transaction for simplicity in tests.
        try:
            yield db_session
        finally:
            pass  # session lifetime managed by the db_session fixture

    main_module.app.dependency_overrides[db_module.get_db] = _override_get_db

    # Skip the lifespan bootstrap (create_all + admin seed) against a
    # separate engine: tests manage the schema themselves.
    from app import main as _m

    async def _noop_lifespan(app):
        yield

    monkeypatch.setattr(_m, "_bootstrap_admin", lambda: None)

    # Fake dynsec so no broker is needed for API tests.
    class FakeDynsec:
        def __init__(self):
            self.clients: dict[str, dict] = {}
            self.roles: dict[str, dict] = {}
            self.groups: dict[str, dict] = {}

        def _command(self, command, **kwargs):
            return {"success": True}

        def create_client(self, username, password, **kwargs):
            self.clients[username] = {"username": username, "enabled": True, **kwargs}

        def list_clients(self):
            return list(self.clients.values())

        def delete_client(self, username):
            self.clients.pop(username, None)

        def disable_client(self, username):
            self.clients[username]["enabled"] = False

        def enable_client(self, username):
            self.clients[username]["enabled"] = True

        def set_client_password(self, username, password):
            pass

        def get_client(self, username):
            return self.clients[username]

        def create_group(self, groupname, roles=None):
            self.groups[groupname] = {"groupname": groupname, "roles": roles or []}

        def list_groups(self):
            return list(self.groups.values())

        def delete_group(self, groupname):
            self.groups.pop(groupname, None)

        def create_role(self, rolename, acls=None):
            self.roles[rolename] = {"rolename": rolename, "acls": acls or []}

        def list_roles(self):
            return list(self.roles.values())

        def delete_role(self, rolename):
            self.roles.pop(rolename, None)

        def set_role_acls(self, rolename, acls):
            self.roles[rolename]["acls"] = acls

        def get_role(self, rolename):
            return self.roles[rolename]

        def connect(self):
            pass

        def close(self):
            pass

    fake = FakeDynsec()
    monkeypatch.setattr(main_module, "dynsec_client", fake)

    from app.routers import mqtt as mqtt_router

    monkeypatch.setattr(mqtt_router, "dynsec_client", fake)

    with TestClient(main_module.app) as test_client:
        yield test_client

    main_module.app.dependency_overrides.clear()


def login(client: TestClient, username: str, password: str):
    response = client.post(
        "/api/v1/auth/login", json={"username": username, "password": password}
    )
    assert response.status_code == 200, response.text
    return response


@pytest.fixture()
def admin_client(client):
    login(client, "admin", "admin-password-123")
    return client


@pytest.fixture()
def auditor_client(client):
    login(client, "auditor", "auditor-password-123")
    return client