gatekeeper/test/conftest.py

import os
from datetime import datetime, timedelta
from time import tzset

import pytest
from alembic.config import Config
from alembic.script import ScriptDirectory
from fastapi.testclient import TestClient
from sqlalchemy import inspect
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine, select

# TEST_ENV must cover every Settings field
# test_settings.py fails with the missing name if it does not.
TEST_ENV = {
    "SECRET_KEY": "ff" * 16,
    "SQLALCHEMY_DATABASE_URL": "sqlite://",
    "MQTT_HOST": "localhost",
    "MQTT_PORT": "1883",
    "MOCK": "True",
    "ALEMBIC_CONFIG": "./alembic.ini",
    "DISABLE_CARDS": "True",
    "MIFARE_APP_MASTER_KEY": "0" * 16,
    "MIFARE_ACL_READ_BASE_KEY": "1" * 16,
    "MIFARE_ACL_WRITE_BASE_KEY": "2" * 16,
    "DISABLE_CALDAV": "True",
    "CALDAV_URL": "http://caldav.invalid/dav.php",
    "CALDAV_USER": "test-user",
    "CALDAV_PASSWORD": "test-password",
    "CALDAV_CALENDARS": '["test-calendar"]',
    "CALDAV_REFRESH_INTERVAL": "3600",
}


for _name in TEST_ENV:
    os.environ.pop(_name, None)
os.environ.update(TEST_ENV)

import app.services.database as database_service  # noqa: E402

TEST_ADMIN_PASSWORD = "admin123"

engine = create_engine(
    TEST_ENV["SQLALCHEMY_DATABASE_URL"],
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)

# inject custom engine
database_service.configure_engine(engine)

# import main after the custom engine is injected
from app.main import app  # noqa: E402
from app.model.models import (  # noqa: E402
    AccessAuthorizationDB,
    Card,
    GroupDB,
    OneShotAccess,
    UserDB,
)
from app.services.settings import settings  # noqa: E402


def alembic_head() -> str:
    """Head revision of the migration scripts."""
    alembic_cfg = Config(settings.alembic_config)
    return ScriptDirectory.from_config(alembic_cfg).get_current_head()


def _seed_alembic_version() -> None:
    """Mark the test database as already migrated.

    Without a version row create_db_and_tables() treats the database as new and
    runs an alembic stamp, which builds its own engine from the configured URL
    instead of using this shared in-memory engine.
    """
    with engine.connect() as connection:
        if "alembic_version" not in inspect(connection).get_table_names():
            connection.exec_driver_sql(
                "CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL)"
            )
        connection.exec_driver_sql(
            "INSERT OR IGNORE INTO alembic_version (version_num) VALUES (?)",
            (_HEAD_REVISION,),
        )
        connection.commit()


def _seed_default_admin() -> None:
    """Seed the admin user the fixtures log in with.
    TestClient runs lifespan, that calls create_first_user()
    which blocks the admin_user fixture. Pre-creating the admin stops that
    """
    from app.services.auth import get_password_hash

    with Session(engine) as session:
        if session.exec(select(UserDB).where(UserDB.name == "admin")).first() is None:
            session.add(
                UserDB(
                    name="admin",
                    passwordhash=get_password_hash(TEST_ADMIN_PASSWORD),
                    is_admin=True,
                )
            )
            session.commit()


_HEAD_REVISION = alembic_head()
_seed_alembic_version()


@pytest.fixture
def system_tz(request):
    """Run a test with the host-local timezone forced to a POSIX TZ string."""
    previous_tz = os.environ.get("TZ")
    os.environ["TZ"] = request.param
    tzset()
    try:
        yield request.param
    finally:
        if previous_tz is None:
            os.environ.pop("TZ", None)
        else:
            os.environ["TZ"] = previous_tz
        tzset()


@pytest.fixture(autouse=True)
def _installed_engine():
    """Undo configure_engine() leaks between tests."""
    previous = database_service.get_engine()
    yield
    database_service.configure_engine(previous)


@pytest.fixture(scope="function")
def db_session():
    """Create a fresh database session for each test."""
    SQLModel.metadata.create_all(engine)
    _seed_alembic_version()
    _seed_default_admin()
    with Session(engine) as session:
        yield session
    SQLModel.metadata.drop_all(engine)


@pytest.fixture(scope="function")
def client(db_session):
    """Create a test client bound to the test engine.

    No FastAPI dependency override is needed: get_session() resolves its engine
    through get_engine(), which already points at the test engine.
    """
    with TestClient(app) as test_client:
        yield test_client


@pytest.fixture
def admin_user(db_session):
    """Get the admin user the seeded database already contains."""
    admin = db_session.exec(select(UserDB).where(UserDB.name == "admin")).one()
    db_session.refresh(admin)
    return admin


@pytest.fixture
def regular_user(db_session):
    """Create a regular user for testing."""
    from app.services.auth import get_password_hash

    user = UserDB(
        name="user", passwordhash=get_password_hash("user123"), is_admin=False
    )
    db_session.add(user)
    db_session.commit()
    db_session.refresh(user)
    return user


@pytest.fixture
def auth_headers(client, admin_user):
    """Get authentication headers for admin user."""
    response = client.post(
        "/api/v1/token",
        data={"username": admin_user.name, "password": TEST_ADMIN_PASSWORD},
    )
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def user_auth_headers(client, regular_user):
    """Get authentication headers for regular user."""
    response = client.post(
        "/api/v1/token", data={"username": regular_user.name, "password": "user123"}
    )
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def test_group(db_session):
    """Create a test group."""
    group = GroupDB(name="Test Group")
    db_session.add(group)
    db_session.commit()
    db_session.refresh(group)
    return group


@pytest.fixture
def test_card(db_session, test_group):
    """Create a test card."""
    card = Card(
        key="test-key-123",
        group_id=test_group.id,
        enabled=True,
        name="test_card",
        card_serial="00:00:00:00:00:00:00",
    )
    db_session.add(card)
    db_session.commit()
    db_session.refresh(card)
    return card


@pytest.fixture
def test_aa_os(db_session):
    oneshot = OneShotAccess(uses=1, ends_at=datetime.now() + timedelta(days=1))
    db_session.add(oneshot)
    aa = AccessAuthorizationDB(name="Test oneshot AA", is_active=True, type="oneshot")
    aa.oneshot = oneshot

    db_session.add(aa)
    db_session.commit()
    db_session.refresh(aa)
    return aa