In tests, rolling back transactions in spite of "autocommit"

[jolly lagoon]

Hi all, new to Litestar and struggling a bit. I'm trying to figure out if there's a way to force tests to roll back their db changes at completion, in spite of SQLAlchemyPlugin being set to before_send_handler="autocommit". I'm using Postgres running in a testcontainer.

I've successfully set up a test db session fixture that uses Postgres savepoints to roll back, which works great as long as the tests update the db directly, through that session. The problem seems to be that when I invoke an actual Litestar route, it uses a different db session, one that actually does commit (provided by the SQLAlchemyPlugin?). I'm curious if the solution is to make the plugin provide the test session explicitly, and if so how that's done. I've spent a day or so wrestling with this and am a bit over my head.

I'd be most grateful for any ideas! Two things I'd rather not do if I don't have to, but will if I do have to:

(1) Drop and recreate the entire starting database for each test, which is what the litestar-fullstack example seems to do;
(2) Duplicate the app creation logic in the tests with a slightly different plugin configuration.

Thanks for any help!

[wanton kraken]

Notes for In tests, rolling back transactions in spite of "autocommit"

At your assistance

@jolly lagoon

No Response?

If no response in a reasonable time, ping @Member.

Closing

To close, type !solve or byte solve.

MCVE

Please include an MCVE so that we can reproduce your issue locally.

[fierce marsh]

Take a look at the fullstack test repository in a bit more detail, I think. If there’s something not working the way you’d expect, drop a link here.

You can modify the fullstack method to do what you want. Just use a session scoped fixture here.

FWIW, Pytest databases creates a single DB for the entire test run, the data is what is seeded per test.

[jolly lagoon]

Thanks for responding @fierce marsh, I will take a closer look and see if I can work it out.

[jolly lagoon]

Maybe you can help me see what I'm missing? I do see that integration tests using AsyncTestClient automatically get a db_cleanup fixture that truncates all tables after the test completes. That's not what I'm ideally hoping to find.

What I don't see is how the tests are directed to use either the test db engine or a custom session. We have the app fixture here:

@pytest.fixture
def app(engine: AsyncEngine, db_schema: None) -> Litestar:
    """Create Litestar app for testing.

    The app uses the same PostgreSQL database as the test session.
    """
    from app.server.asgi import create_app

    return create_app()

which receives the test engine instance, but then doesn't do anything with it. It calls the normal create_app(), which as far as I can tell ultimately installs a normal SQLAlchemyPlugin with no special session handling.

The multiple levels of indirection are throwing me a bit, but I'm pretty stumped. I don't see how the test engine is being used in the test app, let alone any answer to my original question.

[fierce marsh]

Sorry for not giving more details - I was typing from my phone. You are also going to likely need to modify a few things from fullstack to get exactly what you want. Here goes though:

1. Fixtures that handle services are monkeypatched into the settings object so you can control what engine is used. https://github.com/litestar-org/litestar-fullstack/blob/main/src/py/tests/integration/conftest.py and https://github.com/litestar-org/litestar-fullstack/blob/main/src/py/tests/conftest.py#L64
2. Pytest database offers ready to go database fixtures that are scoped to the session level. Meaning the database starts once at the start of the tests and shutdown at the very end: https://github.com/litestar-org/pytest-databases/blob/main/src/pytest_databases/docker/postgres.py#L443
3. If you want to apply DDLs a single time and not on each test, then you simply can create a session scoped fixture in your tests, use the postgres_service (or whatever version you specifically need), and do what you need to prep the DB connection.
4. Keep in mind that if you deploy DDLs a single time this likely leaves you in a spot where you need to cleanup data between the tests. Which is actually why you see the method I use in fullstack.

https://github.com/litestar-org/litestar-fullstack/blob/main/src/py/tests/conftest.py#L140 <- this is what I mean by "truncating" the tables (not an actual truncate operation because that has different transaction semantics). I emtty the tables between tests and reseed it with what is needed. I don't drop the database or the tables between each test

[sleek saddle]

src/py/tests/conftest.py line 140

async def db_cleanup(engine: AsyncEngine, db_schema: None) -> AsyncGenerator[None, None]:

[jolly lagoon]

Thanks — I think we're talking past each other a little bit, maybe it will help if I clarify what I want to do, which is very different from what fullstack is doing. I want to:

• Set up the database schema (DDL) and seed its contents, only once at startup.
• Run each test in a transaction that is rolled back at the end, leaving no changes to the initial data set.

Whereas fullstack is:

• Setting up the database schema (DDL) at startup.
• Automatically clearing all data out of the database at the end of each test (via the db_cleanup fixture).
• (I assume) reseeding the db with data at the beginning of each test, or at least each test that needs seeded data.

I hate to impose further, but I'm still pretty lost. I do need to look at those monkeypatching references — thanks for the link.

The reason I want to use rollbacks rather than tossing and recreating data is that these are intended as unit tests rather than integration tests and I want to keep them fast. Also this model is conceptually nicer.

There's a complication in that SQLAlchemyPlugin is set to autocommit, but I think I know how to work around that with Postgres savepoints, which can prevent those commits from actually committing. I've gotten the rollback approach working, provided my tests talk to the db directly, rather than invoking a Litestar route.

The root problem seems to be that the Litestar route gets its session from the SQLAlchemyPlugin, and isn't using the specially constructed test session. This is where perhaps the monkeypatching you pointed me at will help, so I'll be looking at that now.

Thank you for taking pity on a newb! I'm a pretty experienced developer but I'm new to much of Python and completely new to Python web frameworks.

OK, I think I see what the "monkeypatching" is doing: the engine fixture puts the engine URL in the environment, where it overrides the creation of the AsyncEngine that's then passed to SQLAlchemyPlugin. That explains how the tests are being made to use the test database, so I'm glad to understand that.

I don't think that helps with what I'm trying to do, because I need the routes to use a specially configured db_session — one that's set up to auto-rollback — instead of the one provided by SQLAlchemyPlugin. I wonder if I can hack the plugin's session_maker_class. This is getting very deep for my level of knowledge at this point.

[jolly lagoon]

I've got this half "working," by stashing the modified session I want to use in app.state, and having the route handler explicitly check for it there, and if found, use it instead of the passed-in session from the plugin. I've confirmed that it isolates tests the way I want.

If I could get the plugin to use the session from app.state, that could solve the problem — ugly but functional. It looks like that might not be possible. I'd love to find a less clunky way anyway.

[fierce marsh]

It sounds like. you aren't using the actual SQLAlchemy plugin. it's going to be hard for us to help you without actual code to reference in this case.

[fierce marsh]

the less clunky way would be to use the plugin, IMO. Let ie manage your sessions.

[jolly lagoon]

I am using the actual plugin. But I've come to the conclusion that what I wanted to do is very likely impossible.

[fierce marsh]

you can disable the auto-commit on the sqlalchemy plugin. In fact, it's off by default

[jolly lagoon]

ah, but I'm using the litestar-users plugin as well, which recommends very strongly turning auto-commit on. (It also seemed to be the generally recommended practice, from some poking around.)

and I don't want to turn it off for tests only because that changes the SUT.

[fierce marsh]

And if you want it to use your test session, you are going to have to go the monkey patch route. This is the standard way to address that in my experience.

[fierce marsh]

@unkempt glade What's the idea behind the auto-commit recommendation here?

[jolly lagoon]

Well, that was the part that seemed really hard — if it is in fact possible, I might need to know more about the innards of SQLAlchemyPlugin than I do so far.

[fierce marsh]

do you have an engine or a session maker that you are using in your tests?

if so, you just monkey patch that onto the config. there's an _engine_instance and a session_maker_class.

if you want to share some code, i can probably give you more concrete refrences

[jolly lagoon]

