PlanOpticon

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

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button