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."
28 lines
697 B
Python
28 lines
697 B
Python
import pytest
|
|
from sqlalchemy import inspect
|
|
from sqlalchemy.ext.asyncio import AsyncEngine
|
|
|
|
pytestmark = [
|
|
pytest.mark.integration,
|
|
pytest.mark.postgres,
|
|
pytest.mark.asyncio(loop_scope="session"),
|
|
]
|
|
|
|
EXPECTED_TABLES = {
|
|
"tenants",
|
|
"api_keys",
|
|
"source_files",
|
|
"ingestion_jobs",
|
|
"ingestion_job_events",
|
|
"alembic_version",
|
|
}
|
|
|
|
|
|
async def test_migrations_create_schema_from_empty_database(postgres_engine: AsyncEngine) -> None:
|
|
async with postgres_engine.connect() as connection:
|
|
table_names = await connection.run_sync(
|
|
lambda sync_conn: inspect(sync_conn).get_table_names()
|
|
)
|
|
|
|
assert EXPECTED_TABLES.issubset(set(table_names))
|