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

1from __future__ import annotations 

2 

3import typing 

4 

5from .._models import Request, Response 

6from .base import AsyncBaseTransport, BaseTransport 

7 

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

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

10 

11 

12__all__ = ["MockTransport"] 

13 

14 

15class MockTransport(AsyncBaseTransport, BaseTransport): 

16 def __init__(self, handler: SyncHandler | AsyncHandler) -> None: 

17 self.handler = handler 

18 

19 def handle_request( 

20 self, 

21 request: Request, 

22 ) -> Response: 

23 request.read() 

24 response = self.handler(request) 

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

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

27 return response 

28 

29 async def handle_async_request( 

30 self, 

31 request: Request, 

32 ) -> Response: 

33 await request.aread() 

34 response = self.handler(request) 

35 

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

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

38 # return the result. 

39 

40 if not isinstance(response, Response): 

41 response = await response 

42 

43 return response