Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
159 changes: 159 additions & 0 deletions src/workato_platform_cli/cli/commands/profiles.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import json
import os

from pathlib import Path
from typing import Any

import asyncclick as click
Expand Down Expand Up @@ -345,6 +346,164 @@ async def status(
click.echo(" 💡 Or set WORKATO_API_TOKEN environment variable")


def _format_file_path(file_path: Path) -> str:
"""Format file path for display, showing relative to current directory."""
try:
rel_path = file_path.relative_to(Path.cwd())
return f"./{rel_path}"
except ValueError:
# File is outside current directory (shouldn't happen with cwd search)
return str(file_path)


def _update_workatoenv_files(old_name: str, new_name: str) -> list[Path]:
"""Find and update all .workatoenv files that reference the old profile name.

Searches recursively from current directory.
Returns list of updated file paths.
"""
updated_files = []
current_dir = Path.cwd()

# Search from current directory for .workatoenv files
for workatoenv_file in current_dir.rglob(".workatoenv"):
try:
# Open for reading and writing
with open(workatoenv_file, "r+") as f:
data = json.load(f)

# Check if profile field matches old name
if data.get("profile") == old_name:
# Update to new name
data["profile"] = new_name

# Write back (truncate and write from beginning)
f.seek(0)
f.truncate()
json.dump(data, f, indent=2)
f.write("\n") # Add trailing newline

updated_files.append(workatoenv_file)
except (OSError, json.JSONDecodeError):
# Skip files we can't read or parse
continue

return updated_files


@profiles.command()
@click.argument("old_name")
@click.argument("new_name")
@click.option(
"--output-mode",
type=click.Choice(["table", "json"]),
default="table",
help="Output format: table (default) or json",
)
@click.option(
"--yes",
is_flag=True,
help="Skip confirmation prompt",
)
@handle_cli_exceptions
@inject
async def rename(
old_name: str,
new_name: str,
output_mode: str = "table",
yes: bool = False,
config_manager: ConfigManager = Provide[Container.config_manager],
) -> None:
"""Rename a profile"""
# Check if old profile exists
old_profile = config_manager.profile_manager.get_profile(old_name)
if not old_profile:
if output_mode == "json":
error_msg = f"Profile '{old_name}' not found"
output_data: dict[str, Any] = {"status": "error", "error": error_msg}
click.echo(json.dumps(output_data))
else:
click.echo(f"❌ Profile '{old_name}' not found")
click.echo("💡 Use 'workato profiles list' to see available profiles")
return

# Check if new name already exists
if config_manager.profile_manager.get_profile(new_name):
if output_mode == "json":
error_msg = f"Profile '{new_name}' already exists"
output_data = {"status": "error", "error": error_msg}
click.echo(json.dumps(output_data))
else:
click.echo(f"❌ Profile '{new_name}' already exists")
click.echo(
"💡 Choose a different name or delete the existing profile first"
)
return

# Show confirmation prompt (skip in JSON mode or if --yes flag)
if (
not yes
and output_mode != "json"
and not click.confirm(f"Rename profile '{old_name}' to '{new_name}'?")
):
click.echo("❌ Rename cancelled")
return

# Get the token from keyring
old_token = config_manager.profile_manager._get_token_from_keyring(old_name)

# Create new profile with same data and token
try:
config_manager.profile_manager.set_profile(new_name, old_profile, old_token)
except ValueError as e:
if output_mode == "json":
output_data = {"status": "error", "error": str(e)}
click.echo(json.dumps(output_data))
else:
click.echo(f"❌ Failed to create new profile: {e}")
return

# If old profile was current, set new profile as current
current_profile = config_manager.profile_manager.get_current_profile_name()
was_current = current_profile == old_name
if was_current:
config_manager.profile_manager.set_current_profile(new_name)

# Delete old profile
config_manager.profile_manager.delete_profile(old_name)

# Update all .workatoenv files that reference the old profile
if output_mode == "table":
click.echo("🔄 Updating project configurations...")
updated_files = _update_workatoenv_files(old_name, new_name)

# JSON output mode
if output_mode == "json":
output_data = {
"status": "success",
"old_name": old_name,
"new_name": new_name,
"was_current_profile": was_current,
"updated_files": [str(f) for f in updated_files],
"updated_files_count": len(updated_files),
}
click.echo(json.dumps(output_data))
return

# Table output mode (default)
click.echo("✅ Profile renamed successfully")
if was_current:
click.echo(f"✅ Set '{new_name}' as the active profile")

# Display updated files
if not updated_files:
return

click.echo(f"✅ Updated {len(updated_files)} project configuration(s)")
for file_path in updated_files:
click.echo(f" • {_format_file_path(file_path)}")


@profiles.command()
@click.argument("profile_name")
@click.confirmation_option(prompt="Are you sure you want to delete this profile?")
Expand Down
80 changes: 79 additions & 1 deletion tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
"""Pytest configuration and shared fixtures."""

import json
import tempfile

from collections.abc import Generator
from collections.abc import Callable, Generator
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, Mock, patch
Expand Down Expand Up @@ -144,3 +145,80 @@ def prevent_keyring_errors() -> None:
minimal_keyring.delete_password.return_value = None

sys.modules["keyring"] = minimal_keyring


# Shared test helpers


def parse_json_output(capsys: pytest.CaptureFixture[str]) -> dict[str, Any]:
"""Parse JSON output from capsys."""
output = capsys.readouterr().out
result: dict[str, Any] = json.loads(output)
return result


def create_workatoenv_file(
tmp_path: Path,
dir_name: str,
profile: str,
project_id: int = 123,
**extra_fields: Any,
) -> Path:
"""Create a test .workatoenv file.

Args:
tmp_path: Temporary directory path
dir_name: Name of the project directory to create
profile: Profile name to set in the workatoenv file
project_id: Project ID (default: 123)
**extra_fields: Additional fields to include in the workatoenv file
"""
project_dir = tmp_path / dir_name
project_dir.mkdir()
workatoenv = project_dir / ".workatoenv"

data = {"project_id": project_id, "profile": profile}
data.update(extra_fields)

workatoenv.write_text(json.dumps(data))
return workatoenv


def _make_workatoenv_updater(tmp_path: Path) -> Callable[[str, str], list[Path]]:
"""Create a mock _update_workatoenv_files function for testing.

Returns a function that searches tmp_path instead of home directory.
"""

def mock_update(old_name: str, new_name: str) -> list[Path]:
updated_files = []
for workatoenv_file in tmp_path.rglob(".workatoenv"):
try:
with open(workatoenv_file, "r+") as f:
data = json.load(f)
if data.get("profile") == old_name:
data["profile"] = new_name
f.seek(0)
f.truncate()
json.dump(data, f, indent=2)
f.write("\n")
updated_files.append(workatoenv_file)
except (OSError, json.JSONDecodeError):
continue
return updated_files

return mock_update


def mock_workatoenv_updates(tmp_path: Path) -> Any:
"""Context manager for mocking workatoenv file updates.

Use this to mock the _update_workatoenv_files function in profiles module.
Must import profiles_module in your test file to use this helper.
"""
from workato_platform_cli.cli.commands import profiles as profiles_module

mock_update = _make_workatoenv_updater(tmp_path)
return patch.object(
profiles_module, "_update_workatoenv_files", side_effect=mock_update
)
Loading