Address PR comments

Author: jutyler1

awscli/customizations/ecs/expressgateway/display_strategy.py

@@ -27,7 +27,7 @@ class DisplayStrategy:
     Each strategy controls its own execution model, timing, and output format.
     """
 
-    def execute(self, collector, start_time, timeout_minutes):
+    def execute_monitoring(self, collector, start_time, timeout_minutes):
         """Execute the monitoring loop.
 
         Args:
@@ -47,10 +47,17 @@ class InteractiveDisplayStrategy(DisplayStrategy):
     """
 
     def __init__(self, display, use_color):
+        """Initialize the interactive display strategy.
+
+        Args:
+            display: Display instance from prompt_toolkit_display module
+                providing the interactive terminal interface
+            use_color: Whether to use colored output
+        """
         self.display = display
         self.use_color = use_color
 
-    def execute(self, collector, start_time, timeout_minutes):
+    def execute_monitoring(self, collector, start_time, timeout_minutes):
         """Execute async monitoring with spinner and keyboard controls."""
         final_output, timed_out = asyncio.run(
             self._execute_async(collector, start_time, timeout_minutes)
@@ -143,6 +150,8 @@ async def update_spinner():
                     pass
 
         finally:
+            # Ensure display app is properly shut down
+            self.display.app.exit()
             spinner_task.cancel()
             if display_task is not None and not display_task.done():
                 display_task.cancel()

tests/unit/customizations/ecs/expressgateway/test_display_strategy.py

@@ -28,7 +28,7 @@ def test_base_strategy_not_implemented(self):
         """Test base class raises NotImplementedError."""
         strategy = DisplayStrategy()
         with pytest.raises(NotImplementedError):
-            strategy.execute(None, None, None)
+            strategy.execute_monitoring(None, None, None)
 
 
 @pytest.fixture
@@ -71,7 +71,9 @@ def test_execute_with_mock_display(
         mock_sleep.side_effect = KeyboardInterrupt()
 
         start_time = time.time()
-        strategy.execute(mock_collector, start_time, timeout_minutes=1)
+        strategy.execute_monitoring(
+            mock_collector, start_time, timeout_minutes=1
+        )
 
         # Verify display was called
         assert mock_display.display.called
@@ -106,7 +108,9 @@ def test_completion_message_on_normal_exit(
         mock_sleep.side_effect = KeyboardInterrupt()
 
         start_time = time.time()
-        strategy.execute(mock_collector, start_time, timeout_minutes=1)
+        strategy.execute_monitoring(
+            mock_collector, start_time, timeout_minutes=1
+        )
 
         captured = capsys.readouterr()
         assert "Monitoring Complete!" in captured.out
@@ -128,7 +132,9 @@ def test_collector_output_is_displayed(
         mock_sleep.side_effect = KeyboardInterrupt()
 
         start_time = time.time()
-        strategy.execute(mock_collector, start_time, timeout_minutes=1)
+        strategy.execute_monitoring(
+            mock_collector, start_time, timeout_minutes=1
+        )
 
         captured = capsys.readouterr()
         assert unique_output in captured.out
@@ -157,7 +163,9 @@ def test_execute_handles_service_inactive(
         mock_sleep.side_effect = KeyboardInterrupt()
 
         start_time = time.time()
-        strategy.execute(mock_collector, start_time, timeout_minutes=1)
+        strategy.execute_monitoring(
+            mock_collector, start_time, timeout_minutes=1
+        )
 
         # Strategy should handle the error and set output to "Service is inactive"
         captured = capsys.readouterr()
@@ -190,8 +198,35 @@ def test_execute_other_client_errors_propagate(
 
         # Other client errors should propagate
         with pytest.raises(ClientError) as exc_info:
-            strategy.execute(mock_collector, start_time, timeout_minutes=1)
+            strategy.execute_monitoring(
+                mock_collector, start_time, timeout_minutes=1
+            )
 
         assert (
             exc_info.value.response['Error']['Code'] == 'AccessDeniedException'
         )
+
+    @patch('time.sleep')
+    def test_display_cleanup_on_exception(
+        self, mock_sleep, app_session, mock_display
+    ):
+        """Test display app is properly shut down when exception occurs."""
+        mock_collector = Mock()
+        error = ClientError(
+            error_response={'Error': {'Code': 'ThrottlingException'}},
+            operation_name='DescribeServiceRevisions',
+        )
+        mock_collector.get_current_view = Mock(side_effect=error)
+
+        strategy = InteractiveDisplayStrategy(
+            display=mock_display, use_color=True
+        )
+        mock_sleep.side_effect = KeyboardInterrupt()
+
+        with pytest.raises(ClientError):
+            strategy.execute_monitoring(
+                mock_collector, time.time(), timeout_minutes=1
+            )
+
+        # Verify app.exit() was called in finally block despite exception
+        mock_display.app.exit.assert_called()