azk/chatbot_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(test): share container fixtures across the integration and e2e suites

Browse Source 
Why:
- tests/e2e/ cannot reach fixtures defined in a per-boundary conftest, and
  `pytest_plugins` is only honoured in the root conftest.

Changes:
- move the Postgres/MinIO/Qdrant container fixtures into
  tests/support/containers.py and register it as a root plugin
- fold MinIO bucket creation into `minio_settings`; an autouse fixture in a
  globally registered plugin would pull a container into unit runs
- add a `postgres_settings` fixture so a component can be built from it directly

Impact:
- no behavior change; `pytest -m unit` still needs no Docker

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Ali Zarinkolah
2026-08-20 22:26:02 +03:30
parent 012b44d5f2
commit 2c72688440
5 changed files with 211 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
tests/conftest.py
View File

@@ -10,6 +10,12 @@ from httpx import ASGITransport, AsyncClient
from src.config import Settings
from src.main import create_app
# Disposable Postgres/MinIO/Qdrant containers (ADR-0016). Registered here
# because `pytest_plugins` is only honoured in the root conftest, and both
# `tests/integration/*/` and `tests/e2e/` need the same containers. The
# fixtures are session-scoped but lazy, so unit runs still need no Docker.
pytest_plugins = ["tests.support.containers"]
@pytest.fixture(autouse=True)
def _reset_structlog_after_test() -> Iterator[None]:

39
tests/integration/minio/conftest.py
View File

@@ -1,39 +0,0 @@
from collections.abc import Iterator
import pytest
from testcontainers.community.minio import MinioContainer
from src.config import MinioSettings
from src.infrastructure.minio.client import create_client
_BUCKET = "test-source-files"
@pytest.fixture(scope="session")
def minio_container() -> Iterator[MinioContainer]:
with MinioContainer() as container:
yield container
@pytest.fixture(scope="session")
def minio_settings(minio_container: MinioContainer) -> MinioSettings:
config = minio_container.get_config()
# Pinned to IPv4 for the same reason as postgres_url (see
# tests/integration/postgres/conftest.py): `localhost` resolves to `::1`
# first, but Docker only publishes the mapped port on IPv4, so the
# connection hangs instead of failing.
endpoint = config["endpoint"].replace("localhost:", "127.0.0.1:")
return MinioSettings(
endpoint=endpoint,
access_key=config["access_key"],
secret_key=config["secret_key"],
secure=False,
bucket=_BUCKET,
)
@pytest.fixture(scope="session", autouse=True)
def _ensure_bucket(minio_settings: MinioSettings) -> None:
client = create_client(minio_settings)
if not client.bucket_exists(minio_settings.bucket):
client.make_bucket(minio_settings.bucket)

98
tests/integration/postgres/conftest.py
View File

