From cc6be091bf9796863228159da83c99174f2af51f Mon Sep 17 00:00:00 2001 From: Deep Kumar Singh Kushwah Date: Mon, 8 Jun 2026 10:51:44 +0530 Subject: [PATCH] fix(tests): use asyncio.run in integration reachability fixtures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `asyncio.get_event_loop()` raises `RuntimeError: There is no current event loop` under Python 3.12, so the integration reachability fixtures (pg / redis / qdrant / minio / weaviate / elasticsearch) errored at setup instead of skipping when the Docker backends aren't running — breaking a full-suite run (`pytest tests packages apps/gateway`) at collection. Switch each probe to `asyncio.run(...)`, which manages its own loop, so the fixtures skip cleanly when a backend is unreachable. Co-Authored-By: Claude Opus 4.7 --- tests/integration/conftest.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 2b59112..6557c21 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -115,28 +115,28 @@ async def _elasticsearch_reachable() -> bool: @pytest.fixture(scope="session") def pg_dsn() -> str: - if not asyncio.get_event_loop().run_until_complete(_pg_reachable()): + if not asyncio.run(_pg_reachable()): pytest.skip("Postgres not reachable — run: task dev") return PG_DSN @pytest.fixture(scope="session") def redis_url() -> str: - if not asyncio.get_event_loop().run_until_complete(_redis_reachable()): + if not asyncio.run(_redis_reachable()): pytest.skip("Redis not reachable — run: task dev") return REDIS_URL @pytest.fixture(scope="session") def qdrant_url() -> str: - if not asyncio.get_event_loop().run_until_complete(_qdrant_reachable()): + if not asyncio.run(_qdrant_reachable()): pytest.skip("Qdrant not reachable — run: task dev") return QDRANT_URL @pytest.fixture(scope="session") def minio_config() -> dict[str, str]: - if not asyncio.get_event_loop().run_until_complete(_minio_reachable()): + if not asyncio.run(_minio_reachable()): pytest.skip("MinIO not reachable — run: task dev (MinIO starts with dev stack)") return { "endpoint": MINIO_ENDPOINT, @@ -148,7 +148,7 @@ def minio_config() -> dict[str, str]: @pytest.fixture(scope="session") def weaviate_config() -> dict[str, int | str]: - if not asyncio.get_event_loop().run_until_complete(_weaviate_reachable()): + if not asyncio.run(_weaviate_reachable()): pytest.skip( f"Weaviate not reachable at http://{WEAVIATE_HOST}:{WEAVIATE_HTTP_PORT} — " "start a local instance or set TEST_WEAVIATE_HOST" @@ -162,7 +162,7 @@ def weaviate_config() -> dict[str, int | str]: @pytest.fixture(scope="session") def elasticsearch_url() -> str: - if not asyncio.get_event_loop().run_until_complete(_elasticsearch_reachable()): + if not asyncio.run(_elasticsearch_reachable()): pytest.skip( f"Elasticsearch not reachable at {ELASTICSEARCH_URL} — " "start a local instance or set TEST_ELASTICSEARCH_URL"