Aegis/backend/tests/conftest.py

"""Pytest fixtures and configuration for backend tests.

The conftest intentionally avoids importing ``app.main`` at module level
because that triggers heavy side-effect imports (boto3, APScheduler, etc.)
which are NOT needed for unit tests.  The ``client`` fixture lazily imports
the FastAPI app only when actually requested.
"""

import os

# Set DATABASE_URL to SQLite *before* any app module is imported so that
# the lazy engine in app.database never tries to connect to PostgreSQL.
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")


import pytest
from sqlalchemy import JSON, String, Text, create_engine, event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.database import Base

# ── Patch PostgreSQL-specific column types so SQLite can handle them ─────
# Must run BEFORE importing models, because column type objects are
# instantiated at class-definition time.
from sqlalchemy.dialects.postgresql import UUID as PG_UUID, JSONB as PG_JSONB

# Tell SQLAlchemy: when compiling for SQLite, render JSONB as plain JSON
# and PostgreSQL UUID as CHAR(32).
from sqlalchemy.dialects.sqlite.base import SQLiteTypeCompiler

if not hasattr(SQLiteTypeCompiler, "visit_JSONB"):
    SQLiteTypeCompiler.visit_JSONB = lambda self, type_, **kw: "JSON"

if not hasattr(SQLiteTypeCompiler, "visit_UUID"):
    SQLiteTypeCompiler.visit_UUID = lambda self, type_, **kw: "CHAR(32)"

from app.auth import hash_password
from app.models.user import User

# ── Import all models so Base.metadata knows about every table ──────────
import app.models  # noqa: F401  — triggers model registration via __init__

# Use in-memory SQLite for tests
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"

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

# SQLite needs PRAGMA foreign_keys to enforce FK constraints
@event.listens_for(engine, "connect")
def _set_sqlite_pragma(dbapi_conn, connection_record):
    cursor = dbapi_conn.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def override_get_db():
    """Override the database dependency for testing."""
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()


@pytest.fixture(scope="function")
def db():
    """Create a fresh database for each test."""
    Base.metadata.create_all(bind=engine)
    db = TestingSessionLocal()
    yield db
    db.close()
    Base.metadata.drop_all(bind=engine)


@pytest.fixture(scope="function")
def client(db, monkeypatch):
    """Create a test client with database override.

    Imports ``app.main`` lazily to avoid pulling in boto3 / APScheduler
    when only the ``db`` fixture is needed.

    MinIO and the background scheduler are no-ops here so tests do not
    require Docker services for application startup.

    JWT blacklist uses an in-memory FakeRedis so tests do not require a
    real Redis instance.
    """
    import fakeredis

    _fake_redis = fakeredis.FakeRedis(decode_responses=True)

    def _blacklist_conn():
        return _fake_redis

    monkeypatch.setattr(
        "app.infrastructure.redis_client.get_redis_blacklist",
        _blacklist_conn,
    )

    monkeypatch.setattr("app.main.ensure_bucket_exists", lambda: None)
    monkeypatch.setattr("app.main.start_scheduler", lambda: None)
    monkeypatch.setattr(
        "app.main.scheduler.shutdown",
        lambda wait=False: None,
    )

    from app.main import app
    from app.database import get_db
    import app.database as _db_mod

    _db_mod._engine = engine
    _db_mod._SessionLocal = TestingSessionLocal

    app.dependency_overrides[get_db] = override_get_db
    Base.metadata.create_all(bind=engine)

    from app.limiter import limiter
    limiter.enabled = False

    from fastapi.testclient import TestClient
    with TestClient(app) as test_client:
        yield test_client

    Base.metadata.drop_all(bind=engine)
    app.dependency_overrides.clear()