@@ -1,98 +0,0 @@
from collections.abc import AsyncIterator, Iterator
import pytest
import pytest_asyncio
from alembic.command import upgrade
from alembic.config import Config
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
from testcontainers.community.postgres import PostgresContainer
from src.config import PostgresSettings
from src.infrastructure.postgres.database import create_engine
def _alembic_config(database_url: str) -> Config:
config = Config("alembic.ini")
config.set_main_option("sqlalchemy.url", database_url)
return config
def _settings_from_url(url: str) -> PostgresSettings:
# testcontainers returns postgresql+asyncpg://user:pass@host:port/db ;
# PostgresSettings builds its own dsn from parts, so parse the parts back out.
without_scheme = url.split("://", 1)[1]
creds, hostpart = without_scheme.split("@", 1)
user, password = creds.split(":", 1)
hostport, db = hostpart.split("/", 1)
host, port = hostport.split(":", 1)
return PostgresSettings(host=host, port=int(port), user=user, password=password, db=db)
@pytest.fixture(scope="session")
def postgres_container() -> Iterator[PostgresContainer]:
with PostgresContainer("postgres:17", driver="asyncpg") as container:
yield container
@pytest.fixture(scope="session")
def postgres_url(postgres_container: PostgresContainer) -> str:
"""The container's URL, pinned to IPv4.
Testcontainers reports the host as `localhost`, which resolves to `::1`
before `127.0.0.1`. Docker publishes the mapped port on IPv4 only, and the
IPv6 SYN is dropped rather than refused, so asyncpg blocks on the first
address until its connect timeout instead of falling back to the second --
the connection does not fail, it hangs.
"""
return postgres_container.get_connection_url().replace("@localhost:", "@127.0.0.1:")
@pytest.fixture(scope="session")
def migrated_postgres_url(postgres_url: str) -> str:
"""The container's URL, after Alembic has created the schema on it once."""
upgrade(_alembic_config(postgres_url), "head")
return postgres_url
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def postgres_engine(migrated_postgres_url: str) -> AsyncIterator[AsyncEngine]:
engine = create_engine(_settings_from_url(migrated_postgres_url))
try:
yield engine
finally:
await engine.dispose()
@pytest_asyncio.fixture(loop_scope="session")
async def db_sessionmaker(
postgres_engine: AsyncEngine,
) -> AsyncIterator[async_sessionmaker[AsyncSession]]:
"""A session *factory* per test, bound to a rolled-back outer transaction.
Every session it produces shares one connection/outer transaction, so
writes `commit()`ed by one session are visible to the next -- needed for
code under test that opens more than one session per operation (auth
resolution, the ADR-0017 two-phase upload) -- while the whole test's
writes still roll back together at teardown (ADR-0016: isolate data per
test).
"""
async with postgres_engine.connect() as connection:
outer_transaction = await connection.begin()
sessionmaker = async_sessionmaker(
bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
)
yield sessionmaker
await outer_transaction.rollback()
@pytest_asyncio.fixture(loop_scope="session")
async def db_session(
db_sessionmaker: async_sessionmaker[AsyncSession],
) -> AsyncIterator[AsyncSession]:
"""One session per test, bound to a rolled-back outer transaction.
Isolates each test's writes (ADR-0016: isolate data per test) without
needing a fresh container or unique keys per test.
"""
async with db_sessionmaker() as session:
yield session

52
tests/integration/qdrant/conftest.py
View File

@@ -1,52 +0,0 @@
import uuid
from collections.abc import AsyncIterator, Iterator
import pytest
import pytest_asyncio
from qdrant_client import AsyncQdrantClient
from testcontainers.community.qdrant import QdrantContainer
from src.config import QdrantSettings
from src.infrastructure.qdrant.client import create_client
# Pinned to match the `qdrant-client` major/minor in pyproject.toml. The
# testcontainers default image trails it far enough that the client emits an
# incompatibility warning, and testing against a version we do not deploy is
# the wrong signal anyway.
_QDRANT_IMAGE = "qdrant/qdrant:v1.19.0"
@pytest.fixture(scope="session")
def qdrant_container() -> Iterator[QdrantContainer]:
with QdrantContainer(image=_QDRANT_IMAGE) as container:
yield container
@pytest.fixture(scope="session")
def qdrant_url(qdrant_container: QdrantContainer) -> str:
"""The container's REST URL, pinned to IPv4.
Same gotcha as postgres_url/minio_settings (see their conftests):
testcontainers reports the host as `localhost`, which resolves to `::1`
first, but Docker publishes the mapped port on IPv4 only. The IPv6 SYN is
dropped rather than refused, so the client hangs until its timeout instead
of falling back to the second address -- the connection does not fail, it
hangs.
"""
host = qdrant_container.get_container_host_ip().replace("localhost", "127.0.0.1")
return f"http://{host}:{qdrant_container.get_exposed_port(6333)}"
@pytest.fixture
def qdrant_settings(qdrant_url: str) -> QdrantSettings:
"""Settings naming a collection unique to this test (ADR-0016 isolation)."""
return QdrantSettings(url=qdrant_url, collection=f"chunks_{uuid.uuid4().hex}")
@pytest_asyncio.fixture(loop_scope="session")
async def qdrant_client(qdrant_settings: QdrantSettings) -> AsyncIterator[AsyncQdrantClient]:
client = create_client(qdrant_settings)
try:
yield client
finally:
await client.close()

