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
|
"""Test OAuth2 device code flow implementation."""
import time
from unittest.mock import AsyncMock, patch
import pytest
from actron_neo_api import ActronAirAPI, ActronAirOAuth2DeviceCodeAuth
class TestActronAirOAuth2DeviceCodeAuth:
"""Test OAuth2 device code flow authentication."""
def test_init(self):
"""Test ActronAirOAuth2DeviceCodeAuth initialization."""
auth = ActronAirOAuth2DeviceCodeAuth("https://example.com", "test_client")
assert auth.base_url == "https://example.com"
assert auth.client_id == "test_client"
assert auth.access_token is None
assert auth.refresh_token is None
assert auth.token_type == "Bearer"
assert auth.token_expiry is None
assert not auth.is_token_valid
assert not auth.is_token_expiring_soon
@pytest.mark.asyncio
async def test_request_device_code_success(self):
"""Test successful device code request."""
auth = ActronAirOAuth2DeviceCodeAuth("https://example.com", "test_client")
mock_response = {
"device_code": "test_device_code",
"user_code": "TEST123",
"verification_uri": "https://example.com/device",
"expires_in": 600,
"interval": 5,
}
with patch("aiohttp.ClientSession.post") as mock_post:
mock_post.return_value.__aenter__.return_value.status = 200
mock_post.return_value.__aenter__.return_value.json.return_value = mock_response
result = await auth.request_device_code()
assert result["device_code"] == "test_device_code"
assert result["user_code"] == "TEST123"
assert result["verification_uri"] == "https://example.com/device"
assert "verification_uri_complete" in result
@pytest.mark.asyncio
async def test_poll_for_token_success(self):
"""Test successful token polling."""
auth = ActronAirOAuth2DeviceCodeAuth("https://example.com", "test_client")
mock_response = {
"access_token": "test_access_token",
"refresh_token": "test_refresh_token",
"token_type": "Bearer",
"expires_in": 3600,
}
with patch("aiohttp.ClientSession.post") as mock_post:
mock_post.return_value.__aenter__.return_value.status = 200
mock_post.return_value.__aenter__.return_value.json.return_value = mock_response
result = await auth.poll_for_token("test_device_code")
assert result["access_token"] == "test_access_token"
assert auth.access_token == "test_access_token"
assert auth.refresh_token == "test_refresh_token"
assert auth.is_token_valid
@pytest.mark.asyncio
async def test_poll_for_token_pending(self):
"""Test token polling when authorization is pending and times out."""
auth = ActronAirOAuth2DeviceCodeAuth("https://example.com", "test_client")
mock_response = {"error": "authorization_pending"}
with patch("aiohttp.ClientSession.post") as mock_post, patch("time.time") as mock_time:
# Simulate timeout by advancing time past the threshold
# Provide enough values for: start_time, while condition checks, and logging
mock_time.side_effect = [0, 0, 601, 601, 601]
mock_post.return_value.__aenter__.return_value.status = 400
mock_post.return_value.__aenter__.return_value.json.return_value = mock_response
result = await auth.poll_for_token("test_device_code", interval=1, timeout=1)
assert result is None
@pytest.mark.asyncio
async def test_refresh_access_token(self):
"""Test access token refresh."""
auth = ActronAirOAuth2DeviceCodeAuth("https://example.com", "test_client")
auth.refresh_token = "test_refresh_token"
mock_response = {
"access_token": "new_access_token",
"refresh_token": "new_refresh_token",
"token_type": "Bearer",
"expires_in": 3600,
}
with patch("aiohttp.ClientSession.post") as mock_post:
mock_post.return_value.__aenter__.return_value.status = 200
mock_post.return_value.__aenter__.return_value.json.return_value = mock_response
token, expiry = await auth.refresh_access_token()
assert token == "new_access_token"
assert auth.access_token == "new_access_token"
assert auth.refresh_token == "new_refresh_token"
@pytest.mark.asyncio
async def test_get_user_info(self):
"""Test getting user information."""
auth = ActronAirOAuth2DeviceCodeAuth("https://example.com", "test_client")
auth.access_token = "test_access_token"
auth.refresh_token = "test_refresh_token" # Add refresh token to avoid error
auth.token_expiry = time.time() + 3600 # Set token as valid
mock_response = {"id": "test_user_id", "email": "test@example.com", "name": "Test User"}
with patch("aiohttp.ClientSession.get") as mock_get:
mock_get.return_value.__aenter__.return_value.status = 200
mock_get.return_value.__aenter__.return_value.json.return_value = mock_response
result = await auth.get_user_info()
assert result["id"] == "test_user_id"
assert result["email"] == "test@example.com"
def test_set_tokens(self):
"""Test manually setting tokens."""
auth = ActronAirOAuth2DeviceCodeAuth("https://example.com", "test_client")
auth.set_tokens(
access_token="test_access_token",
refresh_token="test_refresh_token",
expires_in=3600,
token_type="Bearer",
)
assert auth.access_token == "test_access_token"
assert auth.refresh_token == "test_refresh_token"
assert auth.token_type == "Bearer"
assert auth.is_token_valid
class TestActronAirAPIWithOAuth2:
"""Test ActronAirAPI with OAuth2 integration."""
def test_init_default(self):
"""Test ActronAirAPI initialization with default parameters."""
api = ActronAirAPI()
assert api.oauth2_auth is not None
assert api.oauth2_auth.base_url == "https://nimbus.actronair.com.au"
assert api.oauth2_auth.client_id == "home_assistant"
assert api.oauth2_auth.refresh_token is None
def test_init_with_refresh_token(self):
"""Test ActronAirAPI initialization with refresh token."""
api = ActronAirAPI(refresh_token="test_refresh_token")
assert api.oauth2_auth is not None
assert api.oauth2_auth.refresh_token == "test_refresh_token"
def test_init_with_custom_params(self):
"""Test ActronAirAPI initialization with custom parameters."""
api = ActronAirAPI(
oauth2_client_id="custom_client",
refresh_token="custom_token",
platform="neo",
)
assert api.oauth2_auth.client_id == "custom_client"
assert api.oauth2_auth.refresh_token == "custom_token"
assert api.base_url == "https://nimbus.actronair.com.au"
@pytest.mark.asyncio
async def test_oauth2_methods_available(self):
"""Test OAuth2 methods are available."""
api = ActronAirAPI()
# Mock the OAuth2 auth methods
api.oauth2_auth.request_device_code = AsyncMock(return_value={"device_code": "test"})
api.oauth2_auth.poll_for_token = AsyncMock(return_value={"access_token": "test"})
api.oauth2_auth.get_user_info = AsyncMock(return_value={"id": "test"})
# Test methods
device_code = await api.request_device_code()
token_data = await api.poll_for_token("test_device_code")
user_info = await api.get_user_info()
assert device_code["device_code"] == "test"
assert token_data["access_token"] == "test"
assert user_info["id"] == "test"
def test_token_properties(self):
"""Test token properties work correctly."""
api = ActronAirAPI(refresh_token="test_refresh_token")
api.oauth2_auth.access_token = "test_access_token"
assert api.access_token == "test_access_token"
assert api.refresh_token_value == "test_refresh_token"
if __name__ == "__main__":
pytest.main([__file__])
|
