neuronetz-gateway/tests/conftest.py

"""Shared pytest fixtures for the neuronetz-gateway test suite.

Phase 1 scaffold. Provides testcontainers-backed Postgres and Redis fixtures
per SPEC §10, but guards them so collection never fails when Docker or
``testcontainers`` is unavailable (e.g. CI on an empty test suite). When the
dependency or the Docker daemon is missing, tests requesting these fixtures are
skipped with a clear reason rather than erroring at collection time.

Later phases (2+) consume these fixtures for integration tests against a real
Postgres + Redis. Keep this module import-clean: no heavyweight imports at
module top level beyond the standard library and pytest.
"""

from __future__ import annotations

from collections.abc import AsyncIterator, Iterator
from typing import TYPE_CHECKING

import pytest

if TYPE_CHECKING:
    # Imported only for type checkers; never required at runtime/collection.
    import redis.asyncio as aioredis
    from fastapi import FastAPI
    from testcontainers.postgres import PostgresContainer
    from testcontainers.redis import RedisContainer


def _testcontainers_available() -> bool:
    """Return True if ``testcontainers`` is importable and Docker looks usable.

    We deliberately avoid spinning up a container here; we only check that the
    library imports. The actual fixtures attempt to start a container and skip
    gracefully on failure (e.g. Docker daemon not running).
    """
    try:
        import testcontainers.core.container  # noqa: F401
    except ImportError:
        return False
    return True


@pytest.fixture(scope="session")
def postgres_container() -> Iterator[PostgresContainer]:
    """Session-scoped Postgres container.

    Skips (does not error) if testcontainers/Docker are unavailable so that an
    empty or unit-only test run still passes in CI without Docker.

    Yields the started ``PostgresContainer``; later phases derive the async
    SQLAlchemy URL from ``container.get_connection_url()`` (swap the driver to
    ``postgresql+asyncpg``).
    """
    if not _testcontainers_available():
        pytest.skip("testcontainers/Docker unavailable; skipping Postgres-backed test")

    from testcontainers.postgres import PostgresContainer

    try:
        container = PostgresContainer("postgres:16-alpine")
        container.start()
    except Exception as exc:  # noqa: BLE001 - any startup failure means skip, not fail
        pytest.skip(f"could not start Postgres testcontainer: {exc}")

    try:
        yield container
    finally:
        container.stop()


@pytest.fixture(scope="session")
def redis_container() -> Iterator[RedisContainer]:
    """Session-scoped Redis container.

    Skips (does not error) if testcontainers/Docker are unavailable.

    Yields the started ``RedisContainer``; later phases build a ``redis://``
    URL from ``container.get_container_host_ip()`` and the mapped port.
    """
    if not _testcontainers_available():
        pytest.skip("testcontainers/Docker unavailable; skipping Redis-backed test")

    from testcontainers.redis import RedisContainer

    try:
        container = RedisContainer("redis:7-alpine")
        container.start()
    except Exception as exc:  # noqa: BLE001 - any startup failure means skip, not fail
        pytest.skip(f"could not start Redis testcontainer: {exc}")

    try:
        yield container
    finally:
        container.stop()


@pytest.fixture()
def postgres_url(postgres_container: PostgresContainer) -> str:
    """Async SQLAlchemy connection URL for the Postgres testcontainer.

    Rewrites the default psycopg URL to the asyncpg driver the gateway uses.
    """
    url: str = postgres_container.get_connection_url()
    # testcontainers returns e.g. postgresql+psycopg2://...; gateway uses asyncpg.
    for prefix in ("postgresql+psycopg2://", "postgresql+psycopg://", "postgresql://"):
        if url.startswith(prefix):
            return "postgresql+asyncpg://" + url[len(prefix) :]
    return url


@pytest.fixture()
def redis_url(redis_container: RedisContainer) -> str:
    """``redis://`` connection URL for the Redis testcontainer."""
    host = redis_container.get_container_host_ip()
    port = redis_container.get_exposed_port(6379)
    return f"redis://{host}:{port}/0"


@pytest.fixture()
def mock_ollama_app() -> FastAPI:
    """A fresh in-process mock Ollama ASGI app (the integration upstream).

    Defined here at the project root so both unit and integration tests can
    request it. The behaviour lives in ``tests/integration/mock_ollama.py``.
    """
    from tests.integration.mock_ollama import create_mock_ollama

    return create_mock_ollama()


@pytest.fixture()
async def redis_client(redis_url: str) -> AsyncIterator[aioredis.Redis]:
    """A connected async Redis client against the testcontainer, flushed clean.

    Each test gets an empty keyspace (FLUSHDB on entry) so rate-limit and
    budget counters never leak between tests. The client is closed on teardown.
    """
    import redis.asyncio as aioredis

    client: aioredis.Redis = aioredis.from_url(redis_url, decode_responses=True)
    await client.flushdb()
    try:
        yield client
    finally:
        await client.flushdb()
        await client.aclose()