Coverage for tests/test_http.py: 100%

1from typing import TYPE_CHECKING, cast 

2 

3import httpx 

4import pytest 

5 

6from prisma.http import HTTP 

7from prisma.utils import _NoneType 

8from prisma._types import Literal 

9from prisma.errors import HTTPClientClosedError 

10 

11from .utils import patch_method 

12 

13if TYPE_CHECKING: 

14 from _pytest.monkeypatch import MonkeyPatch 

15 

16 

17State = Literal['initial', 'open', 'closed'] 

18 

19 

20def assert_session_state(http: HTTP, state: State) -> None: 

21 if state == 'initial': 

22 assert http._session is _NoneType 

23 elif state == 'open': 

24 assert isinstance(http._session, httpx.AsyncClient) 

25 elif state == 'closed': 

26 with pytest.raises(HTTPClientClosedError): 

27 assert http.session 

28 else: # pragma: no cover 

29 raise ValueError(f'Unknown value {state} for state, must be one of initial, open or closed') 

30 

31 

32@pytest.mark.asyncio 

33async def test_request_on_closed_sessions() -> None: 

34 """Attempting to make a request on a closed session raises an error""" 

35 http = HTTP() 

36 http.open() 

37 assert http.closed is False 

38 await http.close() 

39 

40 # mypy thinks that http.closed is Literal[False] 

41 # when it is in fact a bool 

42 closed = cast(bool, http.closed) 

43 assert closed is True 

44 

45 with pytest.raises(HTTPClientClosedError): 

46 await http.request('GET', '/') 

47 

48 

49@pytest.mark.asyncio 

50async def test_lazy_session_open() -> None: 

51 """Accessing the session property opens the session""" 

52 http = HTTP() 

53 assert_session_state(http, 'initial') 

54 

55 # access the session property, opening the session 

56 # TODO: test using an actual request 

57 assert http.session 

58 

59 assert_session_state(http, 'open') 

60 await http.close() 

61 assert_session_state(http, 'closed') 

62 

63 

64@pytest.mark.asyncio 

65async def test_httpx_default_config(monkeypatch: 'MonkeyPatch') -> None: 

66 """The default timeout is passed to HTTPX""" 

67 http = HTTP() 

68 assert_session_state(http, 'initial') 

69 

70 getter = patch_method(monkeypatch, httpx.AsyncClient, '__init__') 

71 

72 http.open() 

73 assert_session_state(http, 'open') 

74 

75 # hardcode the default config to ensure there are no unintended changes 

76 captured = getter() 

77 assert captured == ( 

78 (), 

79 { 

80 'limits': httpx.Limits(max_connections=1000), 

81 'timeout': httpx.Timeout(30), 

82 }, 

83 )