mirror of
https://github.com/lowobservable/oec.git
synced 2026-03-05 19:09:04 +00:00
Refactor run loop to improve perceived terminal responsiveness
This commit is contained in:
Andrew Kay
@@ -26,6 +26,8 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
|
||||
self.controller.connected_poll_period = 1
|
||||
|
||||
self.controller._update_session = Mock()
|
||||
|
||||
patcher = patch('oec.controller.poll')
|
||||
|
||||
self.poll_mock = patcher.start()
|
||||
@@ -73,34 +75,35 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
self.addCleanup(patch.stopall)
|
||||
|
||||
def test_no_terminal(self):
|
||||
self._assert_run_loop(0, ReceiveTimeout, 0, False)
|
||||
self._assert_run_loop(1, ReceiveTimeout, 4, False)
|
||||
self._assert_run_loop(0, ReceiveTimeout, False, 0, False)
|
||||
self._assert_run_loop(1, ReceiveTimeout, False, 4, False)
|
||||
|
||||
self.assertIsNone(self.controller.terminal)
|
||||
self.assertIsNone(self.controller.session)
|
||||
|
||||
def test_terminal_attached(self):
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
|
||||
self._assert_run_loop(0, None, 0, False)
|
||||
self._assert_run_loop(0.5, None, 0.5, False)
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), False, 0, True)
|
||||
self._assert_run_loop(0, None, False, 0, False)
|
||||
self._assert_run_loop(0.5, None, True, 0.5, False)
|
||||
|
||||
self.assertIsNotNone(self.controller.terminal)
|
||||
self.assertIsNotNone(self.controller.session)
|
||||
|
||||
self.controller.session.handle_host.assert_called()
|
||||
self.controller._update_session.assert_called()
|
||||
|
||||
def test_unsupported_terminal_attached(self):
|
||||
self.read_terminal_ids_mock.return_value = DFT_TERMINAL_IDS
|
||||
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), False, 0, True)
|
||||
|
||||
self.assertIsNone(self.controller.terminal)
|
||||
self.assertIsNone(self.controller.session)
|
||||
|
||||
def test_keystroke(self):
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0110000010), 0, True)
|
||||
self._assert_run_loop(0, None, 0, False)
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), False, 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0110000010), False, 0, True)
|
||||
self._assert_run_loop(0, None, False, 0, False)
|
||||
self._assert_run_loop(0.5, None, True, 0.5, False)
|
||||
|
||||
self.assertIsNotNone(self.controller.terminal)
|
||||
self.assertIsNotNone(self.controller.session)
|
||||
@@ -108,9 +111,9 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
self.controller.session.handle_key.assert_called_with(Key.LOWER_A, KeyboardModifiers.NONE, 96)
|
||||
|
||||
def test_terminal_detached(self):
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
|
||||
self._assert_run_loop(0, None, 0, False)
|
||||
self._assert_run_loop(0.5, ReceiveTimeout, 0.5, False)
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), False, 0, True)
|
||||
self._assert_run_loop(0, None, False, 0, False)
|
||||
self._assert_run_loop(0.5, ReceiveTimeout, True, 0.5, False)
|
||||
|
||||
self.assertIsNone(self.controller.terminal)
|
||||
self.assertIsNone(self.controller.session)
|
||||
@@ -118,12 +121,12 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
self.session_mock.terminate.assert_called()
|
||||
|
||||
def test_session_disconnected(self):
|
||||
self.session_mock.handle_host.side_effect = [None, SessionDisconnectedError, None]
|
||||
self.controller._update_session.side_effect = [None, SessionDisconnectedError, None]
|
||||
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
|
||||
self._assert_run_loop(0, None, 0, False)
|
||||
self._assert_run_loop(0.5, None, 0.5, False)
|
||||
self._assert_run_loop(1.5, None, 0.5, False)
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), False, 0, True)
|
||||
self._assert_run_loop(0, None, False, 0, False)
|
||||
self._assert_run_loop(0.5, None, True, 0.5, False)
|
||||
self._assert_run_loop(1.5, None, True, 0.5, False)
|
||||
|
||||
self.assertIsNotNone(self.controller.terminal)
|
||||
self.assertIsNotNone(self.controller.session)
|
||||
@@ -132,8 +135,8 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
|
||||
def test_alarm(self):
|
||||
# Arrange
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
|
||||
self._assert_run_loop(0, None, 0, False)
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), False, 0, True)
|
||||
self._assert_run_loop(0, None, False, 0, False)
|
||||
|
||||
self.assertIsNotNone(self.controller.terminal)
|
||||
|
||||
@@ -141,18 +144,18 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
self.controller.terminal.sound_alarm()
|
||||
|
||||
# Assert
|
||||
self._assert_run_loop(0.5, None, 0.5, False)
|
||||
self._assert_run_loop(0.5, None, True, 0.5, False)
|
||||
|
||||
self.assertEqual(self.poll_mock.call_args[0][1], PollAction.ALARM)
|
||||
|
||||
self.assertFalse(self.controller.terminal.alarm)
|
||||
|
||||
def test_toggle_cursor_blink(self):
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), False, 0, True)
|
||||
|
||||
self.assertFalse(self.controller.terminal.display.cursor_blink)
|
||||
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101010010), 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101010010), False, 0, True)
|
||||
|
||||
self.assertTrue(self.controller.terminal.display.cursor_blink)
|
||||
|
||||
@@ -162,7 +165,7 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
|
||||
self.load_control_register_mock.reset_mock()
|
||||
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101010010), 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101010010), False, 0, True)
|
||||
|
||||
self.assertFalse(self.controller.terminal.display.cursor_blink)
|
||||
|
||||
@@ -171,13 +174,13 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
self.assertFalse(self.load_control_register_mock.call_args[0][1].cursor_blink)
|
||||
|
||||
def test_toggle_cursor_reverse(self):
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), False, 0, True)
|
||||
|
||||
self.assertFalse(self.controller.terminal.display.cursor_reverse)
|
||||
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0100111110), 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101010010), 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0100111110), 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0100111110), False, 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101010010), False, 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0100111110), False, 0, True)
|
||||
|
||||
self.assertTrue(self.controller.terminal.display.cursor_reverse)
|
||||
|
||||
@@ -187,9 +190,9 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
|
||||
self.load_control_register_mock.reset_mock()
|
||||
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0100111110), 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101010010), 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0100111110), 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0100111110), False, 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101010010), False, 0, True)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0100111110), False, 0, True)
|
||||
|
||||
self.assertFalse(self.controller.terminal.display.cursor_reverse)
|
||||
|
||||
@@ -198,31 +201,33 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
self.assertFalse(self.load_control_register_mock.call_args[0][1].cursor_reverse)
|
||||
|
||||
def test_toggle_clicker(self):
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True)
|
||||
self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), False, 0, True)
|
||||
|
||||
self.assertFalse(self.controller.terminal.keyboard.clicker)
|
||||
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101011110), 0, True)
|
||||
self._assert_run_loop(0, None, 0, False)
|
||||
self._assert_run_loop(0, KeystrokePollResponse(0b0101011110), False, 0, True)
|
||||
self._assert_run_loop(0, None, False, 0, False)
|
||||
|
||||
self.assertTrue(self.controller.terminal.keyboard.clicker)
|
||||
|
||||
self.assertEqual(self.poll_mock.call_args[0][1], PollAction.ENABLE_KEYBOARD_CLICKER)
|
||||
|
||||
self._assert_run_loop(0.5, KeystrokePollResponse(0b0101011110), 0.5, True)
|
||||
self._assert_run_loop(1, None, 0, False)
|
||||
self._assert_run_loop(0.5, KeystrokePollResponse(0b0101011110), True, 0.5, True)
|
||||
self._assert_run_loop(1, None, False, 0, False)
|
||||
|
||||
self.assertFalse(self.controller.terminal.keyboard.clicker)
|
||||
|
||||
self.assertEqual(self.poll_mock.call_args[0][1], PollAction.DISABLE_KEYBOARD_CLICKER)
|
||||
|
||||
def _assert_run_loop(self, poll_time, poll_response, expected_delay, expected_poll_ack):
|
||||
def _assert_run_loop(self, poll_time, poll_response, expected_update_session, expected_poll_delay, expected_poll_ack):
|
||||
# Arrange
|
||||
self.controller._update_session.reset_mock()
|
||||
|
||||
self.poll_mock.side_effect = [poll_response]
|
||||
|
||||
self.poll_ack_mock.reset_mock()
|
||||
|
||||
self.perf_counter_mock.side_effect = [poll_time, poll_time + expected_delay]
|
||||
self.perf_counter_mock.side_effect = [poll_time, poll_time + expected_poll_delay]
|
||||
|
||||
self.sleep_mock.reset_mock()
|
||||
|
||||
@@ -230,12 +235,70 @@ class RunLoopTestCase(unittest.TestCase):
|
||||
self.controller._run_loop()
|
||||
|
||||
# Assert
|
||||
if expected_delay > 0:
|
||||
self.sleep_mock.assert_called_once_with(expected_delay)
|
||||
else:
|
||||
if expected_update_session:
|
||||
self.controller._update_session.assert_called_once_with(expected_poll_delay)
|
||||
self.sleep_mock.assert_not_called()
|
||||
else:
|
||||
self.controller._update_session.assert_not_called()
|
||||
|
||||
if expected_poll_delay > 0:
|
||||
self.sleep_mock.assert_called_once_with(expected_poll_delay)
|
||||
else:
|
||||
self.sleep_mock.assert_not_called()
|
||||
|
||||
if expected_poll_ack:
|
||||
self.poll_ack_mock.assert_called_once()
|
||||
else:
|
||||
self.poll_ack_mock.assert_not_called()
|
||||
|
||||
class UpdateSessionTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.controller = Controller(None, None, None)
|
||||
|
||||
self.controller.session = Mock()
|
||||
|
||||
patcher = patch('oec.controller.time.perf_counter')
|
||||
|
||||
self.perf_counter_mock = patcher.start()
|
||||
|
||||
patcher = patch('oec.controller.select')
|
||||
|
||||
self.select_mock = patcher.start()
|
||||
|
||||
def test_zero_duration(self):
|
||||
# Act
|
||||
self.controller._update_session(0)
|
||||
|
||||
# Assert
|
||||
self.controller.session.handle_host.assert_not_called()
|
||||
|
||||
self.select_mock.assert_not_called()
|
||||
|
||||
def test_select_timeout(self):
|
||||
# Arrange
|
||||
self.select_mock.return_value = ([], [], [])
|
||||
|
||||
# Act
|
||||
self.controller._update_session(1)
|
||||
|
||||
# Assert
|
||||
self.controller.session.handle_host.assert_not_called()
|
||||
|
||||
self.select_mock.assert_called_once()
|
||||
|
||||
def test_select_available(self):
|
||||
# Arrange
|
||||
self.perf_counter_mock.side_effect = [0, 0.75, 0.75]
|
||||
|
||||
self.select_mock.side_effect = [([self.controller.session], [], []), ([], [], [])]
|
||||
|
||||
# Act
|
||||
self.controller._update_session(1)
|
||||
|
||||
# Assert
|
||||
self.controller.session.handle_host.assert_called_once()
|
||||
|
||||
self.assertEqual(self.select_mock.call_count, 2)
|
||||
|
||||
self.assertEqual(self.select_mock.call_args_list[0][0][3], 1)
|
||||
self.assertEqual(self.select_mock.call_args_list[1][0][3], 0.25)
|
||||
|
||||
Reference in New Issue
Block a user