Add insert support to TN3270

oec/display.py

@@ -307,7 +307,13 @@ class StatusLine:
 
         if modifiers.is_shift():
             indicators[0] = 0xda
-        else:
-            indicators[0] = 0x00
 
         self.write(35, indicators)
+
+    def write_keyboard_insert(self, insert):
+        indicators = bytearray(1)
+
+        if insert:
+            indicators[0] = 0xd3
+
+        self.write(45, indicators)

oec/tn3270.py

@@ -5,7 +5,7 @@ oec.tn3270
 
 import logging
 from tn3270 import Telnet, Emulator, AttributeCell, CharacterCell, AID, OperatorError, \
-                   ProtectedCellOperatorError
+                   ProtectedCellOperatorError, FieldOverflowOperatorError
 
 from .session import Session, SessionDisconnectedError
 from .display import encode_ebcdic_character, encode_string
@@ -56,6 +56,7 @@ class TN3270Session(Session):
         self.telnet = None
         self.emulator = None
 
+        self.keyboard_insert = False
         self.waiting_on_host = False
         self.operator_error = None
 
@@ -119,20 +120,26 @@ class TN3270Session(Session):
                 self.emulator.cursor_left()
             elif key == Key.RIGHT:
                 self.emulator.cursor_right()
-            #elif key == Key.INSERT:
+            elif key == Key.INSERT:
+                self._handle_insert_key()
             elif key == Key.DELETE:
                 self.emulator.delete()
             else:
                 byte = get_ebcdic_character_for_key(key)
 
                 if byte:
-                    self.emulator.input(byte)
+                    self.emulator.input(byte, self.keyboard_insert)
         except OperatorError as error:
             self.operator_error = error
 
         self._apply()
         self._flush()
 
+    def _handle_insert_key(self):
+        self.keyboard_insert = not self.keyboard_insert
+
+        self.terminal.display.status_line.write_keyboard_insert(self.keyboard_insert)
+
     def _connect_host(self):
         terminal_type = f'IBM-3278-{self.terminal.terminal_id.model}'
 
@@ -204,6 +211,9 @@ class TN3270Session(Session):
         elif isinstance(self.operator_error, ProtectedCellOperatorError):
             # X SPACE ARROW_LEFT OPERATOR ARROW_RIGHT
             message_area = b'\xf6\x00\xf8\xdb\xd8'
+        elif isinstance(self.operator_error, FieldOverflowOperatorError):
+            # X SPACE OPERATOR >
+            message_area = b'\xf6\x00\xdb' + encode_string('>')
         elif self.emulator.keyboard_locked:
             # X SPACE SYSTEM
             message_area = b'\xf6\x00' + encode_string('SYSTEM')

requirements.txt

@@ -2,7 +2,7 @@ ptyprocess==0.6.0
 pycoax==0.1.2
 pyserial==3.4
 pyte==0.8.0
-pytn3270==0.2.0
+pytn3270==0.3.0
 sliplib==0.3.0
 sortedcontainers==2.1.0
 wcwidth==0.1.7

tests/test_tn3270.py

@@ -6,7 +6,7 @@ import context
 from oec.session import SessionDisconnectedError
 from oec.keyboard import Key, KeyboardModifiers
 from oec.tn3270 import TN3270Session
-from tn3270 import AttributeCell, CharacterCell, AID, ProtectedCellOperatorError
+from tn3270 import AttributeCell, CharacterCell, AID, ProtectedCellOperatorError, FieldOverflowOperatorError
 
 class SessionHandleHostTestCase(unittest.TestCase):
     def setUp(self):
@@ -188,7 +188,20 @@ class SessionHandleKeyTestCase(unittest.TestCase):
         self.session.handle_key(Key.LOWER_A, KeyboardModifiers.NONE, None)
 
         # Assert
-        self.session.emulator.input.assert_called_with(0x81)
+        self.session.emulator.input.assert_called_with(0x81, False)
+
+    def test_insert(self):
+        # Act
+        self.session.handle_key(Key.INSERT, KeyboardModifiers.NONE, None)
+
+        self.session.handle_key(Key.LOWER_A, KeyboardModifiers.NONE, None)
+
+        self.session.handle_key(Key.INSERT, KeyboardModifiers.NONE, None)
+
+        # Assert
+        self.assertFalse(self.session.keyboard_insert)
+
+        self.session.emulator.input.assert_called_with(0x81, True)
 
     def test_protected_cell_operator_error(self):
         # Arrange
@@ -200,6 +213,16 @@ class SessionHandleKeyTestCase(unittest.TestCase):
         # Assert
         self.terminal.display.status_line.write.assert_called_with(8, bytes.fromhex('f600f8dbd800000000'))
 
+    def test_field_overflow_operator_error(self):
+        # Arrange
+        self.session.emulator.input = Mock(side_effect=FieldOverflowOperatorError)
+
+        # Act
+        self.session.handle_key(Key.LOWER_A, KeyboardModifiers.NONE, None)
+
+        # Assert
+        self.terminal.display.status_line.write.assert_called_with(8, bytes.fromhex('f600db080000000000'))
+
 class MockDisplay:
     def __init__(self, rows, columns):
         self.buffer = bytearray(rows * columns)