zhifu-platform/tests/test_integration.py

from uuid import uuid4

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import insert, text
from sqlalchemy.exc import IntegrityError

from zhifu.app import create_app
from zhifu.modules.identity.models import Store, Tenant

pytestmark = pytest.mark.integration


async def test_application_and_database_constraints():
    app = create_app()
    async with app.router.lifespan_context(app):
        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
            live = await client.get("/health/live")
            ready = await client.get("/health/ready")
            assert live.status_code == ready.status_code == 200
            assert live.headers["X-Request-ID"] != ready.headers["X-Request-ID"]
        async with app.state.engine.connect() as connection:
            transaction = await connection.begin()
            try:
                tenant_id = uuid4().bytes
                await connection.execute(insert(Tenant).values(id=tenant_id, name="验收商家"))
                await connection.execute(
                    insert(Store).values(id=uuid4().bytes, tenant_id=tenant_id, name="验收店铺")
                )
                async with connection.begin_nested() as savepoint:
                    with pytest.raises(IntegrityError):
                        await connection.execute(
                            insert(Store).values(
                                id=uuid4().bytes, tenant_id=uuid4().bytes, name="无所属商家"
                            )
                        )
                    await savepoint.rollback()
                assert await connection.scalar(text("SELECT @@session.time_zone")) == "+00:00"
            finally:
                await transaction.rollback()