"""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()