70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
"""Fixtures for MangaCollec requests without network access or real waiting."""
|
|
|
|
import asyncio
|
|
from collections.abc import Iterator
|
|
from unittest.mock import AsyncMock
|
|
|
|
import httpx
|
|
import pytest
|
|
from mangacollec.mangacollec import Mangacollec
|
|
from pytest_mock import MockerFixture
|
|
|
|
|
|
@pytest.fixture
|
|
def responses() -> list[httpx.Response | httpx.RequestError]:
|
|
"""Return the sequence of HTTP responses and transport errors to simulate."""
|
|
return []
|
|
|
|
|
|
@pytest.fixture
|
|
def sent_requests() -> list[httpx.Request]:
|
|
"""Record requests made through the mock transport."""
|
|
return []
|
|
|
|
|
|
@pytest.fixture
|
|
def client(
|
|
responses: list[httpx.Response | httpx.RequestError],
|
|
sent_requests: list[httpx.Request],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> Iterator[Mangacollec]:
|
|
"""Provide an authenticated client with mocked resource and OAuth transports."""
|
|
monkeypatch.setenv("BOOKSYNC_REQUEST_TIMEOUT", "30")
|
|
client = Mangacollec("test-client", "test-secret")
|
|
asyncio.run(client.close())
|
|
|
|
def handle_request(request: httpx.Request) -> httpx.Response:
|
|
"""Consume the next simulated response and record its request."""
|
|
sent_requests.append(request)
|
|
response = responses.pop(0)
|
|
if isinstance(response, httpx.RequestError):
|
|
raise response
|
|
return response
|
|
|
|
transport = httpx.MockTransport(handle_request)
|
|
client._client = httpx.AsyncClient(
|
|
base_url="https://api.mangacollec.com/", transport=transport, trust_env=False,
|
|
)
|
|
client._oauth_client = httpx.AsyncClient(
|
|
base_url="https://api.mangacollec.com/", transport=transport, trust_env=False,
|
|
)
|
|
client._set_token({
|
|
"access_token": "test-token",
|
|
"token_type": "Bearer",
|
|
"expires_in": 86400,
|
|
"created_at": 0,
|
|
})
|
|
|
|
try:
|
|
yield client
|
|
finally:
|
|
asyncio.run(client.close())
|
|
|
|
|
|
@pytest.fixture
|
|
def sleep_mock(mocker: MockerFixture) -> AsyncMock:
|
|
"""Record asynchronous waits while completing them immediately."""
|
|
sleep = AsyncMock(return_value=None)
|
|
mocker.patch("mangacollec.mangacollec.asyncio.sleep", new=sleep)
|
|
return sleep
|