feat(ecs): add mode parameter and validation logic

Add --mode parameter to enable text-only monitoring and extract validation.
Modified: monitorexpressgatewayservice.py (+59), monitormutatinggatewayservice.py
- Add --mode parameter (interactive, text-only)
- Add validation methods for mode and resources
- Auto-detect mode based on TTY availability
- Support both interactive and text-only strategies
Benefits: CI/CD compatibility, better error messages

Author: jutyler1

awscli/customizations/ecs/monitorexpressgatewayservice.py

@@ -18,13 +18,14 @@
 allowing users to track resource creation progress, deployment status, and service health through
 an interactive command-line interface with live updates and visual indicators.
 
-The data collection logic is handled by ServiceViewCollector, which parses AWS resources and
-formats monitoring output. This module focuses on display and user interaction.
-
 The module implements two resource view modes:
 - RESOURCE: Displays all resources associated with the service
 - DEPLOYMENT: Shows resources that have changed in the most recent deployment
 
+And two display modes:
+- INTERACTIVE: Real-time display with spinner and keyboard navigation (requires TTY)
+- TEXT-ONLY: Text output with timestamps and change detection (works without TTY)
+
 Key Features:
 - Real-time progress monitoring with spinner animations
 - Diff-based resource tracking for deployment changes
@@ -36,18 +37,21 @@
     ECSExpressGatewayServiceWatcher: Core monitoring logic and resource tracking
 
 Usage:
-    aws ecs monitor-express-gateway-service --service-arn <arn> [--resource-view RESOURCE|DEPLOYMENT]
+    aws ecs monitor-express-gateway-service --service-arn <arn> [--resource-view RESOURCE|DEPLOYMENT] [--mode INTERACTIVE|TEXT-ONLY]
 """
 
-import asyncio
 import sys
-import threading
 import time
 
 from botocore.exceptions import ClientError
 
 from awscli.customizations.commands import BasicCommand
 from awscli.customizations.ecs.exceptions import MonitoringError
+from awscli.customizations.ecs.expressgateway.display_strategy import (
+    DisplayStrategy,
+    InteractiveDisplayStrategy,
+    TextOnlyDisplayStrategy,
+)
 from awscli.customizations.ecs.prompt_toolkit_display import Display
 from awscli.customizations.ecs.serviceviewcollector import ServiceViewCollector
 from awscli.customizations.utils import uni_print
@@ -64,14 +68,16 @@ class ECSMonitorExpressGatewayService(BasicCommand):
 
     DESCRIPTION = (
         "Monitors the progress of resource creation for an ECS Express Gateway Service. "
-        "This command provides real-time monitoring of service deployments with interactive "
-        "progress display, showing the status of load balancers, security groups, auto-scaling "
+        "This command provides real-time monitoring of service deployments showing the status "
+        "of load balancers, security groups, auto-scaling "
         "configurations, and other AWS resources as they are created or updated. "
         "Use ``--resource-view RESOURCE`` to view all service resources, or ``--resource-view DEPLOYMENT`` to track only "
         "resources that have changed in the most recent deployment. "
-        "The command requires a terminal (TTY) to run and the monitoring session continues "
-        "until manually stopped by the user or the specified timeout is reached. "
-        "Use keyboard shortcuts to navigate: up/down to scroll through resources, 'q' to quit monitoring."
+        "Choose ``--mode INTERACTIVE`` for real-time display with keyboard navigation (requires TTY), "
+        "or ``--mode text-only`` for text output with timestamps (works without TTY). "
+        "The monitoring session continues until manually stopped by the user or the specified timeout is reached. "
+        "In interactive mode, use keyboard shortcuts: up/down to scroll through resources, 'q' to quit. "
+        "In TEXT-ONLY mode, press Ctrl+C to stop monitoring."
     )
 
     ARG_TABLE = [
@@ -96,6 +102,16 @@ class ECSMonitorExpressGatewayService(BasicCommand):
             'default': 'RESOURCE',
             'choices': ['RESOURCE', 'DEPLOYMENT'],
         },
+        {
+            'name': 'mode',
+            'help_text': (
+                "Display mode for monitoring output. "
+                "interactive (default if TTY available) - Real-time display with spinner and keyboard navigation. "
+                "text-only - Text output with timestamps and change detection (works without TTY)."
+            ),
+            'required': False,
+            'choices': ['interactive', 'text-only'],
+        },
         {
             'name': 'timeout',
             'help_text': (
@@ -126,15 +142,12 @@ def _run_main(self, parsed_args, parsed_globals):
             parsed_globals: Global CLI configuration including region and endpoint
         """
         try:
-            # Check if running in a TTY for interactive display
-            if not sys.stdout.isatty():
-                uni_print(
-                    "Error: This command requires a TTY. "
-                    "Please run this command in a terminal.",
-                    sys.stderr,
-                )
-                return 1
+            display_mode = self._determine_display_mode(parsed_args.mode)
+        except ValueError as e:
+            uni_print(str(e), sys.stderr)
+            return 1
 
+        try:
             self._client = self._session.create_client(
                 'ecs',
                 region_name=parsed_globals.region,
@@ -149,14 +162,52 @@ def _run_main(self, parsed_args, parsed_globals):
                 self._client,
                 parsed_args.service_arn,
                 parsed_args.resource_view,
+                display_mode,
                 timeout_minutes=parsed_args.timeout,
                 use_color=use_color,
             ).exec()
         except MonitoringError as e:
             uni_print(f"Error monitoring service: {e}", sys.stderr)
+            return 1
+
+    def _determine_display_mode(self, requested_mode):
+        """Determine and validate the display mode.
+
+        Args:
+            requested_mode: User-requested mode ('interactive', 'text-only', or None)
+
+        Returns:
+            str: Validated display mode ('interactive' or 'text-only')
+
+        Raises:
+            ValueError: If interactive mode is requested without TTY
+        """
+        # Determine display mode with auto-detection
+        if requested_mode is None:
+            # Auto-detect: interactive if TTY available, else text-only
+            return 'interactive' if sys.stdout.isatty() else 'text-only'
+
+        # Validate requested mode
+        if requested_mode == 'interactive':
+            if not sys.stdout.isatty():
+                raise ValueError(
+                    "Error: Interactive mode requires a TTY (terminal). "
+                    "Use --mode text-only for non-interactive environments."
+                )
+            return 'interactive'
+
+        # text-only mode doesn't require TTY
+        return requested_mode
 
     def _should_use_color(self, parsed_globals):
-        """Determine if color output should be used based on global settings."""
+        """Determine if color output should be used based on global settings.
+
+        Args:
+            parsed_globals: Global CLI configuration
+
+        Returns:
+            bool: True if color should be used
+        """
         if parsed_globals.color == 'on':
             return True
         elif parsed_globals.color == 'off':
@@ -175,104 +226,51 @@ class ECSExpressGatewayServiceWatcher:
     Args:
         client: ECS client for API calls
         service_arn (str): ARN of the service to monitor
-        mode (str): Monitoring mode - 'RESOURCE' or 'DEPLOYMENT'
+        resource_view (str): Resource view mode - 'RESOURCE' or 'DEPLOYMENT'
+        display_mode (str): Display mode - 'INTERACTIVE' or 'TEXT-ONLY'
         timeout_minutes (int): Maximum monitoring time in minutes (default: 30)
     """
 
     def __init__(
         self,
         client,
         service_arn,
-        mode,
+        resource_view,
+        display_mode,
         timeout_minutes=30,
         display=None,
         use_color=True,
         collector=None,
     ):
         self._client = client
         self.service_arn = service_arn
-        self.mode = mode
-        self.timeout_minutes = timeout_minutes
+        self.display_mode = display_mode
         self.start_time = time.time()
-        self.use_color = use_color
-        self.display = display or Display()
+        self.timeout_minutes = timeout_minutes
         self.collector = collector or ServiceViewCollector(
-            client, service_arn, mode, use_color
+            client, service_arn, resource_view, use_color
         )
-
-    @staticmethod
-    def is_monitoring_available():
-        """Check if monitoring is available (requires TTY)."""
-        return sys.stdout.isatty()
+        self.display = display or Display()
 
     def exec(self):
-        """Start monitoring the express gateway service with progress display."""
-
-        def monitor_service(spinner_char):
-            return self.collector.get_current_view(spinner_char)
-
-        asyncio.run(self._execute_with_progress_async(monitor_service, 100))
-
-    async def _execute_with_progress_async(
-        self, execution, progress_refresh_millis, execution_refresh_millis=5000
-    ):
-        """Execute monitoring loop with animated progress display."""
-        spinner_chars = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
-        spinner_index = 0
-
-        current_output = "Waiting for initial data"
-
-        async def update_data():
-            nonlocal current_output
-            while True:
-                current_time = time.time()
-                if current_time - self.start_time > self.timeout_minutes * 60:
-                    break
-                try:
-                    loop = asyncio.get_event_loop()
-                    new_output = await loop.run_in_executor(
-                        None, execution, "{SPINNER}"
-                    )
-                    current_output = new_output
-                except ClientError as e:
-                    if (
-                        e.response.get('Error', {}).get('Code')
-                        == 'InvalidParameterException'
-                    ):
-                        error_message = e.response.get('Error', {}).get(
-                            'Message', ''
-                        )
-                        if (
-                            "Cannot call DescribeServiceRevisions for a service that is INACTIVE"
-                            in error_message
-                        ):
-                            current_output = "Service is inactive"
-                        else:
-                            raise
-                    else:
-                        raise
-                await asyncio.sleep(execution_refresh_millis / 1000.0)
-
-        async def update_spinner():
-            nonlocal spinner_index
-            while True:
-                spinner_char = spinner_chars[spinner_index]
-                display_output = current_output.replace(
-                    "{SPINNER}", spinner_char
-                )
-                status_text = f"Getting updates... {spinner_char} | up/down to scroll, q to quit"
-                self.display.display(display_output, status_text)
-                spinner_index = (spinner_index + 1) % len(spinner_chars)
-                await asyncio.sleep(progress_refresh_millis / 1000.0)
+        """Execute monitoring using the appropriate display strategy."""
+        strategy = self._create_display_strategy()
+        strategy.execute(
+            collector=self.collector,
+            start_time=self.start_time,
+            timeout_minutes=self.timeout_minutes,
+        )
 
-        # Start both tasks
-        data_task = asyncio.create_task(update_data())
-        spinner_task = asyncio.create_task(update_spinner())
+    def _create_display_strategy(self):
+        """Create display strategy based on display mode.
 
-        try:
-            await self.display.run()
-        finally:
-            data_task.cancel()
-            spinner_task.cancel()
-            final_output = current_output.replace("{SPINNER}", "")
-            uni_print(final_output + "\nMonitoring Complete!\n")
+        Returns:
+            DisplayStrategy: Appropriate strategy for the selected mode
+        """
+        if self.display_mode == 'text-only':
+            return TextOnlyDisplayStrategy(use_color=self.collector.use_color)
+        else:
+            return InteractiveDisplayStrategy(
+                display=self.display,
+                use_color=self.collector.use_color,
+            )