|
| 1 | +# SPDX-License-Identifier: BSD-2-Clause |
| 2 | + |
| 3 | +import unittest |
| 4 | +import json |
| 5 | +import tempfile |
| 6 | +from pathlib import Path |
| 7 | +from unittest import mock |
| 8 | + |
| 9 | +from chipflow.auth import ( |
| 10 | + get_api_key, |
| 11 | + authenticate_with_github_token, |
| 12 | + authenticate_with_device_flow, |
| 13 | + is_gh_authenticated, |
| 14 | + get_gh_token, |
| 15 | + save_api_key, |
| 16 | + load_saved_api_key, |
| 17 | + logout, |
| 18 | + AuthenticationError, |
| 19 | +) |
| 20 | + |
| 21 | + |
| 22 | +class TestAuthHelpers(unittest.TestCase): |
| 23 | + """Test helper functions in auth module""" |
| 24 | + |
| 25 | + @mock.patch('chipflow.auth.subprocess.run') |
| 26 | + def test_is_gh_authenticated_success(self, mock_run): |
| 27 | + """Test checking if gh is authenticated - success case""" |
| 28 | + mock_run.return_value.returncode = 0 |
| 29 | + self.assertTrue(is_gh_authenticated()) |
| 30 | + mock_run.assert_called_once() |
| 31 | + |
| 32 | + @mock.patch('chipflow.auth.subprocess.run') |
| 33 | + def test_is_gh_authenticated_not_authenticated(self, mock_run): |
| 34 | + """Test checking if gh is authenticated - not authenticated""" |
| 35 | + mock_run.return_value.returncode = 1 |
| 36 | + self.assertFalse(is_gh_authenticated()) |
| 37 | + |
| 38 | + @mock.patch('chipflow.auth.subprocess.run') |
| 39 | + def test_is_gh_authenticated_not_installed(self, mock_run): |
| 40 | + """Test checking if gh is authenticated - not installed""" |
| 41 | + mock_run.side_effect = FileNotFoundError() |
| 42 | + self.assertFalse(is_gh_authenticated()) |
| 43 | + |
| 44 | + @mock.patch('chipflow.auth.subprocess.run') |
| 45 | + def test_get_gh_token_success(self, mock_run): |
| 46 | + """Test getting GitHub token - success""" |
| 47 | + mock_run.return_value.stdout = "ghp_test123\n" |
| 48 | + token = get_gh_token() |
| 49 | + self.assertEqual(token, "ghp_test123") |
| 50 | + |
| 51 | + @mock.patch('chipflow.auth.subprocess.run') |
| 52 | + def test_get_gh_token_failure(self, mock_run): |
| 53 | + """Test getting GitHub token - failure""" |
| 54 | + mock_run.side_effect = FileNotFoundError() |
| 55 | + token = get_gh_token() |
| 56 | + self.assertIsNone(token) |
| 57 | + |
| 58 | + def test_save_and_load_api_key(self): |
| 59 | + """Test saving and loading API key""" |
| 60 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 61 | + # Mock the credentials file path |
| 62 | + with mock.patch('chipflow.auth.get_credentials_file') as mock_creds_file: |
| 63 | + creds_file = Path(tmpdir) / "credentials" |
| 64 | + mock_creds_file.return_value = creds_file |
| 65 | + |
| 66 | + # Save API key |
| 67 | + test_key = "cf_test_12345" |
| 68 | + save_api_key(test_key) |
| 69 | + |
| 70 | + # Verify file exists and has correct permissions |
| 71 | + self.assertTrue(creds_file.exists()) |
| 72 | + # Note: File permissions check skipped as it's platform-specific |
| 73 | + |
| 74 | + # Load API key |
| 75 | + loaded_key = load_saved_api_key() |
| 76 | + self.assertEqual(loaded_key, test_key) |
| 77 | + |
| 78 | + def test_load_api_key_no_file(self): |
| 79 | + """Test loading API key when file doesn't exist""" |
| 80 | + with mock.patch('chipflow.auth.get_credentials_file') as mock_creds_file: |
| 81 | + mock_creds_file.return_value = Path("/nonexistent/credentials") |
| 82 | + key = load_saved_api_key() |
| 83 | + self.assertIsNone(key) |
| 84 | + |
| 85 | + def test_logout(self): |
| 86 | + """Test logout removes credentials file""" |
| 87 | + with tempfile.TemporaryDirectory() as tmpdir: |
| 88 | + with mock.patch('chipflow.auth.get_credentials_file') as mock_creds_file: |
| 89 | + creds_file = Path(tmpdir) / "credentials" |
| 90 | + mock_creds_file.return_value = creds_file |
| 91 | + |
| 92 | + # Create a credentials file |
| 93 | + creds_file.write_text(json.dumps({"api_key": "test"})) |
| 94 | + self.assertTrue(creds_file.exists()) |
| 95 | + |
| 96 | + # Logout |
| 97 | + with mock.patch('builtins.print'): |
| 98 | + logout() |
| 99 | + |
| 100 | + # File should be deleted |
| 101 | + self.assertFalse(creds_file.exists()) |
| 102 | + |
| 103 | + |
| 104 | +class TestGitHubTokenAuth(unittest.TestCase): |
| 105 | + """Test GitHub token authentication""" |
| 106 | + |
| 107 | + @mock.patch('chipflow.auth.save_api_key') |
| 108 | + @mock.patch('chipflow.auth.requests.post') |
| 109 | + @mock.patch('chipflow.auth.get_gh_token') |
| 110 | + @mock.patch('chipflow.auth.is_gh_authenticated') |
| 111 | + @mock.patch('builtins.print') |
| 112 | + def test_github_token_auth_success( |
| 113 | + self, mock_print, mock_is_gh, mock_get_token, mock_post, mock_save |
| 114 | + ): |
| 115 | + """Test successful GitHub token authentication""" |
| 116 | + mock_is_gh.return_value = True |
| 117 | + mock_get_token.return_value = "ghp_test123" |
| 118 | + mock_post.return_value.status_code = 200 |
| 119 | + mock_post.return_value.json.return_value = {"api_key": "cf_test_key"} |
| 120 | + |
| 121 | + api_key = authenticate_with_github_token("https://test.api", interactive=True) |
| 122 | + |
| 123 | + self.assertEqual(api_key, "cf_test_key") |
| 124 | + mock_save.assert_called_once_with("cf_test_key") |
| 125 | + mock_post.assert_called_once() |
| 126 | + |
| 127 | + @mock.patch('chipflow.auth.is_gh_authenticated') |
| 128 | + @mock.patch('builtins.print') |
| 129 | + def test_github_token_auth_not_authenticated(self, mock_print, mock_is_gh): |
| 130 | + """Test GitHub token auth when gh not authenticated""" |
| 131 | + mock_is_gh.return_value = False |
| 132 | + |
| 133 | + api_key = authenticate_with_github_token("https://test.api", interactive=True) |
| 134 | + |
| 135 | + self.assertIsNone(api_key) |
| 136 | + |
| 137 | + @mock.patch('chipflow.auth.requests.post') |
| 138 | + @mock.patch('chipflow.auth.get_gh_token') |
| 139 | + @mock.patch('chipflow.auth.is_gh_authenticated') |
| 140 | + @mock.patch('builtins.print') |
| 141 | + def test_github_token_auth_invalid_token( |
| 142 | + self, mock_print, mock_is_gh, mock_get_token, mock_post |
| 143 | + ): |
| 144 | + """Test GitHub token auth with invalid token""" |
| 145 | + mock_is_gh.return_value = True |
| 146 | + mock_get_token.return_value = "invalid_token" |
| 147 | + mock_post.return_value.status_code = 401 |
| 148 | + mock_post.return_value.json.return_value = { |
| 149 | + "error": "invalid_token", |
| 150 | + "error_description": "Invalid GitHub token" |
| 151 | + } |
| 152 | + |
| 153 | + api_key = authenticate_with_github_token("https://test.api", interactive=True) |
| 154 | + |
| 155 | + self.assertIsNone(api_key) |
| 156 | + |
| 157 | + @mock.patch('chipflow.auth.requests.post') |
| 158 | + @mock.patch('chipflow.auth.get_gh_token') |
| 159 | + @mock.patch('chipflow.auth.is_gh_authenticated') |
| 160 | + @mock.patch('builtins.print') |
| 161 | + def test_github_token_auth_network_error( |
| 162 | + self, mock_print, mock_is_gh, mock_get_token, mock_post |
| 163 | + ): |
| 164 | + """Test GitHub token auth with network error""" |
| 165 | + import requests |
| 166 | + mock_is_gh.return_value = True |
| 167 | + mock_get_token.return_value = "ghp_test123" |
| 168 | + mock_post.side_effect = requests.exceptions.ConnectionError("Network error") |
| 169 | + |
| 170 | + api_key = authenticate_with_github_token("https://test.api", interactive=True) |
| 171 | + |
| 172 | + self.assertIsNone(api_key) |
| 173 | + |
| 174 | + |
| 175 | +class TestDeviceFlowAuth(unittest.TestCase): |
| 176 | + """Test device flow authentication""" |
| 177 | + |
| 178 | + @mock.patch('chipflow.auth.save_api_key') |
| 179 | + @mock.patch('chipflow.auth.time.sleep') |
| 180 | + @mock.patch('chipflow.auth.requests.post') |
| 181 | + @mock.patch('builtins.print') |
| 182 | + def test_device_flow_success(self, mock_print, mock_post, mock_sleep, mock_save): |
| 183 | + """Test successful device flow authentication""" |
| 184 | + # Mock init response |
| 185 | + init_response = mock.Mock() |
| 186 | + init_response.status_code = 200 |
| 187 | + init_response.json.return_value = { |
| 188 | + "device_code": "device123", |
| 189 | + "user_code": "ABCD-1234", |
| 190 | + "verification_uri": "https://test.api/auth", |
| 191 | + "interval": 1, |
| 192 | + "expires_in": 60 |
| 193 | + } |
| 194 | + |
| 195 | + # Mock poll response - success on first try |
| 196 | + poll_response = mock.Mock() |
| 197 | + poll_response.status_code = 200 |
| 198 | + poll_response.json.return_value = {"api_key": "cf_test_key"} |
| 199 | + |
| 200 | + mock_post.side_effect = [init_response, poll_response] |
| 201 | + |
| 202 | + api_key = authenticate_with_device_flow("https://test.api", interactive=True) |
| 203 | + |
| 204 | + self.assertEqual(api_key, "cf_test_key") |
| 205 | + mock_save.assert_called_once_with("cf_test_key") |
| 206 | + |
| 207 | + @mock.patch('chipflow.auth.time.sleep') |
| 208 | + @mock.patch('chipflow.auth.requests.post') |
| 209 | + @mock.patch('builtins.print') |
| 210 | + def test_device_flow_pending_then_success(self, mock_print, mock_post, mock_sleep): |
| 211 | + """Test device flow with pending state then success""" |
| 212 | + # Mock init response |
| 213 | + init_response = mock.Mock() |
| 214 | + init_response.status_code = 200 |
| 215 | + init_response.json.return_value = { |
| 216 | + "device_code": "device123", |
| 217 | + "user_code": "ABCD-1234", |
| 218 | + "verification_uri": "https://test.api/auth", |
| 219 | + "interval": 1, |
| 220 | + "expires_in": 60 |
| 221 | + } |
| 222 | + |
| 223 | + # Mock poll responses - pending, then success |
| 224 | + pending_response = mock.Mock() |
| 225 | + pending_response.status_code = 202 |
| 226 | + pending_response.json.return_value = { |
| 227 | + "error": "authorization_pending" |
| 228 | + } |
| 229 | + |
| 230 | + success_response = mock.Mock() |
| 231 | + success_response.status_code = 200 |
| 232 | + success_response.json.return_value = {"api_key": "cf_test_key"} |
| 233 | + |
| 234 | + mock_post.side_effect = [init_response, pending_response, success_response] |
| 235 | + |
| 236 | + with mock.patch('chipflow.auth.save_api_key'): |
| 237 | + api_key = authenticate_with_device_flow("https://test.api", interactive=True) |
| 238 | + |
| 239 | + self.assertEqual(api_key, "cf_test_key") |
| 240 | + |
| 241 | + @mock.patch('chipflow.auth.time.sleep') |
| 242 | + @mock.patch('chipflow.auth.requests.post') |
| 243 | + @mock.patch('builtins.print') |
| 244 | + def test_device_flow_timeout(self, mock_print, mock_post, mock_sleep): |
| 245 | + """Test device flow timeout""" |
| 246 | + # Mock init response |
| 247 | + init_response = mock.Mock() |
| 248 | + init_response.status_code = 200 |
| 249 | + init_response.json.return_value = { |
| 250 | + "device_code": "device123", |
| 251 | + "user_code": "ABCD-1234", |
| 252 | + "verification_uri": "https://test.api/auth", |
| 253 | + "interval": 1, |
| 254 | + "expires_in": 2 # Very short timeout |
| 255 | + } |
| 256 | + |
| 257 | + # Mock poll response - always pending |
| 258 | + pending_response = mock.Mock() |
| 259 | + pending_response.status_code = 202 |
| 260 | + pending_response.json.return_value = { |
| 261 | + "error": "authorization_pending" |
| 262 | + } |
| 263 | + |
| 264 | + mock_post.side_effect = [init_response, pending_response, pending_response, pending_response] |
| 265 | + |
| 266 | + with self.assertRaises(AuthenticationError) as ctx: |
| 267 | + authenticate_with_device_flow("https://test.api", interactive=True) |
| 268 | + |
| 269 | + self.assertIn("timed out", str(ctx.exception)) |
| 270 | + |
| 271 | + |
| 272 | +class TestGetAPIKey(unittest.TestCase): |
| 273 | + """Test the main get_api_key function with fallback logic""" |
| 274 | + |
| 275 | + def test_get_api_key_from_env_var(self): |
| 276 | + """Test getting API key from environment variable""" |
| 277 | + with mock.patch.dict('os.environ', {'CHIPFLOW_API_KEY': 'env_key'}): |
| 278 | + api_key = get_api_key(interactive=False) |
| 279 | + self.assertEqual(api_key, 'env_key') |
| 280 | + |
| 281 | + @mock.patch('chipflow.auth.load_saved_api_key') |
| 282 | + def test_get_api_key_from_saved_credentials(self, mock_load): |
| 283 | + """Test getting API key from saved credentials""" |
| 284 | + mock_load.return_value = "saved_key" |
| 285 | + with mock.patch.dict('os.environ', {}, clear=True): |
| 286 | + api_key = get_api_key(interactive=False) |
| 287 | + self.assertEqual(api_key, "saved_key") |
| 288 | + |
| 289 | + @mock.patch('chipflow.auth.authenticate_with_github_token') |
| 290 | + @mock.patch('chipflow.auth.load_saved_api_key') |
| 291 | + def test_get_api_key_gh_token_fallback(self, mock_load, mock_gh_auth): |
| 292 | + """Test fallback to GitHub token authentication""" |
| 293 | + mock_load.return_value = None |
| 294 | + mock_gh_auth.return_value = "gh_key" |
| 295 | + |
| 296 | + with mock.patch.dict('os.environ', {}, clear=True): |
| 297 | + api_key = get_api_key(interactive=True) |
| 298 | + self.assertEqual(api_key, "gh_key") |
| 299 | + |
| 300 | + @mock.patch('chipflow.auth.authenticate_with_device_flow') |
| 301 | + @mock.patch('chipflow.auth.authenticate_with_github_token') |
| 302 | + @mock.patch('chipflow.auth.load_saved_api_key') |
| 303 | + def test_get_api_key_device_flow_fallback( |
| 304 | + self, mock_load, mock_gh_auth, mock_device_flow |
| 305 | + ): |
| 306 | + """Test fallback to device flow authentication""" |
| 307 | + mock_load.return_value = None |
| 308 | + mock_gh_auth.return_value = None |
| 309 | + mock_device_flow.return_value = "device_key" |
| 310 | + |
| 311 | + with mock.patch.dict('os.environ', {}, clear=True): |
| 312 | + with mock.patch('builtins.print'): |
| 313 | + api_key = get_api_key(interactive=True) |
| 314 | + self.assertEqual(api_key, "device_key") |
| 315 | + |
| 316 | + @mock.patch('chipflow.auth.authenticate_with_device_flow') |
| 317 | + @mock.patch('chipflow.auth.authenticate_with_github_token') |
| 318 | + @mock.patch('chipflow.auth.load_saved_api_key') |
| 319 | + def test_get_api_key_all_methods_fail( |
| 320 | + self, mock_load, mock_gh_auth, mock_device_flow |
| 321 | + ): |
| 322 | + """Test when all authentication methods fail""" |
| 323 | + mock_load.return_value = None |
| 324 | + mock_gh_auth.return_value = None |
| 325 | + mock_device_flow.side_effect = AuthenticationError("All methods failed") |
| 326 | + |
| 327 | + with mock.patch.dict('os.environ', {}, clear=True): |
| 328 | + with mock.patch('builtins.print'): |
| 329 | + with self.assertRaises(AuthenticationError) as ctx: |
| 330 | + get_api_key(interactive=True) |
| 331 | + self.assertIn("All authentication methods failed", str(ctx.exception)) |
| 332 | + |
| 333 | + @mock.patch('chipflow.auth.load_saved_api_key') |
| 334 | + def test_get_api_key_force_login_ignores_saved(self, mock_load): |
| 335 | + """Test force_login parameter ignores saved credentials""" |
| 336 | + mock_load.return_value = "saved_key" |
| 337 | + |
| 338 | + with mock.patch.dict('os.environ', {}, clear=True): |
| 339 | + with mock.patch('chipflow.auth.authenticate_with_github_token') as mock_gh: |
| 340 | + mock_gh.return_value = "new_key" |
| 341 | + api_key = get_api_key(interactive=True, force_login=True) |
| 342 | + self.assertEqual(api_key, "new_key") |
| 343 | + # Should not have called load_saved_api_key due to force_login |
| 344 | + mock_load.assert_not_called() |
| 345 | + |
| 346 | + |
| 347 | +if __name__ == "__main__": |
| 348 | + unittest.main() |
0 commit comments