Open
Description
I've hit an issue where basically my setup is:
- have an async test that tries to create some things in a postgres database running locally via
testcontainers
- use pytest-asyncio + testcontainers postgres + pytest django
- if I run things synchronously, everything works(migrations are properly created, postgres spawned,
- When I run things async, my migrations are never applied.
- I think it's some weird connection between pytest-asyncio <> pytest-django <> the way django + postgresql works.
This issue is meant as a discussion on what is the right setup, perhaps some users can share.
@pytest.fixture(scope="session")
def postgresql_proc():
with PostgresContainer("postgres:15") as pg:
raw_url = pg.get_connection_url()
clean_url = raw_url.replace("postgresql+psycopg2://", "postgresql://")
cfg = dj_database_url.config(default=clean_url)
proc = SimpleNamespace(
host=pg.get_container_host_ip(),
port=int(pg.get_exposed_port(pg.port)),
user=cfg["USER"],
password=cfg["PASSWORD"],
dbname=cfg["NAME"],
stop=pg.stop,
)
yield proc
@pytest.fixture(scope="session")
def django_db_modify_db_settings(postgresql_proc):
from django.conf import settings
cfg = settings.DATABASES["default"]
cfg.update(
{
"ENGINE": "django.db.backends.postgresql",
"HOST": postgresql_proc.host,
"PORT": postgresql_proc.port,
"NAME": postgresql_proc.dbname,
"USER": postgresql_proc.user,
"PASSWORD": postgresql_proc.password,
"CONN_MAX_AGE": 600,
"CONN_HEALTH_CHECKS": True,
"DISABLE_SERVER_SIDE_CURSORS": True,
}
)
cfg["TEST"]["NAME"] = "test"
settings.DATABASES["default"] = cfg
print(settings.DATABASES["default"])
And then here is a dummy test:
from auth.models import User
@pytest.mark.asyncio
@pytest.mark.django_db
async def test_migration_plan():
out = StringIO()
await sync_to_async(call_command)(
"showmigrations", "--plan", stdout=out, verbosity=1
)
print("\n=== MIGRATION PLAN ===\n", out.getvalue())
recorder = MigrationRecorder(connection)
applied = await sync_to_async(recorder.applied_migrations)()
print(
"\n=== APPLIED MIGRATIONS ===\n",
"\n".join(f"{app}.{name}" for app, name in applied),
)
tables = connection.introspection.table_names()
print("\n=== TABLES IN SCHEMA ===\n", tables)
await DefaultUserFactory() # <----- this fails in async saying user database is not present. works perfectly in sync.
print(User.objects.count())
Let me know if that's not the right forum, and I am happy to move this to pytest-django
perhaps