205
tests/support/containers.py Normal file
View File

@@ -0,0 +1,205 @@
"""Disposable real infrastructure for integration and e2e tests (ADR-0016).
Registered as a plugin from the root `tests/conftest.py` rather than living in
a per-boundary conftest, because `tests/e2e/` needs the same Postgres, MinIO,
and Qdrant containers that `tests/integration/*/` does, and `pytest_plugins`
is only honoured in the root conftest.
Nothing here is autouse: the container fixtures are session-scoped but lazy,
so `pytest -m unit` still runs without Docker.
"""
import uuid
from collections.abc import AsyncIterator, Iterator
import pytest
import pytest_asyncio
from alembic.command import upgrade
from alembic.config import Config
from qdrant_client import AsyncQdrantClient
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
from testcontainers.community.minio import MinioContainer
from testcontainers.community.postgres import PostgresContainer
from testcontainers.community.qdrant import QdrantContainer
from src.config import MinioSettings, PostgresSettings, QdrantSettings
from src.infrastructure.minio.client import create_client as create_minio_client
from src.infrastructure.postgres.database import create_engine
from src.infrastructure.qdrant.client import create_client as create_qdrant_client
_MINIO_BUCKET = "test-source-files"
# Pinned to match the `qdrant-client` major/minor in pyproject.toml. The
# testcontainers default image trails it far enough that the client emits an
# incompatibility warning, and testing against a version we do not deploy is
# the wrong signal anyway.
_QDRANT_IMAGE = "qdrant/qdrant:v1.19.0"
# --- Postgres ---------------------------------------------------------------
def _alembic_config(database_url: str) -> Config:
config = Config("alembic.ini")
config.set_main_option("sqlalchemy.url", database_url)
return config
def _postgres_settings_from_url(url: str) -> PostgresSettings:
# testcontainers returns postgresql+asyncpg://user:pass@host:port/db ;
# PostgresSettings builds its own dsn from parts, so parse the parts back out.
without_scheme = url.split("://", 1)[1]
creds, hostpart = without_scheme.split("@", 1)
user, password = creds.split(":", 1)
hostport, db = hostpart.split("/", 1)
host, port = hostport.split(":", 1)
return PostgresSettings(host=host, port=int(port), user=user, password=password, db=db)
@pytest.fixture(scope="session")
def postgres_container() -> Iterator[PostgresContainer]:
with PostgresContainer("postgres:17", driver="asyncpg") as container:
yield container
@pytest.fixture(scope="session")
def postgres_url(postgres_container: PostgresContainer) -> str:
"""The container's URL, pinned to IPv4.
Testcontainers reports the host as `localhost`, which resolves to `::1`
before `127.0.0.1`. Docker publishes the mapped port on IPv4 only, and the
IPv6 SYN is dropped rather than refused, so asyncpg blocks on the first
address until its connect timeout instead of falling back to the second --
the connection does not fail, it hangs.
"""
return postgres_container.get_connection_url().replace("@localhost:", "@127.0.0.1:")
@pytest.fixture(scope="session")
def migrated_postgres_url(postgres_url: str) -> str:
"""The container's URL, after Alembic has created the schema on it once."""
upgrade(_alembic_config(postgres_url), "head")
return postgres_url
@pytest.fixture(scope="session")
def postgres_settings(migrated_postgres_url: str) -> PostgresSettings:
"""Settings an application component can be constructed from directly."""
return _postgres_settings_from_url(migrated_postgres_url)
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def postgres_engine(postgres_settings: PostgresSettings) -> AsyncIterator[AsyncEngine]:
engine = create_engine(postgres_settings)
try:
yield engine
finally:
await engine.dispose()
@pytest_asyncio.fixture(loop_scope="session")
async def db_sessionmaker(
postgres_engine: AsyncEngine,
) -> AsyncIterator[async_sessionmaker[AsyncSession]]:
"""A session *factory* per test, bound to a rolled-back outer transaction.
Every session it produces shares one connection/outer transaction, so
writes `commit()`ed by one session are visible to the next -- needed for
code under test that opens more than one session per operation (auth
resolution, the ADR-0017 two-phase upload) -- while the whole test's
writes still roll back together at teardown (ADR-0016: isolate data per
test).
"""
async with postgres_engine.connect() as connection:
outer_transaction = await connection.begin()
sessionmaker = async_sessionmaker(
bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
)
yield sessionmaker
await outer_transaction.rollback()
@pytest_asyncio.fixture(loop_scope="session")
async def db_session(
db_sessionmaker: async_sessionmaker[AsyncSession],
) -> AsyncIterator[AsyncSession]:
"""One session per test, bound to a rolled-back outer transaction.
Isolates each test's writes (ADR-0016: isolate data per test) without
needing a fresh container or unique keys per test.
"""
async with db_sessionmaker() as session:
yield session
# --- MinIO ------------------------------------------------------------------
@pytest.fixture(scope="session")
def minio_container() -> Iterator[MinioContainer]:
with MinioContainer() as container:
yield container
@pytest.fixture(scope="session")
def minio_settings(minio_container: MinioContainer) -> MinioSettings:
"""Settings for the container, with the bucket already created.
Bucket creation belongs here rather than in a separate autouse fixture:
autouse in a *globally registered* plugin would pull a MinIO container
into every unit test run.
"""
config = minio_container.get_config()
# Pinned to IPv4 for the same reason as postgres_url above: `localhost`
# resolves to `::1` first, but Docker only publishes the mapped port on
# IPv4, so the connection hangs instead of failing.
endpoint = config["endpoint"].replace("localhost:", "127.0.0.1:")
settings = MinioSettings(
endpoint=endpoint,
access_key=config["access_key"],
secret_key=config["secret_key"],
secure=False,
bucket=_MINIO_BUCKET,
)
client = create_minio_client(settings)
if not client.bucket_exists(settings.bucket):
client.make_bucket(settings.bucket)
return settings
# --- Qdrant -----------------------------------------------------------------
@pytest.fixture(scope="session")
def qdrant_container() -> Iterator[QdrantContainer]:
with QdrantContainer(image=_QDRANT_IMAGE) as container:
yield container
@pytest.fixture(scope="session")
def qdrant_url(qdrant_container: QdrantContainer) -> str:
"""The container's REST URL, pinned to IPv4.
Same gotcha as postgres_url/minio_settings above: testcontainers reports
the host as `localhost`, which resolves to `::1` first, but Docker
publishes the mapped port on IPv4 only. The IPv6 SYN is dropped rather
than refused, so the client hangs until its timeout instead of falling
back to the second address -- the connection does not fail, it hangs.
"""
host = qdrant_container.get_container_host_ip().replace("localhost", "127.0.0.1")
return f"http://{host}:{qdrant_container.get_exposed_port(6333)}"
@pytest.fixture
def qdrant_settings(qdrant_url: str) -> QdrantSettings:
"""Settings naming a collection unique to this test (ADR-0016 isolation)."""
return QdrantSettings(url=qdrant_url, collection=f"chunks_{uuid.uuid4().hex}")
@pytest_asyncio.fixture(loop_scope="session")
async def qdrant_client(qdrant_settings: QdrantSettings) -> AsyncIterator[AsyncQdrantClient]:
client = create_qdrant_client(qdrant_settings)
try:
yield client
finally:
await client.close()