From f0ee256436e184dd8539dd608096bb816e292b90 Mon Sep 17 00:00:00 2001 From: JeyBee Date: Sat, 11 Jul 2026 14:19:52 +0000 Subject: [PATCH 1/5] test: run suite in parallel with pytest-xdist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add pytest-xdist (latest stable, 3.8.0) to the dev extras and default pytest to 4 workers via addopts. Session-scoped fixtures are per-worker under xdist, so each worker starts its own Postgres container and template database — the speedup comes from a few extra Postgres servers rather than aggressive process-level concurrency. Per-test databases already use UUID names, so the suite is parallel-safe as-is. Use `-n 0` to run serially. --- AGENTS.md | 9 +++++++-- pyproject.toml | 5 ++++- uv.lock | 24 ++++++++++++++++++++++++ 3 files changed, 35 insertions(+), 3 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 90aba780..fd1fa947 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -36,9 +36,12 @@ uv run ruff format . --check # Format check uv run lint-imports # Hexagonal architecture boundary validation uv run mypy . # Type checking (strict, targets Python 3.10) -# Run all tests +# Run all tests (parallel by default: 4 xdist workers, one Postgres container each) uv run pytest +# Run tests serially (disable pytest-xdist) +uv run pytest -n 0 + # Run a single test file uv run pytest test/test_queries.py @@ -225,7 +228,9 @@ When a comment explains a non-obvious workaround, keep it short and put it adjac ## Testing Conventions -- **pytest config**: `asyncio_mode = "auto"`, 20s timeout per test, `--durations=15` +- **pytest config**: `asyncio_mode = "auto"`, 20s timeout per test, `--durations=15`, `-n 4` (pytest-xdist) +- Tests run in parallel via **pytest-xdist**: session-scoped fixtures are per-worker, so each worker starts its own Postgres container and template database. Pass `-n 0` to run serially, or `-n ` to change worker count. +- Keep tests xdist-safe: no fixed ports, no shared files, no cross-test state. Database isolation comes from the per-test `CREATE DATABASE ... TEMPLATE` fixture. - Declare async tests as `async def test_...` -- no `@pytest.mark.asyncio` decorator needed - Annotate all test function parameters: `async def test_foo(apgdriver: db.Driver, N: int) -> None:` - Use `@pytest.mark.parametrize` for multi-value testing diff --git a/pyproject.toml b/pyproject.toml index 59d707fb..18eb3151 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -89,6 +89,7 @@ dev = [ "opentelemetry-semantic-conventions>=0.41b0", "pytest-mock>=3.14.0", "pytest-timeout>=2.3.1", + "pytest-xdist>=3.8.0", "ruff", "tqdm", "types-croniter", @@ -154,7 +155,9 @@ explicit_package_bases = true asyncio_mode = "auto" timeout = "20" timeout_func_only = true -addopts = "--durations=15" +# -n 4: modest parallelism; each xdist worker gets its own session-scoped +# Postgres container, so this scales the suite across a few extra servers. +addopts = "--durations=15 -n 4" # --------------------------------------------------------------------------- # Import-linter: enforce hexagonal architecture boundaries diff --git a/uv.lock b/uv.lock index 818f054c..2406bb60 100644 --- a/uv.lock +++ b/uv.lock @@ -572,6 +572,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/36/f4/c6e662dade71f56cd2f3735141b265c3c79293c109549c1e6933b0651ffc/exceptiongroup-1.3.0-py3-none-any.whl", hash = "sha256:4d111e6e0c13d0644cad6ddaa7ed0261a0b36971f6d23e7ec9b4b9097da78a10", size = 16674, upload-time = "2025-05-10T17:42:49.33Z" }, ] +[[package]] +name = "execnet" +version = "2.1.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/bf/89/780e11f9588d9e7128a3f87788354c7946a9cbb1401ad38a48c4db9a4f07/execnet-2.1.2.tar.gz", hash = "sha256:63d83bfdd9a23e35b9c6a3261412324f964c2ec8dcd8d3c6916ee9373e0befcd", size = 166622, upload-time = "2025-11-12T09:56:37.75Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ab/84/02fc1827e8cdded4aa65baef11296a9bbe595c474f0d6d758af082d849fd/execnet-2.1.2-py3-none-any.whl", hash = "sha256:67fba928dd5a544b783f6056f449e5e3931a5c378b128bc18501f7ea79e296ec", size = 40708, upload-time = "2025-11-12T09:56:36.333Z" }, +] + [[package]] name = "executing" version = "2.2.1" @@ -1372,6 +1381,7 @@ dev = [ { name = "pytest-cov" }, { name = "pytest-mock" }, { name = "pytest-timeout" }, + { name = "pytest-xdist" }, { name = "ruff" }, { name = "testcontainers" }, { name = "time-machine" }, @@ -1451,6 +1461,7 @@ requires-dist = [ { name = "pytest-cov", marker = "extra == 'dev'" }, { name = "pytest-mock", marker = "extra == 'dev'", specifier = ">=3.14.0" }, { name = "pytest-timeout", marker = "extra == 'dev'", specifier = ">=2.3.1" }, + { name = "pytest-xdist", marker = "extra == 'dev'", specifier = ">=3.8.0" }, { name = "ruff", marker = "extra == 'dev'" }, { name = "sentry-sdk", marker = "extra == 'sentry'", specifier = ">=2.0.0" }, { name = "tabulate", specifier = ">=0.9.0" }, @@ -1775,6 +1786,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fa/b6/3127540ecdf1464a00e5a01ee60a1b09175f6913f0644ac748494d9c4b21/pytest_timeout-2.4.0-py3-none-any.whl", hash = "sha256:c42667e5cdadb151aeb5b26d114aff6bdf5a907f176a007a30b940d3d865b5c2", size = 14382, upload-time = "2025-05-05T19:44:33.502Z" }, ] +[[package]] +name = "pytest-xdist" +version = "3.8.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "execnet" }, + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/78/b4/439b179d1ff526791eb921115fca8e44e596a13efeda518b9d845a619450/pytest_xdist-3.8.0.tar.gz", hash = "sha256:7e578125ec9bc6050861aa93f2d59f1d8d085595d6551c2c90b6f4fad8d3a9f1", size = 88069, upload-time = "2025-07-01T13:30:59.346Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ca/31/d4e37e9e550c2b92a9cbc2e4d0b7420a27224968580b5a447f420847c975/pytest_xdist-3.8.0-py3-none-any.whl", hash = "sha256:202ca578cfeb7370784a8c33d6d05bc6e13b4f25b5053c30a152269fd10f0b88", size = 46396, upload-time = "2025-07-01T13:30:56.632Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0" From 1c72af09422f718fd19085384cbc73ec046e4ab3 Mon Sep 17 00:00:00 2001 From: JeyBee Date: Sat, 11 Jul 2026 14:40:50 +0000 Subject: [PATCH 2/5] test: fix postgres container startup race under xdist The postgres image logs "database system is ready to accept connections" twice: once from the initdb bootstrap server (unix socket only) and once from the real server. The wait strategy matched the first occurrence, so fixtures connected before TCP was listening and got connection resets. With 4 workers running initdb concurrently on a 4-vCPU runner the bootstrap window stretches to seconds, making the race near-certain; the failed session fixture is then cached by pytest, erroring every DB test in that worker. Wait for the second occurrence, as testcontainers-java does. Also bound per-container memory (shared_buffers 128MB, max_wal_size 512MB, tmpfs capped at 1g) since each xdist worker now runs its own container with its data dir on tmpfs (RAM). --- test/conftest.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/test/conftest.py b/test/conftest.py index 179b8c6e..11f29893 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -62,7 +62,12 @@ def __init__( self.with_env("POSTGRES_USER", self.username) self.with_env("POSTGRES_PASSWORD", self.password) self.with_env("POSTGRES_DB", self.dbname) - self.waiting_for(LogMessageWaitStrategy("database system is ready to accept connections")) + # times=2: the postgres image prints the ready line once from the initdb + # bootstrap server (unix socket only) before the real server listens on + # TCP; waiting for the first occurrence races connection resets. + self.waiting_for( + LogMessageWaitStrategy("database system is ready to accept connections", times=2) + ) def get_connection_url(self, host: str | None = None) -> str: if self._container is None: @@ -94,6 +99,10 @@ async def postgres_container() -> AsyncGenerator[str, None]: # https://postgresqlco.nf/doc/en/param/vacuum_buffer_usage_limit/ # Assume worst case for backwards compatibility + # + # Memory is bounded per container (shared_buffers, max_wal_size, tmpfs size) + # because each xdist worker starts its own container and the data dir lives + # on tmpfs (RAM). commands = [ "postgres", "-c", @@ -107,11 +116,11 @@ async def postgres_container() -> AsyncGenerator[str, None]: "-c", "checkpoint_timeout=30min", "-c", - "max_wal_size=4GB", + "max_wal_size=512MB", "-c", "checkpoint_completion_target=0.9", "-c", - "shared_buffers=256MB", + "shared_buffers=128MB", "-c", "autovacuum_naptime=5s", ] + (["-c", "vacuum_buffer_usage_limit=8MB"] if int(postgres_version) >= 16 else []) @@ -119,7 +128,7 @@ async def postgres_container() -> AsyncGenerator[str, None]: container = ( PgQueuerPostgresContainer(f"postgres:{postgres_version}", driver=None) .with_command(commands) - .with_kwargs(tmpfs={"/var/lib/pg/data": "rw"}) + .with_kwargs(tmpfs={"/var/lib/pg/data": "rw,size=1g"}) .with_envs(PGDATA="/var/lib/pg/data") ) From 5ccd2c7e7184ed32ad4e4b9022883dc8d3b93d30 Mon Sep 17 00:00:00 2001 From: JeyBee Date: Sat, 11 Jul 2026 14:53:09 +0000 Subject: [PATCH 3/5] test: anchor container readiness on real server startup sequence LogMessageWaitStrategy in testcontainers 4.13 ignores its `times` argument, so the previous fix still matched the initdb bootstrap server's ready line and fixtures connected before TCP was listening. Anchor the wait on the bootstrap shutdown line followed by the real ready line instead, and retry the session fixture's first connection for up to a minute as belt and braces. --- test/conftest.py | 32 +++++++++++++++++++++++++++----- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/test/conftest.py b/test/conftest.py index 11f29893..9eaffb1b 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -2,6 +2,7 @@ import asyncio import os +import re import uuid from typing import Any, AsyncGenerator from urllib.parse import quote, urlparse, urlunparse @@ -62,11 +63,19 @@ def __init__( self.with_env("POSTGRES_USER", self.username) self.with_env("POSTGRES_PASSWORD", self.password) self.with_env("POSTGRES_DB", self.dbname) - # times=2: the postgres image prints the ready line once from the initdb - # bootstrap server (unix socket only) before the real server listens on - # TCP; waiting for the first occurrence races connection resets. + # The postgres image prints the ready line once from the initdb + # bootstrap server (unix socket only) before the real server listens + # on TCP, and LogMessageWaitStrategy ignores its `times` argument + # (testcontainers 4.13), so anchor on the bootstrap shutdown line + # followed by the real server's ready line (both on stderr). self.waiting_for( - LogMessageWaitStrategy("database system is ready to accept connections", times=2) + LogMessageWaitStrategy( + re.compile( + r"database system is shut down" + r".*database system is ready to accept connections", + re.DOTALL, + ) + ) ) def get_connection_url(self, host: str | None = None) -> str: @@ -136,13 +145,26 @@ async def postgres_container() -> AsyncGenerator[str, None]: yield running.get_connection_url() +async def create_pool_with_retry(dsn: str, deadline_s: float = 60.0) -> asyncpg.Pool: + """The container port can accept and reset connections while postgres starts up.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + deadline_s + while True: + try: + return await asyncpg.create_pool(dsn=dsn) + except (OSError, asyncpg.CannotConnectNowError): + if loop.time() > deadline: + raise + await asyncio.sleep(0.5) + + @pytest.fixture(scope="session") async def migrated_db(postgres_container: str) -> AsyncGenerator[str, None]: """Ensure the database is migrated before running tests.""" parent = f"parent_{uuid.uuid4().hex}" - async with asyncpg.create_pool(dsn=build_dsn_for(postgres_container, "/postgres")) as pool: + async with await create_pool_with_retry(build_dsn_for(postgres_container, "/postgres")) as pool: await pool.execute(f"CREATE DATABASE {parent}") async with asyncpg.create_pool(dsn=build_dsn_for(postgres_container, f"/{parent}")) as pool: From 70902402be7c912ba9be2f6ca57109351c88fc17 Mon Sep 17 00:00:00 2001 From: JeyBee Date: Sat, 11 Jul 2026 15:01:02 +0000 Subject: [PATCH 4/5] test: wait on init-complete marker for container readiness The bootstrap server's log output is split across stdout and stderr, so both the "ready to accept connections" line and the shutdown+ready sequence are ambiguous stream-wise. "PostgreSQL init process complete" appears exactly once, on stdout, after the bootstrap server is gone; the remaining gap until the real server listens on TCP is bridged by the connection retry in the migrated_db fixture. --- test/conftest.py | 22 ++++++++-------------- 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/test/conftest.py b/test/conftest.py index 9eaffb1b..0d4b43f2 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -63,20 +63,14 @@ def __init__( self.with_env("POSTGRES_USER", self.username) self.with_env("POSTGRES_PASSWORD", self.password) self.with_env("POSTGRES_DB", self.dbname) - # The postgres image prints the ready line once from the initdb - # bootstrap server (unix socket only) before the real server listens - # on TCP, and LogMessageWaitStrategy ignores its `times` argument - # (testcontainers 4.13), so anchor on the bootstrap shutdown line - # followed by the real server's ready line (both on stderr). - self.waiting_for( - LogMessageWaitStrategy( - re.compile( - r"database system is shut down" - r".*database system is ready to accept connections", - re.DOTALL, - ) - ) - ) + # The image's initdb bootstrap server logs its own "ready to accept + # connections" (unix socket only) before the real server listens on + # TCP, and bootstrap output is split across stdout/stderr, so the + # ready line is ambiguous. The init-complete marker appears exactly + # once (stdout) after the bootstrap server is gone; the remaining + # ~100ms until the real server listens is bridged by + # create_pool_with_retry. + self.waiting_for(LogMessageWaitStrategy(re.compile(r"PostgreSQL init process complete"))) def get_connection_url(self, host: str | None = None) -> str: if self._container is None: From 2069cefecf8eafef61d9a4d999aa07bb64a48a8a Mon Sep 17 00:00:00 2001 From: JeyBee Date: Sat, 11 Jul 2026 16:04:29 +0000 Subject: [PATCH 5/5] test: gate container readiness on docker healthcheck MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace log-line matching and the fixture-level connection retry with a docker healthcheck probing pg_isready over TCP, mirroring the service container setup in ci.yml. The initdb bootstrap server only listens on the unix socket, so the container cannot report healthy before the real server accepts TCP connections — readiness is decided by the infrastructure instead of log heuristics or client retries. --- test/conftest.py | 48 ++++++++++++++++++++++-------------------------- 1 file changed, 22 insertions(+), 26 deletions(-) diff --git a/test/conftest.py b/test/conftest.py index 0d4b43f2..fd765aa3 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -2,7 +2,6 @@ import asyncio import os -import re import uuid from typing import Any, AsyncGenerator from urllib.parse import quote, urlparse, urlunparse @@ -22,7 +21,7 @@ uvloop = None # type: ignore[assignment] from testcontainers.core.container import DockerContainer -from testcontainers.core.wait_strategies import LogMessageWaitStrategy +from testcontainers.core.wait_strategies import HealthcheckWaitStrategy from pgqueuer.db import SyncPsycopgDriver @@ -63,14 +62,21 @@ def __init__( self.with_env("POSTGRES_USER", self.username) self.with_env("POSTGRES_PASSWORD", self.password) self.with_env("POSTGRES_DB", self.dbname) - # The image's initdb bootstrap server logs its own "ready to accept - # connections" (unix socket only) before the real server listens on - # TCP, and bootstrap output is split across stdout/stderr, so the - # ready line is ambiguous. The init-complete marker appears exactly - # once (stdout) after the bootstrap server is gone; the remaining - # ~100ms until the real server listens is bridged by - # create_pool_with_retry. - self.waiting_for(LogMessageWaitStrategy(re.compile(r"PostgreSQL init process complete"))) + # Readiness comes from docker's own healthcheck (as in ci.yml service + # containers), probing pg_isready over TCP: the image's initdb + # bootstrap server listens only on the unix socket and logs a + # misleading "ready to accept connections" line, so the container + # cannot report healthy before the real server accepts TCP. + self._kwargs.setdefault( + "healthcheck", + { + "test": ["CMD", "pg_isready", "-h", "127.0.0.1", "-U", self.username], + "interval": 500_000_000, # docker durations are nanoseconds + "timeout": 2_000_000_000, + "retries": 120, + }, + ) + self.waiting_for(HealthcheckWaitStrategy()) def get_connection_url(self, host: str | None = None) -> str: if self._container is None: @@ -129,9 +135,12 @@ async def postgres_container() -> AsyncGenerator[str, None]: ] + (["-c", "vacuum_buffer_usage_limit=8MB"] if int(postgres_version) >= 16 else []) container = ( - PgQueuerPostgresContainer(f"postgres:{postgres_version}", driver=None) + PgQueuerPostgresContainer( + f"postgres:{postgres_version}", + driver=None, + tmpfs={"/var/lib/pg/data": "rw,size=1g"}, + ) .with_command(commands) - .with_kwargs(tmpfs={"/var/lib/pg/data": "rw,size=1g"}) .with_envs(PGDATA="/var/lib/pg/data") ) @@ -139,26 +148,13 @@ async def postgres_container() -> AsyncGenerator[str, None]: yield running.get_connection_url() -async def create_pool_with_retry(dsn: str, deadline_s: float = 60.0) -> asyncpg.Pool: - """The container port can accept and reset connections while postgres starts up.""" - loop = asyncio.get_running_loop() - deadline = loop.time() + deadline_s - while True: - try: - return await asyncpg.create_pool(dsn=dsn) - except (OSError, asyncpg.CannotConnectNowError): - if loop.time() > deadline: - raise - await asyncio.sleep(0.5) - - @pytest.fixture(scope="session") async def migrated_db(postgres_container: str) -> AsyncGenerator[str, None]: """Ensure the database is migrated before running tests.""" parent = f"parent_{uuid.uuid4().hex}" - async with await create_pool_with_retry(build_dsn_for(postgres_container, "/postgres")) as pool: + async with asyncpg.create_pool(dsn=build_dsn_for(postgres_container, "/postgres")) as pool: await pool.execute(f"CREATE DATABASE {parent}") async with asyncpg.create_pool(dsn=build_dsn_for(postgres_container, f"/{parent}")) as pool: