33See: https://github.com/OpenHands/software-agent-sdk/issues/2244
44"""
55
6+ import importlib .util
67import sys
78from unittest import mock
89
@@ -20,19 +21,22 @@ def test_flush_stdin_returns_zero_when_not_tty(self):
2021 result = flush_stdin ()
2122 assert result == 0
2223
23- def test_flush_stdin_returns_zero_on_windows (self ):
24- """flush_stdin should return 0 on Windows where termios is not available."""
25- with mock .patch .object (sys .stdin , "isatty" , return_value = True ):
26- with mock .patch .dict (sys .modules , {"termios" : None }):
27- # Force ImportError by making termios module fail to import
28- with mock .patch (
29- "openhands.sdk.logger.logger.flush_stdin"
30- ) as mock_flush :
31- # Since we can't easily mock ImportError for termios,
32- # we test that the function handles it gracefully
33- mock_flush .return_value = 0
34- result = mock_flush ()
35- assert result == 0
24+ def test_flush_stdin_returns_zero_when_termios_unavailable (self ):
25+ """flush_stdin should return 0 when termios is not available (e.g., Windows).
26+
27+ Note: This test verifies the code path by checking the function's behavior
28+ when termios import fails. On platforms where termios is available, we
29+ simulate its absence by patching the import mechanism.
30+ """
31+ if importlib .util .find_spec ("termios" ) is None :
32+ # On Windows/platforms without termios, test the real behavior
33+ with mock .patch .object (sys .stdin , "isatty" , return_value = True ):
34+ result = flush_stdin ()
35+ assert result == 0
36+ else :
37+ # On Unix, we can't easily unload termios since it's already imported.
38+ # The termios unavailable path is covered by Windows CI.
39+ pytest .skip ("termios is available; Windows CI covers the unavailable path" )
3640
3741 def test_flush_stdin_is_exported (self ):
3842 """flush_stdin should be available in the public API."""
@@ -42,8 +46,6 @@ def test_flush_stdin_is_exported(self):
4246
4347 def test_flush_stdin_handles_oserror_gracefully (self ):
4448 """flush_stdin should handle OSError gracefully."""
45- import importlib .util
46-
4749 if importlib .util .find_spec ("termios" ) is None :
4850 pytest .skip ("termios not available on this platform" )
4951
@@ -54,8 +56,6 @@ def test_flush_stdin_handles_oserror_gracefully(self):
5456
5557 def test_flush_stdin_handles_termios_error_gracefully (self ):
5658 """flush_stdin should handle termios.error gracefully."""
57- import importlib .util
58-
5959 if importlib .util .find_spec ("termios" ) is None :
6060 pytest .skip ("termios not available on this platform" )
6161
@@ -73,14 +73,32 @@ def test_flush_stdin_handles_termios_error_gracefully(self):
7373class TestFlushStdinIntegration :
7474 """Integration tests for flush_stdin in conversation flow."""
7575
76- def test_flush_stdin_imported_in_local_conversation (self ):
77- """flush_stdin should be imported in LocalConversation."""
78- from openhands .sdk .conversation .impl import local_conversation
76+ def test_flush_stdin_called_in_visualizer_on_event (self ):
77+ """Verify flush_stdin is called before rendering events in the visualizer."""
78+ from openhands .sdk .conversation .visualizer .default import (
79+ DefaultConversationVisualizer ,
80+ )
81+ from openhands .sdk .event import PauseEvent
82+
83+ visualizer = DefaultConversationVisualizer ()
84+
85+ with mock .patch (
86+ "openhands.sdk.conversation.visualizer.default.flush_stdin"
87+ ) as mock_flush :
88+ # Create a simple event and trigger on_event
89+ event = PauseEvent ()
90+ visualizer .on_event (event )
91+
92+ # Verify flush_stdin was called
93+ mock_flush .assert_called_once ()
7994
80- assert hasattr (local_conversation , "flush_stdin" )
95+ def test_flush_stdin_registered_with_atexit (self ):
96+ """Verify flush_stdin is registered as an atexit handler."""
8197
82- def test_flush_stdin_imported_in_default_visualizer (self ):
83- """flush_stdin should be imported in DefaultConversationVisualizer."""
84- from openhands .sdk .conversation .visualizer import default
98+ # Get registered atexit functions
99+ # Note: atexit._run_exitfuncs() would run them, but we just check registration
100+ # The atexit module doesn't expose a clean way to inspect handlers,
101+ # but we can verify by checking the module-level flag
102+ from openhands .sdk .logger import logger as logger_module
85103
86- assert hasattr ( default , "flush_stdin" )
104+ assert logger_module . _cleanup_registered is True
0 commit comments