Coverage for tests/test_connect_auth.py: 100%

1"""Tests de l'authentification.""" 

2from __future__ import annotations 

3 

4from unittest.mock import AsyncMock, MagicMock 

5 

6import httpx 

7import pytest 

8 

9from henrri_connect.connect.synchro import SyncHenrriClient 

10from henrri_connect.connect.asynchro import AsyncHenrriClient 

11from henrri_connect.exc import HenrriAuthError 

12from henrri_connect.models import TokenResponse # pylint: disable=W0611 

13 

14 

15def test_sync_headers_and_auth_flow(monkeypatch): # pylint: disable=W0613, C0116 

16 client: SyncHenrriClient = SyncHenrriClient.__new__(SyncHenrriClient) 

17 client._client_id = "cid" # pylint: disable=W0212 

18 client._client_secret = "csecret" # pylint: disable=W0212 

19 client._base_url = "https://api.test" # pylint: disable=W0212 

20 client._access_token = None # pylint: disable=W0212 

21 client._refresh_token_str = None # pylint: disable=W0212 

22 client._http = MagicMock(spec=httpx.Client) # pylint: disable=W0212 

23 

24 # _headers should raise when no token and authenticated=True 

25 with pytest.raises(HenrriAuthError): 

26 client._headers(authenticated=True) # type: ignore[attr-defined] # pylint: disable=W0212 

27 

28 # with token, Authorization header present 

29 client._access_token = "tok" # pylint: disable=W0212 

30 headers = client._headers(authenticated=True) # type: ignore[attr-defined] # pylint: disable=W0212 

31 assert headers["Authorization"] == "Bearer tok" 

32 

33 # Test _request handles 401 by calling _do_refresh when refresh token present 

34 resp1 = MagicMock(spec=httpx.Response) 

35 resp1.status_code = 401 

36 resp1.is_success = False 

37 resp1.json.return_value = {} 

38 

39 resp2 = MagicMock(spec=httpx.Response) 

40 resp2.status_code = 200 

41 resp2.is_success = True 

42 resp2.json.return_value = { 

43 "accessToken": "newtok", 

44 "refreshToken": "newrefresh", 

45 "expiresIn": 3600 

46 } 

47 

48 client._http.request.side_effect = [resp1, resp2] # pylint: disable=W0212 

49 client._access_token = "old" # pylint: disable=W0212 

50 client._refresh_token_str = "r" # pylint: disable=W0212 

51 

52 # replace _do_refresh to set new token 

53 def _do_refresh(): 

54 client._access_token = "newtok" # pylint: disable=W0212 

55 client._do_refresh = _do_refresh # pylint: disable=W0212 

56 

57 out = client._request("GET", "/v1/test", authenticated=True) # type: ignore[attr-defined] # pylint: disable=W0212 

58 assert out is resp2 

59 assert client._access_token == "newtok" # pylint: disable=W0212 

60 assert client._http.request.call_count == 2 # pylint: disable=W0212 

61 

62 

63@pytest.mark.asyncio 

64async def test_async_headers_and_refresh(monkeypatch): # pylint: disable=W0613, C0116 

65 client: AsyncHenrriClient = AsyncHenrriClient.__new__(AsyncHenrriClient) 

66 client._client_id = "cid" # pylint: disable=W0212 

67 client._client_secret = "csecret" # pylint: disable=W0212 

68 client._base_url = "https://api.test" # pylint: disable=W0212 

69 client._access_token = None # pylint: disable=W0212 

70 client._refresh_token_str = None # pylint: disable=W0212 

71 client._http = AsyncMock(spec=httpx.AsyncClient) # pylint: disable=W0212 

72 

73 with pytest.raises(HenrriAuthError): 

74 client._headers(authenticated=True) # type: ignore[attr-defined] # pylint: disable=W0212 

75 

76 client._access_token = "tok" # pylint: disable=W0212 

77 headers = client._headers(authenticated=True) # type: ignore[attr-defined] # pylint: disable=W0212 

78 assert headers["Authorization"] == "Bearer tok" 

79 

80 resp1 = MagicMock(spec=httpx.Response) 

81 resp1.status_code = 401 

82 resp1.is_success = False 

83 resp1.json.return_value = {} 

84 

85 resp2 = MagicMock(spec=httpx.Response) 

86 resp2.status_code = 200 

87 resp2.is_success = True 

88 resp2.json.return_value = { 

89 "accessToken": "newtok", 

90 "refreshToken": "newrefresh", 

91 "expiresIn": 3600 

92 } 

93 

94 # first call: resp1, second call: resp2 

95 client._http.request.side_effect = [resp1, resp2] # pylint: disable=W0212 

96 

97 client._access_token = "old" # pylint: disable=W0212 

98 client._refresh_token_str = "r" # pylint: disable=W0212 

99 

100 async def _do_refresh(): 

101 client._access_token = "newtok" # pylint: disable=W0212 

102 client._do_refresh = _do_refresh # pylint: disable=W0212 

103 

104 out = await client._request("GET", "/v1/test", authenticated=True) # type: ignore[attr-defined] # pylint: disable=W0212 

105 assert out is resp2 

106 assert client._access_token == "newtok" # pylint: disable=W0212 

107 assert client._http.request.call_count == 2 # pylint: disable=W0212