Testing

Directory traversal and versioning

• minversion = 3: This ensures that anyone running this test suite, whether it's a new developer on their local machine or the CI/CD pipeline, is using a compatible version of pytest. It prevents bizarre errors caused by legacy test runners.
• norecursedirs: By default, pytest will aggressively search every folder in your project for tests. We explicitly tell it to ignore directories like settings, urls, and docs. This significantly speeds up the "test discovery" phase, especially as your project grows.

Async context

This is where the heavy lifting for Flama happens.

• asyncio_mode = auto: Historically, to test an async function, you had to manually add the @pytest.mark.asyncio decorator above every single test. This was tedious and highly prone to human error (forgetting the decorator often resulted in the test being skipped or passing silently without executing the coroutine). By setting this to auto, pytest automatically detects any async def test_... function and wraps it in an event loop for you. It drastically reduces boilerplate.
• asyncio_default_fixture_loop_scope = function: This is arguably the most critical setting for database isolation. The "loop scope" defines how long an async event loop stays alive. If you share an event loop across multiple tests (e.g., setting it to session), a lingering background task or unclosed database connection from Test A can crash Test B. By restricting the scope to function, we guarantee that every single test gets its own fresh, isolated event loop that is destroyed immediately after the test concludes.

Categorisation with markers

As your application scales, running the entire test suite on every minor code change becomes impractically slow.

• markers: We explicitly register type_integration and type_unit. This allows developers to selectively run subsets of tests from the command line (e.g., pytest -m type_unit). By registering them here, we also prevent typos, if a developer accidentally types @pytest.mark.type_integraton, pytest will catch the error immediately rather than silently skipping the test.

This is where the magic truly happens. Moving from the configuration to conftest.py is like moving from the blueprints to laying the actual concrete foundation.

In pytest, conftest.py is not just a regular Python module; it acts as a local plugin directory. Any fixtures defined here are automatically discovered and made globally available to all your test files without needing a single import statement. It is the central orchestrator of your test suite’s state.

Session scope

Notice that the fixture is decorated with @pytest.fixture(scope="session"). This is vital for performance. It tells pytest to execute this setup code exactly once per test session, rather than repeating it for every single test function.

Furthermore, we inject a built-in fixture called worker_id. When running tests in parallel using the pytest-xdist plugin, pytest spawns multiple independent processes (workers).

• If you run tests sequentially, worker_id evaluates to "master".
• If you run tests in parallel (e.g., across 4 CPU cores), worker_id evaluates to "gw0", "gw1", "gw2", and "gw3".

If all four workers tried to write to a single test database simultaneously, they would instantly trigger race conditions and table deadlocks. By parsing the worker_id, we dynamically generate unique database names (test-gw0, test-gw1, etc.), ensuring absolute isolation for every parallel process.

The AUTOCOMMIT Imperative

By default, SQLAlchemy wraps all database operations in a transaction (a BEGIN ... COMMIT block) to ensure data integrity. However, PostgreSQL strictly forbids executing structural commands like CREATE DATABASE or DROP DATABASE inside a transaction block. If you try, Postgres will throw a fatal error.

To bypass this, we must explicitly override SQLAlchemy's default behavior by passing isolation_level="AUTOCOMMIT" into our engine. This tells SQLAlchemy to fire the raw SQL commands directly at the server without wrapping them in a transaction, satisfying Postgres's requirements.

The lifecycle

In traditional Python, a return statement immediately ends a function. In pytest fixtures, we use yield instead. The yield statement effectively splits the fixture into two distinct phases:

1. Setup (Before yield): Drops any existing leftover database and creates a fresh one.
2. The Yield: Pauses the fixture, passes the Database namedtuple to the test suite, and allows all the tests in the session to execute.
3. Teardown (After yield): Once all tests finish (or even if they crash), the code resumes execution and guarantees the database is cleanly dropped, preventing orphaned databases from cluttering your local Postgres server.

