diff --git a/cli/casp/README.md b/cli/casp/README.md index cb7637fc23..ae1ee44b05 100644 --- a/cli/casp/README.md +++ b/cli/casp/README.md @@ -63,3 +63,11 @@ For example, to add a `my-command` command, follow these steps: Once you have completed these steps, the new command will be available as `casp my-command`. + +## Running Tests + +To run all unit tests for the `casp` CLI, use the following command from the root of the project: + +```bash +python -m unittest discover -s cli/casp/src/casp/tests -v +``` \ No newline at end of file diff --git a/cli/casp/src/casp/tests/__init__.py b/cli/casp/src/casp/tests/__init__.py new file mode 100644 index 0000000000..c37e93b74b --- /dev/null +++ b/cli/casp/src/casp/tests/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/cli/casp/src/casp/tests/utils/__init__.py b/cli/casp/src/casp/tests/utils/__init__.py new file mode 100644 index 0000000000..c37e93b74b --- /dev/null +++ b/cli/casp/src/casp/tests/utils/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/cli/casp/src/casp/tests/utils/test_gcloud.py b/cli/casp/src/casp/tests/utils/test_gcloud.py new file mode 100644 index 0000000000..6abc36c7cd --- /dev/null +++ b/cli/casp/src/casp/tests/utils/test_gcloud.py @@ -0,0 +1,233 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests for the gcloud utility functions. + + For running, use (from the root of the project): + python -m unittest discover -s cli/casp/src/casp/tests -p test_gcloud.py -v +""" + +import subprocess +import unittest +from unittest.mock import Mock +from unittest.mock import patch + +from casp.utils import gcloud + + +class IsValidCredentialsTest(unittest.TestCase): + """Tests for _is_valid_credentials.""" + + @patch('os.path.exists', return_value=True, autospec=True) + @patch( + 'google.oauth2.credentials.Credentials.from_authorized_user_file', + autospec=True) + def test_valid_credentials(self, mock_from_file, mock_exists): + """Tests with a valid credentials file.""" + mock_from_file.return_value = Mock() + self.assertTrue(gcloud._is_valid_credentials('valid/path')) # pylint: disable=protected-access + mock_exists.assert_called_once_with('valid/path') + mock_from_file.assert_called_once_with('valid/path') + + @patch('os.path.exists', return_value=False, autospec=True) + def test_path_does_not_exist(self, mock_exists): + """Tests with a non-existent path.""" + self.assertFalse(gcloud._is_valid_credentials('invalid/path')) # pylint: disable=protected-access + mock_exists.assert_called_once_with('invalid/path') + + @patch('os.path.exists', return_value=True, autospec=True) + @patch( + 'google.oauth2.credentials.Credentials.from_authorized_user_file', + autospec=True) + def test_auth_error(self, mock_from_file, mock_exists): + """Tests with an auth exception.""" + mock_from_file.side_effect = ValueError + self.assertFalse(gcloud._is_valid_credentials('path')) # pylint: disable=protected-access + mock_exists.assert_called_once_with('path') + mock_from_file.assert_called_once_with('path') + + def test_empty_path(self): + """Tests with an empty path string.""" + self.assertFalse(gcloud._is_valid_credentials('')) # pylint: disable=protected-access + + def test_none_path(self): + """Tests with a None path.""" + self.assertFalse(gcloud._is_valid_credentials(None)) # pylint: disable=protected-access + + +class RunGcloudLoginTest(unittest.TestCase): + """Tests for _run_gcloud_login.""" + + @patch( + 'casp.utils.gcloud._is_valid_credentials', + return_value=True, + autospec=True) + @patch('subprocess.run', autospec=True) + def test_login_success(self, mock_run, mock_is_valid): + """Tests successful gcloud login.""" + self.assertTrue(gcloud._run_gcloud_login()) # pylint: disable=protected-access + mock_run.assert_called_once_with( + ['gcloud', 'auth', 'application-default', 'login'], check=True) + mock_is_valid.assert_called_once_with( + gcloud.DEFAULT_GCLOUD_CREDENTIALS_PATH) + + @patch('subprocess.run', autospec=True) + @patch('click.secho', autospec=True) + def test_gcloud_not_found(self, mock_secho, mock_run): + """Tests with gcloud command not found.""" + mock_run.side_effect = FileNotFoundError + self.assertFalse(gcloud._run_gcloud_login()) # pylint: disable=protected-access + mock_secho.assert_called_once() + args, _ = mock_secho.call_args + self.assertIn('gcloud command not found', args[0]) + + @patch('subprocess.run', autospec=True) + @patch('click.secho', autospec=True) + def test_login_failed(self, mock_secho, mock_run): + """Tests with a failed login command.""" + mock_run.side_effect = subprocess.CalledProcessError(1, 'cmd') + self.assertFalse(gcloud._run_gcloud_login()) # pylint: disable=protected-access + mock_secho.assert_called_once() + args, _ = mock_secho.call_args + self.assertIn('gcloud login failed', args[0]) + + +class PromptForCustomPathTest(unittest.TestCase): + """Tests for _prompt_for_custom_path.""" + + @patch('click.prompt', autospec=True) + @patch( + 'casp.utils.gcloud._is_valid_credentials', + return_value=True, + autospec=True) + def test_valid_path(self, mock_is_valid, mock_prompt): + """Tests with a valid custom path.""" + mock_prompt.return_value = '/valid/path' + self.assertEqual(gcloud._prompt_for_custom_path(), '/valid/path') # pylint: disable=protected-access + mock_is_valid.assert_called_once_with('/valid/path') + + @patch('click.prompt', autospec=True) + @patch( + 'casp.utils.gcloud._is_valid_credentials', + return_value=False, + autospec=True) + @patch('click.secho', autospec=True) + def test_invalid_path(self, mock_secho, mock_is_valid, mock_prompt): + """Tests with an invalid custom path.""" + mock_prompt.return_value = '/invalid/path' + self.assertIsNone(gcloud._prompt_for_custom_path()) # pylint: disable=protected-access + mock_is_valid.assert_called_once_with('/invalid/path') + mock_secho.assert_called_once_with( + 'Error: The provided credentials file is not valid.', fg='red') + + @patch('click.prompt', autospec=True) + def test_empty_path(self, mock_prompt): + """Tests with empty input from prompt.""" + mock_prompt.return_value = '' + self.assertIsNone(gcloud._prompt_for_custom_path()) # pylint: disable=protected-access + + +class GetCredentialsPathTest(unittest.TestCase): + """Tests for get_credentials_path.""" + + @patch( + 'casp.utils.gcloud._is_valid_credentials', + return_value=True, + autospec=True) + def test_default_path_valid(self, mock_is_valid): + """Tests when the default credentials path is valid.""" + self.assertEqual(gcloud.get_credentials_path(), + gcloud.DEFAULT_GCLOUD_CREDENTIALS_PATH) + mock_is_valid.assert_called_once_with( + gcloud.DEFAULT_GCLOUD_CREDENTIALS_PATH) + + @patch('casp.utils.gcloud._prompt_for_custom_path', autospec=True) + @patch( + 'casp.utils.gcloud._run_gcloud_login', return_value=True, autospec=True) + @patch('click.confirm', return_value=True, autospec=True) + @patch( + 'casp.utils.gcloud._is_valid_credentials', + return_value=False, + autospec=True) + def test_login_success(self, mock_is_valid, mock_confirm, mock_login, + mock_prompt): + """Tests successful login after default path fails.""" + self.assertEqual(gcloud.get_credentials_path(), + gcloud.DEFAULT_GCLOUD_CREDENTIALS_PATH) + mock_is_valid.assert_called_once_with( + gcloud.DEFAULT_GCLOUD_CREDENTIALS_PATH) + mock_confirm.assert_called_once() + mock_login.assert_called_once() + mock_prompt.assert_not_called() + + @patch( + 'casp.utils.gcloud._prompt_for_custom_path', + return_value='/custom/path', + autospec=True) + @patch( + 'casp.utils.gcloud._run_gcloud_login', return_value=False, autospec=True) + @patch('click.confirm', return_value=True, autospec=True) + @patch( + 'casp.utils.gcloud._is_valid_credentials', + return_value=False, + autospec=True) + def test_login_fail_then_custom_path(self, mock_is_valid, mock_confirm, + mock_login, mock_prompt): + """Tests providing a custom path after a failed login.""" + self.assertEqual(gcloud.get_credentials_path(), '/custom/path') + mock_is_valid.assert_called_once_with( + gcloud.DEFAULT_GCLOUD_CREDENTIALS_PATH) + mock_confirm.assert_called_once() + mock_login.assert_called_once() + mock_prompt.assert_called_once() + + @patch( + 'casp.utils.gcloud._prompt_for_custom_path', + return_value='/custom/path', + autospec=True) + @patch('casp.utils.gcloud._run_gcloud_login', autospec=True) + @patch('click.confirm', return_value=False, autospec=True) + @patch( + 'casp.utils.gcloud._is_valid_credentials', + return_value=False, + autospec=True) + def test_decline_login_then_custom_path(self, mock_is_valid, mock_confirm, + mock_login, mock_prompt): + """Tests providing a custom path after declining to log in.""" + self.assertEqual(gcloud.get_credentials_path(), '/custom/path') + mock_is_valid.assert_called_once_with( + gcloud.DEFAULT_GCLOUD_CREDENTIALS_PATH) + mock_confirm.assert_called_once() + mock_login.assert_not_called() + mock_prompt.assert_called_once() + + @patch( + 'casp.utils.gcloud._prompt_for_custom_path', + return_value=None, + autospec=True) + @patch('click.confirm', return_value=False, autospec=True) + @patch( + 'casp.utils.gcloud._is_valid_credentials', + return_value=False, + autospec=True) + def test_all_fail(self, mock_is_valid, mock_confirm, mock_prompt): + """Tests when all methods to get credentials fail.""" + self.assertIsNone(gcloud.get_credentials_path()) + mock_is_valid.assert_called_once_with( + gcloud.DEFAULT_GCLOUD_CREDENTIALS_PATH) + mock_confirm.assert_called_once() + mock_prompt.assert_called_once() + + +if __name__ == '__main__': + unittest.main() diff --git a/cli/casp/src/casp/utils/gcloud.py b/cli/casp/src/casp/utils/gcloud.py new file mode 100644 index 0000000000..04b5327e34 --- /dev/null +++ b/cli/casp/src/casp/utils/gcloud.py @@ -0,0 +1,105 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""gcloud utility functions.""" + +import os +import subprocess + +import click +from google.oauth2 import credentials + +DEFAULT_GCLOUD_CREDENTIALS_PATH = os.path.expanduser( + '~/.config/gcloud/application_default_credentials.json') + + +def _is_valid_credentials(path: str) -> bool: + """Returns True if the path points to a valid credentials file.""" + if not path or not os.path.exists(path): + click.secho('Error: No valid credentials file found.', fg='red') + return False + try: + credentials.Credentials.from_authorized_user_file(path) + return True + except ValueError as e: + click.secho(f'Error when checking for valid credentials: {e}', fg='red') + return False + + +def _run_gcloud_login() -> bool: + """ + Runs the gcloud login command and returns True on success. + """ + try: + subprocess.run( + ['gcloud', 'auth', 'application-default', 'login'], check=True) + # After login, re-validate the default file. + return _is_valid_credentials(DEFAULT_GCLOUD_CREDENTIALS_PATH) + except FileNotFoundError: + click.secho( + 'Error: gcloud command not found. Please ensure it is installed and ' + 'in your PATH. ' + 'Or you can mannually run ' + '`gcloud auth application-default login`', + fg='red') + return False + except subprocess.CalledProcessError: + click.secho( + 'Error: gcloud login failed. ' + 'You can mannually run ' + '`gcloud auth application-default login`', + fg='red') + return False + + +def _prompt_for_custom_path() -> str | None: + """ + Prompts the user for a custom credentials path and returns it if valid. + """ + path = click.prompt( + 'Enter path to your credentials file (or press Ctrl+C to cancel)', + default='', + show_default=False, + type=click.Path(exists=True, dir_okay=False, resolve_path=True)) + + if not path: + return None + + if _is_valid_credentials(path): + return path + + click.secho('Error: The provided credentials file is not valid.', fg='red') + return None + + +def get_credentials_path() -> str | None: + """ + Finds a valid gcloud credentials path, prompting the user if needed. + + Returns: + The path to a valid credentials file, or None if one cannot be found. + """ + if _is_valid_credentials(DEFAULT_GCLOUD_CREDENTIALS_PATH): + return DEFAULT_GCLOUD_CREDENTIALS_PATH + + click.secho( + 'Default gcloud credentials not found or are invalid.', fg='yellow') + + if click.confirm('Do you want to log in with gcloud now?'): + if _run_gcloud_login(): + return DEFAULT_GCLOUD_CREDENTIALS_PATH + + click.secho( + '\nLogin was skipped or failed. You can provide a direct path instead.', + fg='yellow') + return _prompt_for_custom_path()