[CLI] Add gcloud utility module

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
+```

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.

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.

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()

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()