From ab122c98d2597bb03ab3bb8bf51f14e492ba153f Mon Sep 17 00:00:00 2001 From: Robert Fujara Date: Tue, 4 Nov 2025 15:51:43 +0100 Subject: [PATCH 1/7] Add reusable credential prompt and validation method MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements DEVP-492: Extract reusable credential prompt and validation method to eliminate code duplication and improve maintainability. Changes: - Add _prompt_and_validate_credentials() async method to ConfigManager - Method prompts for API token with hidden input - Validates credentials via Workato API call - Stores validated token in keyring - Returns ProfileData with workspace information - Comprehensive error handling for empty tokens, API failures, and keyring issues Tests: - Add 4 unit tests covering success, empty token, API failure, and keyring disabled scenarios - All tests pass (79 tests total) - Type checking and linting clean 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- .../cli/utils/config/manager.py | 66 ++++++ tests/unit/config/test_manager.py | 195 ++++++++++++++++++ 2 files changed, 261 insertions(+) diff --git a/src/workato_platform_cli/cli/utils/config/manager.py b/src/workato_platform_cli/cli/utils/config/manager.py index 605c154..7b8d484 100644 --- a/src/workato_platform_cli/cli/utils/config/manager.py +++ b/src/workato_platform_cli/cli/utils/config/manager.py @@ -294,6 +294,72 @@ async def _setup_profile(self) -> str: return profile_name + async def _prompt_and_validate_credentials( + self, profile_name: str, region_info: RegionInfo + ) -> ProfileData: + """Prompt for API credentials and validate them with an API call. + + Args: + profile_name: Name of the profile being configured + region_info: Region information containing the API URL + + Returns: + ProfileData: Validated profile data with workspace information + + Raises: + click.ClickException: If token is empty or validation fails + """ + # Prompt for API token + click.echo("🔐 Enter your API token") + token = await click.prompt("Enter your Workato API token", hide_input=True) + + # Validate token is not empty + if not token.strip(): + raise click.ClickException("API token cannot be empty") + + # Create API configuration with the token and region URL + api_config = Configuration( + access_token=token, host=region_info.url, ssl_ca_cert=certifi.where() + ) + + # Make API call to test authentication and get workspace info + try: + async with Workato(configuration=api_config) as workato_api_client: + user_info = await workato_api_client.users_api.get_workspace_details() + except Exception as e: + raise click.ClickException( + f"Authentication failed: {e}\n" + "Please verify your API token is correct and try again." + ) from e + + # Store validated token in keyring + success = self.profile_manager._store_token_in_keyring(profile_name, token) + if not success: + if self.profile_manager._is_keyring_enabled(): + raise click.ClickException( + "Failed to store token in keyring. " + "Please check your system keyring setup." + ) + # If keyring is disabled, token will need to come from env vars + click.echo( + "⚠️ Keyring is disabled. " + "Set WORKATO_API_TOKEN environment variable to persist token." + ) + + # Create and return ProfileData object with workspace info + if not region_info.url: + raise click.ClickException("Region URL is required") + + profile_data = ProfileData( + region=region_info.region, + region_url=region_info.url, + workspace_id=user_info.id, + ) + + click.echo(f"✅ Authenticated as: {user_info.name}") + + return profile_data + async def _create_new_profile(self, profile_name: str) -> None: """Create a new profile interactively""" # AVAILABLE_REGIONS and RegionInfo already imported at top diff --git a/tests/unit/config/test_manager.py b/tests/unit/config/test_manager.py index b72deb4..23418c2 100644 --- a/tests/unit/config/test_manager.py +++ b/tests/unit/config/test_manager.py @@ -1499,6 +1499,201 @@ async def fake_prompt(message: str, **_: Any) -> str: with pytest.raises(SystemExit): await manager._create_new_profile("dev") + @pytest.mark.asyncio + async def test_prompt_and_validate_credentials_success( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """Test successful credential prompt and validation.""" + from workato_platform_cli.cli.utils.config.models import RegionInfo + + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + + monkeypatch.setattr( + ConfigManager.__module__ + ".Workato", + StubWorkato, + ) + + async def fake_prompt(message: str, **_: Any) -> str: + if "API token" in message: + return "valid-token-123" + return "unused" + + monkeypatch.setattr( + ConfigManager.__module__ + ".click.prompt", + fake_prompt, + ) + + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + region_info = RegionInfo( + region="us", name="US Data Center", url="https://www.workato.com" + ) + profile_data = await manager._prompt_and_validate_credentials( + "test-profile", region_info + ) + + assert profile_data.region == "us" + assert profile_data.region_url == "https://www.workato.com" + assert profile_data.workspace_id == 101 + mock_profile_manager._store_token_in_keyring.assert_called_with( + "test-profile", "valid-token-123" + ) + assert any("Authenticated as" in msg for msg in outputs) + + @pytest.mark.asyncio + async def test_prompt_and_validate_credentials_empty_token( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """Test that empty token raises ClickException.""" + from workato_platform_cli.cli.utils.config.models import RegionInfo + + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + + async def fake_prompt(message: str, **_: Any) -> str: + if "API token" in message: + return " " # Empty/whitespace token + return "unused" + + monkeypatch.setattr( + ConfigManager.__module__ + ".click.prompt", + fake_prompt, + ) + + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + region_info = RegionInfo( + region="us", name="US Data Center", url="https://www.workato.com" + ) + + with pytest.raises(click.ClickException) as excinfo: + await manager._prompt_and_validate_credentials("test-profile", region_info) + + assert "token cannot be empty" in str(excinfo.value) + + @pytest.mark.asyncio + async def test_prompt_and_validate_credentials_api_failure( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """Test that API validation failure raises appropriate exception.""" + from workato_platform_cli.cli.utils.config.models import RegionInfo + + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + + # Create a Workato stub that raises an exception + class FailingWorkato: + def __init__(self, configuration: Configuration) -> None: + self.configuration = configuration + self.users_api = FailingUsersAPI() + + async def __aenter__(self) -> "FailingWorkato": + return self + + async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None: + return None + + class FailingUsersAPI: + async def get_workspace_details(self) -> User: + raise Exception("Authentication failed: Invalid token") + + monkeypatch.setattr( + ConfigManager.__module__ + ".Workato", + FailingWorkato, + ) + + async def fake_prompt(message: str, **_: Any) -> str: + if "API token" in message: + return "invalid-token" + return "unused" + + monkeypatch.setattr( + ConfigManager.__module__ + ".click.prompt", + fake_prompt, + ) + + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + region_info = RegionInfo( + region="us", name="US Data Center", url="https://www.workato.com" + ) + + with pytest.raises(click.ClickException) as excinfo: + await manager._prompt_and_validate_credentials("test-profile", region_info) + + assert "Authentication failed" in str(excinfo.value) + + @pytest.mark.asyncio + async def test_prompt_and_validate_credentials_keyring_disabled( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """Test credential validation with keyring disabled.""" + from workato_platform_cli.cli.utils.config.models import RegionInfo + + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + + # Mock keyring as disabled + mock_profile_manager._store_token_in_keyring.return_value = False + mock_profile_manager._is_keyring_enabled.return_value = False + manager.profile_manager = mock_profile_manager + + monkeypatch.setattr( + ConfigManager.__module__ + ".Workato", + StubWorkato, + ) + + async def fake_prompt(message: str, **_: Any) -> str: + if "API token" in message: + return "valid-token-123" + return "unused" + + monkeypatch.setattr( + ConfigManager.__module__ + ".click.prompt", + fake_prompt, + ) + + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + region_info = RegionInfo( + region="us", name="US Data Center", url="https://www.workato.com" + ) + profile_data = await manager._prompt_and_validate_credentials( + "test-profile", region_info + ) + + assert profile_data.region == "us" + assert any("Keyring is disabled" in msg for msg in outputs) + assert any("WORKATO_API_TOKEN" in msg for msg in outputs) + @pytest.mark.asyncio async def test_setup_profile_existing_create_new_success( self, From fe4da2ff8713e2338400e487008f1764fe0ed217 Mon Sep 17 00:00:00 2001 From: Robert Fujara Date: Tue, 4 Nov 2025 16:04:08 +0100 Subject: [PATCH 2/7] Refactor _create_new_profile() to use _prompt_and_validate_credentials() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Modified _prompt_and_validate_credentials() to return tuple[ProfileData, str] - Removed duplicate credential prompting and token storage logic from _create_new_profile() - Updated _create_new_profile() to use the reusable credential validation method - Updated tests to match new method signature - Token storage now handled consistently by set_profile() method This eliminates code duplication and improves separation of concerns between credential validation and profile storage. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- .../cli/utils/config/manager.py | 47 +++---------------- tests/unit/config/test_manager.py | 20 +++----- 2 files changed, 14 insertions(+), 53 deletions(-) diff --git a/src/workato_platform_cli/cli/utils/config/manager.py b/src/workato_platform_cli/cli/utils/config/manager.py index 7b8d484..a2aea27 100644 --- a/src/workato_platform_cli/cli/utils/config/manager.py +++ b/src/workato_platform_cli/cli/utils/config/manager.py @@ -296,7 +296,7 @@ async def _setup_profile(self) -> str: async def _prompt_and_validate_credentials( self, profile_name: str, region_info: RegionInfo - ) -> ProfileData: + ) -> tuple[ProfileData, str]: """Prompt for API credentials and validate them with an API call. Args: @@ -304,7 +304,7 @@ async def _prompt_and_validate_credentials( region_info: Region information containing the API URL Returns: - ProfileData: Validated profile data with workspace information + tuple[ProfileData, str]: Validated profile data and API token Raises: click.ClickException: If token is empty or validation fails @@ -332,20 +332,6 @@ async def _prompt_and_validate_credentials( "Please verify your API token is correct and try again." ) from e - # Store validated token in keyring - success = self.profile_manager._store_token_in_keyring(profile_name, token) - if not success: - if self.profile_manager._is_keyring_enabled(): - raise click.ClickException( - "Failed to store token in keyring. " - "Please check your system keyring setup." - ) - # If keyring is disabled, token will need to come from env vars - click.echo( - "⚠️ Keyring is disabled. " - "Set WORKATO_API_TOKEN environment variable to persist token." - ) - # Create and return ProfileData object with workspace info if not region_info.url: raise click.ClickException("Region URL is required") @@ -358,7 +344,7 @@ async def _prompt_and_validate_credentials( click.echo(f"✅ Authenticated as: {user_info.name}") - return profile_data + return profile_data, token async def _create_new_profile(self, profile_name: str) -> None: """Create a new profile interactively""" @@ -403,32 +389,13 @@ async def _create_new_profile(self, profile_name: str) -> None: region="custom", name="Custom URL", url=custom_url ) - # Get API token - click.echo("🔐 Enter your API token") - token = await click.prompt("Enter your Workato API token", hide_input=True) - if not token.strip(): - click.echo("❌ No token provided") - sys.exit(1) - - # Test authentication and get workspace info - api_config = Configuration( - access_token=token, host=selected_region.url, ssl_ca_cert=certifi.where() - ) - - async with Workato(configuration=api_config) as workato_api_client: - user_info = await workato_api_client.users_api.get_workspace_details() - - # Create and save profile - if not selected_region.url: - raise click.ClickException("Region URL is required") - profile_data = ProfileData( - region=selected_region.region, - region_url=selected_region.url, - workspace_id=user_info.id, + # Prompt for credentials and validate with API + profile_data, token = await self._prompt_and_validate_credentials( + profile_name, selected_region ) + # Save profile and token self.profile_manager.set_profile(profile_name, profile_data, token) - click.echo(f"✅ Authenticated as: {user_info.name}") async def _setup_project(self, profile_name: str, workspace_root: Path) -> None: """Setup project interactively""" diff --git a/tests/unit/config/test_manager.py b/tests/unit/config/test_manager.py index 23418c2..077b37c 100644 --- a/tests/unit/config/test_manager.py +++ b/tests/unit/config/test_manager.py @@ -1496,7 +1496,7 @@ async def fake_prompt(message: str, **_: Any) -> str: fake_prompt, ) - with pytest.raises(SystemExit): + with pytest.raises(click.ClickException, match="API token cannot be empty"): await manager._create_new_profile("dev") @pytest.mark.asyncio @@ -1536,16 +1536,14 @@ async def fake_prompt(message: str, **_: Any) -> str: region_info = RegionInfo( region="us", name="US Data Center", url="https://www.workato.com" ) - profile_data = await manager._prompt_and_validate_credentials( + profile_data, token = await manager._prompt_and_validate_credentials( "test-profile", region_info ) assert profile_data.region == "us" assert profile_data.region_url == "https://www.workato.com" assert profile_data.workspace_id == 101 - mock_profile_manager._store_token_in_keyring.assert_called_with( - "test-profile", "valid-token-123" - ) + assert token == "valid-token-123" assert any("Authenticated as" in msg for msg in outputs) @pytest.mark.asyncio @@ -1652,14 +1650,10 @@ async def test_prompt_and_validate_credentials_keyring_disabled( monkeypatch: pytest.MonkeyPatch, mock_profile_manager: Mock, ) -> None: - """Test credential validation with keyring disabled.""" + """Test credential validation returns correct data.""" from workato_platform_cli.cli.utils.config.models import RegionInfo manager = ConfigManager(config_dir=tmp_path, skip_validation=True) - - # Mock keyring as disabled - mock_profile_manager._store_token_in_keyring.return_value = False - mock_profile_manager._is_keyring_enabled.return_value = False manager.profile_manager = mock_profile_manager monkeypatch.setattr( @@ -1686,13 +1680,13 @@ async def fake_prompt(message: str, **_: Any) -> str: region_info = RegionInfo( region="us", name="US Data Center", url="https://www.workato.com" ) - profile_data = await manager._prompt_and_validate_credentials( + profile_data, token = await manager._prompt_and_validate_credentials( "test-profile", region_info ) assert profile_data.region == "us" - assert any("Keyring is disabled" in msg for msg in outputs) - assert any("WORKATO_API_TOKEN" in msg for msg in outputs) + assert token == "valid-token-123" + assert any("Authenticated as" in msg for msg in outputs) @pytest.mark.asyncio async def test_setup_profile_existing_create_new_success( From dcd1a635cf2ff0b97bdee7f8ea3593ad02040068 Mon Sep 17 00:00:00 2001 From: Robert Fujara Date: Tue, 4 Nov 2025 16:39:12 +0100 Subject: [PATCH 3/7] Add credential validation to _setup_profile() for existing profiles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When selecting an existing profile, check if credentials exist in keychain. If missing, prompt user to re-enter API token and validate via API call. This fixes the bug where init fails when keychain is missing credentials. Added tests: - test_setup_profile_existing_with_valid_credentials - test_setup_profile_existing_with_missing_credentials 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- .../cli/utils/config/manager.py | 28 +++++ tests/unit/config/test_manager.py | 118 ++++++++++++++++++ 2 files changed, 146 insertions(+) diff --git a/src/workato_platform_cli/cli/utils/config/manager.py b/src/workato_platform_cli/cli/utils/config/manager.py index a2aea27..3691027 100644 --- a/src/workato_platform_cli/cli/utils/config/manager.py +++ b/src/workato_platform_cli/cli/utils/config/manager.py @@ -279,6 +279,34 @@ async def _setup_profile(self) -> str: await self._create_new_profile(profile_name) else: profile_name = answers["profile_choice"] + + # Check if credentials exist in keychain for existing profile + existing_profile = self.profile_manager.get_profile(profile_name) + token = self.profile_manager._get_token_from_keyring(profile_name) + + if existing_profile and not token: + # Credentials missing from keychain - prompt for re-entry + click.echo("⚠️ Credentials not found for this profile") + + # Get region info from existing profile + region_info = RegionInfo( + region=existing_profile.region, + url=existing_profile.region_url, + name=existing_profile.region, + ) + + # Prompt and validate credentials + ( + profile_data, + validated_token, + ) = await self._prompt_and_validate_credentials( + profile_name, region_info + ) + + # Update profile with validated credentials + self.profile_manager.set_profile( + profile_name, profile_data, validated_token + ) else: profile_name = ( await click.prompt("Enter profile name", default="default", type=str) diff --git a/tests/unit/config/test_manager.py b/tests/unit/config/test_manager.py index 077b37c..582f650 100644 --- a/tests/unit/config/test_manager.py +++ b/tests/unit/config/test_manager.py @@ -1394,6 +1394,124 @@ def fake_inquirer_prompt(questions: list[Any]) -> dict[str, str]: assert profile_name == "existing" assert any("Profile:" in line for line in outputs) + @pytest.mark.asyncio + async def test_setup_profile_existing_with_valid_credentials( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """Existing profile with valid credentials should not prompt for re-entry.""" + + mock_profile_manager.set_profile( + "existing", + ProfileData( + region="us", region_url="https://www.workato.com", workspace_id=1 + ), + "existing-token", + ) + mock_profile_manager.set_current_profile("existing") + + # Mock _get_token_from_keyring to return valid token + mock_profile_manager._get_token_from_keyring.return_value = "existing-token" + + monkeypatch.setattr( + ConfigManager.__module__ + ".ProfileManager", + lambda: mock_profile_manager, + ) + + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + def fake_inquirer_prompt(questions: list[Any]) -> dict[str, str]: + assert questions[0].message == "Select a profile" + return {"profile_choice": "existing"} + + monkeypatch.setattr( + ConfigManager.__module__ + ".inquirer.prompt", + fake_inquirer_prompt, + ) + + config_manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + profile_name = await config_manager._setup_profile() + assert profile_name == "existing" + assert any("Profile:" in line for line in outputs) + # Should not show credential warning + assert not any("Credentials not found" in line for line in outputs) + + @pytest.mark.asyncio + async def test_setup_profile_existing_with_missing_credentials( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """Existing profile with missing credentials should prompt for re-entry.""" + + existing_profile = ProfileData( + region="us", region_url="https://www.workato.com", workspace_id=1 + ) + mock_profile_manager.set_profile("existing", existing_profile, None) + mock_profile_manager.set_current_profile("existing") + + # Mock _get_token_from_keyring to return None (missing credentials) + mock_profile_manager._get_token_from_keyring.return_value = None + + # Mock get_profile to return the existing profile + mock_profile_manager.get_profile.return_value = existing_profile + + monkeypatch.setattr( + ConfigManager.__module__ + ".ProfileManager", + lambda: mock_profile_manager, + ) + monkeypatch.setattr( + ConfigManager.__module__ + ".Workato", + StubWorkato, + ) + + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + prompt_answers = { + "Enter your Workato API token": ["new-token"], + } + + async def fake_prompt( + text: str, type: Any = None, hide_input: bool = False + ) -> Any: + return prompt_answers.get(text, [""])[0] + + monkeypatch.setattr( + ConfigManager.__module__ + ".click.prompt", + fake_prompt, + ) + + def fake_inquirer_prompt(questions: list[Any]) -> dict[str, str]: + assert questions[0].message == "Select a profile" + return {"profile_choice": "existing"} + + monkeypatch.setattr( + ConfigManager.__module__ + ".inquirer.prompt", + fake_inquirer_prompt, + ) + + config_manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + profile_name = await config_manager._setup_profile() + assert profile_name == "existing" + + # Should show credential warning + assert any("Credentials not found" in line for line in outputs) + assert any("Please re-enter your API token" in line for line in outputs) + + # Should call set_profile with new credentials + mock_profile_manager.set_profile.assert_called() + @pytest.mark.asyncio async def test_create_new_profile_custom_region( self, From 87d5a0bb90f51c0a83e695f2fe7960991de812e3 Mon Sep 17 00:00:00 2001 From: Robert Fujara Date: Tue, 4 Nov 2025 17:02:18 +0100 Subject: [PATCH 4/7] Add credential validation for non-interactive mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add validation to detect when a profile exists but credentials are missing from the keychain in non-interactive mode. This provides clear, actionable error messages instead of cryptic authentication failures. Changes: - Add credential check after ConfigManager.initialize() for existing profiles - Use SystemExit(1) for JSON mode to ensure proper exit codes - Use click.Abort() for table mode for standard CLI behavior - Skip validation when creating new profiles (region+api_token provided) - Respect WORKATO_API_TOKEN environment variable Tests: - Add test for missing credentials in JSON mode - Add test for missing credentials in table mode - Add test for valid credentials (success path) - Add test for new profile creation (validation skipped) - Update existing tests to mock validate_environment_config() All tests pass (25/25), type checking passes, linting passes. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/workato_platform_cli/cli/commands/init.py | 22 ++ tests/unit/commands/test_init.py | 277 ++++++++++++++++++ 2 files changed, 299 insertions(+) diff --git a/src/workato_platform_cli/cli/commands/init.py b/src/workato_platform_cli/cli/commands/init.py index 2febd39..c4de307 100644 --- a/src/workato_platform_cli/cli/commands/init.py +++ b/src/workato_platform_cli/cli/commands/init.py @@ -129,6 +129,28 @@ async def init( non_interactive=non_interactive, ) + # Validate credentials exist for non-interactive mode with existing profile + if non_interactive and profile and not (region and api_token): + # Only validate if using existing profile (not creating new one) + is_valid, _ = config_manager.validate_environment_config() + if not is_valid: + error_msg = ( + f"Profile '{profile}' exists but credentials not found in keychain. " + "Please run 'workato init' interactively or set WORKATO_API_TOKEN " + "environment variable." + ) + if output_mode == "json": + error_data = { + "status": "error", + "error": error_msg, + "error_code": "MISSING_CREDENTIALS", + } + click.echo(json.dumps(error_data)) + raise SystemExit(1) + else: + click.echo(f"❌ {error_msg}") + raise click.Abort() + # Check if project directory exists and is non-empty # Exclude CLI-managed files from the check cli_managed_files = {".workatoenv", ".gitignore", ".workato-ignore"} diff --git a/tests/unit/commands/test_init.py b/tests/unit/commands/test_init.py index 9f47adc..7ea738c 100644 --- a/tests/unit/commands/test_init.py +++ b/tests/unit/commands/test_init.py @@ -86,6 +86,11 @@ async def test_init_non_interactive_success(monkeypatch: pytest.MonkeyPatch) -> "get_project_directory", return_value=None, ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=(True, []), + ), patch.object( mock_config_manager.profile_manager, "resolve_environment_variables", @@ -437,6 +442,11 @@ async def test_init_non_empty_directory_non_interactive_json( "get_project_directory", return_value=project_dir, ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=(True, []), + ), ): mock_initialize = AsyncMock(return_value=mock_config_manager) monkeypatch.setattr( @@ -487,6 +497,11 @@ async def test_init_non_empty_directory_non_interactive_table( "get_project_directory", return_value=project_dir, ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=(True, []), + ), ): mock_initialize = AsyncMock(return_value=mock_config_manager) monkeypatch.setattr( @@ -532,6 +547,11 @@ async def test_init_non_empty_directory_interactive_cancelled( "get_project_directory", return_value=project_dir, ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=(True, []), + ), ): mock_initialize = AsyncMock(return_value=mock_config_manager) monkeypatch.setattr( @@ -649,6 +669,11 @@ async def test_init_json_output_mode_success( "get_project_directory", return_value=None, ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=(True, []), + ), patch.object( mock_config_manager.profile_manager, "resolve_environment_variables", @@ -803,6 +828,11 @@ async def test_init_cli_managed_files_ignored_non_interactive( "get_project_directory", return_value=project_dir, ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=(True, []), + ), patch.object( mock_config_manager.profile_manager, "resolve_environment_variables", @@ -865,6 +895,11 @@ async def test_init_user_files_with_cli_files_triggers_warning( "get_project_directory", return_value=project_dir, ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=(True, []), + ), ): mock_initialize = AsyncMock(return_value=mock_config_manager) monkeypatch.setattr( @@ -913,6 +948,11 @@ async def test_init_only_workatoenv_file_ignored( "get_project_directory", return_value=project_dir, ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=(True, []), + ), patch.object( mock_config_manager.profile_manager, "resolve_environment_variables", @@ -949,3 +989,240 @@ async def test_init_only_workatoenv_file_ignored( # Should proceed without error mock_pull.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_init_non_interactive_profile_missing_credentials_json( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test non-interactive mode with missing credentials in JSON output mode.""" + mock_config_manager = Mock() + + with ( + patch.object( + mock_config_manager, + "load_config", + return_value=Mock(profile="test-profile"), + ), + patch.object( + mock_config_manager, + "get_project_directory", + return_value=None, + ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=( + False, + ["API token (WORKATO_API_TOKEN or profile credentials)"], + ), + ), + ): + mock_initialize = AsyncMock(return_value=mock_config_manager) + monkeypatch.setattr( + init_module.ConfigManager, + "initialize", + mock_initialize, + ) + + output = StringIO() + monkeypatch.setattr(init_module.click, "echo", lambda msg: output.write(msg)) + + assert init_module.init.callback + with pytest.raises(SystemExit) as exc_info: + await init_module.init.callback( + profile="test-profile", + region=None, + api_token=None, + api_url=None, + project_name=None, + project_id=123, + non_interactive=True, + output_mode="json", + ) + + assert exc_info.value.code == 1 + result = json.loads(output.getvalue()) + assert result["status"] == "error" + assert "credentials not found" in result["error"] + assert "test-profile" in result["error"] + assert result["error_code"] == "MISSING_CREDENTIALS" + + +@pytest.mark.asyncio +async def test_init_non_interactive_profile_missing_credentials_table( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test non-interactive mode with missing credentials in table output mode.""" + mock_config_manager = Mock() + + with ( + patch.object( + mock_config_manager, + "load_config", + return_value=Mock(profile="test-profile"), + ), + patch.object( + mock_config_manager, + "get_project_directory", + return_value=None, + ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=( + False, + ["API token (WORKATO_API_TOKEN or profile credentials)"], + ), + ), + ): + mock_initialize = AsyncMock(return_value=mock_config_manager) + monkeypatch.setattr( + init_module.ConfigManager, + "initialize", + mock_initialize, + ) + + monkeypatch.setattr(init_module.click, "echo", lambda *args, **kwargs: None) + + assert init_module.init.callback + with pytest.raises(click.Abort): + await init_module.init.callback( + profile="test-profile", + region=None, + api_token=None, + api_url=None, + project_name=None, + project_id=123, + non_interactive=True, + output_mode="table", + ) + + +@pytest.mark.asyncio +async def test_init_non_interactive_profile_with_valid_credentials( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test non-interactive mode with valid credentials proceeds normally.""" + mock_config_manager = Mock() + mock_workato_client = Mock() + workato_context = AsyncMock() + + with ( + patch.object( + mock_config_manager, + "load_config", + return_value=Mock(profile="test-profile"), + ), + patch.object( + mock_config_manager, + "get_project_directory", + return_value=None, + ), + patch.object( + mock_config_manager, + "validate_environment_config", + return_value=(True, []), + ), + patch.object( + mock_config_manager.profile_manager, + "resolve_environment_variables", + return_value=("test-token", "https://api.workato.com"), + ), + patch.object(workato_context, "__aenter__", return_value=mock_workato_client), + patch.object(workato_context, "__aexit__", return_value=False), + ): + mock_initialize = AsyncMock(return_value=mock_config_manager) + monkeypatch.setattr( + init_module.ConfigManager, + "initialize", + mock_initialize, + ) + + mock_pull = AsyncMock() + monkeypatch.setattr(init_module, "_pull_project", mock_pull) + + monkeypatch.setattr(init_module, "Workato", lambda **_: workato_context) + monkeypatch.setattr(init_module, "Configuration", lambda **_: Mock()) + + monkeypatch.setattr(init_module.click, "echo", lambda *args, **kwargs: None) + + assert init_module.init.callback + await init_module.init.callback( + profile="test-profile", + region=None, + api_token=None, + api_url=None, + project_name=None, + project_id=123, + non_interactive=True, + output_mode="table", + ) + + # Should proceed without error + mock_pull.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_init_non_interactive_new_profile_with_region_and_token( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test non-interactive mode creating new profile with region and token. + + This test verifies that credential validation is skipped when creating + a new profile with both region and token provided. + """ + mock_config_manager = Mock() + mock_workato_client = Mock() + workato_context = AsyncMock() + + with ( + patch.object( + mock_config_manager, + "load_config", + return_value=Mock(profile="new-profile"), + ), + patch.object( + mock_config_manager, + "get_project_directory", + return_value=None, + ), + patch.object( + mock_config_manager.profile_manager, + "resolve_environment_variables", + return_value=("new-token", "https://api.workato.com"), + ), + patch.object(workato_context, "__aenter__", return_value=mock_workato_client), + patch.object(workato_context, "__aexit__", return_value=False), + ): + mock_initialize = AsyncMock(return_value=mock_config_manager) + monkeypatch.setattr( + init_module.ConfigManager, + "initialize", + mock_initialize, + ) + + mock_pull = AsyncMock() + monkeypatch.setattr(init_module, "_pull_project", mock_pull) + + monkeypatch.setattr(init_module, "Workato", lambda **_: workato_context) + monkeypatch.setattr(init_module, "Configuration", lambda **_: Mock()) + + monkeypatch.setattr(init_module.click, "echo", lambda *args, **kwargs: None) + + # Should not call validate_environment_config when creating new profile + # (region and api_token are both provided) + assert init_module.init.callback + await init_module.init.callback( + profile="new-profile", + region="us", + api_token="new-token", + api_url=None, + project_name=None, + project_id=123, + non_interactive=True, + output_mode="table", + ) + + # Should proceed without error and not check credentials + mock_pull.assert_awaited_once() From c628568345057649049d948dbb18237bc4e6d8bd Mon Sep 17 00:00:00 2001 From: Robert Fujara Date: Tue, 4 Nov 2025 17:08:53 +0100 Subject: [PATCH 5/7] Fix test assertion for credential prompt message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Update test_setup_profile_existing_with_missing_credentials to check for the actual prompt message "Enter your API token" instead of the incorrect "Please re-enter your API token". The implementation in _prompt_and_validate_credentials() outputs "Enter your Workato API token", so the test should verify this actual behavior. Test results: - All 929 tests pass - 97% code coverage maintained - Type checking passes (mypy) - Linting passes (ruff) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- tests/unit/config/test_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/config/test_manager.py b/tests/unit/config/test_manager.py index 582f650..816c8b9 100644 --- a/tests/unit/config/test_manager.py +++ b/tests/unit/config/test_manager.py @@ -1507,7 +1507,7 @@ def fake_inquirer_prompt(questions: list[Any]) -> dict[str, str]: # Should show credential warning assert any("Credentials not found" in line for line in outputs) - assert any("Please re-enter your API token" in line for line in outputs) + assert any("Enter your API token" in line for line in outputs) # Should call set_profile with new credentials mock_profile_manager.set_profile.assert_called() From 3a70678b16640cf322af1a0b52b2f9f27ae15803 Mon Sep 17 00:00:00 2001 From: Robert Fujara Date: Tue, 4 Nov 2025 17:33:17 +0100 Subject: [PATCH 6/7] Fix missing credentials error message in non-interactive mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When running `workato init` in non-interactive mode with an existing profile that has missing keychain credentials, the CLI now shows a clear error message instead of attempting authentication and failing with "Authentication failed". Changes: - Add credential validation in ConfigManager._setup_non_interactive() before attempting API authentication - Add specific ClickException handler to show clean error messages without the "ClickException" label - Error message now clearly states: "Profile 'X' exists but credentials not found in keychain. Please run 'workato init' interactively or set WORKATO_API_TOKEN environment variable." This improves the user experience by providing actionable guidance when credentials are missing in non-interactive scenarios. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/workato_platform_cli/cli/commands/init.py | 22 ---------------- .../cli/utils/config/manager.py | 8 ++++++ .../cli/utils/exception_handler.py | 26 +++++++++++++++++++ 3 files changed, 34 insertions(+), 22 deletions(-) diff --git a/src/workato_platform_cli/cli/commands/init.py b/src/workato_platform_cli/cli/commands/init.py index c4de307..2febd39 100644 --- a/src/workato_platform_cli/cli/commands/init.py +++ b/src/workato_platform_cli/cli/commands/init.py @@ -129,28 +129,6 @@ async def init( non_interactive=non_interactive, ) - # Validate credentials exist for non-interactive mode with existing profile - if non_interactive and profile and not (region and api_token): - # Only validate if using existing profile (not creating new one) - is_valid, _ = config_manager.validate_environment_config() - if not is_valid: - error_msg = ( - f"Profile '{profile}' exists but credentials not found in keychain. " - "Please run 'workato init' interactively or set WORKATO_API_TOKEN " - "environment variable." - ) - if output_mode == "json": - error_data = { - "status": "error", - "error": error_msg, - "error_code": "MISSING_CREDENTIALS", - } - click.echo(json.dumps(error_data)) - raise SystemExit(1) - else: - click.echo(f"❌ {error_msg}") - raise click.Abort() - # Check if project directory exists and is non-empty # Exclude CLI-managed files from the check cli_managed_files = {".workatoenv", ".gitignore", ".workato-ignore"} diff --git a/src/workato_platform_cli/cli/utils/config/manager.py b/src/workato_platform_cli/cli/utils/config/manager.py index 3691027..15c2104 100644 --- a/src/workato_platform_cli/cli/utils/config/manager.py +++ b/src/workato_platform_cli/cli/utils/config/manager.py @@ -131,6 +131,14 @@ async def _setup_non_interactive( project_profile_override=current_profile_name ) + # Validate credentials are available before attempting authentication + if not api_token: + raise click.ClickException( + f"Profile '{current_profile_name}' exists but credentials not found " + "in keychain. Please run 'workato init' interactively or set " + "WORKATO_API_TOKEN environment variable." + ) + # Map region to URL if region == "custom": if not api_url: diff --git a/src/workato_platform_cli/cli/utils/exception_handler.py b/src/workato_platform_cli/cli/utils/exception_handler.py index af2f5eb..9b5b1b8 100644 --- a/src/workato_platform_cli/cli/utils/exception_handler.py +++ b/src/workato_platform_cli/cli/utils/exception_handler.py @@ -145,6 +145,10 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: except click.Abort: # Let Click handle Abort - don't catch it raise + except click.ClickException as e: + # Handle ClickException specifically - show message without error type + _handle_click_exception(e) + raise SystemExit(1) from None except Exception as e: # Catch-all for any exceptions during initialization _handle_generic_cli_error(e) @@ -175,6 +179,10 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any: except click.Abort: # Let Click handle Abort - don't catch it raise + except click.ClickException as e: + # Handle ClickException specifically - show message without error type + _handle_click_exception(e) + raise SystemExit(1) from None except Exception as e: # Catch-all for any exceptions during initialization _handle_generic_cli_error(e) @@ -510,6 +518,24 @@ def _handle_ssl_error(e: aiohttp.ClientSSLError | ssl.SSLError) -> None: click.echo(" • Your network is not intercepting HTTPS connections") +def _handle_click_exception(e: click.ClickException) -> None: + """Handle click.ClickException with clean error message.""" + output_mode = _get_output_mode() + + error_msg = str(e) + + if output_mode == "json": + error_data = { + "status": "error", + "error": error_msg, + "error_code": "CLI_ERROR", + } + click.echo(json.dumps(error_data)) + return + + click.echo(f"❌ {error_msg}") + + def _handle_generic_cli_error(e: Exception) -> None: """Handle any other unexpected CLI errors with a generic message.""" output_mode = _get_output_mode() From b9538573a504b42370f9f63cdb58ef443f2a1261 Mon Sep 17 00:00:00 2001 From: Robert Fujara Date: Tue, 4 Nov 2025 17:42:27 +0100 Subject: [PATCH 7/7] Fix test expectations for credential validation error handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The validation logic was moved from init.py to manager.py in commit 3a70678, which changes how ClickException is raised and handled. The exception handler now converts all ClickException to SystemExit(1) for both JSON and table modes, with appropriate error messages for each. Updated tests to: - Expect SystemExit(1) for both JSON and table modes (not click.Abort) - Mock ConfigManager.initialize to raise ClickException directly - Verify error_code is 'CLI_ERROR' (not 'MISSING_CREDENTIALS') - Mock click.get_current_context to provide output_mode to exception handler All 25 tests in test_init.py now pass. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- tests/unit/commands/test_init.py | 173 +++++++++++++++---------------- 1 file changed, 85 insertions(+), 88 deletions(-) diff --git a/tests/unit/commands/test_init.py b/tests/unit/commands/test_init.py index 7ea738c..5f04bfc 100644 --- a/tests/unit/commands/test_init.py +++ b/tests/unit/commands/test_init.py @@ -996,57 +996,52 @@ async def test_init_non_interactive_profile_missing_credentials_json( monkeypatch: pytest.MonkeyPatch, ) -> None: """Test non-interactive mode with missing credentials in JSON output mode.""" - mock_config_manager = Mock() - - with ( - patch.object( - mock_config_manager, - "load_config", - return_value=Mock(profile="test-profile"), - ), - patch.object( - mock_config_manager, - "get_project_directory", - return_value=None, - ), - patch.object( - mock_config_manager, - "validate_environment_config", - return_value=( - False, - ["API token (WORKATO_API_TOKEN or profile credentials)"], - ), - ), - ): - mock_initialize = AsyncMock(return_value=mock_config_manager) - monkeypatch.setattr( - init_module.ConfigManager, - "initialize", - mock_initialize, + # Mock ConfigManager.initialize to raise ClickException + # (simulating missing credentials) + mock_initialize = AsyncMock( + side_effect=click.ClickException( + "Profile 'test-profile' exists but credentials not found in keychain. " + "Please run 'workato init' interactively or set WORKATO_API_TOKEN " + "environment variable." ) + ) + monkeypatch.setattr( + init_module.ConfigManager, + "initialize", + mock_initialize, + ) - output = StringIO() - monkeypatch.setattr(init_module.click, "echo", lambda msg: output.write(msg)) - - assert init_module.init.callback - with pytest.raises(SystemExit) as exc_info: - await init_module.init.callback( - profile="test-profile", - region=None, - api_token=None, - api_url=None, - project_name=None, - project_id=123, - non_interactive=True, - output_mode="json", - ) + output = StringIO() + monkeypatch.setattr(init_module.click, "echo", lambda msg: output.write(msg)) + + # Mock get_current_context to return output_mode + mock_ctx = Mock() + mock_ctx.params = {"output_mode": "json"} + monkeypatch.setattr( + init_module.click, + "get_current_context", + lambda silent=True: mock_ctx, + ) + + assert init_module.init.callback + with pytest.raises(SystemExit) as exc_info: + await init_module.init.callback( + profile="test-profile", + region=None, + api_token=None, + api_url=None, + project_name=None, + project_id=123, + non_interactive=True, + output_mode="json", + ) - assert exc_info.value.code == 1 - result = json.loads(output.getvalue()) - assert result["status"] == "error" - assert "credentials not found" in result["error"] - assert "test-profile" in result["error"] - assert result["error_code"] == "MISSING_CREDENTIALS" + assert exc_info.value.code == 1 + result = json.loads(output.getvalue()) + assert result["status"] == "error" + assert "credentials not found" in result["error"] + assert "test-profile" in result["error"] + assert result["error_code"] == "CLI_ERROR" @pytest.mark.asyncio @@ -1054,49 +1049,51 @@ async def test_init_non_interactive_profile_missing_credentials_table( monkeypatch: pytest.MonkeyPatch, ) -> None: """Test non-interactive mode with missing credentials in table output mode.""" - mock_config_manager = Mock() - - with ( - patch.object( - mock_config_manager, - "load_config", - return_value=Mock(profile="test-profile"), - ), - patch.object( - mock_config_manager, - "get_project_directory", - return_value=None, - ), - patch.object( - mock_config_manager, - "validate_environment_config", - return_value=( - False, - ["API token (WORKATO_API_TOKEN or profile credentials)"], - ), - ), - ): - mock_initialize = AsyncMock(return_value=mock_config_manager) - monkeypatch.setattr( - init_module.ConfigManager, - "initialize", - mock_initialize, + # Mock ConfigManager.initialize to raise ClickException + # (simulating missing credentials) + mock_initialize = AsyncMock( + side_effect=click.ClickException( + "Profile 'test-profile' exists but credentials not found in keychain. " + "Please run 'workato init' interactively or set WORKATO_API_TOKEN " + "environment variable." ) + ) + monkeypatch.setattr( + init_module.ConfigManager, + "initialize", + mock_initialize, + ) - monkeypatch.setattr(init_module.click, "echo", lambda *args, **kwargs: None) + output = StringIO() + monkeypatch.setattr(init_module.click, "echo", lambda msg: output.write(msg)) + + # Mock get_current_context to return output_mode + mock_ctx = Mock() + mock_ctx.params = {"output_mode": "table"} + monkeypatch.setattr( + init_module.click, + "get_current_context", + lambda silent=True: mock_ctx, + ) + + assert init_module.init.callback + with pytest.raises(SystemExit) as exc_info: + await init_module.init.callback( + profile="test-profile", + region=None, + api_token=None, + api_url=None, + project_name=None, + project_id=123, + non_interactive=True, + output_mode="table", + ) - assert init_module.init.callback - with pytest.raises(click.Abort): - await init_module.init.callback( - profile="test-profile", - region=None, - api_token=None, - api_url=None, - project_name=None, - project_id=123, - non_interactive=True, - output_mode="table", - ) + assert exc_info.value.code == 1 + # Verify error message was shown + output_text = output.getvalue() + assert "credentials not found" in output_text + assert "test-profile" in output_text @pytest.mark.asyncio