Мобильная версияКод: Выделить всё
RuntimeError: Timeout context manager should be used inside a taskЯ попробовал практически все решения и хаки, которые смог найти, и, похоже, ничего не работает (nest_asyncio , плагины pytest и т. д.)
Вот код с ошибкой:
Код: Выделить всё
#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
Код: Выделить всё
#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")
Код: Выделить всё
#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())
Подробнее здесь: https://stackoverflow.com/questions/793 ... ith-pytest
