-
Notifications
You must be signed in to change notification settings - Fork 163
Closed as not planned
Closed as not planned
Copy link
Labels
needsinfoRequires additional information from the issue authorRequires additional information from the issue author
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
Metadata
Metadata
Assignees
Labels
needsinfoRequires additional information from the issue authorRequires additional information from the issue author