Skip to content

Commit

Permalink
Fix flaky tests with in-memory SQLite databases (#569)
Browse files Browse the repository at this point in the history
* Run garbage collection before asserting SQLite database is cleared in test

* Use unique in-memory database name for each SQLite test

* Run garbage collection after every test
  • Loading branch information
zanieb authored Aug 30, 2023
1 parent 27c1699 commit d047d40
Showing 1 changed file with 22 additions and 4 deletions.
26 changes: 22 additions & 4 deletions tests/test_databases.py
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,9 @@ def create_test_database():
engine = sqlalchemy.create_engine(url)
metadata.drop_all(engine)

# Run garbage collection to ensure any in-memory databases are dropped
gc.collect()


def async_adapter(wrapped_func):
"""
Expand Down Expand Up @@ -1549,7 +1552,10 @@ async def test_mapping_property_interface(database_url):

@async_adapter
async def test_should_not_maintain_ref_when_no_cache_param():
async with Database("sqlite:///file::memory:", uri=True) as database:
async with Database(
"sqlite:///file::memory:",
uri=True,
) as database:
query = sqlalchemy.schema.CreateTable(notes)
await database.execute(query)

Expand All @@ -1561,7 +1567,10 @@ async def test_should_not_maintain_ref_when_no_cache_param():

@async_adapter
async def test_should_maintain_ref_when_cache_param():
async with Database("sqlite:///file::memory:?cache=shared", uri=True) as database:
async with Database(
"sqlite:///file::memory:?cache=shared",
uri=True,
) as database:
query = sqlalchemy.schema.CreateTable(notes)
await database.execute(query)

Expand All @@ -1577,15 +1586,24 @@ async def test_should_maintain_ref_when_cache_param():

@async_adapter
async def test_should_remove_ref_on_disconnect():
async with Database("sqlite:///file::memory:?cache=shared", uri=True) as database:
async with Database(
"sqlite:///file::memory:?cache=shared",
uri=True,
) as database:
query = sqlalchemy.schema.CreateTable(notes)
await database.execute(query)

query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)

async with Database("sqlite:///file::memory:?cache=shared", uri=True) as database:
# Run garbage collection to reset the database if we dropped the reference
gc.collect()

async with Database(
"sqlite:///file::memory:?cache=shared",
uri=True,
) as database:
query = notes.select()
with pytest.raises(sqlite3.OperationalError):
await database.fetch_all(query=query)
Expand Down

0 comments on commit d047d40

Please sign in to comment.