[v2] Fix tests to use pipe input

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

@@ -12,8 +12,6 @@
 
 import pytest
 from botocore.exceptions import ClientError
-from prompt_toolkit.application import create_app_session
-from prompt_toolkit.output import DummyOutput
 
 from awscli.customizations.ecs.expressgateway.display_strategy import (
     DisplayStrategy,
@@ -32,13 +30,6 @@ def test_base_strategy_not_implemented(self):
             strategy.execute_monitoring(None, None, None)
 
 
-@pytest.fixture
-def app_session():
-    """Fixture that creates and manages an app session for prompt_toolkit."""
-    with create_app_session(output=DummyOutput()) as session:
-        yield session
-
-
 @pytest.fixture
 def mock_display():
     """Fixture that creates a mock display for testing."""
@@ -57,7 +48,7 @@ class TestInteractiveDisplayStrategy:
 
     @patch('time.sleep')
     def test_execute_with_mock_display(
-        self, mock_sleep, app_session, mock_display
+        self, mock_sleep, ptk_app_session, mock_display
     ):
         """Test strategy executes with mocked display."""
         mock_collector = Mock()
@@ -96,7 +87,7 @@ def test_strategy_uses_provided_color_setting(self):
 
     @patch('time.sleep')
     def test_completion_message_on_normal_exit(
-        self, mock_sleep, app_session, mock_display, capsys
+        self, mock_sleep, ptk_app_session, mock_display, capsys
     ):
         """Test displays completion message when monitoring completes normally."""
         mock_collector = Mock()
@@ -119,7 +110,7 @@ def test_completion_message_on_normal_exit(
 
     @patch('time.sleep')
     def test_collector_output_is_displayed(
-        self, mock_sleep, app_session, mock_display, capsys
+        self, mock_sleep, ptk_app_session, mock_display, capsys
     ):
         """Test that collector output appears in final output."""
         mock_collector = Mock()
@@ -142,7 +133,7 @@ def test_collector_output_is_displayed(
 
     @patch('time.sleep')
     def test_execute_handles_service_inactive(
-        self, mock_sleep, app_session, mock_display, capsys
+        self, mock_sleep, ptk_app_session, mock_display, capsys
     ):
         """Test strategy handles service inactive error."""
         mock_collector = Mock()
@@ -174,7 +165,7 @@ def test_execute_handles_service_inactive(
 
     @patch('time.sleep')
     def test_execute_other_client_errors_propagate(
-        self, mock_sleep, app_session, mock_display
+        self, mock_sleep, ptk_app_session, mock_display
     ):
         """Test strategy propagates non-service-inactive ClientErrors."""
         mock_collector = Mock()
@@ -209,7 +200,7 @@ def test_execute_other_client_errors_propagate(
 
     @patch('time.sleep')
     def test_display_cleanup_on_exception(
-        self, mock_sleep, app_session, mock_display
+        self, mock_sleep, ptk_app_session, mock_display
     ):
         """Test display app is properly shut down when exception occurs."""
         mock_collector = Mock()

tests/unit/customizations/ecs/test_monitorexpressgatewayservice.py

@@ -10,14 +10,16 @@
 
 import pytest
 from botocore.exceptions import ClientError
-from prompt_toolkit.application import create_app_session
-from prompt_toolkit.output import DummyOutput
 
+from awscli.customizations.ecs.expressgateway.display_strategy import (
+    InteractiveDisplayStrategy
+)
 from awscli.customizations.ecs.exceptions import MonitoringError
 from awscli.customizations.ecs.monitorexpressgatewayservice import (
     ECSExpressGatewayServiceWatcher,
     ECSMonitorExpressGatewayService,
 )
+from awscli.customizations.ecs.prompt_toolkit_display import Display
 
 
 class TestECSMonitorExpressGatewayServiceCommand:
@@ -138,13 +140,6 @@ def test_text_only_mode_without_tty(self, mock_isatty, capsys):
         command._run_main(parsed_args, parsed_globals)
 
 
-@pytest.fixture
-def watcher_app_session():
-    """Fixture that creates and manages an app session for watcher tests."""
-    with create_app_session(output=DummyOutput()) as session:
-        yield session
-
-
 @pytest.fixture
 def service_arn():
     """Fixture that provides a test service ARN."""
@@ -191,7 +186,7 @@ def test_init_uses_injected_collector(self):
         assert watcher.collector == mock_collector
 
     def test_exec_calls_display_strategy_with_correct_parameters(
-        self, watcher_app_session
+        self, ptk_app_session
     ):
         """Test exec() calls display strategy with collector, start_time, and timeout"""
         mock_collector = Mock()
@@ -219,7 +214,7 @@ def test_exec_calls_display_strategy_with_correct_parameters(
         assert call_args.kwargs['timeout_minutes'] == 15
 
     def test_exec_propagates_exceptions_from_display_strategy(
-        self, watcher_app_session
+        self, ptk_app_session
     ):
         """Test exec() propagates exceptions from display strategy"""
         mock_display_strategy = Mock()
@@ -267,6 +262,10 @@ def test_monitoring_error_creation(self):
 class TestColorSupport:
     """Test color support functionality"""
 
+    @pytest.fixture
+    def display(self, ptk_app_session):
+        yield Display()
+
     def test_should_use_color_on(self):
         """Test _should_use_color returns True when color is 'on'"""
         command = ECSMonitorExpressGatewayService(Mock())
@@ -303,29 +302,20 @@ def test_should_use_color_auto_no_tty(self, mock_isatty):
 
         assert command._should_use_color(parsed_globals) is False
 
-    def test_watcher_accepts_use_color_parameter(self, watcher_app_session):
+    @pytest.mark.parametrize("use_color", [True, False])
+    def test_watcher_accepts_use_color_parameter(self, display, use_color):
         """Test ECSExpressGatewayServiceWatcher accepts use_color parameter"""
         mock_client = Mock()
 
-        # Test with use_color=True
-        watcher = ECSExpressGatewayServiceWatcher(
-            mock_client,
-            "arn:aws:ecs:us-east-1:123456789012:service/test-service",
-            "ALL",
-            "INTERACTIVE",
-            use_color=True,
-        )
-        assert watcher.collector.use_color is True
-
-        # Test with use_color=False
         watcher = ECSExpressGatewayServiceWatcher(
             mock_client,
             "arn:aws:ecs:us-east-1:123456789012:service/test-service",
             "ALL",
             "INTERACTIVE",
-            use_color=False,
+            use_color=use_color,
+            display_strategy=InteractiveDisplayStrategy(display, use_color),
         )
-        assert watcher.collector.use_color is False
+        assert watcher.collector.use_color is use_color
 
     def test_invalid_display_mode_raises_error(self):
         """Test that invalid display mode raises ValueError"""

tests/unit/customizations/ecs/test_prompt_toolkit_display.py

@@ -1,19 +1,15 @@
 import asyncio
-import sys
 from unittest.mock import Mock, patch
 
 import pytest
-from prompt_toolkit.application import create_app_session
-from prompt_toolkit.output import DummyOutput
 
 from awscli.customizations.ecs.prompt_toolkit_display import Display
 
 
 class TestPromptToolkitDisplay:
     @pytest.fixture
-    def display(self):
-        with create_app_session(output=DummyOutput()):
-            yield Display()
+    def display(self, ptk_app_session):
+        yield Display()
 
     def test_init(self, display):
         """Test Display initialization."""