loyaly-catalogue/backend/tests/conftest.py

"""
Shared test setup.

Every environment variable here must be set BEFORE `app.main` is imported,
because app/infrastructure/settings.py reads the environment once at import
time.

Hermetic by construction:
  * no web search, no LLM calls (USE_DDG_SEARCH / USE_GOOGLE_CSE / USE_OLLAMA off)
  * database tests use their OWN database, electronics_catalog_test, on the
    local Docker server - never the real electronics_catalog data, and the
    settings guard makes any non-local host impossible anyway. They are
    skipped when that server is not running.
"""
from __future__ import annotations

import base64
import hashlib
import os
import secrets
import sys
from pathlib import Path

import pytest

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))

TEST_ADMIN_PASSWORD = "test-admin-password"
TEST_USER_PASSWORD = "test-user-password"
TEST_API_KEY = "test-api-key-value-not-a-real-secret"
TEST_DB_NAME = "electronics_catalog_test"


def _hash(password: str, iterations: int = 20_000) -> str:
    """Byte-compatible with security.hash_password, at a low iteration count."""
    salt = secrets.token_bytes(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, iterations)
    return "$".join(("pbkdf2_sha256", str(iterations), base64.b64encode(salt).decode(),
                     base64.b64encode(digest).decode()))


# Local DB connection details come from backend/.env when it exists (the
# password is generated per machine); the database NAME is always the test one.
try:
    from dotenv import dotenv_values

    _local_env = dotenv_values(BACKEND / ".env")
except Exception:  # noqa: BLE001
    _local_env = {}
os.environ["DB_HOST"] = "127.0.0.1"
os.environ["DB_PORT"] = _local_env.get("DB_PORT") or "5433"
os.environ["DB_USER"] = _local_env.get("DB_USER") or "postgres"
os.environ["DB_PASSWORD"] = _local_env.get("DB_PASSWORD") or "test-password-not-real"
os.environ["DB_NAME"] = TEST_DB_NAME
os.environ["DB_CONNECT_TIMEOUT_SECONDS"] = "3"

os.environ["USE_OLLAMA"] = "false"
os.environ["ELEC_USE_LLM"] = "false"
os.environ["USE_DDG_SEARCH"] = "false"
os.environ["USE_GOOGLE_CSE"] = "false"
os.environ["GOOGLE_API_KEY"] = ""
os.environ["GOOGLE_CSE_ID"] = ""

# Auth is set unconditionally: the suite asserts on the real guards.
os.environ["AUTH_ENABLED"] = "true"
os.environ["AUTH_ALLOW_ANY_LOGIN"] = "false"
os.environ["AUTH_SECRET_KEY"] = "test-secret-key-not-for-production-use-at-all"
os.environ["AUTH_ADMIN_USERNAME"] = "admin"
os.environ["AUTH_ADMIN_PASSWORD_HASH"] = _hash(TEST_ADMIN_PASSWORD)
os.environ["AUTH_USER_USERNAME"] = "user"
os.environ["AUTH_USER_PASSWORD_HASH"] = _hash(TEST_USER_PASSWORD)
os.environ["API_KEYS"] = f"test-machine:user:{TEST_API_KEY}"
os.environ["AUTH_MAX_LOGIN_ATTEMPTS"] = "3"
os.environ["AUTH_LOCKOUT_SECONDS"] = "60"

from fastapi.testclient import TestClient  # noqa: E402

from app.main import app  # noqa: E402


@pytest.fixture(scope="session")
def client() -> TestClient:
    return TestClient(app)


@pytest.fixture(autouse=True)
def _reset_login_throttle():
    from app.api.routers import auth as auth_router

    with auth_router._failures_lock:
        auth_router._failures.clear()
    yield
    with auth_router._failures_lock:
        auth_router._failures.clear()


def _token(client: TestClient, username: str, password: str) -> str:
    resp = client.post("/api/auth/login", json={"username": username, "password": password})
    assert resp.status_code == 200, resp.text
    return resp.json()["access_token"]


@pytest.fixture
def admin_headers(client: TestClient) -> dict:
    return {"Authorization": f"Bearer {_token(client, 'admin', TEST_ADMIN_PASSWORD)}"}


@pytest.fixture
def user_headers(client: TestClient) -> dict:
    return {"Authorization": f"Bearer {_token(client, 'user', TEST_USER_PASSWORD)}"}


_DATA_TABLES = ("product_image, product_listing_map, price_history, source_listing, product, "
                "fetch_log, search_cache, crawl_run")


@pytest.fixture(scope="session")
def test_database():
    """Create/migrate/seed electronics_catalog_test once per session, or skip."""
    import psycopg

    try:
        with psycopg.connect(host="127.0.0.1", port=os.environ["DB_PORT"], dbname="postgres",
                             user=os.environ["DB_USER"], password=os.environ["DB_PASSWORD"],
                             connect_timeout=3, autocommit=True) as admin:
            exists = admin.execute("SELECT 1 FROM pg_database WHERE datname = %s", (TEST_DB_NAME,)).fetchone()
            if not exists:
                admin.execute(f'CREATE DATABASE "{TEST_DB_NAME}"')
    except Exception as exc:  # noqa: BLE001
        pytest.skip(f"local Postgres not reachable ({exc}); run `docker compose up -d`")
    from app.electronics.db import repository as repo
    from app.electronics.db.migrate import run_migrations
    from app.electronics.reference import load_reference

    run_migrations()
    repo.seed_reference(load_reference())
    return TEST_DB_NAME


@pytest.fixture
def db(test_database):
    """A clean test database for one test (reference data kept)."""
    from app.electronics.db.connection import connect

    with connect(autocommit=True) as conn:
        conn.execute(f"TRUNCATE {', '.join('elec.' + t.strip() for t in _DATA_TABLES.split(','))} CASCADE")
        conn.execute("UPDATE elec.site SET probe_outcome = NULL, breaker_until = NULL, breaker_reason = NULL")
    yield test_database