|
1
|
"""Tests for the unified auth module.""" |
|
2
|
|
|
3
|
import json |
|
4
|
import os |
|
5
|
import time |
|
6
|
from pathlib import Path |
|
7
|
from unittest.mock import MagicMock, patch |
|
8
|
|
|
9
|
from video_processor.auth import ( |
|
10
|
KNOWN_CONFIGS, |
|
11
|
AuthConfig, |
|
12
|
AuthResult, |
|
13
|
OAuthManager, |
|
14
|
get_auth_config, |
|
15
|
get_auth_manager, |
|
16
|
) |
|
17
|
|
|
18
|
# ----------------------------------------------------------------------- |
|
19
|
# AuthConfig |
|
20
|
# ----------------------------------------------------------------------- |
|
21
|
|
|
22
|
|
|
23
|
class TestAuthConfig: |
|
24
|
def test_basic_creation(self): |
|
25
|
config = AuthConfig(service="test") |
|
26
|
assert config.service == "test" |
|
27
|
assert config.supports_oauth is False |
|
28
|
|
|
29
|
def test_with_oauth_endpoints(self): |
|
30
|
config = AuthConfig( |
|
31
|
service="test", |
|
32
|
oauth_authorize_url="https://example.com/auth", |
|
33
|
oauth_token_url="https://example.com/token", |
|
34
|
) |
|
35
|
assert config.supports_oauth is True |
|
36
|
|
|
37
|
def test_resolved_client_id_from_env(self): |
|
38
|
config = AuthConfig( |
|
39
|
service="test", |
|
40
|
client_id_env="TEST_CLIENT_ID", |
|
41
|
) |
|
42
|
with patch.dict(os.environ, {"TEST_CLIENT_ID": "my-id"}): |
|
43
|
assert config.resolved_client_id == "my-id" |
|
44
|
|
|
45
|
def test_resolved_client_id_explicit(self): |
|
46
|
config = AuthConfig( |
|
47
|
service="test", |
|
48
|
client_id="explicit-id", |
|
49
|
client_id_env="TEST_CLIENT_ID", |
|
50
|
) |
|
51
|
assert config.resolved_client_id == "explicit-id" |
|
52
|
|
|
53
|
def test_resolved_api_key(self): |
|
54
|
config = AuthConfig(service="test", api_key_env="TEST_API_KEY") |
|
55
|
with patch.dict(os.environ, {"TEST_API_KEY": "sk-123"}): |
|
56
|
assert config.resolved_api_key == "sk-123" |
|
57
|
|
|
58
|
def test_resolved_api_key_empty(self): |
|
59
|
config = AuthConfig(service="test", api_key_env="TEST_API_KEY") |
|
60
|
with patch.dict(os.environ, {}, clear=True): |
|
61
|
assert config.resolved_api_key is None |
|
62
|
|
|
63
|
def test_resolved_token_path_default(self): |
|
64
|
config = AuthConfig(service="zoom") |
|
65
|
assert config.resolved_token_path.name == "zoom_token.json" |
|
66
|
|
|
67
|
def test_resolved_token_path_custom(self): |
|
68
|
config = AuthConfig( |
|
69
|
service="zoom", |
|
70
|
token_path=Path("/tmp/custom.json"), |
|
71
|
) |
|
72
|
assert config.resolved_token_path == Path("/tmp/custom.json") |
|
73
|
|
|
74
|
def test_resolved_account_id(self): |
|
75
|
config = AuthConfig( |
|
76
|
service="test", |
|
77
|
account_id_env="TEST_ACCOUNT_ID", |
|
78
|
) |
|
79
|
with patch.dict(os.environ, {"TEST_ACCOUNT_ID": "acc-123"}): |
|
80
|
assert config.resolved_account_id == "acc-123" |
|
81
|
|
|
82
|
|
|
83
|
# ----------------------------------------------------------------------- |
|
84
|
# AuthResult |
|
85
|
# ----------------------------------------------------------------------- |
|
86
|
|
|
87
|
|
|
88
|
class TestAuthResult: |
|
89
|
def test_success(self): |
|
90
|
result = AuthResult( |
|
91
|
success=True, |
|
92
|
access_token="tok", |
|
93
|
method="api_key", |
|
94
|
) |
|
95
|
assert result.success |
|
96
|
assert result.access_token == "tok" |
|
97
|
|
|
98
|
def test_failure(self): |
|
99
|
result = AuthResult(success=False, error="no key") |
|
100
|
assert not result.success |
|
101
|
assert result.error == "no key" |
|
102
|
|
|
103
|
|
|
104
|
# ----------------------------------------------------------------------- |
|
105
|
# OAuthManager |
|
106
|
# ----------------------------------------------------------------------- |
|
107
|
|
|
108
|
|
|
109
|
class TestOAuthManager: |
|
110
|
def test_api_key_fallback(self): |
|
111
|
config = AuthConfig( |
|
112
|
service="test", |
|
113
|
api_key_env="TEST_KEY", |
|
114
|
) |
|
115
|
manager = OAuthManager(config) |
|
116
|
with patch.dict(os.environ, {"TEST_KEY": "sk-abc"}): |
|
117
|
result = manager.authenticate() |
|
118
|
assert result.success |
|
119
|
assert result.access_token == "sk-abc" |
|
120
|
assert result.method == "api_key" |
|
121
|
|
|
122
|
def test_no_auth_available(self): |
|
123
|
config = AuthConfig(service="test") |
|
124
|
manager = OAuthManager(config) |
|
125
|
with patch.dict(os.environ, {}, clear=True): |
|
126
|
result = manager.authenticate() |
|
127
|
assert not result.success |
|
128
|
assert "No auth method" in result.error |
|
129
|
|
|
130
|
def test_saved_token_valid(self, tmp_path): |
|
131
|
token_file = tmp_path / "token.json" |
|
132
|
token_data = { |
|
133
|
"access_token": "saved-tok", |
|
134
|
"expires_at": time.time() + 3600, |
|
135
|
} |
|
136
|
token_file.write_text(json.dumps(token_data)) |
|
137
|
|
|
138
|
config = AuthConfig( |
|
139
|
service="test", |
|
140
|
token_path=token_file, |
|
141
|
) |
|
142
|
manager = OAuthManager(config) |
|
143
|
result = manager.authenticate() |
|
144
|
assert result.success |
|
145
|
assert result.access_token == "saved-tok" |
|
146
|
assert result.method == "saved_token" |
|
147
|
|
|
148
|
def test_saved_token_expired_no_refresh(self, tmp_path): |
|
149
|
token_file = tmp_path / "token.json" |
|
150
|
token_data = { |
|
151
|
"access_token": "old-tok", |
|
152
|
"expires_at": time.time() - 100, |
|
153
|
} |
|
154
|
token_file.write_text(json.dumps(token_data)) |
|
155
|
|
|
156
|
config = AuthConfig( |
|
157
|
service="test", |
|
158
|
token_path=token_file, |
|
159
|
) |
|
160
|
manager = OAuthManager(config) |
|
161
|
result = manager.authenticate() |
|
162
|
assert not result.success |
|
163
|
|
|
164
|
def test_get_token_convenience(self): |
|
165
|
config = AuthConfig( |
|
166
|
service="test", |
|
167
|
api_key_env="TEST_KEY", |
|
168
|
) |
|
169
|
manager = OAuthManager(config) |
|
170
|
with patch.dict(os.environ, {"TEST_KEY": "sk-xyz"}): |
|
171
|
token = manager.get_token() |
|
172
|
assert token == "sk-xyz" |
|
173
|
|
|
174
|
def test_get_token_none_on_failure(self): |
|
175
|
config = AuthConfig(service="test") |
|
176
|
manager = OAuthManager(config) |
|
177
|
with patch.dict(os.environ, {}, clear=True): |
|
178
|
token = manager.get_token() |
|
179
|
assert token is None |
|
180
|
|
|
181
|
def test_clear_token(self, tmp_path): |
|
182
|
token_file = tmp_path / "token.json" |
|
183
|
token_file.write_text("{}") |
|
184
|
config = AuthConfig(service="test", token_path=token_file) |
|
185
|
manager = OAuthManager(config) |
|
186
|
manager.clear_token() |
|
187
|
assert not token_file.exists() |
|
188
|
|
|
189
|
def test_clear_token_no_file(self, tmp_path): |
|
190
|
config = AuthConfig( |
|
191
|
service="test", |
|
192
|
token_path=tmp_path / "nonexistent.json", |
|
193
|
) |
|
194
|
manager = OAuthManager(config) |
|
195
|
manager.clear_token() # should not raise |
|
196
|
|
|
197
|
def test_save_token_creates_dir(self, tmp_path): |
|
198
|
nested = tmp_path / "deep" / "dir" / "token.json" |
|
199
|
config = AuthConfig(service="test", token_path=nested) |
|
200
|
manager = OAuthManager(config) |
|
201
|
manager._save_token({"access_token": "tok"}) |
|
202
|
assert nested.exists() |
|
203
|
data = json.loads(nested.read_text()) |
|
204
|
assert data["access_token"] == "tok" |
|
205
|
|
|
206
|
def test_saved_token_expired_with_refresh(self, tmp_path): |
|
207
|
token_file = tmp_path / "token.json" |
|
208
|
token_data = { |
|
209
|
"access_token": "old-tok", |
|
210
|
"refresh_token": "ref-tok", |
|
211
|
"expires_at": time.time() - 100, |
|
212
|
"client_id": "cid", |
|
213
|
"client_secret": "csec", |
|
214
|
} |
|
215
|
token_file.write_text(json.dumps(token_data)) |
|
216
|
|
|
217
|
config = AuthConfig( |
|
218
|
service="test", |
|
219
|
oauth_token_url="https://example.com/token", |
|
220
|
token_path=token_file, |
|
221
|
) |
|
222
|
manager = OAuthManager(config) |
|
223
|
|
|
224
|
mock_resp = MagicMock() |
|
225
|
mock_resp.json.return_value = { |
|
226
|
"access_token": "new-tok", |
|
227
|
"refresh_token": "new-ref", |
|
228
|
"expires_in": 7200, |
|
229
|
} |
|
230
|
mock_resp.raise_for_status = MagicMock() |
|
231
|
|
|
232
|
with patch("requests.post", return_value=mock_resp): |
|
233
|
result = manager.authenticate() |
|
234
|
|
|
235
|
assert result.success |
|
236
|
assert result.access_token == "new-tok" |
|
237
|
assert result.method == "saved_token" |
|
238
|
|
|
239
|
def test_oauth_prefers_saved_over_api_key(self, tmp_path): |
|
240
|
"""Saved token should be tried before API key fallback.""" |
|
241
|
token_file = tmp_path / "token.json" |
|
242
|
token_data = { |
|
243
|
"access_token": "saved-tok", |
|
244
|
"expires_at": time.time() + 3600, |
|
245
|
} |
|
246
|
token_file.write_text(json.dumps(token_data)) |
|
247
|
|
|
248
|
config = AuthConfig( |
|
249
|
service="test", |
|
250
|
api_key_env="TEST_KEY", |
|
251
|
token_path=token_file, |
|
252
|
) |
|
253
|
manager = OAuthManager(config) |
|
254
|
with patch.dict(os.environ, {"TEST_KEY": "api-key"}): |
|
255
|
result = manager.authenticate() |
|
256
|
|
|
257
|
assert result.access_token == "saved-tok" |
|
258
|
assert result.method == "saved_token" |
|
259
|
|
|
260
|
|
|
261
|
# ----------------------------------------------------------------------- |
|
262
|
# Known configs and helpers |
|
263
|
# ----------------------------------------------------------------------- |
|
264
|
|
|
265
|
|
|
266
|
class TestKnownConfigs: |
|
267
|
def test_zoom_config(self): |
|
268
|
config = KNOWN_CONFIGS["zoom"] |
|
269
|
assert config.service == "zoom" |
|
270
|
assert config.supports_oauth |
|
271
|
assert config.client_id_env == "ZOOM_CLIENT_ID" |
|
272
|
|
|
273
|
def test_notion_config(self): |
|
274
|
config = KNOWN_CONFIGS["notion"] |
|
275
|
assert config.api_key_env == "NOTION_API_KEY" |
|
276
|
assert config.supports_oauth |
|
277
|
|
|
278
|
def test_github_config(self): |
|
279
|
config = KNOWN_CONFIGS["github"] |
|
280
|
assert config.api_key_env == "GITHUB_TOKEN" |
|
281
|
assert "repo" in config.scopes |
|
282
|
|
|
283
|
def test_dropbox_config(self): |
|
284
|
config = KNOWN_CONFIGS["dropbox"] |
|
285
|
assert config.api_key_env == "DROPBOX_ACCESS_TOKEN" |
|
286
|
|
|
287
|
def test_google_config(self): |
|
288
|
config = KNOWN_CONFIGS["google"] |
|
289
|
assert config.supports_oauth |
|
290
|
|
|
291
|
def test_microsoft_config(self): |
|
292
|
config = KNOWN_CONFIGS["microsoft"] |
|
293
|
assert config.supports_oauth |
|
294
|
|
|
295
|
def test_all_configs_have_service(self): |
|
296
|
for name, config in KNOWN_CONFIGS.items(): |
|
297
|
assert config.service == name |
|
298
|
|
|
299
|
def test_get_auth_config(self): |
|
300
|
assert get_auth_config("zoom") is not None |
|
301
|
assert get_auth_config("nonexistent") is None |
|
302
|
|
|
303
|
def test_get_auth_manager(self): |
|
304
|
mgr = get_auth_manager("zoom") |
|
305
|
assert mgr is not None |
|
306
|
assert isinstance(mgr, OAuthManager) |
|
307
|
|
|
308
|
def test_get_auth_manager_unknown(self): |
|
309
|
assert get_auth_manager("nonexistent") is None |
|
310
|
|