espresso_frame/server/tests/conftest.py

"""Shared pytest fixtures for the server test suite.

DATABASE_URL must be set before app.db (and anything importing it,
transitively including app.main) is first imported -- app.db builds its
engine/SessionLocal at module import time, not lazily -- so this file
sets it as the very first thing it does, ahead of any `from app...`
import below. Importing app.main also runs migration.run_migrations()
as a side effect of that import (see main.py), which is what actually
creates the schema in the fresh temp database this points at.

Each test shares one migrated schema (re-migrating per test would be
needless I/O), but gets a clean slate of *data*: every table is wiped
after each test rather than relying on SQLAlchemy's transaction-rollback
test-isolation pattern (Session bound to a connection-level transaction
via join_transaction_mode="create_savepoint") -- that pattern needs the
"pysqlite serializable" event-listener workaround (see SQLAlchemy's own
docs on pysqlite's implicit-transaction quirks) that app/db.py's engine
doesn't set up, and this suite has no reason to add production-affecting
engine config just to make tests work. Deleting tables in reverse
dependency order (children before parents) satisfies foreign keys
without needing that workaround at all.
"""

from __future__ import annotations

import os
import tempfile
from pathlib import Path

_tmp_dir = tempfile.mkdtemp(prefix="espresso_frame_tests_")
os.environ["DATABASE_URL"] = f"sqlite:///{Path(_tmp_dir) / 'test.db'}"
# Same reasoning as DATABASE_URL above: logging_setup.configure_logging()
# also runs as an app.main import-time side effect and would otherwise
# try to create the real /data directory.
os.environ["LOG_PATH"] = str(Path(_tmp_dir) / "server.log")

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from app import db as db_module
from app import migration
from app.auth import hash_password
from app.main import app
from app.models import Base, Frame, User, UserFrame


def _reset_db() -> None:
    """Wipes every ORM-mapped table, then reseeds the same baseline a
    real fresh install gets (frame #1 + the server-settings singleton --
    see migration.py's _ensure_frame_one/_ensure_server_settings, both
    idempotent and both already called by run_migrations). The raw
    schema_version table isn't part of Base.metadata (see migration.py),
    so it survives the wipe untouched and run_migrations() skips
    straight to that reseed step instead of re-running every ALTER."""
    with db_module.engine.begin() as conn:
        for table in reversed(Base.metadata.sorted_tables):
            conn.execute(table.delete())
    migration.run_migrations()


@pytest.fixture
def db_session():
    session = db_module.SessionLocal()
    try:
        yield session
    finally:
        session.close()
        _reset_db()


@pytest.fixture
def client(db_session):
    """A TestClient whose every request shares this test's own db_session
    -- so anything the test asserts against db_session sees exactly what
    the app just did. Table data from this test is wiped once db_session
    tears down (see _wipe_all_tables above)."""

    def _override_get_db():
        yield db_session

    app.dependency_overrides[db_module.get_db] = _override_get_db
    # follow_redirects=False: nearly every POST route in this app answers
    # success with a 303 (POST/redirect/GET) -- tests assert against that
    # 303 directly, matching what a real browser's network tab would show
    # before it follows the redirect itself.
    with TestClient(app, follow_redirects=False) as c:
        yield c
    app.dependency_overrides.clear()


def make_user(db: Session, username: str, password: str = "testpass123", **kwargs) -> User:
    """A user row directly via the ORM -- bypasses the HTTP signup/claim
    flow for tests that only care about what happens once a user already
    exists (permission boundaries, source ownership, etc.)."""
    import time as _time

    user = User(
        username=username,
        display_name=kwargs.pop("display_name", username.capitalize()),
        password_hash=hash_password(password),
        created_at=_time.time(),
        **kwargs,
    )
    db.add(user)
    db.flush()
    return user


def link_user(db: Session, user: User, frame: Frame) -> None:
    db.add(UserFrame(user_id=user.id, frame_id=frame.id))
    db.flush()


def claim_device(db: Session, frame: Frame, device_id: str = "001122334455",
                  token: str = "devtok-1") -> str:
    """Gives `frame` device credentials and returns the "id=...&token=..."
    query string real firmware always sends -- require_device has no
    fallback for a bare /frame/* request without ?id= (the old shared-
    MANAGEMENT_TOKEN/no-id path this project used to resolve to a single
    legacy frame is gone), so any device-facing test needs this."""
    frame.device_id = device_id
    frame.device_token = token
    db.commit()
    return f"id={device_id}&token={token}"


def login(client: TestClient, username: str, password: str = "testpass123") -> None:
    resp = client.post("/login", data={"username": username, "password": password})
    assert resp.status_code == 303, resp.text


def get_csrf_token(client: TestClient, page_url: str) -> str:
    """Scrapes the csrf_token hidden field out of a rendered page -- the
    same value a real browser's form submit would carry, see auth.py's
    _csrf_ok."""
    import re

    resp = client.get(page_url)
    assert resp.status_code == 200, resp.text
    m = re.search(r'name="csrf_token" value="([^"]+)"', resp.text)
    assert m, f"no csrf_token found on {page_url}"
    return m.group(1)


def csrf_headers(client: TestClient, page_url: str = "/settings") -> dict:
    """X-CSRF-Token header for JSON API POSTs (require_user_api's
    _csrf_ok checks this header, not a form field -- see common.js'
    fetchJson, which reads the same <meta name="csrf-token"> tag this
    scrapes)."""
    import re

    resp = client.get(page_url)
    assert resp.status_code == 200, resp.text
    m = re.search(r'name="csrf-token" content="([^"]+)"', resp.text)
    assert m, f"no csrf-token meta tag found on {page_url}"
    return {"X-CSRF-Token": m.group(1)}