test(postgres): support multi-session integration tests

Why:
- Code under test (auth resolution, the two-phase upload) opens more than
  one session per operation; the existing fixture only exposed one
  rolled-back session.

Changes:
- Add a db_sessionmaker fixture sharing one outer transaction.
- Pin loop_scope="session" -- without it, a second async test against the
  session-scoped Postgres container fails with "Event loop is closed."
This commit is contained in:
2026-08-19 15:00:50 +03:30
parent 3bced65926
commit e70ad13b10
3 changed files with 80 additions and 10 deletions

View File

@@ -1 +1,49 @@
"""Object builders for test fixtures, populated as later phases need them."""
"""Object builders for test fixtures (ADR-0016).
Insert rows via a caller-supplied `AsyncSession` without committing — callers
decide their own transaction boundary (the `db_session` fixture rolls back
after each test).
"""
import uuid
from sqlalchemy.ext.asyncio import AsyncSession
from src.application.auth.keys import generate_api_key, hash_secret
from src.infrastructure.postgres.models.api_key import ApiKey
from src.infrastructure.postgres.models.tenant import Tenant
async def create_tenant(
session: AsyncSession, *, slug: str | None = None, status: str = "active"
) -> Tenant:
slug = slug or f"tenant-{uuid.uuid4().hex[:8]}"
tenant = Tenant(id=uuid.uuid4(), slug=slug, name=slug, status=status)
session.add(tenant)
await session.flush()
return tenant
async def create_api_key(
session: AsyncSession,
*,
tenant: Tenant,
scopes: list[str] | None = None,
status: str = "active",
) -> tuple[ApiKey, str]:
"""Returns `(api_key, full_key)`. `full_key` is the bearer token to send;
only its hash is persisted.
"""
key_prefix, secret, full_key = generate_api_key()
api_key = ApiKey(
id=uuid.uuid4(),
tenant_id=tenant.id,
name="test-key",
key_prefix=key_prefix,
key_hash=hash_secret(secret),
scopes=scopes or ["files:write"],
status=status,
)
session.add(api_key)
await session.flush()
return api_key, full_key