diff --git a/src/pltr/cli.py b/src/pltr/cli.py index 86c0bed..ec32e9e 100644 --- a/src/pltr/cli.py +++ b/src/pltr/cli.py @@ -32,6 +32,7 @@ models, data_health, audit, + widgets, ) from pltr.commands.cp import cp_command @@ -103,6 +104,11 @@ name="audit", help="Audit log operations for compliance and security monitoring", ) +app.add_typer( + widgets.app, + name="widgets", + help="Manage widget sets, releases, and repositories", +) app.add_typer( admin.app, name="admin", diff --git a/src/pltr/commands/widgets.py b/src/pltr/commands/widgets.py new file mode 100644 index 0000000..76198aa --- /dev/null +++ b/src/pltr/commands/widgets.py @@ -0,0 +1,466 @@ +""" +Widget management commands for Foundry. +Provides access to widget sets, releases, repositories, and dev mode settings. + +Note: All Widgets APIs are in Private Beta. +""" + +from typing import Optional + +import typer +from rich.console import Console + +from ..auth.base import MissingCredentialsError, ProfileNotFoundError +from ..services.widgets import WidgetsService +from ..utils.completion import ( + complete_output_format, + complete_profile, + complete_rid, + cache_rid, +) +from ..utils.formatting import OutputFormatter +from ..utils.progress import SpinnerProgressTracker + +app = typer.Typer(help="Widget operations (Private Beta)") +dev_mode_app = typer.Typer(help="Dev mode settings management") +release_app = typer.Typer(help="Widget release management") +repository_app = typer.Typer(help="Widget repository management") + +app.add_typer(dev_mode_app, name="dev-mode") +app.add_typer(release_app, name="release") +app.add_typer(repository_app, name="repository") + +console = Console() +formatter = OutputFormatter(console) + + +# ===== Widget Set Commands ===== + + +@app.command("get") +def get_widget_set( + widget_set_rid: str = typer.Argument( + ..., + help="Widget set RID (e.g., ri.widgetregistry..widget-set.xxx)", + autocompletion=complete_rid, + ), + profile: Optional[str] = typer.Option( + None, + "--profile", + "-p", + help="Profile name", + autocompletion=complete_profile, + ), + format: str = typer.Option( + "table", + "--format", + "-f", + help="Output format (table, json, csv)", + autocompletion=complete_output_format, + ), + output: Optional[str] = typer.Option( + None, "--output", "-o", help="Output file path" + ), +) -> None: + """Get details of a widget set.""" + try: + cache_rid(widget_set_rid) + + service = WidgetsService(profile=profile) + + with SpinnerProgressTracker().track_spinner("Fetching widget set..."): + widget_set = service.get_widget_set(widget_set_rid) + + if output: + formatter.save_to_file([widget_set], output, format) + formatter.print_success(f"Widget set saved to {output}") + else: + formatter.display([widget_set], format) + + except (ProfileNotFoundError, MissingCredentialsError) as e: + formatter.print_error(f"Authentication error: {e}") + raise typer.Exit(1) + except Exception as e: + formatter.print_error(f"Failed to get widget set: {e}") + raise typer.Exit(1) + + +# ===== Dev Mode Commands ===== + + +@dev_mode_app.command("get") +def get_dev_mode_settings( + profile: Optional[str] = typer.Option( + None, + "--profile", + "-p", + help="Profile name", + autocompletion=complete_profile, + ), + format: str = typer.Option( + "table", + "--format", + "-f", + help="Output format (table, json, csv)", + autocompletion=complete_output_format, + ), + output: Optional[str] = typer.Option( + None, "--output", "-o", help="Output file path" + ), +) -> None: + """Get dev mode settings for the current user.""" + try: + service = WidgetsService(profile=profile) + + with SpinnerProgressTracker().track_spinner("Fetching dev mode settings..."): + settings = service.get_dev_mode_settings() + + if output: + formatter.save_to_file([settings], output, format) + formatter.print_success(f"Dev mode settings saved to {output}") + else: + formatter.display([settings], format) + + except (ProfileNotFoundError, MissingCredentialsError) as e: + formatter.print_error(f"Authentication error: {e}") + raise typer.Exit(1) + except Exception as e: + formatter.print_error(f"Failed to get dev mode settings: {e}") + raise typer.Exit(1) + + +@dev_mode_app.command("enable") +def enable_dev_mode( + profile: Optional[str] = typer.Option( + None, + "--profile", + "-p", + help="Profile name", + autocompletion=complete_profile, + ), + format: str = typer.Option( + "table", + "--format", + "-f", + help="Output format (table, json, csv)", + autocompletion=complete_output_format, + ), +) -> None: + """Enable dev mode for the current user.""" + try: + service = WidgetsService(profile=profile) + + with SpinnerProgressTracker().track_spinner("Enabling dev mode..."): + settings = service.enable_dev_mode() + + formatter.print_success("Dev mode enabled") + formatter.display([settings], format) + + except (ProfileNotFoundError, MissingCredentialsError) as e: + formatter.print_error(f"Authentication error: {e}") + raise typer.Exit(1) + except Exception as e: + formatter.print_error(f"Failed to enable dev mode: {e}") + raise typer.Exit(1) + + +@dev_mode_app.command("disable") +def disable_dev_mode( + profile: Optional[str] = typer.Option( + None, + "--profile", + "-p", + help="Profile name", + autocompletion=complete_profile, + ), + format: str = typer.Option( + "table", + "--format", + "-f", + help="Output format (table, json, csv)", + autocompletion=complete_output_format, + ), +) -> None: + """Disable dev mode for the current user.""" + try: + service = WidgetsService(profile=profile) + + with SpinnerProgressTracker().track_spinner("Disabling dev mode..."): + settings = service.disable_dev_mode() + + formatter.print_success("Dev mode disabled") + formatter.display([settings], format) + + except (ProfileNotFoundError, MissingCredentialsError) as e: + formatter.print_error(f"Authentication error: {e}") + raise typer.Exit(1) + except Exception as e: + formatter.print_error(f"Failed to disable dev mode: {e}") + raise typer.Exit(1) + + +@dev_mode_app.command("pause") +def pause_dev_mode( + profile: Optional[str] = typer.Option( + None, + "--profile", + "-p", + help="Profile name", + autocompletion=complete_profile, + ), + format: str = typer.Option( + "table", + "--format", + "-f", + help="Output format (table, json, csv)", + autocompletion=complete_output_format, + ), +) -> None: + """Pause dev mode for the current user.""" + try: + service = WidgetsService(profile=profile) + + with SpinnerProgressTracker().track_spinner("Pausing dev mode..."): + settings = service.pause_dev_mode() + + formatter.print_success("Dev mode paused") + formatter.display([settings], format) + + except (ProfileNotFoundError, MissingCredentialsError) as e: + formatter.print_error(f"Authentication error: {e}") + raise typer.Exit(1) + except Exception as e: + formatter.print_error(f"Failed to pause dev mode: {e}") + raise typer.Exit(1) + + +# ===== Release Commands ===== + + +@release_app.command("list") +def list_releases( + widget_set_rid: str = typer.Argument( + ..., + help="Widget set RID (e.g., ri.widgetregistry..widget-set.xxx)", + autocompletion=complete_rid, + ), + page_size: Optional[int] = typer.Option( + None, + "--page-size", + help="Number of results per page", + ), + profile: Optional[str] = typer.Option( + None, + "--profile", + "-p", + help="Profile name", + autocompletion=complete_profile, + ), + format: str = typer.Option( + "table", + "--format", + "-f", + help="Output format (table, json, csv)", + autocompletion=complete_output_format, + ), + output: Optional[str] = typer.Option( + None, "--output", "-o", help="Output file path" + ), +) -> None: + """List releases for a widget set.""" + try: + cache_rid(widget_set_rid) + + service = WidgetsService(profile=profile) + + with SpinnerProgressTracker().track_spinner("Fetching releases..."): + releases = service.list_releases( + widget_set_rid=widget_set_rid, + page_size=page_size, + ) + + if not releases: + formatter.print_info("No releases found for this widget set") + return + + formatter.print_info(f"Found {len(releases)} releases") + + if output: + formatter.save_to_file(releases, output, format) + formatter.print_success(f"Releases saved to {output}") + else: + formatter.display(releases, format) + + except (ProfileNotFoundError, MissingCredentialsError) as e: + formatter.print_error(f"Authentication error: {e}") + raise typer.Exit(1) + except Exception as e: + formatter.print_error(f"Failed to list releases: {e}") + raise typer.Exit(1) + + +@release_app.command("get") +def get_release( + widget_set_rid: str = typer.Argument( + ..., + help="Widget set RID (e.g., ri.widgetregistry..widget-set.xxx)", + autocompletion=complete_rid, + ), + release_version: str = typer.Argument( + ..., + help="Release version (semver, e.g., 1.2.0)", + ), + profile: Optional[str] = typer.Option( + None, + "--profile", + "-p", + help="Profile name", + autocompletion=complete_profile, + ), + format: str = typer.Option( + "table", + "--format", + "-f", + help="Output format (table, json, csv)", + autocompletion=complete_output_format, + ), + output: Optional[str] = typer.Option( + None, "--output", "-o", help="Output file path" + ), +) -> None: + """Get details of a specific release.""" + try: + cache_rid(widget_set_rid) + + service = WidgetsService(profile=profile) + + with SpinnerProgressTracker().track_spinner( + f"Fetching release {release_version}..." + ): + release = service.get_release( + widget_set_rid=widget_set_rid, + release_version=release_version, + ) + + if output: + formatter.save_to_file([release], output, format) + formatter.print_success(f"Release saved to {output}") + else: + formatter.display([release], format) + + except (ProfileNotFoundError, MissingCredentialsError) as e: + formatter.print_error(f"Authentication error: {e}") + raise typer.Exit(1) + except Exception as e: + formatter.print_error(f"Failed to get release: {e}") + raise typer.Exit(1) + + +@release_app.command("delete") +def delete_release( + widget_set_rid: str = typer.Argument( + ..., + help="Widget set RID (e.g., ri.widgetregistry..widget-set.xxx)", + autocompletion=complete_rid, + ), + release_version: str = typer.Argument( + ..., + help="Release version to delete (semver, e.g., 1.2.0)", + ), + yes: bool = typer.Option( + False, + "--yes", + "-y", + help="Skip confirmation prompt", + ), + profile: Optional[str] = typer.Option( + None, + "--profile", + "-p", + help="Profile name", + autocompletion=complete_profile, + ), +) -> None: + """Delete a specific release.""" + try: + cache_rid(widget_set_rid) + + if not yes: + confirm = typer.confirm( + f"Are you sure you want to delete release {release_version}?" + ) + if not confirm: + formatter.print_info("Operation cancelled") + raise typer.Exit(0) + + service = WidgetsService(profile=profile) + + with SpinnerProgressTracker().track_spinner( + f"Deleting release {release_version}..." + ): + service.delete_release( + widget_set_rid=widget_set_rid, + release_version=release_version, + ) + + formatter.print_success(f"Release {release_version} deleted successfully") + + except typer.Exit: + raise # Re-raise Exit exceptions (including cancellation) + except (ProfileNotFoundError, MissingCredentialsError) as e: + formatter.print_error(f"Authentication error: {e}") + raise typer.Exit(1) from e + except Exception as e: + formatter.print_error(f"Failed to delete release: {e}") + raise typer.Exit(1) from e + + +# ===== Repository Commands ===== + + +@repository_app.command("get") +def get_repository( + repository_rid: str = typer.Argument( + ..., + help="Repository RID (e.g., ri.stemma.main.repository.xxx)", + autocompletion=complete_rid, + ), + profile: Optional[str] = typer.Option( + None, + "--profile", + "-p", + help="Profile name", + autocompletion=complete_profile, + ), + format: str = typer.Option( + "table", + "--format", + "-f", + help="Output format (table, json, csv)", + autocompletion=complete_output_format, + ), + output: Optional[str] = typer.Option( + None, "--output", "-o", help="Output file path" + ), +) -> None: + """Get details of a widget repository.""" + try: + cache_rid(repository_rid) + + service = WidgetsService(profile=profile) + + with SpinnerProgressTracker().track_spinner("Fetching repository..."): + repository = service.get_repository(repository_rid) + + if output: + formatter.save_to_file([repository], output, format) + formatter.print_success(f"Repository saved to {output}") + else: + formatter.display([repository], format) + + except (ProfileNotFoundError, MissingCredentialsError) as e: + formatter.print_error(f"Authentication error: {e}") + raise typer.Exit(1) + except Exception as e: + formatter.print_error(f"Failed to get repository: {e}") + raise typer.Exit(1) diff --git a/src/pltr/services/widgets.py b/src/pltr/services/widgets.py new file mode 100644 index 0000000..39390aa --- /dev/null +++ b/src/pltr/services/widgets.py @@ -0,0 +1,293 @@ +""" +Widgets service wrapper for Foundry SDK. +Provides access to widget set operations for managing Foundry widgets. + +Note: All Widgets APIs are in Private Beta and require preview=True. +""" + +from typing import Any, Dict, List, Optional + +from .base import BaseService + + +class WidgetsService(BaseService): + """Service wrapper for Foundry Widgets operations.""" + + def _get_service(self) -> Any: + """Get the Foundry Widgets service.""" + return self.client.widgets + + # ===== DevModeSettings ===== + + def get_dev_mode_settings(self) -> Dict[str, Any]: + """ + Get the dev mode settings for the current user. + + Returns: + Dictionary containing dev mode settings: + - enabled: Whether dev mode is enabled + - paused: Whether dev mode is paused + - widgetSetSettings: Settings per widget set + + Raises: + RuntimeError: If the operation fails + + Example: + >>> service = WidgetsService() + >>> settings = service.get_dev_mode_settings() + """ + try: + settings = self.service.DevModeSettings.get(preview=True) + return self._serialize_response(settings) + except Exception as e: + raise RuntimeError(f"Failed to get dev mode settings: {e}") from e + + def enable_dev_mode(self) -> Dict[str, Any]: + """ + Enable dev mode for the current user. + + Returns: + Dictionary containing updated dev mode settings + + Raises: + RuntimeError: If the operation fails + + Example: + >>> service = WidgetsService() + >>> settings = service.enable_dev_mode() + """ + try: + settings = self.service.DevModeSettings.enable(preview=True) + return self._serialize_response(settings) + except Exception as e: + raise RuntimeError(f"Failed to enable dev mode: {e}") from e + + def disable_dev_mode(self) -> Dict[str, Any]: + """ + Disable dev mode for the current user. + + Returns: + Dictionary containing updated dev mode settings + + Raises: + RuntimeError: If the operation fails + + Example: + >>> service = WidgetsService() + >>> settings = service.disable_dev_mode() + """ + try: + settings = self.service.DevModeSettings.disable(preview=True) + return self._serialize_response(settings) + except Exception as e: + raise RuntimeError(f"Failed to disable dev mode: {e}") from e + + def pause_dev_mode(self) -> Dict[str, Any]: + """ + Pause dev mode for the current user. + + Returns: + Dictionary containing updated dev mode settings + + Raises: + RuntimeError: If the operation fails + + Example: + >>> service = WidgetsService() + >>> settings = service.pause_dev_mode() + """ + try: + settings = self.service.DevModeSettings.pause(preview=True) + return self._serialize_response(settings) + except Exception as e: + raise RuntimeError(f"Failed to pause dev mode: {e}") from e + + # ===== WidgetSet ===== + + def get_widget_set(self, widget_set_rid: str) -> Dict[str, Any]: + """ + Get a widget set by RID. + + Args: + widget_set_rid: Widget set Resource Identifier + Expected format: ri.widgetregistry..widget-set. + + Returns: + Dictionary containing widget set details: + - rid: Widget set RID + - name: Widget set name + - widgets: List of widgets in the set + + Raises: + RuntimeError: If the operation fails + + Example: + >>> service = WidgetsService() + >>> widget_set = service.get_widget_set( + ... "ri.widgetregistry..widget-set.abc123" + ... ) + """ + try: + widget_set = self.service.WidgetSet.get( + widget_set_rid=widget_set_rid, + preview=True, + ) + return self._serialize_response(widget_set) + except Exception as e: + raise RuntimeError( + f"Failed to get widget set '{widget_set_rid}': {e}" + ) from e + + # ===== Releases ===== + + def list_releases( + self, + widget_set_rid: str, + page_size: Optional[int] = None, + ) -> List[Dict[str, Any]]: + """ + List releases for a widget set. + + Args: + widget_set_rid: Widget set Resource Identifier + page_size: Number of results per page (optional) + + Returns: + List of release dictionaries containing: + - version: Release version (semver) + - createdAt: Creation timestamp + - createdBy: User who created the release + + Raises: + RuntimeError: If the operation fails + + Example: + >>> service = WidgetsService() + >>> releases = service.list_releases( + ... widget_set_rid="ri.widgetregistry..widget-set.abc123" + ... ) + """ + try: + kwargs: Dict[str, Any] = {} + if page_size is not None: + kwargs["page_size"] = page_size + + releases = self.service.WidgetSet.Release.list( + widget_set_rid=widget_set_rid, + preview=True, + **kwargs, + ) + return [self._serialize_response(r) for r in releases] + except Exception as e: + raise RuntimeError( + f"Failed to list releases for '{widget_set_rid}': {e}" + ) from e + + def get_release( + self, + widget_set_rid: str, + release_version: str, + ) -> Dict[str, Any]: + """ + Get a specific release of a widget set. + + Args: + widget_set_rid: Widget set Resource Identifier + release_version: Semantic version of the release (e.g., "1.2.0") + + Returns: + Dictionary containing release details: + - version: Release version + - createdAt: Creation timestamp + - widgets: Widgets included in this release + + Raises: + RuntimeError: If the operation fails + + Example: + >>> service = WidgetsService() + >>> release = service.get_release( + ... widget_set_rid="ri.widgetregistry..widget-set.abc123", + ... release_version="1.0.0" + ... ) + """ + try: + release = self.service.WidgetSet.Release.get( + widget_set_rid=widget_set_rid, + release_version=release_version, + preview=True, + ) + return self._serialize_response(release) + except Exception as e: + raise RuntimeError( + f"Failed to get release '{release_version}' for '{widget_set_rid}': {e}" + ) from e + + def delete_release( + self, + widget_set_rid: str, + release_version: str, + ) -> None: + """ + Delete a specific release of a widget set. + + Args: + widget_set_rid: Widget set Resource Identifier + release_version: Semantic version of the release to delete + + Raises: + RuntimeError: If the operation fails + + Example: + >>> service = WidgetsService() + >>> service.delete_release( + ... widget_set_rid="ri.widgetregistry..widget-set.abc123", + ... release_version="1.0.0" + ... ) + """ + try: + self.service.WidgetSet.Release.delete( + widget_set_rid=widget_set_rid, + release_version=release_version, + preview=True, + ) + except Exception as e: + raise RuntimeError( + f"Failed to delete release '{release_version}' for '{widget_set_rid}': {e}" + ) from e + + # ===== Repository ===== + + def get_repository(self, repository_rid: str) -> Dict[str, Any]: + """ + Get a widget repository by RID. + + Args: + repository_rid: Repository Resource Identifier + Expected format: ri.stemma.main.repository. + + Returns: + Dictionary containing repository details: + - rid: Repository RID + - name: Repository name + - widgetSetRid: Associated widget set RID + + Raises: + RuntimeError: If the operation fails + + Example: + >>> service = WidgetsService() + >>> repo = service.get_repository( + ... "ri.stemma.main.repository.abc123" + ... ) + """ + try: + repository = self.service.Repository.get( + repository_rid=repository_rid, + preview=True, + ) + return self._serialize_response(repository) + except Exception as e: + raise RuntimeError( + f"Failed to get repository '{repository_rid}': {e}" + ) from e diff --git a/tests/test_commands/test_widgets.py b/tests/test_commands/test_widgets.py new file mode 100644 index 0000000..62dacb2 --- /dev/null +++ b/tests/test_commands/test_widgets.py @@ -0,0 +1,560 @@ +""" +Tests for widget management commands. +""" + +from unittest.mock import Mock, patch + +import pytest +from typer.testing import CliRunner + +from pltr.cli import app + + +class TestWidgetsCommands: + """Test widgets CLI commands.""" + + @pytest.fixture + def runner(self): + """Create CLI test runner.""" + return CliRunner() + + @pytest.fixture + def mock_service(self): + """Create mock WidgetsService.""" + with patch("pltr.commands.widgets.WidgetsService") as MockService: + mock_svc = Mock() + MockService.return_value = mock_svc + yield mock_svc + + # ===== Widget Set Get Command Tests ===== + + def test_get_widget_set_success(self, runner, mock_service) -> None: + """Test successful get widget set command.""" + # Setup + widget_set_result = { + "rid": "ri.widgetregistry..widget-set.abc123", + "name": "my-widgets", + "widgets": [{"id": "widget1", "name": "Widget One"}], + } + mock_service.get_widget_set.return_value = widget_set_result + + result = runner.invoke( + app, + [ + "widgets", + "get", + "ri.widgetregistry..widget-set.abc123", + "--format", + "json", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.get_widget_set.assert_called_once_with( + "ri.widgetregistry..widget-set.abc123" + ) + + def test_get_widget_set_error(self, runner, mock_service) -> None: + """Test get widget set command with error.""" + # Setup + mock_service.get_widget_set.side_effect = RuntimeError("Widget set not found") + + result = runner.invoke( + app, + [ + "widgets", + "get", + "ri.widgetregistry..widget-set.invalid", + ], + ) + + # Assert + assert result.exit_code == 1 + assert "Failed to get widget set" in result.stdout + + # ===== Dev Mode Get Command Tests ===== + + def test_dev_mode_get_success(self, runner, mock_service) -> None: + """Test successful get dev mode settings command.""" + # Setup + settings_result = { + "enabled": True, + "paused": False, + "widgetSetSettings": {}, + } + mock_service.get_dev_mode_settings.return_value = settings_result + + result = runner.invoke( + app, + [ + "widgets", + "dev-mode", + "get", + "--format", + "json", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.get_dev_mode_settings.assert_called_once() + + def test_dev_mode_get_error(self, runner, mock_service) -> None: + """Test get dev mode settings command with error.""" + # Setup + mock_service.get_dev_mode_settings.side_effect = RuntimeError("API error") + + result = runner.invoke( + app, + [ + "widgets", + "dev-mode", + "get", + ], + ) + + # Assert + assert result.exit_code == 1 + assert "Failed to get dev mode settings" in result.stdout + + # ===== Dev Mode Enable Command Tests ===== + + def test_dev_mode_enable_success(self, runner, mock_service) -> None: + """Test successful enable dev mode command.""" + # Setup + settings_result = { + "enabled": True, + "paused": False, + } + mock_service.enable_dev_mode.return_value = settings_result + + result = runner.invoke( + app, + [ + "widgets", + "dev-mode", + "enable", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.enable_dev_mode.assert_called_once() + assert "enabled" in result.stdout.lower() + + def test_dev_mode_enable_error(self, runner, mock_service) -> None: + """Test enable dev mode command with error.""" + # Setup + mock_service.enable_dev_mode.side_effect = RuntimeError("Permission denied") + + result = runner.invoke( + app, + [ + "widgets", + "dev-mode", + "enable", + ], + ) + + # Assert + assert result.exit_code == 1 + assert "Failed to enable dev mode" in result.stdout + + # ===== Dev Mode Disable Command Tests ===== + + def test_dev_mode_disable_success(self, runner, mock_service) -> None: + """Test successful disable dev mode command.""" + # Setup + settings_result = { + "enabled": False, + "paused": False, + } + mock_service.disable_dev_mode.return_value = settings_result + + result = runner.invoke( + app, + [ + "widgets", + "dev-mode", + "disable", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.disable_dev_mode.assert_called_once() + assert "disabled" in result.stdout.lower() + + # ===== Dev Mode Pause Command Tests ===== + + def test_dev_mode_pause_success(self, runner, mock_service) -> None: + """Test successful pause dev mode command.""" + # Setup + settings_result = { + "enabled": True, + "paused": True, + } + mock_service.pause_dev_mode.return_value = settings_result + + result = runner.invoke( + app, + [ + "widgets", + "dev-mode", + "pause", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.pause_dev_mode.assert_called_once() + assert "paused" in result.stdout.lower() + + # ===== Release List Command Tests ===== + + def test_release_list_success(self, runner, mock_service) -> None: + """Test successful list releases command.""" + # Setup + releases_result = [ + {"version": "1.0.0", "createdAt": "2024-01-01T00:00:00Z"}, + {"version": "1.1.0", "createdAt": "2024-02-01T00:00:00Z"}, + ] + mock_service.list_releases.return_value = releases_result + + result = runner.invoke( + app, + [ + "widgets", + "release", + "list", + "ri.widgetregistry..widget-set.abc123", + "--format", + "json", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.list_releases.assert_called_once_with( + widget_set_rid="ri.widgetregistry..widget-set.abc123", + page_size=None, + ) + + def test_release_list_empty(self, runner, mock_service) -> None: + """Test list releases command with no results.""" + # Setup + mock_service.list_releases.return_value = [] + + result = runner.invoke( + app, + [ + "widgets", + "release", + "list", + "ri.widgetregistry..widget-set.abc123", + ], + ) + + # Assert + assert result.exit_code == 0 + assert "No releases found" in result.stdout + + def test_release_list_with_page_size(self, runner, mock_service) -> None: + """Test list releases command with page size.""" + # Setup + mock_service.list_releases.return_value = [{"version": "1.0.0"}] + + result = runner.invoke( + app, + [ + "widgets", + "release", + "list", + "ri.widgetregistry..widget-set.abc123", + "--page-size", + "10", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.list_releases.assert_called_once_with( + widget_set_rid="ri.widgetregistry..widget-set.abc123", + page_size=10, + ) + + def test_release_list_error(self, runner, mock_service) -> None: + """Test list releases command with error.""" + # Setup + mock_service.list_releases.side_effect = RuntimeError("Widget set not found") + + result = runner.invoke( + app, + [ + "widgets", + "release", + "list", + "ri.widgetregistry..widget-set.invalid", + ], + ) + + # Assert + assert result.exit_code == 1 + assert "Failed to list releases" in result.stdout + + # ===== Release Get Command Tests ===== + + def test_release_get_success(self, runner, mock_service) -> None: + """Test successful get release command.""" + # Setup + release_result = { + "version": "1.0.0", + "createdAt": "2024-01-01T00:00:00Z", + "widgets": [{"id": "widget1"}], + } + mock_service.get_release.return_value = release_result + + result = runner.invoke( + app, + [ + "widgets", + "release", + "get", + "ri.widgetregistry..widget-set.abc123", + "1.0.0", + "--format", + "json", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.get_release.assert_called_once_with( + widget_set_rid="ri.widgetregistry..widget-set.abc123", + release_version="1.0.0", + ) + + def test_release_get_error(self, runner, mock_service) -> None: + """Test get release command with error.""" + # Setup + mock_service.get_release.side_effect = RuntimeError("Release not found") + + result = runner.invoke( + app, + [ + "widgets", + "release", + "get", + "ri.widgetregistry..widget-set.abc123", + "99.99.99", + ], + ) + + # Assert + assert result.exit_code == 1 + assert "Failed to get release" in result.stdout + + # ===== Release Delete Command Tests ===== + + def test_release_delete_success(self, runner, mock_service) -> None: + """Test successful delete release command.""" + # Setup + mock_service.delete_release.return_value = None + + result = runner.invoke( + app, + [ + "widgets", + "release", + "delete", + "ri.widgetregistry..widget-set.abc123", + "1.0.0", + "--yes", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.delete_release.assert_called_once_with( + widget_set_rid="ri.widgetregistry..widget-set.abc123", + release_version="1.0.0", + ) + assert "deleted" in result.stdout.lower() + + def test_release_delete_cancelled(self, runner, mock_service) -> None: + """Test delete release command cancelled by user.""" + result = runner.invoke( + app, + [ + "widgets", + "release", + "delete", + "ri.widgetregistry..widget-set.abc123", + "1.0.0", + ], + input="n\n", + ) + + # Assert + assert result.exit_code == 0 + mock_service.delete_release.assert_not_called() + assert "cancelled" in result.stdout.lower() + + def test_release_delete_error(self, runner, mock_service) -> None: + """Test delete release command with error.""" + # Setup + mock_service.delete_release.side_effect = RuntimeError("Cannot delete release") + + result = runner.invoke( + app, + [ + "widgets", + "release", + "delete", + "ri.widgetregistry..widget-set.abc123", + "1.0.0", + "--yes", + ], + ) + + # Assert + assert result.exit_code == 1 + assert "Failed to delete release" in result.stdout + + # ===== Repository Get Command Tests ===== + + def test_repository_get_success(self, runner, mock_service) -> None: + """Test successful get repository command.""" + # Setup + repository_result = { + "rid": "ri.stemma.main.repository.abc123", + "name": "my-widget-repo", + "widgetSetRid": "ri.widgetregistry..widget-set.def456", + } + mock_service.get_repository.return_value = repository_result + + result = runner.invoke( + app, + [ + "widgets", + "repository", + "get", + "ri.stemma.main.repository.abc123", + "--format", + "json", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_service.get_repository.assert_called_once_with( + "ri.stemma.main.repository.abc123" + ) + + def test_repository_get_error(self, runner, mock_service) -> None: + """Test get repository command with error.""" + # Setup + mock_service.get_repository.side_effect = RuntimeError("Repository not found") + + result = runner.invoke( + app, + [ + "widgets", + "repository", + "get", + "ri.stemma.main.repository.invalid", + ], + ) + + # Assert + assert result.exit_code == 1 + assert "Failed to get repository" in result.stdout + + # ===== Help Command Tests ===== + + def test_help_command(self, runner) -> None: + """Test help output for commands.""" + # Test main help + result = runner.invoke(app, ["widgets", "--help"]) + assert result.exit_code == 0 + assert "widgets" in result.stdout.lower() + + # Test dev-mode help + result = runner.invoke(app, ["widgets", "dev-mode", "--help"]) + assert result.exit_code == 0 + assert "dev" in result.stdout.lower() + + # Test release help + result = runner.invoke(app, ["widgets", "release", "--help"]) + assert result.exit_code == 0 + assert "release" in result.stdout.lower() + + # Test repository help + result = runner.invoke(app, ["widgets", "repository", "--help"]) + assert result.exit_code == 0 + assert "repository" in result.stdout.lower() + + # ===== File Output Tests ===== + + def test_get_widget_set_with_file_output( + self, runner, mock_service, tmp_path + ) -> None: + """Test get widget set command with file output.""" + # Setup + widget_set_result = { + "rid": "ri.widgetregistry..widget-set.abc123", + "name": "my-widgets", + } + mock_service.get_widget_set.return_value = widget_set_result + output_file = tmp_path / "widget_set.json" + + with patch("pltr.commands.widgets.formatter") as mock_formatter: + result = runner.invoke( + app, + [ + "widgets", + "get", + "ri.widgetregistry..widget-set.abc123", + "--output", + str(output_file), + "--format", + "json", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_formatter.save_to_file.assert_called_once_with( + [widget_set_result], str(output_file), "json" + ) + + def test_release_list_with_file_output( + self, runner, mock_service, tmp_path + ) -> None: + """Test list releases command with file output.""" + # Setup + releases_result = [{"version": "1.0.0"}] + mock_service.list_releases.return_value = releases_result + output_file = tmp_path / "releases.json" + + with patch("pltr.commands.widgets.formatter") as mock_formatter: + result = runner.invoke( + app, + [ + "widgets", + "release", + "list", + "ri.widgetregistry..widget-set.abc123", + "--output", + str(output_file), + "--format", + "json", + ], + ) + + # Assert + assert result.exit_code == 0 + mock_formatter.save_to_file.assert_called_once_with( + releases_result, str(output_file), "json" + )