IndexError on force_rollback of Transaction in tests.
#570
Comments
It would appear, this isn't entirely true or behaving as expected? |
Based on the instantiation of the transaction, it would seem the above connection changing should be impossible. |
|
The transaction |
Reproducible with low_level transaction as well. |
|
Thanks for all the details! Patches welcome :) otherwise I'll investigate when I get a chance |
|
There is a similar issue when force_rollback is set on a database level at least when using postgresql backend. i noticed you also tested this with a postgresql backend @EdgyNortal. I'm thinking this might be a problem on the db integration layer as when asked for connection its always returning a new object (which i dont think is correct when i see how core is trying to reuse the same connection?) From what i was able to capture its the PostgressConnection that is changing while the Core connection sometimes changed sometimes stayed the same... (in both cases the exact same error was raised) I might look at this some more but i am a bit out of my depth here so help is appreciated |
|
I have a similar problem. traceback: An example to reproduce the problem: conftest.py import asyncio
import typing
import pytest
from databases import Database
async def clear_database(database: Database) -> None:
await database.execute("DELETE FROM person")
async def create_tables(database: Database) -> None:
await database.execute("CREATE TABLE IF NOT EXISTS person (id INT)")
@pytest.fixture(name="connected_database", scope="session")
async def _connected_database() -> typing.AsyncGenerator[Database, None]:
_database: Database = Database(
url="postgres://user:password@localhost:5432/db",
force_rollback=True,
)
try:
await _database.connect()
await create_tables(_database)
await clear_database(_database)
yield _database
finally:
await _database.disconnect()
@pytest.fixture(name="database")
async def _database(connected_database: Database) -> typing.AsyncGenerator[Database, None]:
async with connected_database.transaction(force_rollback=True):
yield connected_database
@pytest.fixture(scope="session")
def event_loop() -> typing.Generator[asyncio.AbstractEventLoop, None, None]:
loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
yield loop
loop.close()test_insert_person.py from databases import Database
from loguru import logger
async def test_insert_person(database: Database) -> None:
person_id: int = 10
actual = await database.fetch_val("INSERT INTO person (id) VALUES(:person_id) RETURNING id", values={"person_id": person_id})
assert actual == person_id
logger.debug("Test success")dependecies python = "^3.12"
databases = {extras = ["asyncpg"], version = "^0.9.0"}
pytest = "^8.1.0"
pytest-asyncio = "^0.21.0"
loguru = "^0.7.2"The query ( This error is only in new versions (started 0.8.0). It works on 0.7.0 |
were you able to find a solution? |
|
@zekiblue I had to switch back to the old version (0.7.0). |
|
I'm facing the exact same problem as #570 (comment), tough my code is simpler: Basic database.py setup: from databases import Database
database = Database(
url=str(settings.database_dsn), # some DSN
force_rollback=settings.unit_testing # True
)conftest.py @pytest.fixture()
async def db():
await database.connect()
# here database._global_transaction._transaction holds a databases.backends.postgres.PostgresTransaction object
yield database
await database.disconnect()test_nothing.py Somewhere along the way that transaction is lost, maybe it's something pytest-asyncio is doing? If it's not clear the Tested on: 0.7.0 does not have the assert in question, but I guess that only hides another problem Lines 407 to 412 in 6b0c767 |
|
Seems to be a pytest-asyncio issue. No idea if there's something encode devs can do to improve the situation. |
|
I believe anyio's pytest plugin might solve this problem? Some prior discussion at agronholm/anyio#497 |
|
Have been banging my head against this same problem. Have come up with a hacky stop-gap decorator that just overrides the fixture to ensure the context manager hasn't been exited at the point of test execution: @fixture
def database(): ...
def with_database(fn):
@functools.wraps(fn)
async def wrapper(*args, **kwargs):
async with Database(DATABASE_URL, force_rollback=True) as db:
kwargs.update(database=db)
return await fn(*args, **kwargs)
return wrapper
@pytest.mark.asyncio
@with_database
async def test_bulk_create_events(database: Database):
...It's late on Friday though so I may have overlooked something -- it's working for now at least. I found I had to create an empty
Hopefully this gets merged soon ^ 🤞🏻 |
MRE:
I have included prints in the
__aenter__and__aexit__of the Transaction class.Somehow the transaction connection on entry != the connection on exit.
Can someone point out what I am doing wrong, or how this could be possible?
The text was updated successfully, but these errors were encountered: