Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/httpx/_transports/mock.py: 50%

1import typing 

2 

3from .._models import Request, Response 

4from .base import AsyncBaseTransport, BaseTransport 

5 

6SyncHandler = typing.Callable[[Request], Response] 

7AsyncHandler = typing.Callable[[Request], typing.Coroutine[None, None, Response]] 

8 

9 

10class MockTransport(AsyncBaseTransport, BaseTransport): 

11 def __init__(self, handler: typing.Union[SyncHandler, AsyncHandler]) -> None: 

12 self.handler = handler 

13 

14 def handle_request( 

15 self, 

16 request: Request, 

17 ) -> Response: 

18 request.read() 

19 response = self.handler(request) 

20 if not isinstance(response, Response): # pragma: no cover 

21 raise TypeError("Cannot use an async handler in a sync Client") 

22 return response 

23 

24 async def handle_async_request( 

25 self, 

26 request: Request, 

27 ) -> Response: 

28 await request.aread() 

29 response = self.handler(request) 

30 

31 # Allow handler to *optionally* be an `async` function. 

32 # If it is, then the `response` variable need to be awaited to actually 

33 # return the result. 

34 

35 if not isinstance(response, Response): 

36 response = await response 

37 

38 return response