flowdeck/tests/conftest.py

"""FlowDeck — pytest fixtures and configuration.

Each test gets a fresh, isolated SQLite database and backup directory so the
suite is safe to run in parallel (``pytest -n auto``): workers never share a
database file, and no state leaks between tests.
"""
import os
import re
import tempfile
from pathlib import Path

import httpx
import pytest
from fastapi.testclient import TestClient


class _TestSessionAuth(httpx.Auth):
    """Session + CSRF injectés à la volée (jamais dans le cookie jar du client).

    - `flowdeck_session` ajouté seulement s'il est absent de la requête (un test
      peut fournir la sienne via `cookies=`) ;
    - `csrf_token` idem, et l'en-tête `X-CSRF-Token` suit TOUJOURS le cookie
      courant (le token tourne quand `/api/csrf-token` est appelé) ;
    - un test qui veut l'anonymat fait `anon(client)` → `client.auth = None`.
    """

    CSRF_FALLBACK = "csrf-test-token"

    def __init__(self, session_token: str):
        self.session_token = session_token

    def auth_flow(self, request):
        ch = request.headers.get("cookie", "")
        add = []
        if "flowdeck_session=" not in ch:
            add.append(f"flowdeck_session={self.session_token}")
        if "csrf_token=" not in ch:
            add.append(f"csrf_token={self.CSRF_FALLBACK}")
        if add:
            request.headers["cookie"] = "; ".join(([ch] if ch else []) + add)
        if "X-CSRF-Token" not in request.headers:
            m = re.search(r"csrf_token=([^;]+)", request.headers.get("cookie", ""))
            if m:
                request.headers["X-CSRF-Token"] = m.group(1)
        yield request


def login_test_client(tc, user_id: int = 1, login: str = "tester", is_admin: int = 1):
    """Connecte un TestClient (A3–A7 : les routes testées exigent une session)."""
    from app.auth.session import SessionManager
    from app.db import get_conn

    with get_conn() as conn:
        conn.execute(
            "INSERT OR IGNORE INTO users (id, login, full_name, is_admin) VALUES (?,?,?,?)",
            (user_id, login, login.title(), is_admin),
        )
        conn.commit()
    tc.auth = _TestSessionAuth(
        SessionManager.create_session(
            {"id": user_id, "login": login, "full_name": login.title(), "is_admin": is_admin}
        )
    )
    return tc


def anon(client):
    """Test d'anonymat : plus de session, plus de CSRF par défaut."""
    client.cookies.clear()
    client.headers.pop("X-CSRF-Token", None)
    client.auth = None
    return client


@pytest.fixture
def client():
    """FastAPI TestClient with a fresh temporary SQLite database."""
    db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
    db_path = db_file.name
    db_file.close()

    backup_dir = tempfile.mkdtemp(prefix="fd_backups_")
    data_dir = tempfile.mkdtemp(prefix="fd_data_")

    # Set env BEFORE importing app modules (config reads at import time).
    os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
    os.environ["APP_SECRET_KEY"] = "test-secret-for-tests"
    os.environ["RATE_LIMIT_ENABLED"] = "false"
    os.environ["PUBLIC_API_INSECURE_OK"] = "true"
    os.environ["BACKUP_ENABLED"] = "true"
    os.environ["BACKUP_DIR"] = backup_dir
    os.environ["PROJECT_SYNC_ENABLED"] = "false"
    # Point data-root (uploads/, emoji/, covers/) at a writable temp dir so tests
    # don't depend on the container's /data path existing on a dev host.
    os.environ["FLOWDECK_DATA_DIR"] = data_dir

    # IMPORTANT: mutate the existing Settings singleton in place — do NOT rebind
    # `app.config.settings`. Modules such as `app.services.backup` and
    # `app.routers.auth` hold a direct reference imported at load time, so
    # rebinding would leave them pointing at the stale defaults (this was the
    # cause of the previously-skipped flaky backup tests).
    import app.config
    s = app.config.settings
    s.database_url = f"sqlite:///{db_path}"
    s.app_secret_key = "test-secret-for-tests"
    s.rate_limit_enabled = False
    s.public_api_insecure_ok = True
    s.backup_enabled = True
    s.backup_dir = backup_dir
    s.backup_interval_hours = 24
    s.backup_keep = 30
    s.project_sync_enabled = False

    from app.db import init_db
    from app.main import app
    init_db()

    yield login_test_client(TestClient(app))

    # Cleanup
    try:
        os.unlink(db_path)
    except PermissionError:
        pass  # Windows: file may still be open in another thread
    for p in Path(backup_dir).glob("*.db"):
        try:
            p.unlink()
        except PermissionError:
            pass
    try:
        Path(backup_dir).rmdir()
    except OSError:
        pass