(I can answer for @unkempt glade — the explanation is actually pretty well documented for why litestar-users recommends autocommit. I could probably get away without it but I'm trying to follow good practices and pretend the app might someday scale up, for learning and practice reasons.)

[fierce marsh]

is there a link that i can read?

[jolly lagoon]

sure, one sec

[fierce marsh]

Also, are you using postgres - because it sounds like some of your concern is on speed of setting up and tearing down tests. I'm not suure how much data your are looking to see/start with, but postgres templates are often used here to make this process much easier

instead of worrying about this in your code, you just restore the postgres template between tests

but again, these really only provide value when you need large data sets for testing

or you have many tests that run for a long duration

[jolly lagoon]

I learned about those yesterday, gonna do more reading. The truth is this is a stupid small db and the actual costs are going to be basically zero — but this app is largely, again, a learning experience, and I'm pretending it's big. I might not have bothered if I had realized how hard this would be.

Still looking for that link about litestar-users, then I'll come back with some details about what I had and maybe some code.

[fierce marsh]

sounds good

[jolly lagoon]

I'm VERY grateful for your generosity with your time here

[fierce marsh]

of course - we are all happy to help. i'm sure we can figure out something that you are going to be happy with

[jolly lagoon]

OK, here's the litestar-users link, search for "Warning" https://mvbosch.github.io/litestar-users/usage/0-configuration/

Now that I reread that, even the backup solution there still requires some level of autocommit

As for my code, unfortunately I never committed the key routine and it's gone now, as I began switching over to the approach used by fullstack. but let me describe the key bits

[fierce marsh]

Have you looked at doing fixtures similar to what is done in the litestar-users tests? https://github.com/mvbosch/litestar-users/blob/main/tests/integration/conftest.py#L172

[sleek saddle]

tests/integration/conftest.py line 172

def sqlalchemy_plugin_config(docker_ip: str) -> SQLAlchemyAsyncConfig:

[jolly lagoon]

No, lemme take a peek!

Yes, I think so. Let me share a little code:

@pytest.fixture
def test_app(test_db_container: PostgresContainer) -> Iterator[Litestar]:
    """Yields a testable app instance using a Postgres testcontainer as its db, overriding
    .env
    """

    with temporarily_modified_environ(
        APP_ENV="testing",
        DB_DRIVER="postgresql+asyncpg",
        DB_USER=test_db_container.username,
        DB_PASS=test_db_container.password,
        DB_HOST=test_db_container.get_container_host_ip(),
        DB_PORT=str(test_db_container.get_exposed_port(5432)),
        DB_NAME=test_db_container.dbname,
    ):
        yield create_app()


@pytest.fixture
def test_client(test_app: Litestar) -> Iterator[TestClient[Litestar]]:
    """Yields a TestClient set up to test the app instance provided by test_app"""

    with TestClient(app=test_app) as test_client:
        yield test_client

These are fixtures that create my testing app and client. It uses my "real" create_app function but overrides the env vars to point at the temporary Dockerized test db.

The part that's gone was another fixture that created an AsyncSession, with setup and cleanup behavior that handled rolling back the test's changes (overridding the autocommit behavior). That part seemed to work.

[fierce marsh]

with temporarily_modified_environ(
APP_ENV="testing",
DB_DRIVER="postgresql+asyncpg",
DB_USER=test_db_container.username,
DB_PASS=test_db_container.password,
DB_HOST=test_db_container.get_container_host_ip(),
DB_PORT=str(test_db_container.get_exposed_port(5432)),
DB_NAME=test_db_container.dbname,
):

where do these become an actual SQLAlchemy config?

[jolly lagoon]

Here:

plugins=[
            SQLAlchemyPlugin(config=_getSQLAlchemyConfig()),
            # litestar-users plugin implements user management and authentication endpoints
            setup.litestar_users.plugin.configure_litestar_users_plugin(
                app_config.get_jwt_encoding_secret()
            ),
        ],

and here:

def _getSQLAlchemyConfig() -> SQLAlchemyAsyncConfig:
    return SQLAlchemyAsyncConfig(
        connection_string=app_config.get_db_url(),
        session_config=AsyncSessionConfig(expire_on_commit=False),
        before_send_handler="autocommit",  # semi-required by litestar-users; good practice anyway
    )

and here

ef get_db_url() -> str:
    """Construct DB url from environment vars"""
    check_loaded()
    attrs: dict[str, str] = {}
    for env_name in [
        "DB_DRIVER",
        "DB_USER",
        "DB_PASS",
        "DB_HOST",
        "DB_PORT",
        "DB_NAME",
    ]:
        value = os.getenv(env_name)
        if value is None:
            raise ConfigurationError(
                f"Database configuration must be fully set in the environment ({env_name} is missing)"
            )
        attrs[env_name] = value
    return (
        f"{attrs['DB_DRIVER']}://"
        f"{attrs['DB_USER']}:{attrs['DB_PASS']}"
        f"@{attrs['DB_HOST']}:{attrs['DB_PORT']}/"
        f"{attrs['DB_NAME']}"
    )

[fierce marsh]

def _getSQLAlchemyConfig() -> SQLAlchemyAsyncConfig:
return SQLAlchemyAsyncConfig(
connection_string=app_config.get_db_url(),
session_config=AsyncSessionConfig(expire_on_commit=False),
before_send_handler="autocommit", # semi-required by litestar-users; good practice anyway
) <- my guess is that you are calling this multiple times g etting a new sqlalchemy config instead of memoizing the result and reusing the same config that comes back. Your issue was that you were having trouble getting "the same" connection right?

[jolly lagoon]

Not from the SQLAlchemyPlugin, that was consistently providing the right connection.

The problem was that I was separately creating a session in a test fixture; this session was specially configured for the auto-rollback behavior.

So I could run tests using that test session, and they worked fine if the tests didn't actually invoke the TestClient (i.e. just talked to the db directly).

[fierce marsh]

can i throw out a counter point and ask how you intend to handle a particular scenario in intgration tests with a rollback?

[jolly lagoon]

But the TestClient used a db_session provided by SQLAlchemyPlugin instead of the one created in my test fixture.

hit me!

[fierce marsh]

let's say you need to test the account registration flow as an integration tests. you can't do this correctly in a single transaction.

your integration test must 1) create the user and then depending on the test, log in with user, reset the password, etc