import collectionsimport reimport pytestimport sqlalchemyfrom src import config
@pytest.fixture(scope="session")def sqlalchemy_database(worker_id):    # Generates unique DB names per parallel worker (e.g., test-gw0, test-gw1)    # ensuring parallel workers never share the same database state.    database = (        "test"        if worker_id == "master"        else f"test-{int(re.match(r'gw(\d+)', worker_id).group(1))}"    )
    # We must use AUTOCOMMIT because Postgres rejects CREATE/DROP DATABASE    # commands if they are wrapped inside standard SQL transactions.    engine = sqlalchemy.create_engine(        config.DATABASE.url(            config.APP.database, config.APP.database_user, config.APP.database_password        ),        isolation_level="AUTOCOMMIT",    )
    # SETUP PHASE: Create the completely ephemeral database    with engine.connect() as connection:        connection.execute(sqlalchemy.text(f'DROP DATABASE IF EXISTS "{database}"'))        connection.execute(sqlalchemy.text(f'CREATE DATABASE "{database}"'))
    # YIELD PHASE: Pause here and let the test session run    # We pass down both the test DB name and the original prod DB name     # so we can dynamically patch configs later.    yield collections.namedtuple("Database", ["test", "prod"])(        test=database, prod=config.APP.database    )
    # TEARDOWN PHASE: Clean up the database after the test session concludes    with engine.connect() as connection:        connection.execute(sqlalchemy.text(f'DROP DATABASE "{database}"'))

By leveraging this architecture, you guarantee that no matter how complex your test suite becomes, it will always execute against a flawlessly clean, isolated environment.

Patching

By default, Alembic looks at your application's configuration to determine which database to connect to. If we don't intercept this, Alembic will happily connect to your default database (which might be your local development DB, or worse, production!) and run the migrations there.

To prevent this, we use unittest.mock.patch.object. We dynamically overwrite the database attribute inside our config.APP module in memory. We point it strictly to the sqlalchemy_database.test name we generated in the previous fixture. Because patch is used as a context manager (the with statement), this override is safely isolated to the scope of the setup phase.

Mutex (Mutual Exclusion) file lock

When running in parallel, we need a "traffic light" to ensure workers take turns reading the Alembic directory. We achieve this using filelock.FileLock.

• The Global Lock (migration.lock): This is our primary mutex. When Worker 0 hits this line, it creates a physical file on your hard drive and claims ownership of it. If Worker 1 arrives a millisecond later, it sees the lock file is taken and pauses its execution. It simply waits in line. Once Worker 0 finishes running command.upgrade("head") and releases the lock, Worker 1 acquires it and proceeds. This strictly serialises the Alembic execution, preventing I/O race conditions while reading the migration scripts.
• The Worker Lock (migration.lock.{worker_id}): This lock serves a completely different purpose: it acts as a tracker. By creating a lock file named after the specific worker (e.g., migration.lock.gw0), we leave a footprint on the filesystem proving that this specific worker is currently alive and using the database.

The teardown

A robust testing suite doesn't just test if you can build the database; it tests if you can tear it down. Broken rollback scripts are a massive liability in production. We explicitly call command.downgrade(alembic_cfg, "base") to verify that every down_revision in your Alembic history is mathematically sound and syntactically correct.

The "Last Worker Standing" Logic: In a parallel environment, how do we know when the entire test suite is finished so we can run the downgrade? We can't just run it when Worker 0 finishes, because Worker 1 might still be executing tests.

We solve this using the worker tracking locks we created earlier. During the teardown phase (after the yield), each worker deletes its specific lock file. Then, it uses Path(".").glob("migration.lock.*") to scan the directory. If it finds any other worker lock files, it silently exits, meaning other workers are still running tests. Only the absolute last worker to finish will see an empty directory. That final worker then claims the global lock one last time and executes the downgrade.

Here is the annotated code implementing this architecture:

