#!/usr/bin/env python3
"""Pytest configuration for AgentForms tests.
Ensures per-worker isolated test DB when running with pytest-xdist
(-n auto or -n NUM), preventing cross-worker SQLite lock contention.
Without xdist: uses a single shared temp DB (same as before).
With xdist: each worker gets its own temp DB file for isolation.
Sets up the test environment BEFORE app modules are imported, since
RELAY_DB_PATH is read at module import time by app.db.
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# ─── Set test env vars BEFORE any app imports ──────────────────────────
# These are read at module import time by app.db and cached
os.environ.setdefault("AGENTFORMS_SECRET_KEY", "test-secret-key-for-testing")
os.environ.setdefault("STRIPE_WEBHOOK_SECRET", "whsec_test_secret")
os.environ.setdefault("ENCRYPTION_KEY", "5c321ae59453d7bae2e05a9e19e97347411a033d9858a0456d5cc63acfcb0369")
# ─── Per-worker isolated test DB ────────────────────────────────────────
# Create a unique temp DB file per pytest worker to avoid SQLite
# database-is-locked errors when running with pytest-xdist.
#
# Key: we set RELAY_DB_PATH before importing app modules so that
# app.db.DB_PATH picks up the correct path.
import tempfile
# Module-level tracking so we only set up once per process
_test_db_setup_done = False
_test_db_path = None
def _setup_test_db(worker_id=None):
"""Create an isolated test DB file for this worker/process.
Args:
worker_id: pytest-xdist worker identifier (e.g., 'gw0', 'gw1')
or 'master' for non-xdist runs.
"""
global _test_db_setup_done, _test_db_path
if _test_db_setup_done:
return _test_db_path
# Create a temp DB file with a name that includes the worker ID
# This ensures parallel workers don't share the same file
suffix = f"_{worker_id or 'master'}.db"
_temp_db = tempfile.NamedTemporaryFile(
suffix=suffix,
delete=False,
prefix="agentforms_test_",
)
_test_db_path = _temp_db.name
_temp_db.close()
# Set env var BEFORE importing app modules
os.environ["RELAY_DB_PATH"] = _test_db_path
# Import and patch DB_PATH for modules that already imported it
try:
import app.db
app.db.DB_PATH = _test_db_path
except ImportError:
pass
_test_db_setup_done = True
return _test_db_path
def pytest_configure(config):
"""Called by pytest before test collection — set up test DB per worker."""
# Detect xdist worker ID if available
worker_id = getattr(config, "workerinput", None)
if worker_id is not None:
# Running under pytest-xdist in a worker process
worker_id = worker_id.get("workerid", "gw0")
else:
worker_id = "master"
_setup_test_db(worker_id)
@pytest.fixture(autouse=True)
def _flush_rate_limits():
"""Flush Redis rate limit keys before each test to avoid cross-test pollution."""
from app.services.ratelimit import limiter
limiter.flush("rate:*")
# Also reset in-memory token buckets from the API module
try:
from app.routes.api import _RATE_LIMIT, _SITE_RATE_LIMIT, _DYNAMIC_SITE_RATE_LIMIT
_RATE_LIMIT.reset()
_SITE_RATE_LIMIT.reset()
_DYNAMIC_SITE_RATE_LIMIT.reset()
except Exception:
pass # Graceful degradation if API module not loaded
@pytest.fixture(scope="session")
def _test_db_cleanup():
"""Clean up the test DB file after all tests in this worker complete."""
yield
global _test_db_path
if _test_db_path and os.path.exists(_test_db_path):
try:
os.unlink(_test_db_path)
except OSError:
pass # File might already be cleaned up by another process