Merge pull request #122 from jkonecny12/1.1-add-disable-concurrency-check

Add support to disable concurrency check

Author: jkonecny12

simpleline/input/input_handler.py

@@ -52,6 +52,9 @@ def __init__(self, callback=None, source=None):
         self._skip_concurrency_check = False
         self._source = source
 
+        self._register_input_ready_signal()
+
+    def _register_input_ready_signal(self):
         App.get_event_loop().register_signal_handler(InputReadySignal,
                                                      self._input_received_handler)
 

simpleline/render/screen/__init__.py

@@ -87,6 +87,14 @@ def title(self, title):
         """
         self._title = title
 
+    @property
+    def input_manager(self):
+        """Get input manager.
+
+        The input manager could be used to tweak input settings.
+        """
+        return self._input_manager
+
     @property
     def password_func(self):
         """Get password function.

simpleline/render/screen/input_manager.py

@@ -46,6 +46,7 @@ def __init__(self, ui_screen):
         self._ui_screen = ui_screen
         self._input_error_counter = 0
         self._input_error_threshold = 5
+        self._skip_concurrency_check = False
         self._input_args = None
 
     @property
@@ -62,6 +63,25 @@ def input_error_threshold_exceeded(self):
         errors = self._input_error_counter % self._input_error_threshold
         return errors == 0
 
+    @property
+    def skip_concurrency_check(self):
+        """Should the concurrency check be skipped?
+
+        :returns bool: True if the check should be skipped.
+        """
+        return self._skip_concurrency_check
+
+    @skip_concurrency_check.setter
+    def skip_concurrency_check(self, value):
+        """Set if the concurrency check should be skipped when asking for user input.
+
+        WARNING: Use this option with caution. When the concurrency check is disabled you
+                 can easily get to unexpected behavior which is hard to debug.
+
+        :param bool value: True to skip the concurrency check.
+        """
+        self._skip_concurrency_check = value
+
     def get_input_blocking(self, message, hidden):
         """Get blocking input from the user.
 
@@ -78,6 +98,7 @@ def get_input_blocking(self, message, hidden):
         else:
             handler = InputHandler(source=self)
 
+        handler.skip_concurrency_check = self._skip_concurrency_check
         handler.get_input(message)
         handler.wait_on_input()
         return handler.value
@@ -101,6 +122,7 @@ def get_input(self, args=None):
             if self._ui_screen.password_func:
                 handler.set_pass_func(self._ui_screen.password_func)
 
+        handler.skip_concurrency_check = self._skip_concurrency_check
         handler.set_callback(self.process_input)
         handler.get_input(prompt)
 

tests/simpleline_tests/adv_widgets_test.py

@@ -23,10 +23,98 @@
 from io import StringIO
 
 from . import UtilityMixin
-
+from simpleline.render.screen import UIScreen
+from simpleline.input.input_handler import InputHandler
 from simpleline.render.adv_widgets import GetInputScreen, GetPasswordInputScreen
 
 
+class InputHandlerMock(InputHandler):
+    """Dummy InputHandler instance.
+
+    This class based on InputHandler has all threading logic disabled for simpler testing.
+    """
+    def _register_input_ready_signal(self):
+        pass
+
+    def get_input(self, prompt):
+        pass
+
+    def wait_on_input(self):
+        pass
+
+
+class UIScreen_TestCase(unittest.TestCase):
+    @patch('simpleline.render.screen.input_manager.InputHandler')
+    def test_uiscreen_disable_concurrency_check(self, input_handler_class_mock):
+        # Replace default InputHandler instance created by InputManager by our mock to have
+        # it stored after use.
+        input_handler_instance_mock = InputHandlerMock()
+        input_handler_class_mock.return_value = input_handler_instance_mock
+
+        screen = UIScreen()
+        input_manager = screen.input_manager
+        input_manager.skip_concurrency_check = True
+        screen.get_user_input("test")
+
+        self.assertTrue(screen.input_manager.skip_concurrency_check)
+        # Check that the created InputHandler instance has the flag correctly set.
+        # We don't need to check the concurrency check functionality, it is tested
+        # elsewhere already.
+        self.assertTrue(input_handler_instance_mock.skip_concurrency_check)
+
+    @patch('simpleline.render.screen.input_manager.InputHandler')
+    def test_uiscreen_password_disable_concurrency_check(self, input_handler_class_mock):
+        # Replace default InputHandler instance created by InputManager by our mock to have
+        # it stored after use.
+        input_handler_instance_mock = InputHandlerMock()
+        input_handler_class_mock.return_value = input_handler_instance_mock
+
+        screen = GetPasswordInputScreen(message="Test prompt")
+        input_manager = screen.input_manager
+        input_manager.skip_concurrency_check = True
+        screen.get_user_input("test")
+
+        self.assertTrue(screen.input_manager.skip_concurrency_check)
+        # Check that the created InputHandler instance has the flag correctly set.
+        # We don't need to check the concurrency check functionality, it is tested
+        # elsewhere already.
+        self.assertTrue(input_handler_instance_mock.skip_concurrency_check)
+
+    @patch('simpleline.render.screen.input_manager.InputHandler')
+    def test_uiscreen_non_blocking_input_disable_concurrency_check(self, input_handler_class_mock):
+        # Replace default InputHandler instance created by InputManager by our mock to have
+        # it stored after use.
+        input_handler_instance_mock = InputHandlerMock()
+        input_handler_class_mock.return_value = input_handler_instance_mock
+
+        screen = UIScreen()
+        input_manager = screen.input_manager
+        input_manager.skip_concurrency_check = True
+        screen.get_input_with_error_check("test")
+
+        self.assertTrue(screen.input_manager.skip_concurrency_check)
+        # Check that the created InputHandler instance has the flag correctly set.
+        # We don't need to check the concurrency check functionality, it is tested
+        # elsewhere already.
+        self.assertTrue(input_handler_instance_mock.skip_concurrency_check)
+
+    @patch('simpleline.render.screen.input_manager.InputHandler')
+    def test_uiscreen_default_concurrency_check(self, input_handler_class_mock):
+        # Replace default InputHandler instance created by InputManager by our mock to have
+        # it stored after use.
+        input_handler_instance_mock = InputHandlerMock()
+        input_handler_class_mock.return_value = input_handler_instance_mock
+
+        screen = UIScreen()
+        screen.get_user_input("test")
+
+        self.assertFalse(screen.input_manager.skip_concurrency_check)
+        # Check that the created InputHandler instance has the flag correctly set.
+        # We don't need to check the concurrency check functionality, it is tested
+        # elsewhere already.
+        self.assertFalse(input_handler_instance_mock.skip_concurrency_check)
+
+
 @patch('simpleline.input.input_handler.InputHandlerRequest._get_input')
 @patch('sys.stdout', new_callable=StringIO)
 class AdvWidgets_TestCase(unittest.TestCase, UtilityMixin):