tests/conftest.py

"""Shared test fixtures.

Unit tests run against an in-memory sqlite DB (shared across sessions via
StaticPool) so the suite needs no external services. The Postgres/pgvector
path is exercised by guarded integration tests (``TEST_DATABASE_URL``).
"""

from __future__ import annotations

import os

import pytest
import pytest_asyncio
from sqlalchemy.pool import StaticPool

# Ensure env is sane before any app import reads Settings.
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite://")
os.environ.setdefault("TESTING", "1")
os.environ.setdefault("ADMIN_TOKEN", "test-admin-token")
os.environ.setdefault("SHOPIFY_APP_PROXY_SECRET", "test-proxy-secret")

from app import db  # noqa: E402


@pytest_asyncio.fixture
async def db_session():
    """A fresh in-memory sqlite DB with all tables created, per test."""
    db.init_engine(
        "sqlite+aiosqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    # Import models so they register on Base.metadata before create_all.
    try:
        import app.models  # noqa: F401
    except ModuleNotFoundError:
        pass
    await db.create_all()
    async with db.get_sessionmaker()() as session:
        yield session
    await db.get_engine().dispose()


@pytest.fixture
def is_postgres() -> bool:
    return db.is_postgres(os.environ.get("DATABASE_URL", ""))


@pytest_asyncio.fixture
async def tenant(db_session):
    """A committed tenant for route tests (slug 't1')."""
    from app.models import Tenant

    t = Tenant(slug="t1", name="Tienda 1", brand_name="Asistente T1")
    db_session.add(t)
    await db_session.commit()
    return t


@pytest_asyncio.fixture
async def default_tenant(db_session):
    """A committed 'default' tenant (used by the App Proxy fallback)."""
    from app.models import Tenant

    t = Tenant(slug="default", name="Default", brand_name="Asistente")
    db_session.add(t)
    await db_session.commit()
    return t


@pytest.fixture(autouse=True)
def _reset_order_limiter():
    """Isolate the in-process rate limiters between tests."""
    from app.ratelimit import get_order_limiter
    from app.routes import chat

    get_order_limiter().clear()
    chat._chat_limiter.clear()
    yield
    get_order_limiter().clear()
    chat._chat_limiter.clear()


@pytest_asyncio.fixture
async def app_client(db_session):
    """ASGI test client whose DB dependency is bound to the test session."""
    import httpx

    from app.db import get_session
    from app.main import create_app

    app = create_app()

    async def _override_session():
        yield db_session

    app.dependency_overrides[get_session] = _override_session
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
        yield app, client
    app.dependency_overrides.clear()