diff --git a/src/workato_platform_cli/cli/commands/init.py b/src/workato_platform_cli/cli/commands/init.py index 2febd39..27de3ea 100644 --- a/src/workato_platform_cli/cli/commands/init.py +++ b/src/workato_platform_cli/cli/commands/init.py @@ -1,6 +1,7 @@ """Initialize Workato CLI for a new project""" import json +import os from typing import Any @@ -54,7 +55,10 @@ async def init( non_interactive: bool = False, output_mode: str = "table", ) -> None: - """Initialize Workato CLI for a new project""" + """Initialize Workato CLI for a new project + + Detects WORKATO_API_TOKEN and WORKATO_HOST environment variables if present. + """ # Validate that --output-mode json requires --non-interactive if output_mode == "json" and not non_interactive: @@ -68,15 +72,40 @@ async def init( return if non_interactive: + # Check for environment variables if flags not provided + env_token = os.environ.get("WORKATO_API_TOKEN") + env_host = os.environ.get("WORKATO_HOST") + + # Track which env vars are actually used for visibility + used_env_token = False + used_env_host = False + + # Use env vars as fallback if CLI flags not provided + if not api_token and env_token: + api_token = env_token + used_env_token = True + if not region and not api_url and env_host: + api_url = env_host + used_env_host = True + # Will detect region in _setup_non_interactive + + # Show which env vars are being used (only in non-JSON mode) + if output_mode != "json" and (used_env_token or used_env_host): + if used_env_token: + click.echo("Using WORKATO_API_TOKEN from environment") + if used_env_host: + click.echo("Using WORKATO_HOST from environment") + click.echo() + # Validate required parameters for non-interactive mode error_msg = None error_code = None - # Either profile OR individual attributes (region, api_token) are required - if not profile and not (region and api_token): + # Either profile OR individual attributes are required + if not profile and not ((region or api_url) and api_token): error_msg = ( - "Either --profile or both --region and --api-token are required " - "in non-interactive mode" + "Either --profile or both (--region or --api-url/WORKATO_HOST) and " + "--api-token/WORKATO_API_TOKEN are required in non-interactive mode" ) error_code = "MISSING_REQUIRED_OPTIONS" elif region == "custom" and not api_url: diff --git a/src/workato_platform_cli/cli/utils/config/manager.py b/src/workato_platform_cli/cli/utils/config/manager.py index 605c154..cd52cc1 100644 --- a/src/workato_platform_cli/cli/utils/config/manager.py +++ b/src/workato_platform_cli/cli/utils/config/manager.py @@ -1,6 +1,7 @@ """Main configuration manager with simplified workspace rules.""" import json +import os import sys from pathlib import Path @@ -127,19 +128,32 @@ async def _setup_non_interactive( profile = self.profile_manager.get_profile(current_profile_name) if profile and profile.region: region = profile.region - api_token, api_url = self.profile_manager.resolve_environment_variables( + env_token, env_url = self.profile_manager.resolve_environment_variables( project_profile_override=current_profile_name ) + # Use provided values or fall back to env vars + if not api_token: + api_token = env_token + if not api_url: + api_url = env_url + + # Detect region from api_url if region not provided + if not region and api_url: + region_info = self._match_host_to_region(api_url) + region = region_info.region + # Map region to URL if region == "custom": if not api_url: raise click.ClickException("--api-url is required when region=custom") region_info = RegionInfo(region="custom", name="Custom URL", url=api_url) - else: + elif region: if region not in AVAILABLE_REGIONS: raise click.ClickException(f"Invalid region: {region}") region_info = AVAILABLE_REGIONS[region] + else: + raise click.ClickException("Region could not be determined") # Test authentication and get workspace info api_config = Configuration( @@ -251,9 +265,57 @@ async def _setup_profile(self) -> str: """Setup or select profile""" click.echo("šŸ“‹ Step 1: Configure profile") - existing_profiles = self.profile_manager.list_profiles() + # Check for environment variables FIRST + env_token = os.environ.get("WORKATO_API_TOKEN") + env_host = os.environ.get("WORKATO_HOST") + profile_name: str | None = None + if env_token or env_host: + # Show what was detected and what's missing + if env_token: + click.echo("āœ“ Found WORKATO_API_TOKEN in environment") + else: + click.echo("āœ— WORKATO_API_TOKEN not found - will prompt for token") + + if env_host: + region_info = self._match_host_to_region(env_host) + click.echo(f"āœ“ Found WORKATO_HOST in environment ({region_info.name})") + else: + click.echo("āœ— WORKATO_HOST not found - will prompt for region") + + # Ask user if they want to use env vars + if env_token and env_host: + prompt_msg = "Use environment variables for authentication?" + elif env_token: + prompt_msg = "Use WORKATO_API_TOKEN from environment?" + else: # only env_host + prompt_msg = "Use WORKATO_HOST from environment?" + + use_env_vars = click.confirm( + prompt_msg, + default=True, + ) + + if use_env_vars: + # Just select/create profile NAME (credentials from env vars) + click.echo() + profile_name = await self._select_profile_name_for_env_vars() + + # Create profile with env var credentials + await self._create_profile_with_env_vars( + profile_name, env_token, env_host + ) + + # Set as current profile + self.profile_manager.set_current_profile(profile_name) + click.echo(f"āœ… Profile: {profile_name}") + + return profile_name + + # Normal flow - no env vars or user declined + existing_profiles = self.profile_manager.list_profiles() + if existing_profiles: choices = list(existing_profiles.keys()) + ["Create new profile"] questions = [ @@ -294,48 +356,129 @@ async def _setup_profile(self) -> str: return profile_name - async def _create_new_profile(self, profile_name: str) -> None: - """Create a new profile interactively""" - # AVAILABLE_REGIONS and RegionInfo already imported at top + def _match_host_to_region(self, host: str) -> RegionInfo: + """Match host URL to known region or return custom""" + for region_info in AVAILABLE_REGIONS.values(): + if region_info.url and region_info.url in host: + return region_info + # Return custom region with provided host + return RegionInfo(region="custom", name="Custom URL", url=host) - # Region selection - click.echo("šŸ“ Select your Workato region") - regions = list(AVAILABLE_REGIONS.values()) - choices = [] + async def _select_profile_name_for_env_vars(self) -> str: + """Let user choose profile name when using env var credentials""" + existing_profiles = self.profile_manager.list_profiles() + + if existing_profiles: + choices = list(existing_profiles.keys()) + ["Create new profile"] + questions = [ + inquirer.List( + "profile_choice", + message="Select a profile name to use", + choices=choices, + ) + ] + + answers: dict[str, str] = inquirer.prompt(questions) + if not answers: + click.echo("āŒ No profile selected") + sys.exit(1) - for region in regions: - if region.region == "custom": - choice_text = "Custom URL" + selected_choice: str = answers["profile_choice"] + if selected_choice == "Create new profile": + new_profile_input: str = await click.prompt( + "Enter new profile name", type=str + ) + profile_name = new_profile_input.strip() + if not profile_name: + click.echo("āŒ Profile name cannot be empty") + sys.exit(1) + return profile_name else: - choice_text = f"{region.name} ({region.url})" - choices.append(choice_text) + # Warn user about overwriting existing profile + click.echo( + f"\nāš ļø This will overwrite the existing profile " + f"'{selected_choice}' with the environment variables." + ) + if not click.confirm("Continue?", default=True): + click.echo("āŒ Cancelled") + sys.exit(1) + return selected_choice + else: + default_profile_input: str = await click.prompt( + "Enter profile name", default="default", type=str + ) + profile_name = default_profile_input.strip() + if not profile_name: + click.echo("āŒ Profile name cannot be empty") + sys.exit(1) + return profile_name + + async def _create_profile_with_env_vars( + self, + profile_name: str, + env_token: str | None, + env_host: str | None, + ) -> None: + """Create profile using environment variable credentials""" + # Detect region from env_host + selected_region: RegionInfo + if env_host: + selected_region = self._match_host_to_region(env_host) + else: + # No env_host, need to ask for region + click.echo() + click.echo("šŸ“ Select your Workato region") + region_result = await self.profile_manager.select_region_interactive() + + if not region_result: + click.echo("āŒ Setup cancelled") + sys.exit(1) + + selected_region = region_result + + # Get token from env or prompt + if env_token: + token = env_token + else: + click.echo() + 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 + click.echo("šŸ”„ Testing authentication with environment variables...") + 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, + ) - questions = [ - inquirer.List( - "region", - message="Select your Workato region", - choices=choices, - ), - ] + self.profile_manager.set_profile(profile_name, profile_data, token) + click.echo(f"āœ… Authenticated as: {user_info.name}") + click.echo("āœ“ Using environment variables for authentication") - answers = inquirer.prompt(questions) - if not answers: + async def _create_new_profile(self, profile_name: str) -> None: + """Create a new profile interactively""" + # Select region + click.echo("šŸ“ Select your Workato region") + region_result = await self.profile_manager.select_region_interactive() + + if not region_result: click.echo("āŒ Setup cancelled") sys.exit(1) - selected_index = choices.index(answers["region"]) - selected_region = regions[selected_index] - - # Handle custom URL - if selected_region.region == "custom": - custom_url = await click.prompt( - "Enter your custom Workato base URL", - type=str, - default="https://www.workato.com", - ) - selected_region = RegionInfo( - region="custom", name="Custom URL", url=custom_url - ) + selected_region = region_result # Get API token click.echo("šŸ” Enter your API token") diff --git a/tests/unit/commands/test_init.py b/tests/unit/commands/test_init.py index 9f47adc..c6c3dc6 100644 --- a/tests/unit/commands/test_init.py +++ b/tests/unit/commands/test_init.py @@ -949,3 +949,333 @@ 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_with_env_vars_regular_mode( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test non-interactive mode shows messages when using env vars (regular mode).""" + mock_config_manager = Mock() + mock_workato_client = Mock() + workato_context = AsyncMock() + + # Set environment variables + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-123") + monkeypatch.setenv("WORKATO_HOST", "https://www.workato.com") + + 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.profile_manager, + "resolve_environment_variables", + return_value=("env-token-123", "https://www.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()) + + # Capture output + output = StringIO() + monkeypatch.setattr( + init_module.click, "echo", lambda msg="": output.write(str(msg) + "\n") + ) + + assert init_module.init.callback + await init_module.init.callback( + profile="test-profile", + region=None, + api_token=None, # Not provided, should use env var + api_url=None, # Not provided, should use env var + project_name="test-project", + project_id=None, + non_interactive=True, + output_mode="table", + ) + + output_text = output.getvalue() + # Verify messages about env var usage + assert "Using WORKATO_API_TOKEN from environment" in output_text + assert "Using WORKATO_HOST from environment" in output_text + mock_pull.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_init_non_interactive_with_env_vars_json_mode( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test non-interactive mode with env vars in JSON output.""" + mock_config_manager = Mock() + mock_workato_client = Mock() + workato_context = AsyncMock() + + # Set environment variables + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-123") + monkeypatch.setenv("WORKATO_HOST", "https://www.workato.com") + + with ( + patch.object( + mock_config_manager, + "load_config", + return_value=Mock( + profile="test-profile", + project_name="test-project", + project_id=123, + folder_id=456, + ), + ), + patch.object( + mock_config_manager, + "get_project_directory", + return_value=None, + ), + patch.object( + mock_config_manager.profile_manager, + "resolve_environment_variables", + return_value=("env-token-123", "https://www.workato.com"), + ), + patch.object( + mock_config_manager.profile_manager, + "get_profile", + return_value=Mock( + region="us", + region_name="US", + region_url="https://www.workato.com", + workspace_id=789, + ), + ), + patch.object( + mock_config_manager.profile_manager, + "get_current_profile_name", + return_value="test-profile", + ), + 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()) + + output = StringIO() + monkeypatch.setattr( + init_module.click, "echo", lambda msg: output.write(str(msg)) + ) + + assert init_module.init.callback + await init_module.init.callback( + profile="test-profile", + region=None, + api_token=None, # Not provided, should use env var + api_url=None, # Not provided, should use env var + project_name="test-project", + project_id=None, + non_interactive=True, + output_mode="json", + ) + + result = json.loads(output.getvalue()) + assert result["status"] == "success" + assert "Using WORKATO_API_TOKEN" not in output.getvalue() + + +@pytest.mark.asyncio +async def test_init_non_interactive_cli_flags_take_precedence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test that CLI flags take precedence over env vars with no env var messages.""" + mock_config_manager = Mock() + mock_workato_client = Mock() + workato_context = AsyncMock() + + # Set environment variables + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-123") + monkeypatch.setenv("WORKATO_HOST", "https://www.workato.com") + + with ( + patch.object( + mock_config_manager, + "load_config", + return_value=Mock( + profile="test-profile", + project_name="test-project", + project_id=123, + folder_id=456, + ), + ), + patch.object( + mock_config_manager, + "get_project_directory", + return_value=None, + ), + patch.object( + mock_config_manager.profile_manager, + "resolve_environment_variables", + return_value=("cli-token", "https://api.workato.com"), + ), + patch.object( + mock_config_manager.profile_manager, + "get_profile", + return_value=Mock( + region="us", + region_name="US", + region_url="https://api.workato.com", + workspace_id=789, + ), + ), + patch.object( + mock_config_manager.profile_manager, + "get_current_profile_name", + return_value="test-profile", + ), + 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()) + + output = StringIO() + monkeypatch.setattr( + init_module.click, "echo", lambda msg="": output.write(str(msg) + "\n") + ) + + assert init_module.init.callback + await init_module.init.callback( + profile="test-profile", + region="us", + api_token="cli-token", # CLI flag provided + api_url=None, + project_name="test-project", + project_id=None, + non_interactive=True, + output_mode="table", + ) + + output_text = output.getvalue() + # Should NOT show env var messages since CLI flags were provided + assert "Using WORKATO_API_TOKEN from environment" not in output_text + assert "Using WORKATO_HOST from environment" not in output_text + mock_pull.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_init_non_interactive_partial_env_vars_json( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test JSON mode with partial env vars (token from env, region from CLI).""" + mock_config_manager = Mock() + mock_workato_client = Mock() + workato_context = AsyncMock() + + # Only set WORKATO_API_TOKEN, not WORKATO_HOST + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-123") + + with ( + patch.object( + mock_config_manager, + "load_config", + return_value=Mock( + profile="test-profile", + project_name="test-project", + project_id=123, + folder_id=456, + ), + ), + patch.object( + mock_config_manager, + "get_project_directory", + return_value=None, + ), + patch.object( + mock_config_manager.profile_manager, + "resolve_environment_variables", + return_value=("env-token-123", "https://www.workato.com"), + ), + patch.object( + mock_config_manager.profile_manager, + "get_profile", + return_value=Mock( + region="us", + region_name="US", + region_url="https://www.workato.com", + workspace_id=789, + ), + ), + patch.object( + mock_config_manager.profile_manager, + "get_current_profile_name", + return_value="test-profile", + ), + 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()) + + output = StringIO() + monkeypatch.setattr( + init_module.click, "echo", lambda msg: output.write(str(msg)) + ) + + assert init_module.init.callback + await init_module.init.callback( + profile="test-profile", + region="us", # CLI flag provided for region + api_token=None, # Not provided, should use env var + api_url=None, + project_name="test-project", + project_id=None, + non_interactive=True, + output_mode="json", + ) + + result = json.loads(output.getvalue()) + assert result["status"] == "success" + # Verify JSON output is valid and contains expected profile data + assert "profile" in result + assert result["profile"]["region"] == "us" diff --git a/tests/unit/config/test_manager.py b/tests/unit/config/test_manager.py index b72deb4..1ab2295 100644 --- a/tests/unit/config/test_manager.py +++ b/tests/unit/config/test_manager.py @@ -451,6 +451,593 @@ async def test_setup_non_interactive_requires_project_selection( assert "No project selected" in str(excinfo.value) + @pytest.mark.asyncio + async def test_setup_non_interactive_with_env_vars( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """Non-interactive mode should use environment variables automatically.""" + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-ni") + monkeypatch.setenv("WORKATO_HOST", "https://www.workato.com") + + monkeypatch.setattr( + ConfigManager.__module__ + ".ProfileManager", + lambda: mock_profile_manager, + ) + monkeypatch.setattr( + ConfigManager.__module__ + ".Workato", + StubWorkato, + ) + StubProjectManager.available_projects = [] + StubProjectManager.created_projects = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".ProjectManager", + StubProjectManager, + ) + + # Mock resolve_environment_variables to return None + # so it uses the api_token/api_url params + mock_profile_manager.resolve_environment_variables.return_value = (None, None) + mock_profile_manager.get_current_profile_name.return_value = "env-profile" + mock_profile_manager.get_profile.return_value = None # No existing profile + + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + manager.workspace_manager = WorkspaceManager(start_path=tmp_path) + monkeypatch.chdir(tmp_path) + + # Should use env vars without requiring --region or --api-token flags + await manager._setup_non_interactive( + profile_name="env-profile", + api_token="env-token-ni", + api_url="https://www.workato.com", + project_name="EnvProject", + ) + + # Verify profile was created + mock_profile_manager.set_profile.assert_called_once() + call_args = mock_profile_manager.set_profile.call_args + profile_name, profile_data, token = call_args[0] + + assert profile_name == "env-profile" + assert token == "env-token-ni" + assert profile_data.region == "us" + + @pytest.mark.asyncio + async def test_setup_non_interactive_detects_eu_from_env( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """Non-interactive mode should detect EU region from WORKATO_HOST.""" + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-eu") + monkeypatch.setenv("WORKATO_HOST", "https://app.eu.workato.com") + + monkeypatch.setattr( + ConfigManager.__module__ + ".ProfileManager", + lambda: mock_profile_manager, + ) + monkeypatch.setattr( + ConfigManager.__module__ + ".Workato", + StubWorkato, + ) + StubProjectManager.available_projects = [] + StubProjectManager.created_projects = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".ProjectManager", + StubProjectManager, + ) + + mock_profile_manager.resolve_environment_variables.return_value = (None, None) + mock_profile_manager.get_current_profile_name.return_value = "eu-profile" + mock_profile_manager.get_profile.return_value = None # No existing profile + + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + manager.workspace_manager = WorkspaceManager(start_path=tmp_path) + monkeypatch.chdir(tmp_path) + + await manager._setup_non_interactive( + profile_name="eu-profile", + api_token="env-token-eu", + api_url="https://app.eu.workato.com", + project_name="EuProject", + ) + + # Verify EU region was detected + mock_profile_manager.set_profile.assert_called_once() + call_args = mock_profile_manager.set_profile.call_args + profile_name, profile_data, token = call_args[0] + + assert profile_name == "eu-profile" + assert profile_data.region == "eu" + assert profile_data.region_url == "https://app.eu.workato.com" + + @pytest.mark.asyncio + async def test_setup_profile_with_both_env_vars_user_accepts_new_profile( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """When both env vars detected and user accepts, profile should use env vars.""" + # Setup environment variables + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-123") + monkeypatch.setenv("WORKATO_HOST", "https://www.workato.com") + + # Setup stubs and mocks + monkeypatch.setattr(ConfigManager.__module__ + ".Workato", StubWorkato) + + # Mock that no profiles exist yet + mock_profile_manager.list_profiles.return_value = {} + + # Mock user confirms using env vars + monkeypatch.setattr( + ConfigManager.__module__ + ".click.confirm", + lambda message, **k: True, + ) + + # Mock profile name prompt (message is "Enter profile name") + prompt_responses = {"Enter profile name": "test-env-profile"} + + async def fake_prompt(message: str, **kwargs: Any) -> str: + return prompt_responses.get(message, "default") + + monkeypatch.setattr(ConfigManager.__module__ + ".click.prompt", fake_prompt) + + # Capture outputs + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + # Execute method under test + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + profile_name = await manager._setup_profile() + + # Assertions + assert profile_name == "test-env-profile" + + # Verify env vars were detected and shown to user + assert any( + "āœ“ Found WORKATO_API_TOKEN in environment" in output for output in outputs + ) + assert any( + "āœ“ Found WORKATO_HOST in environment" in output for output in outputs + ) + + # Verify profile was created with env var credentials + mock_profile_manager.set_profile.assert_called_once() + call_args = mock_profile_manager.set_profile.call_args + created_profile_name, profile_data, token = call_args[0] + + assert created_profile_name == "test-env-profile" + assert token == "env-token-123" + assert profile_data.region == "us" + assert profile_data.region_url == "https://www.workato.com" + + # Verify profile was set as current + mock_profile_manager.set_current_profile.assert_called_once_with( + "test-env-profile" + ) + + @pytest.mark.asyncio + async def test_setup_profile_with_env_vars_user_declines( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """When env vars detected but user declines, should follow normal flow.""" + # Setup environment variables + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-decline") + monkeypatch.setenv("WORKATO_HOST", "https://www.workato.com") + + # Setup stubs and mocks + monkeypatch.setattr(ConfigManager.__module__ + ".Workato", StubWorkato) + + # Mock that profiles exist + mock_profile_manager.list_profiles.return_value = { + "default": ProfileData( + region="us", + region_url="https://www.workato.com", + workspace_id=1, + ) + } + + # Mock user DECLINES using env vars + monkeypatch.setattr( + ConfigManager.__module__ + ".click.confirm", + lambda message, **k: False, + ) + + # Mock inquirer.prompt for profile selection (normal flow) + def fake_inquirer_prompt(questions: list[Any]) -> dict[str, str]: + # User selects existing profile in normal flow + return {"profile_choice": "default"} + + monkeypatch.setattr( + ConfigManager.__module__ + ".inquirer.prompt", + fake_inquirer_prompt, + ) + + # Capture outputs + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + # Execute method under test + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + profile_name = await manager._setup_profile() + + # Assertions + assert profile_name == "default" + + # Verify env vars were detected + assert any( + "āœ“ Found WORKATO_API_TOKEN in environment" in output for output in outputs + ) + + # Verify profile was NOT created with env vars (normal flow was followed) + # The set_profile should not have been called for env vars + # Instead, set_current_profile should be called for existing profile + mock_profile_manager.set_current_profile.assert_called_once_with("default") + + @pytest.mark.asyncio + async def test_setup_profile_with_only_token_prompts_for_region( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """When only WORKATO_API_TOKEN set, should prompt for region selection.""" + # Setup only token env var (no host) + monkeypatch.setenv("WORKATO_API_TOKEN", "token-only") + + # Setup stubs and mocks + monkeypatch.setattr(ConfigManager.__module__ + ".Workato", StubWorkato) + + # Mock that no profiles exist yet + mock_profile_manager.list_profiles.return_value = {} + + # Mock user confirms using env vars + monkeypatch.setattr( + ConfigManager.__module__ + ".click.confirm", + lambda message, **k: True, + ) + + # Mock profile name prompt (message is "Enter profile name") + prompt_responses = {"Enter profile name": "token-only-profile"} + + async def fake_prompt(message: str, **kwargs: Any) -> str: + return prompt_responses.get(message, "default") + + monkeypatch.setattr(ConfigManager.__module__ + ".click.prompt", fake_prompt) + + # Mock region selection (since no host provided) + from workato_platform_cli.cli.utils.config.models import RegionInfo + + region_selected = False + + async def mock_select_region() -> RegionInfo: + nonlocal region_selected + region_selected = True + return RegionInfo( + region="us", + name="United States", + url="https://www.workato.com", + ) + + mock_profile_manager.select_region_interactive = mock_select_region + + # Capture outputs + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + # Execute method under test + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + profile_name = await manager._setup_profile() + + # Assertions + assert profile_name == "token-only-profile" + + # Verify only token was detected (not host) + assert any( + "āœ“ Found WORKATO_API_TOKEN in environment" in output for output in outputs + ) + assert not any( + "āœ“ Found WORKATO_HOST in environment" in output for output in outputs + ) + + # Verify missing host message was shown + assert any( + "āœ— WORKATO_HOST not found - will prompt for region" in output + for output in outputs + ) + + # Verify region was prompted + assert region_selected + + # Verify profile was created + mock_profile_manager.set_profile.assert_called_once() + call_args = mock_profile_manager.set_profile.call_args + created_profile_name, profile_data, token = call_args[0] + + assert created_profile_name == "token-only-profile" + assert token == "token-only" + assert profile_data.region == "us" + + @pytest.mark.asyncio + async def test_setup_profile_with_only_host_prompts_for_token( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """When only WORKATO_HOST set, should prompt for token input.""" + # Setup only host env var (no token) + monkeypatch.setenv("WORKATO_HOST", "https://app.eu.workato.com") + + # Setup stubs and mocks + monkeypatch.setattr(ConfigManager.__module__ + ".Workato", StubWorkato) + + # Mock that no profiles exist yet + mock_profile_manager.list_profiles.return_value = {} + + # Mock user confirms using env vars + confirm_calls = [] + + def fake_confirm(message: str, **kwargs: Any) -> bool: + confirm_calls.append(message) + return True + + monkeypatch.setattr( + ConfigManager.__module__ + ".click.confirm", + fake_confirm, + ) + + # Mock prompts for both profile name and token + token_prompted = False + prompt_responses = { + "Enter profile name": "host-only-profile", + "Enter your Workato API token": "prompted-token-456", + } + + async def fake_prompt(message: str, **kwargs: Any) -> str: + nonlocal token_prompted + if "API token" in message: + token_prompted = True + return prompt_responses.get(message, "default") + + monkeypatch.setattr(ConfigManager.__module__ + ".click.prompt", fake_prompt) + + # Capture outputs + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + # Execute method under test + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + profile_name = await manager._setup_profile() + + # Assertions + assert profile_name == "host-only-profile" + + # Verify only host was detected (not token) + assert any( + "āœ“ Found WORKATO_HOST in environment (EU Data Center)" in output + for output in outputs + ) + assert not any( + "āœ“ Found WORKATO_API_TOKEN in environment" in output for output in outputs + ) + + # Verify missing token message was shown + assert any( + "āœ— WORKATO_API_TOKEN not found - will prompt for token" in output + for output in outputs + ) + + # Verify token was prompted + assert token_prompted + + # Verify profile was created with prompted token and env host + mock_profile_manager.set_profile.assert_called_once() + call_args = mock_profile_manager.set_profile.call_args + created_profile_name, profile_data, token = call_args[0] + + assert created_profile_name == "host-only-profile" + assert token == "prompted-token-456" + assert profile_data.region == "eu" + assert profile_data.region_url == "https://app.eu.workato.com" + + @pytest.mark.asyncio + async def test_select_profile_name_overwrites_existing_with_confirmation( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """When existing profile selected and user confirms, should overwrite.""" + # Setup environment variables + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-overwrite") + monkeypatch.setenv("WORKATO_HOST", "https://www.workato.com") + + # Setup stubs and mocks + monkeypatch.setattr(ConfigManager.__module__ + ".Workato", StubWorkato) + + # Mock that profiles exist + mock_profile_manager.list_profiles.return_value = { + "dev": ProfileData( + region="us", + region_url="https://www.workato.com", + workspace_id=1, + ), + "prod": ProfileData( + region="us", + region_url="https://www.workato.com", + workspace_id=2, + ), + } + + # Track confirm calls to verify both prompts + confirm_calls: list[str] = [] + + def fake_confirm(message: str, **kwargs: Any) -> bool: + confirm_calls.append(message) + return True # User accepts both times + + monkeypatch.setattr( + ConfigManager.__module__ + ".click.confirm", + fake_confirm, + ) + + # Mock inquirer to select existing profile + def fake_inquirer_prompt(questions: list[Any]) -> dict[str, str]: + message = questions[0].message + if "Select a profile name" in message: + # Select existing profile "dev" + return {"profile_choice": "dev"} + return {} + + monkeypatch.setattr( + ConfigManager.__module__ + ".inquirer.prompt", + fake_inquirer_prompt, + ) + + # Capture outputs + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + # Execute method under test + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + profile_name = await manager._setup_profile() + + # Assertions + assert profile_name == "dev" + + # Verify both confirm prompts were called + assert len(confirm_calls) == 2 + assert "Use environment variables for authentication?" in confirm_calls[0] + assert "Continue?" in confirm_calls[1] + + # Verify overwrite warning was shown + assert any( + "āš ļø" in output and "overwrite" in output.lower() for output in outputs + ) + + # Verify profile was created/overwritten + mock_profile_manager.set_profile.assert_called_once() + call_args = mock_profile_manager.set_profile.call_args + created_profile_name, profile_data, token = call_args[0] + + assert created_profile_name == "dev" + assert token == "env-token-overwrite" + + @pytest.mark.asyncio + async def test_select_profile_name_cancels_on_overwrite_decline( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mock_profile_manager: Mock, + ) -> None: + """When existing profile selected but user declines overwrite, should exit.""" + # Setup environment variables + monkeypatch.setenv("WORKATO_API_TOKEN", "env-token-cancel") + monkeypatch.setenv("WORKATO_HOST", "https://www.workato.com") + + # Setup stubs and mocks + monkeypatch.setattr(ConfigManager.__module__ + ".Workato", StubWorkato) + + # Mock that profiles exist + mock_profile_manager.list_profiles.return_value = { + "dev": ProfileData( + region="us", + region_url="https://www.workato.com", + workspace_id=1, + ), + "prod": ProfileData( + region="us", + region_url="https://www.workato.com", + workspace_id=2, + ), + } + + # Track confirm calls - accept env vars, but decline overwrite + confirm_calls: list[str] = [] + + def fake_confirm(message: str, **kwargs: Any) -> bool: + confirm_calls.append(message) + if "Use environment variables" in message: + return True # Accept using env vars + elif "Continue?" in message: + return False # Decline overwrite + return True + + monkeypatch.setattr( + ConfigManager.__module__ + ".click.confirm", + fake_confirm, + ) + + # Mock inquirer to select existing profile + def fake_inquirer_prompt(questions: list[Any]) -> dict[str, str]: + message = questions[0].message + if "Select a profile name" in message: + # Select existing profile "dev" + return {"profile_choice": "dev"} + return {} + + monkeypatch.setattr( + ConfigManager.__module__ + ".inquirer.prompt", + fake_inquirer_prompt, + ) + + # Capture outputs + outputs: list[str] = [] + monkeypatch.setattr( + ConfigManager.__module__ + ".click.echo", + lambda msg="": outputs.append(str(msg)), + ) + + # Execute method under test - should raise SystemExit + manager = ConfigManager(config_dir=tmp_path, skip_validation=True) + manager.profile_manager = mock_profile_manager + + with pytest.raises(SystemExit): + await manager._setup_profile() + + # Verify both confirm prompts were called + assert len(confirm_calls) == 2 + assert "Use environment variables for authentication?" in confirm_calls[0] + assert "Continue?" in confirm_calls[1] + + # Verify overwrite warning was shown + assert any( + "āš ļø" in output and "overwrite" in output.lower() for output in outputs + ) + + # Verify profile was NOT created (user cancelled) + mock_profile_manager.set_profile.assert_not_called() + def test_init_with_explicit_config_dir(self, tmp_path: Path) -> None: """Test ConfigManager respects explicit config_dir.""" config_dir = tmp_path / "explicit" @@ -1412,8 +1999,19 @@ async def test_create_new_profile_custom_region( StubWorkato, ) + # Mock select_region_interactive to return a custom region + from workato_platform_cli.cli.utils.config.models import RegionInfo + + async def mock_select_region() -> RegionInfo: + return RegionInfo( + region="custom", + name="Custom URL", + url="https://custom.workato.test", + ) + + mock_profile_manager.select_region_interactive = mock_select_region + prompt_answers = { - "Enter your custom Workato base URL": ["https://custom.workato.test"], "Enter your Workato API token": ["custom-token"], } @@ -1427,15 +2025,6 @@ async def fake_prompt(message: str, **_: Any) -> str: fake_prompt, ) - def custom_region_prompt(questions: list[Any]) -> dict[str, str]: - assert questions[0].message == "Select your Workato region" - return {"region": "Custom URL"} - - monkeypatch.setattr( - ConfigManager.__module__ + ".inquirer.prompt", - custom_region_prompt, - ) - config_manager = ConfigManager(config_dir=tmp_path, skip_validation=True) await config_manager._create_new_profile("custom") @@ -1461,10 +2050,11 @@ async def test_create_new_profile_cancelled( manager = ConfigManager(config_dir=tmp_path, skip_validation=True) manager.profile_manager = mock_profile_manager - monkeypatch.setattr( - ConfigManager.__module__ + ".inquirer.prompt", - lambda _questions: None, - ) + # Mock select_region_interactive to return None (cancelled) + async def mock_select_region_cancelled() -> None: + return None + + mock_profile_manager.select_region_interactive = mock_select_region_cancelled with pytest.raises(SystemExit): await manager._create_new_profile("dev")