svcforge/tests/integration/conftest.py

"""Integration fixtures: a real Postgres, real SQL, no mocks.

The DB is never mocked. A mocked database proves your mock returns what you told it to.
Every bug worth catching here — SKIP LOCKED semantics, CAS rowcounts, transaction
rollback, `timestamptz` round-tripping — lives in the part a mock replaces.

Two ways to get a database, in priority order:

1. `SVCFORGE_TEST_DSN` in the environment — an already-running Postgres. This is the
   path on a host that has no Docker daemon (for example one whose containerd belongs
   to a Kubernetes kubelet, where installing Docker would evict the runtime).
2. testcontainers, which starts `postgres:16-alpine` and throws it away after. This is
   the CI path.

Same tests either way.
"""

from __future__ import annotations

import os
from collections.abc import AsyncIterator, Iterator
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit

import psycopg
import pytest
import pytest_asyncio

from svcforge_core.repo.db import DictPool, make_pool

MIGRATIONS = Path(__file__).resolve().parents[2] / "migrations"


def _apply_migrations(dsn: str) -> None:
    """Run every migration in lexical order, one transaction each.

    No `create table if not exists` and no reset: every caller hands this a database that
    was created moments ago (a per-process clone, or a fresh container), so the schema is
    always empty and the migrations always apply cleanly from zero.
    """
    with psycopg.connect(dsn, autocommit=True) as conn:
        for path in sorted(MIGRATIONS.glob("*.sql")):
            with conn.transaction(), conn.cursor() as cur:
                cur.execute(path.read_text(encoding="utf-8"))


def _private_database(admin_dsn: str) -> Iterator[str]:
    """Clone a scratch database for THIS pytest process only, and drop it after.

    The `pool` fixture truncates between tests. That is correct within one process and
    catastrophic across several: two pytest runs sharing a database truncate each other's
    rows mid-test, and the failures look like real bugs in the code under test rather than
    like the harness eating itself. Isolating per process makes concurrent runs
    (several agents, or pytest-xdist -n auto) simply work.
    """
    name = f"svcforge_test_{os.getpid()}"
    parsed = urlsplit(admin_dsn)

    with psycopg.connect(admin_dsn, autocommit=True) as conn:
        conn.execute(f'drop database if exists "{name}"')
        conn.execute(f'create database "{name}"')

    dsn = urlunsplit(parsed._replace(path=f"/{name}"))
    try:
        _apply_migrations(dsn)
        yield dsn
    finally:
        with psycopg.connect(admin_dsn, autocommit=True) as conn:
            # Boot any lingering connections, or the drop blocks forever.
            conn.execute(
                "select pg_terminate_backend(pid) from pg_stat_activity where datname = %s",
                (name,),
            )
            conn.execute(f'drop database if exists "{name}"')


@pytest.fixture(scope="session")
def pg_dsn() -> Iterator[str]:
    """A migrated Postgres, private to this process, from the environment or a container."""
    env_dsn = os.getenv("SVCFORGE_TEST_DSN")
    if env_dsn:
        yield from _private_database(env_dsn)
        return

    try:
        from testcontainers.postgres import PostgresContainer
    except ImportError:  # pragma: no cover - CI always has it
        pytest.skip("set SVCFORGE_TEST_DSN or install testcontainers")

    # A container is already private to this process; no need to clone inside it.
    with PostgresContainer("postgres:16-alpine", driver=None) as pg:
        dsn = pg.get_connection_url()
        _apply_migrations(dsn)
        yield dsn


@pytest_asyncio.fixture
async def pool(pg_dsn: str) -> AsyncIterator[DictPool]:
    """A clean database and an open pool, per test.

    max_size=60 is a correctness requirement. test_skip_locked_claims_each_task_exactly_once
    races 50 concurrent claims; a pool smaller than that serialises them at the pool
    instead of at the database, and the test passes while proving nothing.
    """
    async with await psycopg.AsyncConnection.connect(pg_dsn, autocommit=True) as conn:
        # catalog_versions joins the list because a halted rollout is sticky by design:
        # leave it behind and every later test in the session sees an empty work list.
        await conn.execute("truncate tasks, instances, catalog_versions restart identity cascade")

    p = make_pool(pg_dsn, min_size=1, max_size=60)
    await p.open(wait=True)
    try:
        yield p
    finally:
        await p.close()