63 lines
1.6 KiB
Python
63 lines
1.6 KiB
Python
"""Pytest fixtures shared across asf-core tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from asgi_lifespan import LifespanManager
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from app.main import app
|
|
from app.middleware.correlation import (
|
|
request_context_var,
|
|
RequestContext,
|
|
)
|
|
|
|
|
|
TENANT_ID = uuid.UUID("00000000-0000-0000-0000-000000000001")
|
|
USER_ID = uuid.UUID("00000000-0000-0000-0000-000000000002")
|
|
|
|
|
|
@pytest.fixture
|
|
def tenant_context():
|
|
"""Bind a dummy authenticated tenant+user context for the duration of a test."""
|
|
|
|
ctx = RequestContext(
|
|
tenant_id=TENANT_ID,
|
|
user_id=USER_ID,
|
|
correlation_id="test-corr",
|
|
roles=["admin"],
|
|
)
|
|
token = request_context_var.set(ctx)
|
|
yield ctx
|
|
request_context_var.reset(token)
|
|
|
|
|
|
@pytest.fixture
|
|
async def asf_client(tenant_context): # noqa: ARG001 - side effect
|
|
"""Async httpx client pointed at the FastAPI app.
|
|
|
|
For unit tests in bootstrap mode we skip external infra by NOT starting
|
|
the full lifespan; we instead instantiate routes with the tenant
|
|
context bound so requests succeed without Mongo/Rabbit reachable.
|
|
"""
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with LifespanManager(app, startup_timeout=10):
|
|
async with AsyncClient(
|
|
transport=transport,
|
|
base_url="http://testserver",
|
|
headers={
|
|
"X-ASF-Tenant-Id": str(TENANT_ID),
|
|
"X-ASF-User-Id": str(USER_ID),
|
|
},
|
|
) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
def utc_now():
|
|
return datetime(2026, 9, 7, 12, 0, 0, tzinfo=timezone.utc)
|