mirror of
https://github.com/evroon/bracket.git
synced 2026-07-31 18:26:23 -04:00
4835aa7311
- Rewrite database.py with asyncpg pool-based DatabasePool class - Support named params (:param) → positional ($N) conversion - Transaction context manager using contextvars for connection reuse - Replace all SQLAlchemy query building with raw SQL in routes and sql modules - Update utils/db.py to use raw SQL strings instead of SQLAlchemy Select - Update utils/db_init.py to use local engine creation for DDL only - Update all test files to use string table names and raw SQL - Remove databases and sqlalchemy-stubs from dependencies - Keep sqlalchemy only for schema.py definitions and alembic migrations
64 lines
1.8 KiB
Python
64 lines
1.8 KiB
Python
# pylint: disable=redefined-outer-name
|
|
import os
|
|
from collections.abc import AsyncIterator
|
|
from time import sleep
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from bracket.database import DatabasePool, database
|
|
from bracket.utils.db_init import _create_all_tables, _drop_all_tables
|
|
from tests.integration_tests.models import AuthContext
|
|
from tests.integration_tests.sql import inserted_auth_context
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session", autouse=True)
|
|
async def reinit_database(worker_id: str) -> AsyncIterator[DatabasePool]:
|
|
"""
|
|
Creates the test database on the first test run in the session.
|
|
|
|
When running in parallel, the first test runner (gw0) creates a "lock" file and initializes the
|
|
database. The other runners poll this file and wait until it has been removed by gw0.
|
|
When running tests sequentially, the master worker just creates the test database and that's it.
|
|
"""
|
|
await database.connect()
|
|
|
|
if worker_id == "master":
|
|
_drop_all_tables()
|
|
_create_all_tables()
|
|
|
|
try:
|
|
yield database
|
|
finally:
|
|
await database.disconnect()
|
|
|
|
return
|
|
|
|
lock_path = "/tmp/tm_test_lock"
|
|
|
|
if worker_id == "gw0":
|
|
try:
|
|
with open(lock_path, mode="w") as file:
|
|
file.write("")
|
|
|
|
_drop_all_tables()
|
|
_create_all_tables()
|
|
finally:
|
|
os.remove(lock_path)
|
|
else:
|
|
for _ in range(50):
|
|
sleep(0.1)
|
|
if not os.path.exists(lock_path):
|
|
break
|
|
|
|
try:
|
|
yield database
|
|
finally:
|
|
await database.disconnect()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def auth_context(reinit_database: DatabasePool) -> AsyncIterator[AuthContext]:
|
|
async with reinit_database, inserted_auth_context() as auth_context:
|
|
yield auth_context
|