those are all multi-step processes to test in integrations

assuming you want to rollback the request session, how did you intend to test these multi-step integration tests?

[jolly lagoon]

I hadn't gotten to that test yet so I'm not 100% of my ground here, but: I learned that Postgres supports savepoints, which are like "save game," and SQLAlchemy supports them as well. My test session, from the fixture, created a savepoint at the beginning and rolled back to that savepoint at the end. Any commits in between stay visible only within that savepoint — it's similar to a nested transaction and in fact appears to be Postgres's version of nested transactions.

Only if the savepoint is ultimately "released" to the commits get persisted to the full database context.

I did small tests and confirmed that a commit within the savepoint was visible within the savepoint, but could still be rolled back by rolling back to the savepoint.

[fierce marsh]

Makes sense. In theory this works. In practice, it may be a bit more cumbersome than you want it to be.

and it's also in theroy "the right" way to do something like this

but it does require a bit more effort to get your harnesses right

[jolly lagoon]

So I had a fixture something like this (this is pseudocode):

@pytest.fixture
def test_session(db_connection):
  db_session = AsyncSession(db_connection, join_transaction_mode="create_savepoint")

yield db_session

rollback_to_savepoint()

But that session only applied to code running in my tests, because SQLAlchemy provided a different session to the route handler that was not part of that savepoint block.

[fierce marsh]

did you do ayththing to mock test_session to be returned from the SQLAlchemy config when you request a session?

[jolly lagoon]

I wonder if I could have created a function that returned that specific session, and then passed that to create_app to use in the SQLAlchemyPlugin config

[fierce marsh]

yes - you can just mock the return of the session maker class to return your test_session

i'm looking for an example now

[jolly lagoon]

Would it even have to be a mock? I can pass in a session_maker function directly to the plugin config, right?

[fierce marsh]

no - probably doesn't have to be a mock

[jolly lagoon]

cool

[fierce marsh]

https://github.com/litestar-org/litestar-fullstack-inertia/blob/main/tests/integration/conftest.py#L105-L116

[jolly lagoon]

well, I guess I've got today's project set then. I was halfway through converting to the drop-data-and-reseed approach, thank god for git

[sleek saddle]

tests/integration/conftest.py lines 105 to 116

