anti/DECNET
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(live tests): use @pytest_asyncio.fixture for module-scoped async fixtures

Browse Source
This commit is contained in:
anti
2026-05-10 22:30:56 -04:00
parent ab18cd7797
commit b41a7e3115
3 changed files with 10 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
tests/live/test_health_live.py
View File

@@ -14,6 +14,7 @@ from unittest.mock import MagicMock
import httpx import httpx
import pytest import pytest
import pytest_asyncio
# Must be set before any decnet import # Must be set before any decnet import
os.environ.setdefault("DECNET_JWT_SECRET", "test-secret-key-at-least-32-chars-long!!") os.environ.setdefault("DECNET_JWT_SECRET", "test-secret-key-at-least-32-chars-long!!")
@@ -34,7 +35,7 @@ from sqlalchemy.pool import StaticPool # noqa: E402
import uuid as _uuid # noqa: E402 import uuid as _uuid # noqa: E402
@pytest.fixture(scope="module", loop_scope="module", autouse=True) @pytest_asyncio.fixture(scope="module", loop_scope="module", autouse=True)
async def live_db(): async def live_db():
"""Spin up an in-memory SQLite for the live test module.""" """Spin up an in-memory SQLite for the live test module."""
engine = create_async_engine( engine = create_async_engine(
@@ -68,7 +69,7 @@ async def live_db():
await engine.dispose() await engine.dispose()
@pytest.fixture(scope="module", loop_scope="module") @pytest_asyncio.fixture(scope="module", loop_scope="module")
async def live_client(live_db): async def live_client(live_db):
async with httpx.AsyncClient( async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app), transport=httpx.ASGITransport(app=app),
@@ -77,7 +78,7 @@ async def live_client(live_db):
yield ac yield ac
@pytest.fixture(scope="module", loop_scope="module") @pytest_asyncio.fixture(scope="module", loop_scope="module")
async def token(live_client): async def token(live_client):
resp = await live_client.post("/api/v1/auth/login", json={ resp = await live_client.post("/api/v1/auth/login", json={
"username": DECNET_ADMIN_USER, "username": DECNET_ADMIN_USER,

3
tests/live/test_mysql_backend_live.py
View File

@@ -28,6 +28,7 @@ from datetime import datetime, timedelta, timezone
from urllib.parse import urlparse, urlunparse from urllib.parse import urlparse, urlunparse
import pytest import pytest
import pytest_asyncio
from sqlalchemy import text from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.ext.asyncio import create_async_engine
@@ -74,7 +75,7 @@ def _url_with_db(server_url: str, db_name: str) -> str:
return urlunparse(parsed._replace(path=f"/{db_name}")) return urlunparse(parsed._replace(path=f"/{db_name}"))
@pytest.fixture(scope="module", loop_scope="module") @pytest_asyncio.fixture(scope="module", loop_scope="module")
async def mysql_test_db_url(): async def mysql_test_db_url():
"""Create a per-worker throwaway database, yield its URL, drop it on teardown. """Create a per-worker throwaway database, yield its URL, drop it on teardown.

7
tests/live/test_service_isolation_live.py
View File

@@ -23,6 +23,7 @@ from pathlib import Path
import httpx import httpx
import pytest import pytest
import pytest_asyncio
pytestmark = pytest.mark.skipif( pytestmark = pytest.mark.skipif(
os.environ.get("CI") == "true", os.environ.get("CI") == "true",
@@ -66,7 +67,7 @@ from sqlalchemy.pool import StaticPool # noqa: E402
# ─── Shared fixtures ──────────────────────────────────────────────────────── # ─── Shared fixtures ────────────────────────────────────────────────────────
@pytest.fixture(scope="module", loop_scope="module", autouse=True) @pytest_asyncio.fixture(scope="module", loop_scope="module", autouse=True)
async def live_db(): async def live_db():
"""Real in-memory SQLite — shared across this module.""" """Real in-memory SQLite — shared across this module."""
engine = create_async_engine( engine = create_async_engine(
@@ -104,7 +105,7 @@ async def live_db():
await engine.dispose() await engine.dispose()
@pytest.fixture(scope="module", loop_scope="module") @pytest_asyncio.fixture(scope="module", loop_scope="module")
async def live_client(live_db): async def live_client(live_db):
async with httpx.AsyncClient( async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app), transport=httpx.ASGITransport(app=app),
@@ -113,7 +114,7 @@ async def live_client(live_db):
yield ac yield ac
@pytest.fixture(scope="module", loop_scope="module") @pytest_asyncio.fixture(scope="module", loop_scope="module")
async def token(live_client): async def token(live_client):
resp = await live_client.post( resp = await live_client.post(
"/api/v1/auth/login", "/api/v1/auth/login",