Files

81 lines
2.8 KiB
Python

"""Shared fixtures for the Denya OneCare test suite.
Sets DATABASE_URL to an isolated temp SQLite file BEFORE importing any app
module (the engine is created at import time), then provisions tables and
seed data per test.
"""
from __future__ import annotations
import os
import tempfile
_TMP_DIR = tempfile.mkdtemp(prefix="denya-test-")
os.environ["DATABASE_URL"] = f"sqlite+aiosqlite:///{_TMP_DIR}/test.db"
# HARDENING.md P0.1/P0.2: the app now fails closed without a real SECRET_KEY
# and an explicit CORS allow-list — tests must satisfy both.
os.environ.setdefault("SECRET_KEY", "test-secret-key-not-for-production-0123456789abcdef")
os.environ.setdefault("CORS_ORIGINS", "http://test")
import pytest # noqa: E402
import pytest_asyncio # noqa: E402 (DATABASE_URL must be set before app imports)
from httpx import ASGITransport, AsyncClient # noqa: E402
from app.core.database import Base, async_session_factory, engine # noqa: E402
from app.core.ratelimit import login_rate_limiter # noqa: E402
from app.main import app # noqa: E402
from app.models.ticket import Ticket # noqa: E402
from app.services.seed import seed_categories, seed_units, seed_users # noqa: E402
@pytest.fixture
def _reset_login_rate_limiter():
"""Isolate login rate-limit state between tests (shared in-process store)."""
login_rate_limiter.reset()
yield
login_rate_limiter.reset()
@pytest_asyncio.fixture
async def client(_reset_login_rate_limiter):
"""Async test client with a fresh, seeded database per test."""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with async_session_factory() as session:
await seed_users(session)
await session.commit()
# json_path=None → built-in fallback units (same data as the committed
# apartment_mapping.json, kept deterministic for tests)
await seed_units(session, json_path=None)
await session.commit()
await seed_categories(session)
await session.commit()
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest_asyncio.fixture
async def seed_tickets():
"""Insert `n` tickets directly into the DB; returns the count inserted."""
async def _seed(n: int) -> int:
async with async_session_factory() as session:
for i in range(n):
session.add(
Ticket(
ticket_number=f"PAV-TEST-{i:05d}",
status="Logged",
priority="medium",
description=f"Test ticket {i}",
)
)
await session.commit()
return n
return _seed