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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
|
"""Test cases for initialisation of the Kamereon client."""
from typing import cast
import aiohttp
import pytest
from aioresponses import aioresponses
from tests import fixtures
from tests.const import TEST_COUNTRY
from tests.const import TEST_LOCALE
from tests.const import TEST_LOCALE_DETAILS
from tests.const import TEST_LOGIN_TOKEN
from tests.const import TEST_PASSWORD
from tests.const import TEST_PERSON_ID
from tests.const import TEST_USERNAME
from tests.test_credential_store import get_logged_in_credential_store
from renault_api.credential import JWTCredential
from renault_api.exceptions import NotAuthenticatedException
from renault_api.exceptions import RenaultException
from renault_api.gigya import GIGYA_JWT
from renault_api.gigya import GIGYA_LOGIN_TOKEN
from renault_api.renault_session import RenaultSession
def get_logged_in_session(websession: aiohttp.ClientSession) -> RenaultSession:
"""Get initialised RenaultSession."""
return RenaultSession(
websession=websession,
country=TEST_COUNTRY,
locale=TEST_LOCALE,
locale_details=TEST_LOCALE_DETAILS,
credential_store=get_logged_in_credential_store(),
)
@pytest.fixture
def session(websession: aiohttp.ClientSession) -> RenaultSession:
"""Fixture for testing RenaultSession."""
return RenaultSession(
websession=websession,
country=TEST_COUNTRY,
locale_details=TEST_LOCALE_DETAILS,
)
@pytest.mark.asyncio
async def test_init_locale_only(websession: aiohttp.ClientSession) -> None:
"""Test initialisation with locale only."""
session = RenaultSession(
websession=websession,
locale=TEST_LOCALE,
)
assert await session._get_country()
assert await session._get_gigya_api_key()
assert await session._get_gigya_root_url()
assert await session._get_kamereon_api_key()
assert await session._get_kamereon_root_url()
@pytest.mark.asyncio
async def test_init_country_only(websession: aiohttp.ClientSession) -> None:
"""Test initialisation with country only."""
session = RenaultSession(
websession=websession,
country=TEST_COUNTRY,
)
assert await session._get_country()
with pytest.raises(
RenaultException,
match="Credential `gigya-api-key` not found in credential cache.",
):
assert await session._get_gigya_api_key()
with pytest.raises(
RenaultException,
match="Credential `gigya-root-url` not found in credential cache.",
):
assert await session._get_gigya_root_url()
with pytest.raises(
RenaultException,
match="Credential `kamereon-api-key` not found in credential cache.",
):
assert await session._get_kamereon_api_key()
with pytest.raises(
RenaultException,
match="Credential `kamereon-root-url` not found in credential cache.",
):
assert await session._get_kamereon_root_url()
@pytest.mark.asyncio
async def test_init_locale_details_only(websession: aiohttp.ClientSession) -> None:
"""Test initialisation with locale_details only."""
session = RenaultSession(
websession=websession,
locale_details=TEST_LOCALE_DETAILS,
)
with pytest.raises(
RenaultException,
match="Credential `country` not found in credential cache.",
):
assert await session._get_country()
assert await session._get_gigya_api_key()
assert await session._get_gigya_root_url()
assert await session._get_kamereon_api_key()
assert await session._get_kamereon_root_url()
@pytest.mark.asyncio
async def test_init_locale_and_details(websession: aiohttp.ClientSession) -> None:
"""Test initialisation with locale and locale_details."""
session = RenaultSession(
websession=websession,
locale=TEST_LOCALE,
locale_details=TEST_LOCALE_DETAILS,
)
assert await session._get_country()
assert await session._get_gigya_api_key()
assert await session._get_gigya_root_url()
assert await session._get_kamereon_api_key()
assert await session._get_kamereon_root_url()
@pytest.mark.asyncio
async def test_init_locale_country(websession: aiohttp.ClientSession) -> None:
"""Test initialisation with locale and country."""
session = RenaultSession(
websession=websession,
locale=TEST_LOCALE,
country=TEST_COUNTRY,
)
assert await session._get_country()
assert await session._get_gigya_api_key()
assert await session._get_gigya_root_url()
assert await session._get_kamereon_api_key()
assert await session._get_kamereon_root_url()
@pytest.mark.asyncio
async def test_not_logged_in(session: RenaultSession) -> None:
"""Test errors when not logged in."""
with pytest.raises(
NotAuthenticatedException,
match="Gigya login token not available.",
):
await session._get_login_token()
with pytest.raises(
NotAuthenticatedException,
match="Gigya login token not available.",
):
await session._get_person_id()
with pytest.raises(
NotAuthenticatedException,
match="Gigya login token not available.",
):
await session._get_jwt()
@pytest.mark.asyncio
async def test_login(session: RenaultSession, mocked_responses: aioresponses) -> None:
"""Test login/person/jwt response."""
fixtures.inject_gigya_all(mocked_responses)
await session.login(TEST_USERNAME, TEST_PASSWORD)
assert await session._get_login_token() == TEST_LOGIN_TOKEN
assert len(mocked_responses.requests) == 1
assert await session._get_person_id() == TEST_PERSON_ID
assert len(mocked_responses.requests) == 2
assert await session._get_jwt()
assert len(mocked_responses.requests) == 3
# Ensure further requests use cache
assert await session._get_person_id() == TEST_PERSON_ID
assert await session._get_jwt()
assert len(mocked_responses.requests) == 3
@pytest.mark.asyncio
async def test_expired_login_token(
websession: aiohttp.ClientSession, mocked_responses: aioresponses
) -> None:
"""Test _get_jwt response on expired login token."""
session = get_logged_in_session(websession=websession)
fixtures.inject_gigya(
mocked_responses,
urlpath="accounts.getJWT",
filename="error/get_jwt.403005.json",
)
# First attempt uses cached values
assert await session._get_jwt()
assert len(mocked_responses.requests) == 0
assert GIGYA_JWT in session._credentials
assert GIGYA_LOGIN_TOKEN in session._credentials
# mark JWT as expired
jwt_credential = cast(JWTCredential, session._credentials.get(GIGYA_JWT))
jwt_credential.expiry = 1
# first attempt show authentication as expired
with pytest.raises(
NotAuthenticatedException,
match="Authentication expired.",
):
assert await session._get_jwt()
assert len(mocked_responses.requests) == 1
assert GIGYA_JWT not in session._credentials
assert GIGYA_LOGIN_TOKEN not in session._credentials
# subsequent attempts just show not authenticated
with pytest.raises(
NotAuthenticatedException,
match="Gigya login token not available.",
):
assert await session._get_jwt()
assert len(mocked_responses.requests) == 1
|
