1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
from __future__ import annotations
import typing
from .._models import Request, Response
from .base import AsyncBaseTransport, BaseTransport
SyncHandler = typing.Callable[[Request], Response]
AsyncHandler = typing.Callable[[Request], typing.Coroutine[None, None, Response]]
class MockTransport(AsyncBaseTransport, BaseTransport):
def __init__(self, handler: SyncHandler | AsyncHandler) -> None:
self.handler = handler
def handle_request(
self,
request: Request,
) -> Response:
request.read()
response = self.handler(request)
if not isinstance(response, Response): # pragma: no cover
raise TypeError("Cannot use an async handler in a sync Client")
return response
async def handle_async_request(
self,
request: Request,
) -> Response:
await request.aread()
response = self.handler(request)
# Allow handler to *optionally* be an `async` function.
# If it is, then the `response` variable need to be awaited to actually
# return the result.
if not isinstance(response, Response):
response = await response
return response
|
