Add init command and config util (#5004)

This PR introduces the `casp init` command, which performs the essential
first-time setup for the CLI tool.

Key Changes:

* Environment Validation: The command checks for a valid Docker setup
and `gcloud application-default` credentials, exiting with a clear error
message if either is missing.
* Configuration: It automatically creates a configuration file at
`~/.casp/config.json`, saving the path to the gcloud credentials and an
optional custom config path provided by the user.
* Dependency Fetching: It pulls the required Docker image needed for
subsequent operations.
   * Unit Testing.

Author: PauloVLB

cli/casp/src/casp/commands/init.py

@@ -7,16 +7,115 @@
 #      http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
-# distributed under the License is is "AS IS" BASIS,
+# 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.
 """Init command."""
 
+import os
+import sys
+from typing import Any
+from typing import Dict
+
+from casp.utils import config
+from casp.utils import docker_utils
+from casp.utils import gcloud
 import click
 
 
+def _setup_docker():
+  """Sets up Docker."""
+  click.echo('Checking Docker setup...')
+  if not docker_utils.check_docker_setup():
+    click.secho(
+        'Docker setup check failed. Please resolve the issues above.', fg='red')
+    sys.exit(1)
+  click.secho('Docker setup is correct.', fg='green')
+
+
+def _setup_gcloud_credentials(cfg: Dict[str, Any]):
+  """"Setup gcloud credentials. Prompts the user if not found.
+
+  Args:
+    cfg: The configuration dictionary to update.
+  """
+  click.echo('Checking gcloud authentication...')
+  credentials_path = gcloud.get_credentials_path()
+
+  if not credentials_path:
+    click.secho('gcloud authentication check failed.', fg='red')
+    sys.exit(1)
+
+  click.echo(f'Using credentials found in {credentials_path}')
+  cfg['gcloud_credentials_path'] = credentials_path
+  click.secho('gcloud authentication is configured correctly.', fg='green')
+
+
+def _setup_custom_config(cfg: Dict[str, Any]):
+  """Sets up optional custom configuration directory path.
+
+  Args:
+    cfg: The configuration dictionary to update.
+  """
+  custom_config_path = click.prompt(
+      'Enter path to custom config directory (optional)',
+      default='',
+      show_default=False,
+      type=click.Path())
+
+  if not custom_config_path:
+    # Handle case where user wants to clear the path
+    if 'custom_config_path' in cfg:
+      del cfg['custom_config_path']
+      click.echo('Cleared custom config path.')
+    return
+
+  if not os.path.exists(custom_config_path):
+    click.secho(
+        f'Custom config path "{custom_config_path}" does not exist. '
+        'Skipping.',
+        fg='yellow')
+    return
+
+  cfg['custom_config_path'] = custom_config_path
+  click.secho(f'Custom config path set to: {custom_config_path}', fg='green')
+
+
+def _pull_image():
+  """Pulls the docker image."""
+  click.echo(f'Pulling Docker image: {docker_utils.DOCKER_IMAGE}...')
+  if not docker_utils.pull_image():
+    click.secho(
+        f'\nError: Failed to pull Docker image {docker_utils.DOCKER_IMAGE}.',
+        fg='red')
+    click.secho('Initialization failed.', fg='red')
+    sys.exit(1)
+
+
 @click.command(name='init', help='Initializes the CLI')
 def cli():
-  """Initializes the CLI"""
-  click.echo('To be implemented...')
+  """Initializes the CASP CLI.
+
+  This is done by:
+    1. Checking the Docker setup
+    2. Setting up the gcloud credentials for later use
+    3. Optionally setting up a custom configuration directory path
+    4. Saving the configuration to the config file
+    5. Pulling the Docker image
+  """
+  _setup_docker()
+
+  cfg = config.load_config()
+  if not cfg:
+    click.echo('Config file not found, creating a new one...')
+    cfg = {}
+
+  _setup_gcloud_credentials(cfg)
+  _setup_custom_config(cfg)
+
+  config.save_config(cfg)
+  click.secho(f'Configuration saved to {config.CONFIG_FILE}.', fg='green')
+
+  _pull_image()
+  click.secho('Initialization complete.', fg='green')

cli/casp/src/casp/tests/commands/__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/commands/test_init.py

@@ -0,0 +1,182 @@
+# 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 init command.
+
+   For running all the tests, use (from the root of the project):
+   python -m unittest discover -s cli/casp/src/casp/tests -p test_init.py -v
+"""
+
+import os
+import unittest
+from unittest.mock import patch
+
+from casp.commands import init
+from click.testing import CliRunner
+
+
+class InitCliTest(unittest.TestCase):
+  """Tests for the init command."""
+
+  def setUp(self):
+    super().setUp()
+    self.runner = CliRunner()
+
+    # Patch dependencies
+    self.mock_docker_utils = self.enterContext(
+        patch.object(init, 'docker_utils', autospec=True))
+    self.mock_gcloud = self.enterContext(
+        patch.object(init, 'gcloud', autospec=True))
+    self.mock_config = self.enterContext(
+        patch.object(init, 'config', autospec=True))
+    self.mock_os_path_exists = self.enterContext(
+        patch.object(os.path, 'exists', autospec=True))
+
+    # Default mock behaviors for success paths
+    self.mock_docker_utils.check_docker_setup.return_value = True
+    self.mock_docker_utils.pull_image.return_value = True
+    self.mock_docker_utils.DOCKER_IMAGE = 'gcr.io/casp/runner:latest'
+    credentials_path = '/fake/path/credentials.json'
+    self.mock_gcloud.get_credentials_path.return_value = credentials_path
+    self.mock_config.load_config.return_value = {}
+    self.mock_config.CONFIG_FILE = '~/.casp/config.json'
+
+  def test_init_success_all_steps(self):
+    """Tests successful initialization through all steps, no custom config."""
+    result = self.runner.invoke(
+        init.cli, input='\n')  # Enter for optional prompt
+
+    self.assertEqual(0, result.exit_code)
+    self.assertIn('Docker setup is correct.', result.output)
+    self.assertIn('gcloud authentication is configured correctly.',
+                  result.output)
+    self.assertIn(
+        f'Pulling Docker image: {self.mock_docker_utils.DOCKER_IMAGE}',
+        result.output)
+    self.assertIn('Initialization complete.', result.output)
+
+    self.mock_docker_utils.check_docker_setup.assert_called_once()
+    self.mock_gcloud.get_credentials_path.assert_called_once()
+    expected_config = {'gcloud_credentials_path': '/fake/path/credentials.json'}
+    self.mock_config.save_config.assert_called_once_with(expected_config)
+    self.mock_docker_utils.pull_image.assert_called_once()
+
+  def test_init_docker_setup_fails(self):
+    """Tests when Docker setup check fails."""
+    self.mock_docker_utils.check_docker_setup.return_value = False
+    result = self.runner.invoke(init.cli)
+
+    self.assertNotEqual(0, result.exit_code)  # Should indicate failure
+    self.assertIn('Docker setup check failed.', result.output)
+    self.assertNotIn('Initialization complete.', result.output)
+    self.mock_gcloud.get_credentials_path.assert_not_called()
+
+  def test_init_gcloud_auth_fails(self):
+    """Tests when gcloud authentication fails."""
+    self.mock_gcloud.get_credentials_path.return_value = None
+    result = self.runner.invoke(init.cli)
+
+    self.assertNotEqual(0, result.exit_code)
+    self.assertIn('gcloud authentication check failed.', result.output)
+    self.assertNotIn('Initialization complete.', result.output)
+    self.mock_config.save_config.assert_not_called()
+    self.mock_docker_utils.pull_image.assert_not_called()
+
+  def test_init_docker_pull_fails(self):
+    """Tests when Docker image pull fails."""
+    self.mock_docker_utils.pull_image.return_value = False
+    result = self.runner.invoke(init.cli, input='\n')
+
+    self.assertNotEqual(0, result.exit_code)
+    self.assertIn('Error: Failed to pull Docker image', result.output)
+    self.assertIn('Initialization failed.', result.output)
+    self.assertNotIn('Initialization complete.', result.output)
+
+  def test_init_with_existing_config(self):
+    """Tests that existing config is loaded and updated."""
+    self.mock_config.load_config.return_value = {
+        'existing_key': 'existing_value'
+    }
+    result = self.runner.invoke(init.cli, input='\n')
+
+    self.assertEqual(0, result.exit_code)
+    expected_config = {
+        'existing_key': 'existing_value',
+        'gcloud_credentials_path': '/fake/path/credentials.json'
+    }
+    self.mock_config.save_config.assert_called_once_with(expected_config)
+
+  def test_init_custom_config_path_success(self):
+    """Tests providing a valid custom config path."""
+    custom_path = '/my/custom/config/dir'
+    self.mock_os_path_exists.return_value = True  # Path exists
+    self.mock_config.load_config.return_value = {}
+
+    result = self.runner.invoke(init.cli, input=f'{custom_path}\n')
+
+    self.assertEqual(0, result.exit_code)
+    self.mock_os_path_exists.assert_any_call(custom_path)
+    expected_config = {
+        'gcloud_credentials_path': '/fake/path/credentials.json',
+        'custom_config_path': custom_path
+    }
+    self.mock_config.save_config.assert_called_once_with(expected_config)
+    self.assertIn(f'Custom config path set to: {custom_path}', result.output)
+    self.assertIn('Initialization complete.', result.output)
+
+  def test_init_custom_config_path_not_exists(self):
+    """Tests providing a custom config path that does not exist."""
+    custom_path = '/non/existent/dir'
+    self.mock_os_path_exists.return_value = False  # Path does not exist
+    self.mock_config.load_config.return_value = {}
+
+    result = self.runner.invoke(init.cli, input=f'{custom_path}\n')
+
+    self.assertEqual(0, result.exit_code)
+    self.mock_os_path_exists.assert_any_call(custom_path)
+    expected_config = {'gcloud_credentials_path': '/fake/path/credentials.json'}
+    self.mock_config.save_config.assert_called_once_with(expected_config)
+    self.assertIn(f'Custom config path "{custom_path}" does not exist.',
+                  result.output)
+    self.assertIn('Skipping.', result.output)
+    self.assertNotIn('Custom config path set to', result.output)
+    self.assertIn('Initialization complete.', result.output)
+
+  def test_init_custom_config_path_empty(self):
+    """Tests providing an empty custom config path (skipping)."""
+    self.mock_config.load_config.return_value = {}
+    result = self.runner.invoke(init.cli, input='\n')  # Just press Enter
+
+    self.assertEqual(0, result.exit_code)
+    expected_config = {'gcloud_credentials_path': '/fake/path/credentials.json'}
+    self.mock_config.save_config.assert_called_once_with(expected_config)
+    self.assertNotIn('Custom config path set to', result.output)
+    self.assertNotIn('Cleared custom config path', result.output)
+    self.assertIn('Initialization complete.', result.output)
+
+  def test_init_custom_config_path_empty_clears_existing(self):
+    """Tests that an empty input for custom 
+    config path clears an existing one."""
+    self.mock_config.load_config.return_value = {
+        'custom_config_path': '/my/old/path'
+    }
+    result = self.runner.invoke(init.cli, input='\n')
+
+    self.assertEqual(0, result.exit_code)
+    expected_config = {'gcloud_credentials_path': '/fake/path/credentials.json'}
+    self.mock_config.save_config.assert_called_once_with(expected_config)
+    self.assertIn('Cleared custom config path.', result.output)
+
+
+if __name__ == '__main__':
+  unittest.main()