RuntimeError: „Timeout-Kontextmanager sollte innerhalb einer Aufgabe verwendet werden“ mit pytest-asyncio, funktioniert ⇐ Python

[Guest]

Ich verwende Asyncio und Aiohttp in einer Anwendung, die ich erstelle, und kann anscheinend nicht herausfinden, wie ich Pytest gut zum Laufen bringen kann. Wenn ich Pytest verwende, erhalte ich immer ein

Code: Select all

RuntimeError: Timeout context manager should be used inside a task

Wenn ich dieselben Funktionen, die Pytest aufruft, nur in main() ausführe, scheint das Problem zu verschwinden. Ich habe ein Repo zur einfachen Reproduktion unter https://github.com/bcherb2/async_bug hochgeladen.
Ich habe fast jede Lösung und jeden Hack ausprobiert, die ich finden kann, aber nichts scheint zu funktionieren (nest_asyncio , Pytest-Plugins usw.)
Hier ist der fehlerhafte Code:

Code: Select all

#api_client.py
import aiohttp
import uuid
import json
from enum import Enum
from typing import Optional, Dict, Any
from loguru import logger

class RetCode(Enum):
NO_ERROR = 200
BAD_REQUEST = 400
UNAUTHORIZED = 401
NOT_FOUND = 404

class DemoAPIClient:
"""Demo REST client that simulates behavior similar to ANTServerRESTClient."""

def __init__(
self,
base_url: str = "https://jsonplaceholder.typicode.com",
timeout: int = 30
):
"""Initialize the API client.

Args:
base_url: Base URL for the API
timeout: Request timeout in seconds
"""
self.base_url = base_url
self.timeout = timeout

# Session management
self._session: Optional[aiohttp.ClientSession] = None
self._session_token: Optional[str] = None

async def _ensure_session(self) -> aiohttp.ClientSession:
"""Ensure we have an active session, creating one if necessary."""
if self._session is None or self._session.closed:
connector = aiohttp.TCPConnector(force_close=True)
self._session = aiohttp.ClientSession(
connector=connector,
timeout=aiohttp.ClientTimeout(total=self.timeout)
)
return self._session

async def close(self) -> None:
"""Close the client session."""
if self._session:
await self._session.close()
self._session = None
logger.debug("Session closed")

async def login(self) -> None:
"""Simulate login by making a test request."""
try:

test_url = f"{self.base_url}/posts/1"
session = await self._ensure_session()

async with session.get(test_url) as response:
if response.status != 200:
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=response.history,
status=response.status,
message=f"Login failed with status {response.status}"
)

# Simulate session token
self._session_token = str(uuid.uuid4())
logger.info("Successfully logged in to API")

except Exception as e:
logger.error(f"Login failed: {str(e)}")
raise

async def rest(
self,
endpoint: str,
method: str,
data: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Execute a REST request.

Args:
endpoint: The endpoint path (e.g., '/posts')
method: HTTP method (GET, POST, etc.)
data: Optional request body data

Returns:
Dict containing the parsed response data
"""
if not self._session_token:
raise RuntimeError("Not logged in.  Call login() first")

session = await self._ensure_session()
request_id = str(uuid.uuid4())[:8]
url = f"{self.base_url}{endpoint}"

try:
logger.debug(f"[{request_id}] {method} {url}")
if data:
logger.debug(f"[{request_id}] Request body: {data}")

headers = {"Authorization": f"Bearer {self._session_token}"}

async with session.request(
method=method,
url=url,
json=data,
headers=headers
) as response:
response_text = await response.text()
logger.debug(f"[{request_id}] Response: {response_text}")

if response.status >= 400:
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=response.history,
status=response.status,
message=f"Request failed: {response_text}"
)

return json.loads(response_text)

except Exception as e:
logger.error(f"[{request_id}] Request failed: {str(e)}")
raise

Code: Select all

#conftest.py

import pytest_asyncio
from loguru import logger
from api_client import DemoAPIClient

def pytest_configure(config):
config.option.asyncio_mode = "auto"

@pytest_asyncio.fixture(scope="module")
async def api_client():
"""Fixture to provide an authenticated API client."""
logger.info("Setting up API client")
client = DemoAPIClient()

try:
await client.login()
logger.info("API client logged in successfully")
yield client
finally:
await client.close()
logger.info("API client closed")

Code: Select all

#test_api_client.py

import pytest
import asyncio
from loguru import logger
from api_client import DemoAPIClient

async def ensure_task_context():
"""Helper to ensure we're in a task context."""
if asyncio.current_task() is None:
task = asyncio.create_task(asyncio.sleep(0))
await task

@pytest.mark.asyncio
async def test_client_setup(api_client):
"""Test basic client setup."""
logger.debug("Testing client setup")
assert api_client._session_token is not None
assert api_client._session is not None
logger.debug("Client setup verified")

@pytest.mark.asyncio
async def test_get_post(api_client):
"""Test retrieving a post."""
await ensure_task_context()  # Try to ensure task context

try:
response = await api_client.rest("/posts/1", "GET")
assert response is not None
assert "id" in response
assert response["id"] == 1
except Exception as e:
logger.error(f"Test failed: {str(e)}")
raise

@pytest.mark.asyncio
async def test_create_post(api_client):
"""Test creating a new post."""
await ensure_task_context()  # Try to ensure task context

try:
new_post = {
"title": "Test Post",
"body": "Test Content",
"userId": 1
}
response = await api_client.rest("/posts", "POST", new_post)
assert response is not None
assert "id"  in response
assert response["title"] == "Test Post"
except Exception as e:
logger.error(f"Test failed: {str(e)}")
raise

async def main():
"""Main function to run tests directly without pytest."""
logger.info("Starting direct test execution")

client = DemoAPIClient()

try:
await client.login()
logger.info("Client logged in")

logger.info("Running test_client_setup")
await test_client_setup(client)
logger.info("Client setup test passed")

logger.info("Running test_get_post")
await test_get_post(client)
logger.info("Get post test passed")

logger.info("Running test_create_post")
await test_create_post(client)
logger.info("Create post test passed")

except Exception as e:
logger.error(f"Test execution failed: {str(e)}")
raise
finally:
logger.info("Cleaning up client")
await client.close()
logger.info("Client closed")

if __name__ == "__main__":
asyncio.run(main())

vollständiger Trace eines der fehlgeschlagenen Tests:

Code: Select all

________________________________________________________________________________________ test_create_post _________________________________________________________________________________________

api_client = 

@pytest.mark.asyncio
async def test_create_post(api_client):
"""Test creating a new post."""
await ensure_task_context()  # Try to ensure task context

try:
new_post = {
"title": "Test Post",
"body": "Test Content",
"userId": 1
}
>           response = await api_client.rest("/posts", "POST", new_post)

test_api_client.py:51:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
api_client.py:110: in rest
async with session.request(
.venv/lib/python3.12/site-packages/aiohttp/client.py:1425: in __aenter__
self._resp: _RetType = await self._coro
.venv/lib/python3.12/site-packages/aiohttp/client.py:607: in _request
with timer:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = 

def __enter__(self) -> BaseTimerContext:
task = asyncio.current_task(loop=self._loop)
if task is None:
>           raise RuntimeError("Timeout context manager should be used inside a task")
E           RuntimeError: Timeout context manager should be used inside a task

.venv/lib/python3.12/site-packages/aiohttp/helpers.py:636: RuntimeError

Dann führen Sie einfach pytest test_api_client.py und python test_api_client.py aus. Warum scheitert das? Gibt es eine Möglichkeit, das zu beheben?