diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1332d70..24800a2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,6 +43,7 @@ jobs: run: uv run pytest - name: Upload coverage to Coveralls + if: matrix.python-version == '3.13' uses: coverallsapp/github-action@v2 with: github-token: ${{ secrets.GITHUB_TOKEN }} diff --git a/CLAUDE.md b/CLAUDE.md index 0830023..a562020 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -270,9 +270,30 @@ def test_update_service_status(): ## Code Style Guidelines -### Comments +### Comments and Docstrings -Remove obvious comments that repeat what the code already says. Only add comments for complex business logic or non-obvious decisions. Test names should be self-documenting, no docstrings needed. +**CRITICAL: Remove ALL obvious docstrings that just repeat the function signature.** + +Functions with clear names don't need docstrings. Only add docstrings for complex business logic or non-obvious behavior. + +**Bad (obvious docstring):** + +```python +def build_log_group_arn(region: str, account_id: str, log_group: str) -> str: + """Build CloudWatch log group ARN from components.""" + return f"arn:aws:logs:{region}:{account_id}:log-group:{log_group}" +``` + +**Good (no docstring needed):** + +```python +def build_log_group_arn(region: str, account_id: str, log_group: str) -> str: + return f"arn:aws:logs:{region}:{account_id}:log-group:{log_group}" +``` + +Test names should be self-documenting, no docstrings needed. + +Remove obvious comments that repeat what the code already says. Only add comments for complex business logic or non-obvious decisions. **Bad:** diff --git a/pyproject.toml b/pyproject.toml index d28c9be..7b73e7e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lazy-ecs" -version = "0.7.2" +version = "0.7.3" description = "A CLI tool for working with AWS services" readme = "README.md" authors = [ diff --git a/src/lazy_ecs/__init__.py b/src/lazy_ecs/__init__.py index 272d974..58b5a80 100644 --- a/src/lazy_ecs/__init__.py +++ b/src/lazy_ecs/__init__.py @@ -12,9 +12,7 @@ from mypy_boto3_sts.client import STSClient from .aws_service import ECSService -from .core.navigation import handle_navigation, parse_selection -from .core.types import TaskDetails -from .core.utils import show_spinner +from .core.app import navigate_clusters from .ui import ECSNavigator console = Console() @@ -37,7 +35,7 @@ def main() -> None: ecs_service = ECSService(ecs_client, sts_client, logs_client, cloudwatch_client) navigator = ECSNavigator(ecs_service) - _navigate_clusters(navigator, ecs_service) + navigate_clusters(navigator, ecs_service) except Exception as e: console.print(f"\n❌ Error: {e}", style="red") @@ -88,130 +86,5 @@ def _create_cloudwatch_client(profile_name: str | None) -> "CloudWatchClient": return session.client("cloudwatch", config=config) -def _navigate_clusters(navigator: ECSNavigator, ecs_service: ECSService) -> None: - """Handle cluster-level navigation with back support.""" - while True: - selected_cluster = navigator.select_cluster() - - if not selected_cluster: - console.print("\n❌ No cluster selected. Goodbye!", style="yellow") - break - - console.print(f"\n✅ Selected cluster: {selected_cluster}", style="green") - - if _navigate_services(navigator, ecs_service, selected_cluster): - continue # Back to cluster selection - break # Exit was chosen - - -def _navigate_services(navigator: ECSNavigator, ecs_service: ECSService, cluster_name: str) -> bool: - """Handle service-level navigation. Returns True if back was chosen, False if exit.""" - service_selection = navigator.select_service(cluster_name) - - # Handle navigation responses (back/exit) - should_continue, should_exit = handle_navigation(service_selection) - if not should_continue: - return not should_exit # True for back, False for exit - - selection_type, selected_service, _ = parse_selection(service_selection) - if selection_type != "service": - return True - - console.print(f"\n✅ Selected service: {selected_service}", style="green") - - while True: - selection = navigator.select_service_action(cluster_name, selected_service) - - # Handle navigation responses - should_continue, should_exit = handle_navigation(selection) - if not should_continue: - return not should_exit # True for back, False for exit - - selection_type, action_name, task_arn = parse_selection(selection) - if selection_type == "task" and action_name == "show_details": - with show_spinner(): - task_details = ecs_service.get_task_details(cluster_name, selected_service, task_arn) - if task_details: - navigator.display_task_details(task_details) - # Navigate to task features, handle back navigation - if _handle_task_features(navigator, cluster_name, task_arn, task_details, selected_service): - continue # Back to service selection - return False # Exit was chosen - console.print(f"\n⚠️ Could not fetch task details for {task_arn}", style="yellow") - - elif selection_type == "action" and action_name == "force_deployment": - navigator.handle_force_deployment(cluster_name, selected_service) - # Continue the loop to show the menu again - - elif selection_type == "action" and action_name == "show_events": - navigator.show_service_events(cluster_name, selected_service) - # Continue the loop to show the menu again - - elif selection_type == "action" and action_name == "show_metrics": - navigator.show_service_metrics(cluster_name, selected_service) - # Continue the loop to show the menu again - - elif selection_type == "action" and action_name == "open_console": - navigator.open_service_in_console(cluster_name, selected_service) - # Continue the loop to show the menu again - - -_CONTAINER_ACTIONS = { - "tail_logs": lambda nav, cluster, task_arn, container: nav.show_container_logs_live_tail( - cluster, - task_arn, - container, - ), - "show_env": lambda nav, cluster, task_arn, container: nav.show_container_environment_variables( - cluster, - task_arn, - container, - ), - "show_secrets": lambda nav, cluster, task_arn, container: nav.show_container_secrets(cluster, task_arn, container), - "show_ports": lambda nav, cluster, task_arn, container: nav.show_container_port_mappings( - cluster, - task_arn, - container, - ), - "show_volumes": lambda nav, cluster, task_arn, container: nav.show_container_volume_mounts( - cluster, - task_arn, - container, - ), -} - -_TASK_ACTIONS = { - "show_history": lambda nav, cluster, service, _task_arn, _task_details: nav.show_task_history(cluster, service), - "show_details": lambda nav, _cluster, _service, _task_arn, task_details: nav.display_task_details(task_details), - "compare_definitions": lambda nav, _cluster, _service, _task_arn, task_details: nav.show_task_definition_comparison( - task_details, - ), - "open_console": lambda nav, cluster, _service, task_arn, _task_details: nav.open_task_in_console(cluster, task_arn), -} - - -def _handle_task_features( - navigator: ECSNavigator, - cluster_name: str, - task_arn: str, - task_details: TaskDetails | None, - service_name: str, -) -> bool: - """Handle task feature selection and execution. Returns True if back was chosen, False if exit.""" - while True: - selection = navigator.select_task_feature(task_details) - - should_continue, should_exit = handle_navigation(selection) - if not should_continue: - return not should_exit - - selection_type, action_name, container_name = parse_selection(selection) - - if selection_type == "container_action" and action_name in _CONTAINER_ACTIONS: - _CONTAINER_ACTIONS[action_name](navigator, cluster_name, task_arn, container_name) - elif selection_type == "task_action" and action_name in _TASK_ACTIONS: - _TASK_ACTIONS[action_name](navigator, cluster_name, service_name, task_arn, task_details) - - if __name__ == "__main__": main() diff --git a/src/lazy_ecs/core/app.py b/src/lazy_ecs/core/app.py new file mode 100644 index 0000000..5178194 --- /dev/null +++ b/src/lazy_ecs/core/app.py @@ -0,0 +1,203 @@ +"""Main application logic for lazy-ecs CLI.""" + +from typing import TYPE_CHECKING + +from rich.console import Console + +from ..aws_service import ECSService +from ..core.navigation import handle_navigation, parse_selection +from ..core.types import TaskDetails +from ..core.utils import show_spinner +from ..ui import ECSNavigator + +if TYPE_CHECKING: + from collections.abc import Callable + +console = Console() + + +def navigate_clusters(navigator: ECSNavigator, ecs_service: ECSService) -> None: + """Handle cluster-level navigation with back support.""" + while True: + selected_cluster = navigator.select_cluster() + + if not selected_cluster: + console.print("\n❌ No cluster selected. Goodbye!", style="yellow") + break + + console.print(f"\n✅ Selected cluster: {selected_cluster}", style="green") + + if navigate_services(navigator, ecs_service, selected_cluster): + continue # Back to cluster selection + break # Exit was chosen + + +def navigate_services(navigator: ECSNavigator, ecs_service: ECSService, cluster_name: str) -> bool: + """Handle service-level navigation. Returns True if back was chosen, False if exit.""" + service_selection = navigator.select_service(cluster_name) + + # Handle navigation responses (back/exit) + should_continue, should_exit = handle_navigation(service_selection) + if not should_continue: + return not should_exit # True for back, False for exit + + selection_type, selected_service, _ = parse_selection(service_selection) + if selection_type != "service": + return True + + console.print(f"\n✅ Selected service: {selected_service}", style="green") + + while True: + selection = navigator.select_service_action(cluster_name, selected_service) + + # Handle navigation responses + should_continue, should_exit = handle_navigation(selection) + if not should_continue: + return not should_exit # True for back, False for exit + + selection_type, action_name, task_arn = parse_selection(selection) + + if selection_type == "task" and action_name == "show_details": + if not handle_task_selection(navigator, ecs_service, cluster_name, selected_service, task_arn): + return False # Exit was chosen + continue # Back to service selection + + if selection_type == "action": + dispatch_service_action(navigator, cluster_name, selected_service, action_name) + + +def handle_task_selection( + navigator: ECSNavigator, + ecs_service: ECSService, + cluster_name: str, + service_name: str, + task_arn: str, +) -> bool: + """Handle task selection and navigation. Returns True if back was chosen, False if exit.""" + with show_spinner(): + task_details = ecs_service.get_task_details(cluster_name, service_name, task_arn) + + if not task_details: + console.print(f"\n⚠️ Could not fetch task details for {task_arn}", style="yellow") + return True + + navigator.display_task_details(task_details) + return handle_task_features(navigator, cluster_name, task_arn, task_details, service_name) + + +def dispatch_service_action(navigator: ECSNavigator, cluster_name: str, service_name: str, action_name: str) -> None: + """Dispatch service action to appropriate handler.""" + service_actions = get_service_action_handlers() + if action_name in service_actions: + service_actions[action_name](navigator, cluster_name, service_name) + + +def handle_task_features( + navigator: ECSNavigator, + cluster_name: str, + task_arn: str, + task_details: TaskDetails | None, + service_name: str, +) -> bool: + """Handle task feature selection and execution. Returns True if back was chosen, False if exit.""" + while True: + selection = navigator.select_task_feature(task_details) + + should_continue, should_exit = handle_navigation(selection) + if not should_continue: + return not should_exit + + selection_type, action_name, container_name = parse_selection(selection) + + if selection_type == "container_action": + dispatch_container_action(navigator, cluster_name, task_arn, container_name, action_name) + elif selection_type == "task_action": + dispatch_task_action(navigator, cluster_name, service_name, task_arn, task_details, action_name) + + +def dispatch_container_action( + navigator: ECSNavigator, + cluster_name: str, + task_arn: str, + container_name: str, + action_name: str, +) -> bool: + """Dispatch container action to appropriate handler. Returns True if action was found and executed.""" + container_actions = get_container_action_handlers() + if action_name in container_actions: + container_actions[action_name](navigator, cluster_name, task_arn, container_name) + return True + return False + + +def dispatch_task_action( + navigator: ECSNavigator, + cluster_name: str, + service_name: str, + task_arn: str, + task_details: TaskDetails | None, + action_name: str, +) -> bool: + """Dispatch task action to appropriate handler. Returns True if action was found and executed.""" + task_actions = get_task_action_handlers() + if action_name in task_actions: + task_actions[action_name](navigator, cluster_name, service_name, task_arn, task_details) + return True + return False + + +def get_container_action_handlers() -> dict[str, "Callable"]: + """Get mapping of container action names to their handlers.""" + return { + "tail_logs": lambda nav, cluster, task_arn, container: nav.show_container_logs_live_tail( + cluster, + task_arn, + container, + ), + "show_env": lambda nav, cluster, task_arn, container: nav.show_container_environment_variables( + cluster, + task_arn, + container, + ), + "show_secrets": lambda nav, cluster, task_arn, container: nav.show_container_secrets( + cluster, task_arn, container + ), + "show_ports": lambda nav, cluster, task_arn, container: nav.show_container_port_mappings( + cluster, + task_arn, + container, + ), + "show_volumes": lambda nav, cluster, task_arn, container: nav.show_container_volume_mounts( + cluster, + task_arn, + container, + ), + } + + +def get_task_action_handlers() -> dict[str, "Callable"]: + """Get mapping of task action names to their handlers.""" + return { + "show_history": lambda nav, cluster, service, _task_arn, _task_details: nav.show_task_history(cluster, service), + "show_details": lambda nav, _cluster, _service, _task_arn, task_details: nav.display_task_details(task_details), + "compare_definitions": lambda nav, + _cluster, + _service, + _task_arn, + task_details: nav.show_task_definition_comparison( + task_details, + ), + "open_console": lambda nav, cluster, _service, task_arn, _task_details: nav.open_task_in_console( + cluster, task_arn + ), + } + + +def get_service_action_handlers() -> dict[str, "Callable"]: + """Get mapping of service action names to their handlers.""" + return { + "force_deployment": lambda nav, cluster, service: nav.handle_force_deployment(cluster, service), + "show_events": lambda nav, cluster, service: nav.show_service_events(cluster, service), + "show_metrics": lambda nav, cluster, service: nav.show_service_metrics(cluster, service), + "open_console": lambda nav, cluster, service: nav.open_service_in_console(cluster, service), + } diff --git a/src/lazy_ecs/features/container/container.py b/src/lazy_ecs/features/container/container.py index 218356f..842d4ed 100644 --- a/src/lazy_ecs/features/container/container.py +++ b/src/lazy_ecs/features/container/container.py @@ -26,6 +26,14 @@ from ..task.task import TaskService +def build_log_group_arn(region: str, account_id: str, log_group: str) -> str: + return f"arn:aws:logs:{region}:{account_id}:log-group:{log_group}" + + +def build_log_stream_name(stream_prefix: str, container_name: str, task_id: str) -> str: + return f"{stream_prefix}/{container_name}/{task_id}" + + class ContainerService(BaseAWSService): """Service for ECS container operations.""" @@ -90,7 +98,7 @@ def get_log_config(self, cluster_name: str, task_arn: str, container_name: str) if not log_group: return None - log_stream = f"{stream_prefix}/{container_name}/{context.task_id}" + log_stream = build_log_stream_name(stream_prefix, container_name, context.task_id) return {"log_group": log_group, "log_stream": log_stream} @@ -146,7 +154,7 @@ def get_live_container_logs_tail( ) if not region or not aws_account_id: return - log_group_arn = f"arn:aws:logs:{region}:{aws_account_id}:log-group:{log_group}" + log_group_arn = build_log_group_arn(region, aws_account_id, log_group) response = self.logs_client.start_live_tail( logGroupIdentifiers=[log_group_arn], logStreamNames=[log_stream], diff --git a/src/lazy_ecs/features/task/task.py b/src/lazy_ecs/features/task/task.py index 4144a99..580400f 100644 --- a/src/lazy_ecs/features/task/task.py +++ b/src/lazy_ecs/features/task/task.py @@ -68,7 +68,9 @@ def get_task_and_definition( task = tasks[0] task_def_arn = task["taskDefinitionArn"] task_def_response = self.ecs_client.describe_task_definition(taskDefinition=task_def_arn) - task_definition = task_def_response["taskDefinition"] + task_definition = task_def_response.get("taskDefinition") + if not task_definition: + return None return task, task_definition diff --git a/tests/conftest.py b/tests/conftest.py index 5a9ae36..b2caa82 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,17 +7,6 @@ @pytest.fixture def mock_paginated_client(): - """Create a mock AWS client with paginator support. - - Returns a factory function that creates clients with specified pagination pages. - - Example: - def test_something(mock_paginated_client): - pages = [{"clusterArns": ["arn1", "arn2"]}] - client = mock_paginated_client(pages) - - """ - def _create_client(pages: list[dict]) -> Mock: client = Mock() paginator = Mock() @@ -26,3 +15,8 @@ def _create_client(pages: list[dict]) -> Mock: return client return _create_client + + +@pytest.fixture +def mock_ecs_client(): + return Mock() diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..a0aa7db --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,214 @@ +"""Tests for core application logic.""" + +from unittest.mock import Mock, patch + +from lazy_ecs.core.app import ( + dispatch_container_action, + dispatch_service_action, + dispatch_task_action, + get_container_action_handlers, + get_service_action_handlers, + get_task_action_handlers, + handle_task_features, + handle_task_selection, + navigate_services, +) + + +def test_get_container_action_handlers_returns_all_actions(): + handlers = get_container_action_handlers() + + assert "tail_logs" in handlers + assert "show_env" in handlers + assert "show_secrets" in handlers + assert "show_ports" in handlers + assert "show_volumes" in handlers + assert len(handlers) == 5 + + +def test_get_task_action_handlers_returns_all_actions(): + handlers = get_task_action_handlers() + + assert "show_history" in handlers + assert "show_details" in handlers + assert "compare_definitions" in handlers + assert "open_console" in handlers + assert len(handlers) == 4 + + +def test_get_service_action_handlers_returns_all_actions(): + handlers = get_service_action_handlers() + + assert "force_deployment" in handlers + assert "show_events" in handlers + assert "show_metrics" in handlers + assert "open_console" in handlers + assert len(handlers) == 4 + + +def test_dispatch_container_action_calls_handler_for_valid_action(): + mock_navigator = Mock() + + result = dispatch_container_action(mock_navigator, "cluster", "task-arn", "container", "show_env") + + assert result is True + mock_navigator.show_container_environment_variables.assert_called_once_with("cluster", "task-arn", "container") + + +def test_dispatch_container_action_returns_false_for_invalid_action(): + mock_navigator = Mock() + + result = dispatch_container_action(mock_navigator, "cluster", "task-arn", "container", "invalid_action") + + assert result is False + + +def test_dispatch_task_action_calls_handler_for_valid_action(): + mock_navigator = Mock() + + result = dispatch_task_action(mock_navigator, "cluster", "service", "task-arn", None, "show_history") + + assert result is True + mock_navigator.show_task_history.assert_called_once_with("cluster", "service") + + +def test_dispatch_task_action_returns_false_for_invalid_action(): + mock_navigator = Mock() + + result = dispatch_task_action(mock_navigator, "cluster", "service", "task-arn", None, "invalid_action") + + assert result is False + + +def test_dispatch_service_action_calls_handler_for_valid_action(): + mock_navigator = Mock() + + dispatch_service_action(mock_navigator, "cluster", "service", "show_events") + + mock_navigator.show_service_events.assert_called_once_with("cluster", "service") + + +def test_dispatch_service_action_ignores_invalid_action(): + mock_navigator = Mock() + + dispatch_service_action(mock_navigator, "cluster", "service", "invalid_action") + + assert not mock_navigator.method_calls + + +def test_navigate_services_returns_true_on_back(): + mock_navigator = Mock() + mock_navigator.select_service.return_value = "navigation:back" + mock_ecs_service = Mock() + + result = navigate_services(mock_navigator, mock_ecs_service, "cluster") + + assert result is True + + +def test_navigate_services_returns_false_on_exit(): + mock_navigator = Mock() + mock_navigator.select_service.return_value = "navigation:exit" + mock_ecs_service = Mock() + + result = navigate_services(mock_navigator, mock_ecs_service, "cluster") + + assert result is False + + +def test_navigate_services_returns_true_on_non_service_selection(): + mock_navigator = Mock() + mock_navigator.select_service.return_value = "unknown:value" + mock_ecs_service = Mock() + + result = navigate_services(mock_navigator, mock_ecs_service, "cluster") + + assert result is True + + +@patch("lazy_ecs.core.app.console") +def test_navigate_services_handles_service_action_back(_mock_console): + mock_navigator = Mock() + mock_navigator.select_service.return_value = "service:web-api" + mock_navigator.select_service_action.return_value = "navigation:back" + mock_ecs_service = Mock() + + result = navigate_services(mock_navigator, mock_ecs_service, "cluster") + + assert result is True + + +@patch("lazy_ecs.core.app.console") +def test_navigate_services_handles_service_action_exit(_mock_console): + mock_navigator = Mock() + mock_navigator.select_service.return_value = "service:web-api" + mock_navigator.select_service_action.return_value = "navigation:exit" + mock_ecs_service = Mock() + + result = navigate_services(mock_navigator, mock_ecs_service, "cluster") + + assert result is False + + +@patch("lazy_ecs.core.app.console") +def test_handle_task_selection_returns_true_when_no_task_details(_mock_console): + mock_navigator = Mock() + mock_ecs_service = Mock() + mock_ecs_service.get_task_details.return_value = None + + result = handle_task_selection(mock_navigator, mock_ecs_service, "cluster", "service", "task-arn") + + assert result is True + + +@patch("lazy_ecs.core.app.console") +def test_handle_task_selection_calls_handle_task_features(_mock_console): + mock_navigator = Mock() + mock_ecs_service = Mock() + task_details = {"taskArn": "arn:task"} + mock_ecs_service.get_task_details.return_value = task_details + + with patch("lazy_ecs.core.app.handle_task_features", return_value=True) as mock_handle: + result = handle_task_selection(mock_navigator, mock_ecs_service, "cluster", "service", "task-arn") + + mock_handle.assert_called_once_with(mock_navigator, "cluster", "task-arn", task_details, "service") + assert result is True + + +def test_handle_task_features_returns_true_on_back(): + mock_navigator = Mock() + mock_navigator.select_task_feature.return_value = "navigation:back" + + assert handle_task_features(mock_navigator, "cluster", "task-arn", None, "service") is True + + +def test_handle_task_features_returns_false_on_exit(): + mock_navigator = Mock() + mock_navigator.select_task_feature.return_value = "navigation:exit" + + assert handle_task_features(mock_navigator, "cluster", "task-arn", None, "service") is False + + +def test_handle_task_features_dispatches_container_action(): + mock_navigator = Mock() + mock_navigator.select_task_feature.side_effect = ["container_action:show_env:web", "navigation:back"] + + with patch("lazy_ecs.core.app.dispatch_container_action", return_value=True) as mock_dispatch: + result = handle_task_features(mock_navigator, "cluster", "task-arn", None, "service") + + mock_dispatch.assert_called_once_with(mock_navigator, "cluster", "task-arn", "web", "show_env") + assert result is True + + +def test_handle_task_features_dispatches_task_action(): + mock_navigator = Mock() + mock_navigator.select_task_feature.side_effect = ["task_action:show_history", "navigation:back"] + task_details = {"taskArn": "arn:task"} + + with patch("lazy_ecs.core.app.dispatch_task_action", return_value=True) as mock_dispatch: + result = handle_task_features(mock_navigator, "cluster", "task-arn", task_details, "service") # type: ignore[arg-type] + + mock_dispatch.assert_called_once_with( + mock_navigator, "cluster", "service", "task-arn", task_details, "show_history" + ) + assert result is True diff --git a/tests/test_cli.py b/tests/test_cli.py index 2e78a44..d84e62c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -4,6 +4,7 @@ from lazy_ecs import _create_aws_client, main +@patch("lazy_ecs.core.app.console") @patch("lazy_ecs._create_cloudwatch_client") @patch("lazy_ecs._create_sts_client") @patch("lazy_ecs._create_logs_client") @@ -17,8 +18,8 @@ def test_main_successful_flow( mock_create_logs_client, mock_create_sts_client, mock_create_cloudwatch_client, + mock_app_console, ) -> None: - """Test main function with successful cluster selection.""" mock_navigator = Mock() mock_navigator.select_cluster.return_value = "production" mock_navigator_class.return_value = mock_navigator @@ -32,9 +33,10 @@ def test_main_successful_flow( mock_create_cloudwatch_client.assert_called_once_with(None) mock_navigator.select_cluster.assert_called_once() mock_console.print.assert_any_call("🚀 Welcome to lazy-ecs!", style="bold cyan") - mock_console.print.assert_any_call("\n✅ Selected cluster: production", style="green") + mock_app_console.print.assert_any_call("\n✅ Selected cluster: production", style="green") +@patch("lazy_ecs.core.app.console") @patch("lazy_ecs._create_cloudwatch_client") @patch("lazy_ecs._create_sts_client") @patch("lazy_ecs._create_logs_client") @@ -42,14 +44,14 @@ def test_main_successful_flow( @patch("lazy_ecs.ECSNavigator") @patch("lazy_ecs.console") def test_main_no_cluster_selected( - mock_console, + _mock_console, mock_navigator_class, _mock_create_client, _mock_create_logs_client, _mock_create_sts_client, _mock_create_cloudwatch_client, + mock_app_console, ) -> None: - """Test main function when no cluster is selected.""" mock_navigator = Mock() mock_navigator.select_cluster.return_value = None mock_navigator_class.return_value = mock_navigator @@ -57,7 +59,7 @@ def test_main_no_cluster_selected( with patch.object(sys, "argv", ["lazy-ecs"]): main() - mock_console.print.assert_any_call("\n❌ No cluster selected. Goodbye!", style="yellow") + mock_app_console.print.assert_any_call("\n❌ No cluster selected. Goodbye!", style="yellow") @patch("lazy_ecs._create_cloudwatch_client") @@ -72,7 +74,6 @@ def test_main_aws_error( _mock_create_sts_client, _mock_create_cloudwatch_client, ) -> None: - """Test main function with AWS connection error.""" mock_create_client.side_effect = Exception("No credentials found") with patch.object(sys, "argv", ["lazy-ecs"]): @@ -96,7 +97,6 @@ def test_main_with_profile_argument( mock_create_sts_client, mock_create_cloudwatch_client, ) -> None: - """Test main function with --profile argument.""" mock_navigator = Mock() mock_navigator.select_cluster.return_value = "production" mock_navigator_class.return_value = mock_navigator @@ -111,10 +111,8 @@ def test_main_with_profile_argument( def test_create_aws_client_without_profile(): - """Test _create_aws_client without profile returns default client.""" with patch("lazy_ecs.boto3.client") as mock_client: _create_aws_client(None) - # Should be called with ECS and config for connection pooling assert mock_client.call_count == 1 args, kwargs = mock_client.call_args assert args[0] == "ecs" @@ -122,7 +120,6 @@ def test_create_aws_client_without_profile(): def test_create_aws_client_with_profile(): - """Test _create_aws_client with profile uses Session.""" mock_session = Mock() mock_client = Mock() mock_session.client.return_value = mock_client @@ -131,7 +128,6 @@ def test_create_aws_client_with_profile(): result = _create_aws_client("my-profile") mock_session_class.assert_called_once_with(profile_name="my-profile") - # Should be called with ECS and config for connection pooling assert mock_session.client.call_count == 1 args, kwargs = mock_session.client.call_args assert args[0] == "ecs" diff --git a/tests/test_container_service.py b/tests/test_container_service.py new file mode 100644 index 0000000..ec99d06 --- /dev/null +++ b/tests/test_container_service.py @@ -0,0 +1,253 @@ +"""Tests for container service functions.""" + +from unittest.mock import Mock + +import pytest + +from lazy_ecs.features.container.container import ContainerService, build_log_group_arn, build_log_stream_name + + +@pytest.fixture +def mock_task_service(): + return Mock() + + +@pytest.fixture +def container_service(mock_task_service): + mock_ecs_client = Mock() + return ContainerService(mock_ecs_client, mock_task_service) + + +@pytest.fixture +def mock_task_with_awslogs(): + return { + "taskArn": "arn:aws:ecs:us-east-1:123:task/cluster/abc123", + "taskDefinitionArn": "arn:task-def:1", + } + + +@pytest.fixture +def mock_task_definition_with_awslogs(): + return { + "containerDefinitions": [ + { + "name": "web", + "logConfiguration": { + "logDriver": "awslogs", + "options": {"awslogs-group": "/ecs/my-app"}, + }, + } + ] + } + + +def test_get_container_context_returns_none_when_task_not_found(container_service, mock_task_service): + mock_task_service.get_task_and_definition.return_value = None + + result = container_service.get_container_context("cluster", "task-arn", "container-name") + + assert result is None + + +def test_get_container_context_returns_none_when_container_not_found(container_service, mock_task_service): + mock_task = {"taskArn": "task-arn"} + mock_task_definition = { + "containerDefinitions": [ + {"name": "other-container", "image": "nginx:latest"}, + ] + } + mock_task_service.get_task_and_definition.return_value = (mock_task, mock_task_definition) + + result = container_service.get_container_context("cluster", "task-arn", "missing-container") + + assert result is None + + +def test_get_container_context_success(container_service, mock_task_service): + mock_task = {"taskArn": "task-arn"} + mock_task_definition = { + "containerDefinitions": [ + {"name": "web", "image": "nginx:latest"}, + ] + } + mock_task_service.get_task_and_definition.return_value = (mock_task, mock_task_definition) + + result = container_service.get_container_context("cluster", "task-arn", "web") + + assert result.container_name == "web" + assert result.cluster_name == "cluster" + assert result.task_arn == "task-arn" + + +def test_get_container_definition_not_found(container_service): + task_definition = { + "containerDefinitions": [ + {"name": "web", "image": "nginx:latest"}, + ] + } + + result = container_service.get_container_definition(task_definition, "missing") + + assert result is None + + +def test_get_container_definition_success(container_service): + task_definition = { + "containerDefinitions": [ + {"name": "web", "image": "nginx:latest"}, + {"name": "worker", "image": "python:3.11"}, + ] + } + + result = container_service.get_container_definition(task_definition, "worker") + + assert result["name"] == "worker" + assert result["image"] == "python:3.11" + + +def test_build_log_group_arn(): + arn = build_log_group_arn("us-east-1", "123456789012", "my-log-group") + + assert arn == "arn:aws:logs:us-east-1:123456789012:log-group:my-log-group" + + +def test_build_log_stream_name(): + stream = build_log_stream_name("ecs", "web-container", "abc123def456") + + assert stream == "ecs/web-container/abc123def456" + + +def test_build_log_stream_name_with_custom_prefix(): + stream = build_log_stream_name("my-app", "worker", "task-id-123") + + assert stream == "my-app/worker/task-id-123" + + +def test_get_log_config_returns_none_when_context_not_found(container_service, mock_task_service): + mock_task_service.get_task_and_definition.return_value = None + + result = container_service.get_log_config("cluster", "task-arn", "container") + + assert result is None + + +def test_get_log_config_returns_none_for_non_awslogs_driver( + container_service, mock_task_service, mock_task_with_awslogs +): + mock_task_definition = {"containerDefinitions": [{"name": "web", "logConfiguration": {"logDriver": "splunk"}}]} + mock_task_service.get_task_and_definition.return_value = (mock_task_with_awslogs, mock_task_definition) + + result = container_service.get_log_config("cluster", "task-arn", "web") + + assert result is None + + +def test_get_log_config_returns_none_when_no_log_group(container_service, mock_task_service, mock_task_with_awslogs): + mock_task_definition = { + "containerDefinitions": [{"name": "web", "logConfiguration": {"logDriver": "awslogs", "options": {}}}] + } + mock_task_service.get_task_and_definition.return_value = (mock_task_with_awslogs, mock_task_definition) + + result = container_service.get_log_config("cluster", "task-arn", "web") + + assert result is None + + +def test_get_log_config_success_with_defaults( + container_service, mock_task_service, mock_task_with_awslogs, mock_task_definition_with_awslogs +): + mock_task_service.get_task_and_definition.return_value = (mock_task_with_awslogs, mock_task_definition_with_awslogs) + + result = container_service.get_log_config("cluster", "arn:aws:ecs:us-east-1:123:task/cluster/abc123", "web") + + assert result["log_group"] == "/ecs/my-app" + assert result["log_stream"] == "ecs/web/abc123" + + +def test_get_log_config_success_with_custom_prefix(container_service, mock_task_service): + mock_task = {"taskArn": "arn:aws:ecs:us-east-1:123:task/cluster/task-id-456"} + mock_task_definition = { + "containerDefinitions": [ + { + "name": "worker", + "logConfiguration": { + "logDriver": "awslogs", + "options": {"awslogs-group": "/ecs/workers", "awslogs-stream-prefix": "app"}, + }, + } + ] + } + mock_task_service.get_task_and_definition.return_value = (mock_task, mock_task_definition) + + result = container_service.get_log_config("cluster", "arn:aws:ecs:us-east-1:123:task/cluster/task-id-456", "worker") + + assert result["log_group"] == "/ecs/workers" + assert result["log_stream"] == "app/worker/task-id-456" + + +def test_get_container_logs_returns_empty_when_no_client(mock_task_service): + container_service = ContainerService(Mock(), mock_task_service, logs_client=None) + + result = container_service.get_container_logs("/ecs/app", "stream") + + assert result == [] + + +def test_get_container_logs_success(): + mock_logs_client = Mock() + mock_logs_client.get_log_events.return_value = {"events": [{"message": "log1"}, {"message": "log2"}]} + container_service = ContainerService(Mock(), Mock(), logs_client=mock_logs_client) + + result = container_service.get_container_logs("/ecs/app", "stream", lines=100) + + assert len(result) == 2 + mock_logs_client.get_log_events.assert_called_once_with( + logGroupName="/ecs/app", logStreamName="stream", limit=100, startFromHead=False + ) + + +def test_get_container_logs_filtered_returns_empty_when_no_client(mock_task_service): + container_service = ContainerService(Mock(), mock_task_service, logs_client=None) + + result = container_service.get_container_logs_filtered("/ecs/app", "stream", "ERROR") + + assert result == [] + + +def test_get_container_logs_filtered_success(): + mock_logs_client = Mock() + mock_logs_client.filter_log_events.return_value = {"events": [{"message": "ERROR: failed"}]} + container_service = ContainerService(Mock(), Mock(), logs_client=mock_logs_client) + + result = container_service.get_container_logs_filtered("/ecs/app", "stream", "ERROR", lines=25) + + assert len(result) == 1 + mock_logs_client.filter_log_events.assert_called_once_with( + logGroupName="/ecs/app", logStreamNames=["stream"], filterPattern="ERROR", limit=25 + ) + + +def test_list_log_groups_returns_empty_when_no_client(mock_task_service): + container_service = ContainerService(Mock(), mock_task_service, logs_client=None) + + result = container_service.list_log_groups("production", "web") + + assert result == [] + + +def test_list_log_groups_filters_by_cluster_and_container(): + mock_logs_client = Mock() + mock_logs_client.describe_log_groups.return_value = { + "logGroups": [ + {"logGroupName": "/ecs/production-web"}, + {"logGroupName": "/ecs/staging-api"}, + {"logGroupName": "/aws/lambda/function"}, + {"logGroupName": "/ecs/production-worker"}, + ] + } + container_service = ContainerService(Mock(), Mock(), logs_client=mock_logs_client) + + result = container_service.list_log_groups("production", "web") + + assert "/ecs/production-web" in result + assert "/ecs/staging-api" in result # Contains "ecs" so it's included diff --git a/tests/test_container_ui.py b/tests/test_container_ui.py index d27f305..9130e39 100644 --- a/tests/test_container_ui.py +++ b/tests/test_container_ui.py @@ -9,11 +9,6 @@ from lazy_ecs.features.container.ui import ContainerUI -@pytest.fixture -def mock_ecs_client(): - return Mock() - - @pytest.fixture def mock_task_service(): return Mock() diff --git a/tests/test_core_utils.py b/tests/test_core_utils.py index aa0b351..fd50898 100644 --- a/tests/test_core_utils.py +++ b/tests/test_core_utils.py @@ -2,67 +2,59 @@ import time -from lazy_ecs.core.utils import determine_service_status, extract_name_from_arn, paginate_aws_list, show_spinner +from lazy_ecs.core.utils import ( + batch_items, + determine_service_status, + extract_name_from_arn, + paginate_aws_list, + print_error, + print_info, + print_success, + print_warning, + show_spinner, +) def test_extract_name_from_arn(): - """Test extracting resource names from AWS ARNs.""" - # Test cluster ARN - cluster_arn = "arn:aws:ecs:us-east-1:123456789012:cluster/production" - assert extract_name_from_arn(cluster_arn) == "production" - - # Test service ARN - service_arn = "arn:aws:ecs:us-east-1:123456789012:service/production/web-api" - assert extract_name_from_arn(service_arn) == "web-api" - - # Test task ARN - task_arn = "arn:aws:ecs:us-east-1:123456789012:task/production/abc123def456" - assert extract_name_from_arn(task_arn) == "abc123def456" - - # Test simple case - simple_name = "just-a-name" - assert extract_name_from_arn(simple_name) == "just-a-name" + assert extract_name_from_arn("arn:aws:ecs:us-east-1:123456789012:cluster/production") == "production" + assert extract_name_from_arn("arn:aws:ecs:us-east-1:123456789012:service/production/web-api") == "web-api" + assert extract_name_from_arn("arn:aws:ecs:us-east-1:123456789012:task/production/abc123def456") == "abc123def456" + assert extract_name_from_arn("just-a-name") == "just-a-name" def test_determine_service_status_healthy(): - """Test service status determination for healthy service.""" icon, status = determine_service_status(running_count=3, desired_count=3, pending_count=0) assert icon == "✅" assert status == "HEALTHY" def test_determine_service_status_scaling(): - """Test service status determination for scaling service.""" icon, status = determine_service_status(running_count=1, desired_count=3, pending_count=2) assert icon == "⚠️" assert status == "SCALING" def test_determine_service_status_over_scaled(): - """Test service status determination for over-scaled service.""" icon, status = determine_service_status(running_count=5, desired_count=3, pending_count=0) assert icon == "🔴" assert status == "OVER_SCALED" def test_determine_service_status_pending(): - """Test service status determination for pending service.""" icon, status = determine_service_status(running_count=3, desired_count=3, pending_count=1) assert icon == "🟡" assert status == "PENDING" def test_determine_service_status_zero_counts(): - """Test service status determination with zero counts.""" icon, status = determine_service_status(running_count=0, desired_count=0, pending_count=0) assert icon == "✅" assert status == "HEALTHY" def test_show_spinner(): - """Test spinner context manager works without errors.""" with show_spinner(): - time.sleep(0.01) # Brief pause to simulate work + time.sleep(0.01) def test_paginate_aws_list_single_page(mock_paginated_client): @@ -105,3 +97,50 @@ def test_paginate_aws_list_missing_key(mock_paginated_client): result = paginate_aws_list(mock_client, "list_clusters", "clusterArns") assert result == [] + + +def test_print_success(capsys): + print_success("Operation completed") + captured = capsys.readouterr() + assert "Operation completed" in captured.out + + +def test_print_error(capsys): + print_error("Something went wrong") + captured = capsys.readouterr() + assert "Something went wrong" in captured.out + + +def test_print_warning(capsys): + print_warning("Be careful") + captured = capsys.readouterr() + assert "Be careful" in captured.out + + +def test_print_info(capsys): + print_info("Informational message") + captured = capsys.readouterr() + assert "Informational message" in captured.out + + +def test_batch_items_empty_list(): + result = list(batch_items([], 5)) + assert result == [] + + +def test_batch_items_single_batch(): + items = [1, 2, 3] + result = list(batch_items(items, 5)) + assert result == [[1, 2, 3]] + + +def test_batch_items_multiple_batches(): + items = [1, 2, 3, 4, 5, 6, 7, 8, 9] + result = list(batch_items(items, 3)) + assert result == [[1, 2, 3], [4, 5, 6], [7, 8, 9]] + + +def test_batch_items_partial_last_batch(): + items = [1, 2, 3, 4, 5, 6, 7] + result = list(batch_items(items, 3)) + assert result == [[1, 2, 3], [4, 5, 6], [7]] diff --git a/tests/test_service_service.py b/tests/test_service_service.py new file mode 100644 index 0000000..e368cba --- /dev/null +++ b/tests/test_service_service.py @@ -0,0 +1,43 @@ +"""Tests for service service.""" + +from unittest.mock import Mock + +import pytest + +from lazy_ecs.features.service.service import ServiceService + + +@pytest.fixture +def mock_ecs_client(mock_paginated_client): + pages = [{"serviceArns": []}] + return mock_paginated_client(pages) + + +def test_get_service_info_returns_empty_when_no_services(mock_ecs_client): + service_service = ServiceService(mock_ecs_client) + + result = service_service.get_service_info("cluster") + + assert result == [] + + +def test_get_desired_task_definition_arn_returns_none_when_no_services(): + mock_ecs_client = Mock() + mock_ecs_client.describe_services.return_value = {"services": []} + service_service = ServiceService(mock_ecs_client) + + result = service_service.get_desired_task_definition_arn("cluster", "service") + + assert result is None + + +def test_get_desired_task_definition_arn_success(): + mock_ecs_client = Mock() + mock_ecs_client.describe_services.return_value = { + "services": [{"serviceName": "web", "taskDefinition": "arn:task-def:5"}] + } + service_service = ServiceService(mock_ecs_client) + + result = service_service.get_desired_task_definition_arn("cluster", "web") + + assert result == "arn:task-def:5" diff --git a/tests/test_service_ui.py b/tests/test_service_ui.py index 91602d4..7838f04 100644 --- a/tests/test_service_ui.py +++ b/tests/test_service_ui.py @@ -5,20 +5,14 @@ import pytest +from lazy_ecs.core.types import ServiceMetrics from lazy_ecs.features.service.actions import ServiceActions from lazy_ecs.features.service.service import ServiceService from lazy_ecs.features.service.ui import ServiceUI -@pytest.fixture -def mock_ecs_client(): - """Create a mock ECS client.""" - return Mock() - - @pytest.fixture def service_ui(mock_ecs_client): - """Create a ServiceUI instance with mocked services.""" service_service = ServiceService(mock_ecs_client) service_actions = ServiceActions(mock_ecs_client) return ServiceUI(service_service, service_actions) @@ -139,9 +133,8 @@ def test_select_service_action_show_events(mock_select, service_ui): assert selected == "action:show_events" mock_select.assert_called_once() - # Verify that show events option is in the choices call_args = mock_select.call_args - choices = call_args[0][1] # Second positional argument is choices + choices = call_args[0][1] show_events_choice = next((choice for choice in choices if choice.get("value") == "action:show_events"), None) assert show_events_choice is not None assert "Show service events" in show_events_choice["name"] @@ -222,7 +215,16 @@ def test_service_name_truncation_shows_end(mock_print, service_ui): service_ui.display_service_events("test-cluster", "test-service") - # The service name should be truncated to show the end (suffix-v2) mock_print.assert_called_once() - # We can't easily test the exact truncation without complex argument inspection, - # but we know the logic truncates to show the last 15 chars with "..." prefix + + +@patch("lazy_ecs.features.service.ui.console") +def test_display_service_metrics(mock_console, service_ui): + metrics: ServiceMetrics = { + "cpu": {"current": 45.5, "average": 42.0, "maximum": 78.0, "minimum": 35.0}, + "memory": {"current": 62.3, "average": 58.0, "maximum": 85.0, "minimum": 50.0}, + } + + service_ui.display_service_metrics("web-api", metrics) + + assert mock_console.print.called diff --git a/tests/test_task_comparison_ui.py b/tests/test_task_comparison_ui.py new file mode 100644 index 0000000..283cdb1 --- /dev/null +++ b/tests/test_task_comparison_ui.py @@ -0,0 +1,197 @@ +"""Tests for task comparison UI.""" + +from unittest.mock import Mock, patch + +from lazy_ecs.features.task.ui import TaskUI + + +@patch("lazy_ecs.features.task.ui.console") +def test_show_task_definition_comparison_no_service(_mock_console): + task_ui = TaskUI(Mock(), None) + task_details = { + "task_arn": "arn:task", + "task_definition_name": "web", + "task_definition_revision": "5", + } + + task_ui.show_task_definition_comparison(task_details) # type: ignore[arg-type] + + +@patch("lazy_ecs.features.task.ui.console") +@patch("lazy_ecs.features.task.ui.print_warning") +def test_show_task_definition_comparison_not_enough_revisions(_mock_warning, _mock_console): + mock_comparison_service = Mock() + mock_comparison_service.list_task_definition_revisions.return_value = [{"revision": 1, "arn": "arn:1"}] + task_ui = TaskUI(Mock(), mock_comparison_service) + task_details = { + "task_arn": "arn:task", + "task_definition_name": "web", + "task_definition_revision": "1", + } + + task_ui.show_task_definition_comparison(task_details) # type: ignore[arg-type] + + +@patch("lazy_ecs.features.task.ui.select_with_auto_pagination") +@patch("lazy_ecs.features.task.ui.console") +def test_show_task_definition_comparison_user_cancels(_mock_console, mock_select): + mock_comparison_service = Mock() + mock_comparison_service.list_task_definition_revisions.return_value = [ + {"revision": 5, "arn": "arn:5"}, + {"revision": 4, "arn": "arn:4"}, + ] + mock_select.return_value = None + task_ui = TaskUI(Mock(), mock_comparison_service) + task_details = { + "task_arn": "arn:task", + "task_definition_name": "web", + "task_definition_revision": "5", + } + + task_ui.show_task_definition_comparison(task_details) # type: ignore[arg-type] + + mock_select.assert_called_once() + + +@patch("lazy_ecs.features.task.ui.compare_task_definitions") +@patch("lazy_ecs.features.task.ui.select_with_auto_pagination") +@patch("lazy_ecs.features.task.ui.console") +def test_show_task_definition_comparison_success(_mock_console, mock_select, mock_compare): + mock_comparison_service = Mock() + mock_comparison_service.list_task_definition_revisions.return_value = [ + {"revision": 5, "arn": "arn:5"}, + {"revision": 4, "arn": "arn:4"}, + ] + source_def = {"family": "web", "revision": 5} + target_def = {"family": "web", "revision": 4} + mock_comparison_service.get_task_definitions_for_comparison.return_value = (source_def, target_def) + mock_compare.return_value = [{"type": "image_changed", "old": "nginx:1.0", "new": "nginx:2.0"}] + mock_select.return_value = "arn:4" + + task_ui = TaskUI(Mock(), mock_comparison_service) + task_ui._display_comparison_results = Mock() + task_details = { + "task_arn": "arn:task", + "task_definition_name": "web", + "task_definition_revision": "5", + } + + task_ui.show_task_definition_comparison(task_details) # type: ignore[arg-type] + + task_ui._display_comparison_results.assert_called_once() + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_comparison_results_no_changes(_mock_console): + task_ui = TaskUI(Mock()) + + task_ui._display_comparison_results( + {"family": "web", "revision": 5}, + {"family": "web", "revision": 4}, + [], + ) + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_comparison_results_with_changes(_mock_console): + task_ui = TaskUI(Mock()) + task_ui._display_change = Mock() + changes = [ + {"type": "image_changed", "old": "nginx:1.0", "new": "nginx:2.0"}, + {"type": "cpu_changed", "old": "256", "new": "512"}, + ] + + task_ui._display_comparison_results( + {"family": "web", "revision": 5}, + {"family": "web", "revision": 4}, + changes, + ) + + assert task_ui._display_change.call_count == 2 + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_change_environment_added(_mock_console): + task_ui = TaskUI(Mock()) + change = {"type": "environment_added", "container": "web", "key": "DEBUG", "value": "true"} + + task_ui._display_change(change) + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_change_environment_removed(_mock_console): + task_ui = TaskUI(Mock()) + change = {"type": "environment_removed", "container": "web", "key": "OLD_VAR", "value": "old"} + + task_ui._display_change(change) + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_change_environment_changed(_mock_console): + task_ui = TaskUI(Mock()) + change = {"type": "environment_changed", "container": "web", "key": "PORT", "old": "8080", "new": "3000"} + + task_ui._display_change(change) + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_change_secret_changed(_mock_console): + task_ui = TaskUI(Mock()) + change = {"type": "secret_changed", "container": "web", "key": "API_KEY", "old": "arn:1", "new": "arn:2"} + + task_ui._display_change(change) + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_change_ports_changed(_mock_console): + task_ui = TaskUI(Mock()) + change = { + "type": "ports_changed", + "container": "web", + "old": [{"containerPort": 8080}], + "new": [{"containerPort": 3000}], + } + + task_ui._display_change(change) + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_change_command_changed(_mock_console): + task_ui = TaskUI(Mock()) + change = { + "type": "command_changed", + "container": "web", + "old": ["npm", "start"], + "new": ["node", "server.js"], + } + + task_ui._display_change(change) + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_change_volumes_changed(_mock_console): + task_ui = TaskUI(Mock()) + change = { + "type": "volumes_changed", + "container": "web", + "old": [{"sourceVolume": "data", "containerPath": "/data"}], + "new": [{"sourceVolume": "logs", "containerPath": "/logs"}], + } + + task_ui._display_change(change) + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_change_generic(_mock_console): + task_ui = TaskUI(Mock()) + change = {"type": "memory_changed", "old": "256", "new": "512"} + + task_ui._display_change(change) + + +@patch("lazy_ecs.features.task.ui.console") +def test_display_change_unknown_type_ignored(_mock_console): + task_ui = TaskUI(Mock()) + change = {"type": "unknown_change_type"} + + task_ui._display_change(change) diff --git a/tests/test_task_history.py b/tests/test_task_history.py index 26c63f1..c08604f 100644 --- a/tests/test_task_history.py +++ b/tests/test_task_history.py @@ -2,6 +2,7 @@ from datetime import datetime from typing import Any +from unittest.mock import Mock import boto3 import pytest @@ -14,8 +15,7 @@ class TestTaskHistoryParsing: """Test parsing task history and failure information.""" def test_parse_stopped_task_with_stop_code(self): - """Test parsing stopped task with stop code.""" - _mock_task_data: dict[str, Any] = { + task_data: dict[str, Any] = { "taskArn": "arn:aws:ecs:us-east-1:123456789012:task/cluster/task-id", "lastStatus": "STOPPED", "desiredStatus": "STOPPED", @@ -36,7 +36,7 @@ def test_parse_stopped_task_with_stop_code(self): ], } - _expected_history = { + expected = { "task_arn": "arn:aws:ecs:us-east-1:123456789012:task/cluster/task-id", "task_definition_name": "web-api", "task_definition_revision": "5", @@ -58,12 +58,12 @@ def test_parse_stopped_task_with_stop_code(self): ], } - result = TaskService._parse_task_history(_mock_task_data) # type: ignore - assert result == _expected_history + result = TaskService._parse_task_history(task_data) # type: ignore + assert result == expected def test_parse_running_task_no_failure_info(self): """Test parsing running task with no failure information.""" - _mock_task_data: dict[str, Any] = { + task_data: dict[str, Any] = { "taskArn": "arn:aws:ecs:us-east-1:123456789012:task/cluster/task-id-2", "lastStatus": "RUNNING", "desiredStatus": "RUNNING", @@ -73,7 +73,7 @@ def test_parse_running_task_no_failure_info(self): "containers": [{"name": "web-api", "healthStatus": "HEALTHY", "lastStatus": "RUNNING"}], } - _expected_history = { + expected = { "task_arn": "arn:aws:ecs:us-east-1:123456789012:task/cluster/task-id-2", "task_definition_name": "web-api", "task_definition_revision": "6", @@ -95,8 +95,8 @@ def test_parse_running_task_no_failure_info(self): ], } - result = TaskService._parse_task_history(_mock_task_data) # type: ignore - assert result == _expected_history + result = TaskService._parse_task_history(task_data) # type: ignore + assert result == expected def test_analyze_oom_kill_failure(self): """Test analysis of OOM kill failure.""" @@ -117,17 +117,109 @@ def test_analyze_successful_completion(self): assert "✅" in result assert "completed successfully" in result.lower() - def test_analyze_timeout_failure(self): - """Test analysis of timeout failure.""" - result = TaskService._analyze_container_failure( - "web-api", - 137, - "Task killed", - "TaskFailedToStart", - "Task timed out", - ) - assert "⏰" in result - assert "timeout" in result.lower() + @pytest.mark.parametrize( + ("exit_code", "reason", "expected_emoji", "expected_text"), + [ + (137, "Task killed", "⏰", "timeout"), + (139, None, "💥", "segmentation fault"), + (143, None, "🛑", "gracefully stopped"), + (1, "Application crashed", "❌", "application error"), + (42, "Unknown error", "🔴", "exit code 42"), + ], + ) + def test_analyze_container_failure_exit_codes(self, exit_code, reason, expected_emoji, expected_text): + result = TaskService._analyze_container_failure("container", exit_code, reason, None, None) + assert expected_emoji in result + assert expected_text in result.lower() + + @pytest.mark.parametrize( + ("stop_code", "reason", "expected_emoji", "expected_text"), + [ + ("TaskFailedToStart", "CannotPullContainerError: image not found", "📦", "pull container image"), + ("TaskFailedToStart", "ResourcesNotAvailable: insufficient memory", "⚠️", "insufficient resources"), + ("TaskFailedToStart", "Some other reason", "🚫", "failed to start"), + ("ServiceSchedulerInitiated", "Service scaling", "🔄", "service scheduler"), + ("SpotInterruption", "EC2 spot instance reclaimed", "💸", "spot instance interruption"), + ("UserInitiated", "Stopped by admin", "👤", "manually stopped"), + ], + ) + def test_analyze_task_failure_stop_codes(self, stop_code, reason, expected_emoji, expected_text): + result = TaskService._analyze_task_failure(stop_code, reason) + assert expected_emoji in result + assert expected_text in result.lower() + + def test_get_task_failure_analysis_for_running_task(self): + task_history = { + "task_arn": "arn:task", + "task_definition_name": "app", + "task_definition_revision": "1", + "last_status": "RUNNING", + "desired_status": "RUNNING", + "stop_code": None, + "stopped_reason": None, + "created_at": None, + "started_at": None, + "stopped_at": None, + "containers": [], + } + service = TaskService(Mock()) + + result = service.get_task_failure_analysis(task_history) # type: ignore[arg-type] + + assert "✅" in result + assert "running" in result.lower() + + def test_get_task_failure_analysis_container_failure(self): + task_history = { + "task_arn": "arn:task", + "task_definition_name": "app", + "task_definition_revision": "1", + "last_status": "STOPPED", + "desired_status": "STOPPED", + "stop_code": "TaskFailedToStart", + "stopped_reason": "Essential container exited", + "created_at": None, + "started_at": None, + "stopped_at": None, + "containers": [ + { + "name": "web", + "exit_code": 1, + "reason": "App crashed", + "health_status": None, + "last_status": "STOPPED", + } + ], + } + service = TaskService(Mock()) + + result = service.get_task_failure_analysis(task_history) # type: ignore[arg-type] + + assert "❌" in result + assert "application error" in result.lower() + + def test_get_task_failure_analysis_task_level_failure(self): + task_history = { + "task_arn": "arn:task", + "task_definition_name": "app", + "task_definition_revision": "1", + "last_status": "STOPPED", + "desired_status": "STOPPED", + "stop_code": "TaskFailedToStart", + "stopped_reason": "CannotPullContainerError", + "created_at": None, + "started_at": None, + "stopped_at": None, + "containers": [ + {"name": "web", "exit_code": None, "reason": None, "health_status": None, "last_status": "STOPPED"} + ], + } + service = TaskService(Mock()) + + result = service.get_task_failure_analysis(task_history) # type: ignore[arg-type] + + assert "📦" in result + assert "pull container image" in result.lower() class TestTaskHistoryService: @@ -162,9 +254,9 @@ def mock_ecs_client(self, mock_paginated_client): def test_get_task_history_includes_stopped_tasks(self, mock_ecs_client): """Test getting task history includes stopped tasks.""" - _service = TaskService(mock_ecs_client) + service = TaskService(mock_ecs_client) - result = _service.get_task_history("test-cluster", "web-service") + result = service.get_task_history("test-cluster", "web-service") assert len(result) > 0 assert any(task["last_status"] == "STOPPED" for task in result) @@ -173,9 +265,9 @@ def test_get_task_history_handles_no_stopped_tasks(self, mock_paginated_client): pages = [{"taskArns": []}] client = mock_paginated_client(pages) - _service = TaskService(client) + service = TaskService(client) - result = _service.get_task_history("test-cluster", "web-service") + result = service.get_task_history("test-cluster", "web-service") assert result == [] diff --git a/tests/test_task_service.py b/tests/test_task_service.py new file mode 100644 index 0000000..f64eba4 --- /dev/null +++ b/tests/test_task_service.py @@ -0,0 +1,38 @@ +"""Tests for task service.""" + +from unittest.mock import Mock + +from lazy_ecs.features.task.task import TaskService + + +def test_get_task_details_returns_none_when_no_tasks(): + mock_ecs_client = Mock() + mock_ecs_client.describe_tasks.return_value = {"tasks": []} + task_service = TaskService(mock_ecs_client) + + result = task_service.get_task_details("cluster", "task-arn", None) + + assert result is None + + +def test_get_task_and_definition_returns_none_when_no_tasks(): + mock_ecs_client = Mock() + mock_ecs_client.describe_tasks.return_value = {"tasks": []} + task_service = TaskService(mock_ecs_client) + + result = task_service.get_task_and_definition("cluster", "task-arn") + + assert result is None + + +def test_get_task_and_definition_returns_none_when_no_task_definition(): + mock_ecs_client = Mock() + mock_ecs_client.describe_tasks.return_value = { + "tasks": [{"taskArn": "arn:task", "taskDefinitionArn": "arn:task-def:1"}] + } + mock_ecs_client.describe_task_definition.return_value = {} + task_service = TaskService(mock_ecs_client) + + result = task_service.get_task_and_definition("cluster", "task-arn") + + assert result is None diff --git a/tests/test_task_ui.py b/tests/test_task_ui.py index 85a1424..4ad39db 100644 --- a/tests/test_task_ui.py +++ b/tests/test_task_ui.py @@ -8,15 +8,8 @@ from lazy_ecs.features.task.ui import TaskUI -@pytest.fixture -def mock_ecs_client(): - """Create a mock ECS client.""" - return Mock() - - @pytest.fixture def task_ui(mock_ecs_client): - """Create a TaskUI instance with mocked service.""" task_service = TaskService(mock_ecs_client) return TaskUI(task_service) diff --git a/tests/test_task_ui_formatters.py b/tests/test_task_ui_formatters.py new file mode 100644 index 0000000..58fb15d --- /dev/null +++ b/tests/test_task_ui_formatters.py @@ -0,0 +1,50 @@ +"""Tests for task UI formatting functions.""" + +import pytest + +from lazy_ecs.features.task.ui import _format_ports, _format_volumes + + +@pytest.mark.parametrize( + ("ports", "expected"), + [ + ([], "none"), + ([{"containerPort": 8080, "protocol": "tcp"}], "8080/tcp"), + ([{"containerPort": 8080, "hostPort": 80, "protocol": "tcp"}], "80:8080/tcp"), + ( + [ + {"containerPort": 8080, "hostPort": 80, "protocol": "tcp"}, + {"containerPort": 443, "protocol": "tcp"}, + {"containerPort": 53, "protocol": "udp"}, + ], + "80:8080/tcp, 443/tcp, 53/udp", + ), + ([{"containerPort": 3000}], "3000/tcp"), + ], +) +def test_format_ports(ports, expected): + assert _format_ports(ports) == expected + + +@pytest.mark.parametrize( + ("volumes", "expected"), + [ + ([], "none"), + ([{"sourceVolume": "data", "containerPath": "/app/data"}], "data:/app/data"), + ([{"sourceVolume": "config", "containerPath": "/etc/config", "readOnly": True}], "config:/etc/config:ro"), + ( + [ + {"sourceVolume": "data", "containerPath": "/app/data"}, + {"sourceVolume": "logs", "containerPath": "/var/log", "readOnly": True}, + {"sourceVolume": "cache", "containerPath": "/app/cache"}, + ], + "data:/app/data, logs:/var/log:ro, cache:/app/cache", + ), + ], +) +def test_format_volumes(volumes, expected): + assert _format_volumes(volumes) == expected + + +def test_format_volumes_missing_fields(): + assert "data:?" in _format_volumes([{"sourceVolume": "data"}]) diff --git a/tests/test_ui.py b/tests/test_ui.py index aaed3eb..f197ad8 100644 --- a/tests/test_ui.py +++ b/tests/test_ui.py @@ -9,12 +9,10 @@ @pytest.fixture def mock_ecs_service() -> Mock: - """Create a mock ECS service.""" return Mock() def test_navigator_initialization(mock_ecs_service) -> None: - """Test that ECSNavigator properly initializes all UI components.""" navigator = ECSNavigator(mock_ecs_service) assert navigator._cluster_ui is not None @@ -24,7 +22,6 @@ def test_navigator_initialization(mock_ecs_service) -> None: def test_select_cluster_delegates_to_cluster_ui(mock_ecs_service) -> None: - """Test that select_cluster delegates to ClusterUI.""" navigator = ECSNavigator(mock_ecs_service) navigator._cluster_ui.select_cluster = Mock(return_value="production") @@ -35,7 +32,6 @@ def test_select_cluster_delegates_to_cluster_ui(mock_ecs_service) -> None: def test_select_service_delegates_to_service_ui(mock_ecs_service) -> None: - """Test that select_service delegates to ServiceUI.""" navigator = ECSNavigator(mock_ecs_service) navigator._service_ui.select_service = Mock(return_value="service:web-api") @@ -46,7 +42,6 @@ def test_select_service_delegates_to_service_ui(mock_ecs_service) -> None: def test_select_service_action_integration(mock_ecs_service) -> None: - """Test that select_service_action integrates ECSService and ServiceUI.""" mock_ecs_service.get_task_info.return_value = [{"name": "task-1", "value": "task-arn-1"}] navigator = ECSNavigator(mock_ecs_service) @@ -171,10 +166,95 @@ def test_container_methods_delegate_to_container_ui(mock_ecs_service) -> None: def test_handle_force_deployment_delegates_to_service_ui(mock_ecs_service) -> None: - """Test that handle_force_deployment delegates to ServiceUI.""" navigator = ECSNavigator(mock_ecs_service) navigator._service_ui.handle_force_deployment = Mock() navigator.handle_force_deployment("cluster", "service") navigator._service_ui.handle_force_deployment.assert_called_once_with("cluster", "service") + + +def test_show_service_events_delegates_to_service_ui(mock_ecs_service): + navigator = ECSNavigator(mock_ecs_service) + navigator._service_ui.display_service_events = Mock() + + navigator.show_service_events("cluster", "service") + + navigator._service_ui.display_service_events.assert_called_once_with("cluster", "service") + + +@patch("lazy_ecs.ui.console") +def test_show_service_metrics_with_data(_mock_console, mock_ecs_service): + mock_ecs_service.get_service_metrics.return_value = {"cpu": 50.0, "memory": 60.0} + navigator = ECSNavigator(mock_ecs_service) + navigator._service_ui.display_service_metrics = Mock() + + navigator.show_service_metrics("cluster", "service") + + navigator._service_ui.display_service_metrics.assert_called_once_with("service", {"cpu": 50.0, "memory": 60.0}) + + +@patch("lazy_ecs.ui.console") +def test_show_service_metrics_no_data(mock_console, mock_ecs_service): + mock_ecs_service.get_service_metrics.return_value = None + navigator = ECSNavigator(mock_ecs_service) + + navigator.show_service_metrics("cluster", "service") + + mock_console.print.assert_any_call("\n⚠️ No metrics available for service 'service'", style="yellow") + + +def test_show_task_history_delegates_to_task_ui(mock_ecs_service): + navigator = ECSNavigator(mock_ecs_service) + navigator._task_ui.display_task_history = Mock() + + navigator.show_task_history("cluster", "service") + + navigator._task_ui.display_task_history.assert_called_once_with("cluster", "service") + + +def test_show_task_definition_comparison_with_details(mock_ecs_service): + navigator = ECSNavigator(mock_ecs_service) + navigator._task_ui.show_task_definition_comparison = Mock() + task_details = {"taskArn": "arn:task"} + + navigator.show_task_definition_comparison(task_details) # type: ignore[arg-type] + + navigator._task_ui.show_task_definition_comparison.assert_called_once_with(task_details) + + +def test_show_task_definition_comparison_without_details(mock_ecs_service): + navigator = ECSNavigator(mock_ecs_service) + navigator._task_ui.show_task_definition_comparison = Mock() + + navigator.show_task_definition_comparison(None) + + navigator._task_ui.show_task_definition_comparison.assert_not_called() + + +def test_open_service_in_console(mock_ecs_service): + with patch("webbrowser.open") as mock_webbrowser: + mock_ecs_service.get_region.return_value = "us-east-1" + navigator = ECSNavigator(mock_ecs_service) + + navigator.open_service_in_console("production", "web-api") + + mock_webbrowser.assert_called_once() + url_arg = mock_webbrowser.call_args[0][0] + assert "us-east-1" in url_arg + assert "production" in url_arg + assert "web-api" in url_arg + + +def test_open_task_in_console(mock_ecs_service): + with patch("webbrowser.open") as mock_webbrowser: + mock_ecs_service.get_region.return_value = "us-west-2" + navigator = ECSNavigator(mock_ecs_service) + + navigator.open_task_in_console("staging", "task-arn-123") + + mock_webbrowser.assert_called_once() + url_arg = mock_webbrowser.call_args[0][0] + assert "us-west-2" in url_arg + assert "staging" in url_arg + assert "task-arn-123" in url_arg diff --git a/uv.lock b/uv.lock index 7f547d3..2e89a5c 100644 --- a/uv.lock +++ b/uv.lock @@ -404,7 +404,7 @@ wheels = [ [[package]] name = "lazy-ecs" -version = "0.7.2" +version = "0.7.3" source = { editable = "." } dependencies = [ { name = "boto3" },