from alembic import commandfrom alembic.config import Config as AlembicConfigfrom filelock import FileLockfrom pathlib import Pathfrom unittest.mock import patch
# autouse=True ensures this runs automatically; no test needs to manually request it.@pytest.fixture(scope="session", autouse=True)def migrations(worker_id, sqlalchemy_database):    alembic_cfg = AlembicConfig("alembic.ini")
    # Dynamically point Alembic to the ephemeral database    with patch.object(config.APP, "database", sqlalchemy_database.test):                # If running sequentially (no pytest-xdist), just run the migrations.        if worker_id == "master":            command.upgrade(alembic_cfg, "head")            yield  # Let the test session execute            command.downgrade(alembic_cfg, "base")                    # If running in parallel, orchestrate the file locks        else:            # Create a tracking lock to prove this worker is currently active            with FileLock(f"migration.lock.{worker_id}"):                                # Wait in line for the global lock to prevent I/O collisions                with FileLock("migration.lock"):                    command.upgrade(alembic_cfg, "head")                                # The yield statement sits OUTSIDE the global lock.             # This allows all workers to run their tests concurrently now that             # their individual databases are fully migrated.            yield                        # Teardown: Scan the directory. If no other worker locks exist,            # this is the final worker. It is safe to run the downgrade.            if not any(Path(".").glob("migration.lock.*")):                with FileLock("migration.lock"):                    command.downgrade(alembic_cfg, "base")

By combining unittest.mock for dynamic routing and filelock for Inter-Process Communication, we have created a bulletproof, parallel-safe database initialisation pipeline.

Bypassing the socket

When you make a request using flama.client.Client, it does not open a network socket. Instead, it translates your HTTP request (e.g., client.get("/users")) directly into an ASGI dictionary (the standard format that ASGI apps consume) and invokes your application's root callable directly in Python's memory. This entirely bypasses the TCP/IP stack, making requests orders of magnitude faster.

Dynamic dependency injection

Before we can use the client, we must ensure the application it wraps is actually pointing to our sqlalchemy_database.test created in the previous steps.

We do this by intercepting the initialised Flama app and mutating its internal SQLAlchemyModule state. Notice that the scope of these fixtures is scope="function". This guarantees that every single test function gets a fresh instance of the client, preventing internal client state (like cookies or session headers) from leaking between tests.

import src.appfrom flama.client import Clientimport pytest
@pytest.fixture(scope="function")async def app(sqlalchemy_database):    # Import the actual, production-ready ASGI application instance    _app = src.app.app
    # Overwrite the SQLAlchemy database connection string in memory.    # This guarantees the app connects to 'test-gw0' instead of 'my_prod_db'.    _app.sqlalchemy.database = config.DATABASE.url(        sqlalchemy_database.test, config.APP.database_user, config.APP.database_password    )        return _app
@pytest.fixture(scope="function")async def client(app):    # Use an async context manager to ensure the ASGI app lifecycle     # (startup and shutdown events) is properly triggered and cleanly closed.    async with Client(app=app) as _client:        yield _client

The "autouse" safety net

We define this fixture with @pytest.fixture(scope="function", autouse=True). The autouse=True flag is crucial: it means pytest will invisibly apply this fixture to every single test function in your suite, even if the developer forgets to include connection in the test's arguments. This eliminates human error. You can never accidentally commit test data and pollute the database.

BEGIN ... ROLLBACK Lifecycle

1. The Setup (begin_transaction): Before the test executes, we ask Flama's native SQLAlchemyModule for a raw database connection. We then explicitly issue a BEGIN statement to start a new transaction block.
2. The Execution (yield): We yield the active connection to the test. The test runs, hits the API endpoints, and inserts/updates/deletes data. To the application, these changes appear completely real and permanent. It can query the data back immediately.
3. The Erasure (rollback=True): The moment the test finishes, the fixture resumes execution. We call end_transaction(transaction, rollback=True). Instead of committing the changes to the disk, PostgreSQL simply discards the transaction log in memory. Every single mutation the test performed vanishes instantly.

The database is restored to the exact pristine state it was in after the Alembic migrations ran, taking effectively zero I/O time.

@pytest.fixture(scope="function", autouse=True)async def connection(client):    # Acquire a dedicated connection from the application's connection pool    connection = await client.app.sqlalchemy.open_connection()        # Start an active transaction block (BEGIN)    transaction = await client.app.sqlalchemy.begin_transaction(connection)
    # Yield the connection so the test can use it (and the app can use it)    yield connection
    # The test has finished. Force a ROLLBACK.    # Postgres discards all changes. The database is clean instantly.    await client.app.sqlalchemy.end_transaction(transaction, rollback=True)        # Return the connection to the pool    await client.app.sqlalchemy.close_connection(connection)

By successfully setting up the ephemeral database, handling the parallel migrations with filelock, and implementing this transactional rollback fixture, you have constructed an enterprise-grade testing environment.