@pytest.fixture(autouse=True)
async def _patch_db(
    app: "Litestar",
    engine: AsyncEngine,
    sessionmaker: async_sessionmaker[AsyncSession],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(config.alchemy, "session_maker", sessionmaker)
    monkeypatch.setattr(config.alchemy, "engine_instance", engine)
    # Also patch app state for tests that check app.state directly
    app.state[config.alchemy.engine_app_state_key] = engine
    app.state[config.alchemy.session_maker_app_state_key] = sessionmaker

[fierce marsh]

monkeypatch.setattr(config.alchemy, "session_maker", sessionmaker)
monkeypatch.setattr(config.alchemy, "engine_instance", engine)

[jolly lagoon]

ohhh, interesting, I did not find that on my own

[fierce marsh]

this is from the inertia variant of fullstack. i'm not sure it's in the other one too, but it should be

[jolly lagoon]

I see it, good to know. This is awesome. I'm going to have to study that later today because I'm almost out of time for the morning, but it sounds like one or both of these approaches might work

that would make me happy

Thank you again — I will be back later today with either good news or more questions 🙂

Oh I see, it's monkey patching the config objects specifically, which eventually get rolled up into the plugin initializer

[fierce marsh]

correct

[jolly lagoon]

interesting

[fierce marsh]

Feel free to drop more questions here as they come to you

[jolly lagoon]

Certainly will!

[jolly lagoon]

Well, I got the Session override to work beautifully — thanks for the tips, they were on point. I was so happy! Unfortunately I then ran into a different, deeper problem with async SQLAlchemy calls, which might well be unsolvable. I'll try to explain what I've gathered — my understanding of asyncio is very rudimentary at present:

• To allow rolling back commits made by a test, I need an outer transaction (or Postgres's equivalent of one).
• To create that outer transaction in my fixture, I need to create an explicit AsyncConnection.
• That, unfortunately, binds the connection to the fixture's event loop. (For some reason, when I create the AsyncSession directly from the AsyncEngine, this isn't a problem.)
• Later, when the TestClient runs a route that executes a query, it tries to do it on a different event loop owned by something else (the TestClient?) and everything go boom. I get a "Future...attached to a different loop" error.

I wonder if I could move all this out of fixtures and into the session factory I'm already handing to SQLAlchemyPlugin? Then the connection logic would presumably all be on the right event loop...

Already spent way more time on this than it's worth... but it's for LEARNING

[jolly lagoon]

I tried (yuck) subclassing AsyncSession to make a session that cleaned up its own commits. That didn't work for two reasons: (1) I needed a session per test, but I was getting a session per test client invocation, which broke when I had multiple endpoints accessed in one test; (2) the second session failed anyway with SQLAlchemy internal errors.

Time to give up and admit that SQLAlchemy and Litestar really don't want me to write my tests this way. I'm going back to the branch where I'd started converting to the fullstack example approach. Sigh.

[unkempt glade]

Hi, I have spent a lot of time trying to figure this out in the past and think I can help here.
Disclaimer: I did not read this whole thread
Various attempts always led to the issue of the TestClient being bound to a different event loop than the one defined by us in fixtures.
I'll just share my fixture as it's documented and commented
Okay discord wants to use an attachment

This sped up my integration tests a lot, I never bothered measuring but it's a minutes-to-seconds type of deal

Using this pattern I seed the test database at the sqlalchemy engine fixture creation time

[jolly lagoon]

@unkempt glade Many thanks, I will take a close look at this later today when I have time!! And also thanks for your work on litestar-users, while you're here.

[jolly lagoon]

[Update at the bottom: I got it to work!]

@unkempt glade this is great stuff and I've learned a lot from it. I have not successfully gotten it to run. I don't know if you have any more time to spend on this, but the sticking point is that app and db_config are referenced but unsourced. I assume that they're meant to be imported from the main app, so that's what I tried, with import app and corresponding tweaks to the code.

It's currently failing to find the engine in the app state, I haven't a clue why:

engine = app.app.state[app.sql_alchemy_config.engine_app_state_key]

failing with:

tests/unit/conftest.py:68: in _
    engine = app.app.state[app.sql_alchemy_config.engine_app_state_key]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <litestar.datastructures.state.State object at 0x106e94280>
key = 'db_engine_2'

>       return self._state[key]
               ^^^^^^^^^^^^^^^^
E       KeyError: 'db_engine_2'

I'm a bit out of my depth and don't love my chances of troubleshooting this, although I'm going to keep trying a bit longer.

UPDATE: amazingly, I got it working! Fixed my prior assumption, and I'm now fetching the app properly from the TestClient and getting the SQLAlchemyPlugin's config directly from the app's plugin registry. Incredibly, that seems to be all that I'd gotten wrong. All my integration tests are now passing, including the one designed to fail if the rollbacks fail.

Thanks a million for this. I'm off to study the code some more and port it over to my main branch.