From d02f9844a4c9c645f2674dcaeff212e8fd0d6ea4 Mon Sep 17 00:00:00 2001 From: Andrew Kay Date: Fri, 27 Dec 2019 14:18:51 -0600 Subject: [PATCH] Add support for TN3270 alarm and VT100 bell --- oec/controller.py | 6 +++-- oec/terminal.py | 16 +++++++++++- oec/tn3270.py | 2 ++ oec/vt100.py | 8 ++++++ requirements.txt | 4 +-- tests/test_controller.py | 19 +++++++++++++- tests/test_terminal.py | 11 ++++++++ tests/test_vt100.py | 56 ++++++++++++++++++++++++++-------------- 8 files changed, 96 insertions(+), 26 deletions(-) create mode 100644 tests/test_terminal.py diff --git a/oec/controller.py b/oec/controller.py index 04c156e..4e80653 100644 --- a/oec/controller.py +++ b/oec/controller.py @@ -5,7 +5,7 @@ oec.controller import time import logging -from coax import poll, poll_ack, KeystrokePollResponse, ReceiveTimeout, \ +from coax import poll, poll_ack, PollAction, KeystrokePollResponse, ReceiveTimeout, \ ReceiveError, ProtocolError from .terminal import Terminal, read_terminal_ids @@ -166,7 +166,9 @@ class Controller: self.last_poll_time = time.perf_counter() - poll_response = poll(self.interface, timeout=1) + poll_action = self.terminal.get_poll_action() if self.terminal else PollAction.NONE + + poll_response = poll(self.interface, poll_action, timeout=1) if poll_response: try: diff --git a/oec/terminal.py b/oec/terminal.py index c3832a8..c7867b9 100644 --- a/oec/terminal.py +++ b/oec/terminal.py @@ -5,7 +5,8 @@ oec.terminal import time import logging -from coax import read_terminal_id, read_extended_id, ReceiveError, ProtocolError +from coax import read_terminal_id, read_extended_id, PollAction, ReceiveError, \ + ProtocolError from .display import Dimensions, Display from .keyboard import Keyboard @@ -67,3 +68,16 @@ class Terminal: self.display = Display(interface, dimensions) self.keyboard = Keyboard(keymap) + + self.alarm = False + + def sound_alarm(self): + self.alarm = True + + def get_poll_action(self): + if self.alarm: + self.alarm = False + + return PollAction.ALARM + + return PollAction.NONE diff --git a/oec/tn3270.py b/oec/tn3270.py index 0d2d28d..cfb5506 100644 --- a/oec/tn3270.py +++ b/oec/tn3270.py @@ -71,6 +71,8 @@ class TN3270Session(Session): self.emulator = Emulator(self.telnet, rows, columns) + self.emulator.alarm = lambda: self.terminal.sound_alarm() + def terminate(self): if self.telnet: self._disconnect_host() diff --git a/oec/vt100.py b/oec/vt100.py index 7cf4673..33a4dc8 100644 --- a/oec/vt100.py +++ b/oec/vt100.py @@ -86,6 +86,14 @@ class VT100Session(Session): self.vt100_screen.write_process_input = lambda data: self.host_process.write(data.encode()) + # Unfortunately multiple VT100 bells will be replaced with a single 3270 terminal + # alarm - also because the alarm is only sounded on terminal POLL the alarm sound + # may appear out of sync with the terminal. + # + # A better approach may be to perform a flush when the bell is encountered but + # that does not appear possible with the standard pyte ByteStream. + self.vt100_screen.bell = lambda: self.terminal.sound_alarm() + self.vt100_stream = pyte.ByteStream(self.vt100_screen) def start(self): diff --git a/requirements.txt b/requirements.txt index 6886420..2ec460a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,8 @@ ptyprocess==0.6.0 -pycoax==0.1.2 +pycoax==0.2.0 pyserial==3.4 pyte==0.8.0 -pytn3270==0.3.0 +pytn3270==0.4.0 sliplib==0.3.0 sortedcontainers==2.1.0 wcwidth==0.1.7 diff --git a/tests/test_controller.py b/tests/test_controller.py index f7bc5cd..1250eac 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -1,6 +1,6 @@ import unittest from unittest.mock import Mock, PropertyMock, patch -from coax import PowerOnResetCompletePollResponse, KeystrokePollResponse, ReceiveTimeout +from coax import PollAction, PowerOnResetCompletePollResponse, KeystrokePollResponse, ReceiveTimeout from coax.protocol import TerminalId import context @@ -97,6 +97,23 @@ class RunLoopTestCase(unittest.TestCase): self.assertEqual(self.create_session_mock.call_count, 2) + def test_alarm(self): + # Arrange + self._assert_run_loop(0, PowerOnResetCompletePollResponse(0xa), 0, True) + self._assert_run_loop(0, None, 0, False) + + self.assertIsNotNone(self.controller.terminal) + + # Act + self.controller.terminal.sound_alarm() + + # Assert + self._assert_run_loop(0.5, None, 0.5, False) + + self.assertEqual(self.poll_mock.call_args[0][1], PollAction.ALARM) + + self.assertFalse(self.controller.terminal.alarm) + def _assert_run_loop(self, poll_time, poll_response, expected_delay, expected_poll_ack): # Arrange self.poll_mock.side_effect = [poll_response] diff --git a/tests/test_terminal.py b/tests/test_terminal.py new file mode 100644 index 0000000..df4001f --- /dev/null +++ b/tests/test_terminal.py @@ -0,0 +1,11 @@ +import unittest + +import context + +from oec.keymap_3278_2 import KEYMAP as KEYMAP_3278_2 + +class TerminalGetPollActionTestCase(unittest.TestCase): + def setUp(self): + self.interface = Mock() + + self.terminal = Terminal(self.interface, TerminalId(0b11110100), 'c1348300', KEYMAP_3278_2) diff --git a/tests/test_vt100.py b/tests/test_vt100.py index b974d32..333d8cc 100644 --- a/tests/test_vt100.py +++ b/tests/test_vt100.py @@ -8,34 +8,50 @@ from oec.keyboard import Key, KeyboardModifiers from oec.vt100 import VT100Session, select class SessionHandleHostTestCase(unittest.TestCase): - @patch('oec.vt100.select') - def test(self, select_mock): + def setUp(self): + self.terminal = Mock() + + self.terminal.display.dimensions = Dimensions(24, 80) + + self.session = VT100Session(self.terminal, None) + + self.session.host_process = Mock() + + patcher = patch('oec.vt100.select') + + select_mock = patcher.start() + + select_mock.return_value = [[self.session.host_process]] + + self.addCleanup(patch.stopall) + + def test(self): # Arrange - terminal = Mock() - - terminal.display.dimensions = Dimensions(24, 80) - - session = VT100Session(terminal, None) - - session.host_process = Mock() - - session.host_process.read = Mock(return_value=b'abc') - - select_mock.return_value = [[session.host_process]] + self.session.host_process.read = Mock(return_value=b'abc') # Act - session.handle_host() + self.session.handle_host() # Assert - terminal.display.buffered_write.assert_any_call(0x80, row=0, column=0) - terminal.display.buffered_write.assert_any_call(0x81, row=0, column=1) - terminal.display.buffered_write.assert_any_call(0x82, row=0, column=2) + self.terminal.display.buffered_write.assert_any_call(0x80, row=0, column=0) + self.terminal.display.buffered_write.assert_any_call(0x81, row=0, column=1) + self.terminal.display.buffered_write.assert_any_call(0x82, row=0, column=2) - terminal.display.flush.assert_called() + self.terminal.display.flush.assert_called() - terminal.display.move_cursor.assert_called_with(row=0, column=3) + self.terminal.display.move_cursor.assert_called_with(row=0, column=3) - self.assertFalse(session.vt100_screen.dirty) + self.assertFalse(self.session.vt100_screen.dirty) + + def test_bell(self): + # Arrange + self.session.host_process.read = Mock(return_value=b'\a') + + # Act + self.session.handle_host() + + # Assert + self.terminal.sound_alarm.assert_called() class SessionHandleKeyTestCase(unittest.TestCase): def setUp(self):