@pytest.fixture(scope="function")
def admin_user(db):
    """Create an admin user for testing."""
    user = User(
        username="admin",
        email="admin@test.com",
        hashed_password=hash_password("admin123"),
        role="admin",
        is_active=True,
        must_change_password=False,
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


@pytest.fixture(scope="function")
def red_tech_user(db):
    """Create a red_tech user for testing."""
    user = User(
        username="redtech",
        email="redtech@test.com",
        hashed_password=hash_password("redtech123"),
        role="red_tech",
        is_active=True,
        must_change_password=False,
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


@pytest.fixture(scope="function")
def blue_tech_user(db):
    """Create a blue_tech user for testing."""
    user = User(
        username="bluetech",
        email="bluetech@test.com",
        hashed_password=hash_password("bluetech123"),
        role="blue_tech",
        is_active=True,
        must_change_password=False,
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


@pytest.fixture(scope="function")
def red_lead_user(db):
    """Create a red_lead user for testing."""
    user = User(
        username="redlead",
        email="redlead@test.com",
        hashed_password=hash_password("redlead123"),
        role="red_lead",
        is_active=True,
        must_change_password=False,
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


@pytest.fixture(scope="function")
def blue_lead_user(db):
    """Create a blue_lead user for testing."""
    user = User(
        username="bluelead",
        email="bluelead@test.com",
        hashed_password=hash_password("bluelead123"),
        role="blue_lead",
        is_active=True,
        must_change_password=False,
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


@pytest.fixture(scope="function")
def manager_user(db):
    """Create a manager user for testing."""
    user = User(
        username="manager",
        email="manager@test.com",
        hashed_password=hash_password("manager123"),
        role="manager",
        is_active=True,
        must_change_password=False,
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


@pytest.fixture(scope="function")
def admin_token(client, admin_user):
    """Get an auth token for the admin user."""
    response = client.post(
        "/api/v1/auth/login",
        data={"username": "admin@test.com", "password": "admin123"},
    )
    return response.json()["access_token"]


@pytest.fixture(scope="function")
def red_tech_token(client, red_tech_user):
    """Get an auth token for the red_tech user."""
    response = client.post(
        "/api/v1/auth/login",
        data={"username": "redtech@test.com", "password": "redtech123"},
    )
    return response.json()["access_token"]


@pytest.fixture(scope="function")
def auth_headers(admin_token):
    """Return authorization headers for admin user."""
    return {"Authorization": f"Bearer {admin_token}"}


@pytest.fixture(scope="function")
def red_tech_headers(red_tech_token):
    """Return authorization headers for red_tech user."""
    return {"Authorization": f"Bearer {red_tech_token}"}


@pytest.fixture(scope="function")
def blue_tech_token(client, blue_tech_user):
    """Get an auth token for the blue_tech user."""
    response = client.post(
        "/api/v1/auth/login",
        data={"username": "bluetech@test.com", "password": "bluetech123"},
    )
    return response.json()["access_token"]


@pytest.fixture(scope="function")
def blue_tech_headers(blue_tech_token):
    """Return authorization headers for blue_tech user."""
    return {"Authorization": f"Bearer {blue_tech_token}"}


@pytest.fixture(scope="function")
def red_lead_token(client, red_lead_user):
    """Get an auth token for the red_lead user."""
    response = client.post(
        "/api/v1/auth/login",
        data={"username": "redlead@test.com", "password": "redlead123"},
    )
    return response.json()["access_token"]


@pytest.fixture(scope="function")
def red_lead_headers(red_lead_token):
    """Return authorization headers for red_lead user."""
    return {"Authorization": f"Bearer {red_lead_token}"}


@pytest.fixture(scope="function")
def blue_lead_token(client, blue_lead_user):
    """Get an auth token for the blue_lead user."""
    response = client.post(
        "/api/v1/auth/login",
        data={"username": "bluelead@test.com", "password": "bluelead123"},
    )
    return response.json()["access_token"]


@pytest.fixture(scope="function")
def blue_lead_headers(blue_lead_token):
    """Return authorization headers for blue_lead user."""
    return {"Authorization": f"Bearer {blue_lead_token}"}


@pytest.fixture(scope="function")
def manager_token(client, manager_user):
    """Get an auth token for the manager user."""
    response = client.post(
        "/api/v1/auth/login",
        data={"username": "manager@test.com", "password": "manager123"},
    )
    return response.json()["access_token"]


@pytest.fixture(scope="function")
def manager_headers(manager_token):
    """Return authorization headers for manager user."""
    return {"Authorization": f"Bearer {manager_token}"}


@pytest.fixture(scope="function")
def api(client):
    """Issue an authenticated request while avoiding stale-cookie role bleed.

    ``client``'s cookie jar persists across requests within a test, and
    ``get_current_user`` prefers the ``aegis_token`` cookie over the
    ``Authorization`` header. A test that uses more than one role's
    ``*_headers`` fixture (e.g. submit as red_lead, then approve as
    manager) would otherwise have its *first* request silently
    authenticate as whichever role's fixture happens to log in last,
    since pytest resolves all fixtures before the test body runs. Use
    this instead of ``client.post``/``client.get`` directly whenever a
    test mixes more than one role.

    Usage: ``api("post", url, headers, json=payload)``.
    """
    def _request(method: str, url: str, headers: dict, **kwargs):
        client.cookies.clear()
        return getattr(client, method)(url, headers=headers, **kwargs)
    return _request