Adding test for debounce method

Author: suoto

hdl_checker/tests/test_lsp_utils.py

@@ -19,17 +19,20 @@
 """
 
 import logging
+import time
+from multiprocessing import Queue
 from tempfile import mkdtemp
 from typing import Any
 
+import mock
 import parameterized  # type: ignore
 import unittest2  # type: ignore
-from mock import Mock
 from pygls.types import DiagnosticSeverity, MessageType, Position, Range
 
 from hdl_checker import lsp
 from hdl_checker.diagnostics import CheckerDiagnostic, DiagType
 from hdl_checker.path import Path, TemporaryPath
+from hdl_checker.utils import debounce
 
 _logger = logging.getLogger(__name__)
 
@@ -51,7 +54,7 @@ class TestCheckerDiagToLspDict(unittest2.TestCase):
             (DiagType.NONE, DiagnosticSeverity.Error),
         ]
     )
-    def test_converting_to_lsp(self, diag_type, severity):
+    def test_convertingDiagnosticType(self, diag_type, severity):
         # type: (...) -> Any
         """
         Test conversion between hdl_checker.DiagType to pygls.types.DiagnosticSeverity
@@ -81,12 +84,12 @@ def test_converting_to_lsp(self, diag_type, severity):
             ),
         )
 
-    def test_workspace_notify(self) -> None:  # pylint: disable=no-self-use
+    def test_workspaceNotify(self) -> None:  # pylint: disable=no-self-use
         """
         Test server notification messages call the appropriate LS methods
         """
-        workspace = Mock()
-        workspace.show_message = Mock()
+        workspace = mock.Mock()
+        workspace.show_message = mock.Mock()
 
         server = lsp.Server(
             workspace, root_dir=TemporaryPath(mkdtemp(prefix="hdl_checker_"))
@@ -105,3 +108,59 @@ def test_workspace_notify(self) -> None:  # pylint: disable=no-self-use
 
         server._handleUiError("some error")  # pylint: disable=protected-access
         workspace.show_message.assert_called_once_with("some error", MessageType.Error)
+
+    def test_debounceWithoutKey(self):
+        _logger.info("#" * 100)
+        interval = 0.1
+
+        queue = Queue()
+
+        def func(arg):
+            _logger.info("Called with %s", arg)
+            queue.put(arg)
+
+        wrapped = debounce(interval)(func)
+        self.assertTrue(queue.empty())
+
+        wrapped(1)
+        wrapped(2)
+        self.assertTrue(queue.empty())
+
+        time.sleep(2 * interval)
+
+        self.assertEqual(queue.get(1), 2)
+        self.assertTrue(queue.empty())
+
+    def test_debounceWithKey(self):
+        _logger.info("#" * 100)
+        interval = 0.1
+
+        obj = mock.Mock()
+
+        def func(arg):
+            _logger.info("Called with %s", arg)
+            obj(arg)
+
+        wrapped = debounce(interval, "arg")(func)
+        obj.assert_not_called()
+
+        wrapped(1)
+        wrapped(2)
+        wrapped(3)
+
+        obj.assert_not_called()
+
+        time.sleep(2 * interval)
+
+        obj.assert_has_calls(
+            [mock.call(1), mock.call(2), mock.call(3),], any_order=True
+        )
+
+        wrapped(4)
+        wrapped(4)
+        wrapped(4)
+        time.sleep(2 * interval)
+
+        obj.assert_has_calls(
+            [mock.call(1), mock.call(2), mock.call(3), mock.call(4),], any_order=True
+        )

hdl_checker/utils.py

@@ -474,10 +474,11 @@ def wrapper(func):
 
         @functools.wraps(func)
         def debounced(*args, **kwargs):
-            if keyed_by:
-                key = inspect.signature(func).bind(*args, **kwargs).arguments[keyed_by]
-            else:
-                key = None
+            key = (
+                inspect.signature(func).bind(*args, **kwargs).arguments[keyed_by]
+                if keyed_by
+                else None
+            )
 
             # Could not find a way to disable debouncing whilist testing, so
             # we'll use a module var