Dramatically simplify import approach in keyboard.py

This also has the side effect of making test_keyboard_module_fallback_import much nicer

Towards #22

Author: sjmf

kvm_serial/backend/keyboard.py

@@ -1,16 +1,26 @@
 import threading
+from importlib import import_module
 from serial import Serial
 from enum import Enum
 from .inputhandler import InputHandler
 
 try:
-    from kvm_serial.backend.implementations.baseop import BaseOp
+    import_module("kvm_serial.backend.implementations")
 except ModuleNotFoundError:
-    # Allow running as a script directly
+    # Allow running as a script directly: add parent to path so
+    # _load_implementation's fallback "backend.implementations.*" resolves
     import os, sys
 
     sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-    from implementations.baseop import BaseOp
+
+
+def _load_implementation(module_name, class_name):
+    """Load a handler class, trying package import first then script-mode fallback."""
+    try:
+        mod = import_module(f"kvm_serial.backend.implementations.{module_name}")
+    except ModuleNotFoundError:
+        mod = import_module(f"backend.implementations.{module_name}")
+    return getattr(mod, class_name)
 
 
 class Mode(Enum):
@@ -21,6 +31,14 @@ class Mode(Enum):
     CURSES = 4
 
 
+_MODE_IMPLEMENTATIONS = {
+    Mode.USB: ("pyusbop", "PyUSBOp"),
+    Mode.PYNPUT: ("pynputop", "PynputOp"),
+    Mode.TTY: ("ttyop", "TtyOp"),
+    Mode.CURSES: ("cursesop", "CursesOp"),
+}
+
+
 class KeyboardListener(InputHandler):
     def __init__(
         self,
@@ -56,43 +74,14 @@ def stop(self):
         self.thread.join()
 
     def run_keyboard(self):
-        # Select operation mode
-        keyboard_handler: BaseOp
-
         if self.mode is Mode.NONE:
             return  # noop
-        elif self.mode is Mode.USB:
-            try:
-                from kvm_serial.backend.implementations.pyusbop import PyUSBOp
-            except ModuleNotFoundError:
-                from backend.implementations.pyusbop import PyUSBOp
-
-            keyboard_handler = PyUSBOp(self.serial_port, layout=self.layout)
-        elif self.mode is Mode.PYNPUT:
-            try:
-                from kvm_serial.backend.implementations.pynputop import PynputOp
-            except ModuleNotFoundError:
-                from backend.implementations.pynputop import PynputOp
-
-            keyboard_handler = PynputOp(self.serial_port, layout=self.layout)
-        elif self.mode is Mode.TTY:
-            try:
-                from kvm_serial.backend.implementations.ttyop import TtyOp
-            except ModuleNotFoundError:
-                from backend.implementations.ttyop import TtyOp
-
-            keyboard_handler = TtyOp(self.serial_port, layout=self.layout)
-        elif self.mode is Mode.CURSES:
-            try:
-                from kvm_serial.backend.implementations.cursesop import CursesOp
-            except ModuleNotFoundError:
-                from backend.implementations.cursesop import CursesOp
-
-            keyboard_handler = CursesOp(self.serial_port, layout=self.layout)
-        else:
+        if self.mode not in _MODE_IMPLEMENTATIONS:
             raise ValueError(f"Unknown keyboard mode: {self.mode!r}")
 
-        keyboard_handler.run()
+        module_name, class_name = _MODE_IMPLEMENTATIONS[self.mode]
+        Impl = _load_implementation(module_name, class_name)
+        Impl(self.serial_port, layout=self.layout).run()
 
 
 def keyboard_main():

tests/backend/test_keyboard.py

@@ -66,68 +66,32 @@ def test_keyboard_main_invokes_listener(self, monkeypatch, sys_modules_patch):
 
     def test_keyboard_module_fallback_import(self, monkeypatch, sys_modules_patch):
         """
-        Test that keyboard.py falls back to importing implementations.baseop.BaseOp
-        if kvm_serial.backend.implementations.baseop import fails with ModuleNotFoundError.
-
-        This is a horrible test that by necessity has to mess around with both
-        __import__ and sys.modules. It makes an effort to restore these afterwards.
+        Test that _load_implementation falls back to backend.implementations.*
+        when kvm_serial.backend.implementations.* import fails.
 
         Mocks:
-            - builtins.__import__: Raises ModuleNotFoundError for kvm_serial.backend.implementations.baseop
-            - sys.modules: Injects a mock implementations.baseop.BaseOp
-            - KeyboardListener: Mocked to avoid running threads
-            - logging.basicConfig: Avoids side effects
+            - keyboard.import_module: Raises ModuleNotFoundError for kvm_serial.*,
+              returns a mock module for the backend.* fallback path
         Asserts:
-            - The fallback import path is used (implementations.baseop.BaseOp)
-            - KeyboardListener can still be constructed and used
+            - The fallback import path is used and returns the expected class
         """
-        import importlib
-
-        # Save original sys.modules entries to restore after test
-        orig_kvm_serial = sys.modules.get("kvm_serial")
-        orig_baseop = sys.modules.get("kvm_serial.backend.implementations.baseop")
-
-        try:
-            # Remove relevant modules from sys.modules
-            sys.modules.pop("kvm_serial.backend.implementations.baseop", None)
-            sys.modules.pop("kvm_serial", None)
-
-            # Patch __import__ to raise ModuleNotFoundError for the primary import
-            orig_import = __import__
-
-            def import_side_effect(name, *args, **kwargs):
-                if name == "kvm_serial.backend.implementations.baseop":
-                    raise ModuleNotFoundError
-                return orig_import(name, *args, **kwargs)
-
-            monkeypatch.setattr("builtins.__import__", import_side_effect)
-
-            # Inject a mock implementations.baseop.BaseOp
-            mock_baseop = MagicMock()
-            sys.modules["implementations.baseop"] = MagicMock(BaseOp=mock_baseop)
-
-            # Patch KeyboardListener and logging.basicConfig
-            from kvm_serial.backend import keyboard as kb_mod
-
-            # Ensure module is in sys.modules before reload
-            sys.modules["kvm_serial.backend.keyboard"] = kb_mod
-
-            with (
-                patch.object(kb_mod, "KeyboardListener", MagicMock()),
-                patch("logging.basicConfig", lambda *a, **k: None),
-            ):
-                # Reload the module to trigger the import logic
-                importlib.reload(kb_mod)
-                # Now, BaseOp should be the mock from implementations.baseop
-                assert hasattr(kb_mod, "BaseOp")
-                assert kb_mod.BaseOp is mock_baseop
-        finally:
-            # Restore sys.modules to its original state
-            sys.modules.pop("implementations.baseop", None)
-            if orig_kvm_serial is not None:
-                sys.modules["kvm_serial"] = orig_kvm_serial
-            if orig_baseop is not None:
-                sys.modules["kvm_serial.backend.implementations.baseop"] = orig_baseop
+        from importlib import import_module as real_import_module
+        from kvm_serial.backend import keyboard as kb_mod
+
+        mock_module = MagicMock()
+        mock_handler = MagicMock()
+        mock_module.TestHandler = mock_handler
+
+        def import_side_effect(name, *args, **kwargs):
+            if name.startswith("kvm_serial.backend.implementations."):
+                raise ModuleNotFoundError(name)
+            if name == "backend.implementations.testmod":
+                return mock_module
+            return real_import_module(name, *args, **kwargs)
+
+        with patch.object(kb_mod, "import_module", side_effect=import_side_effect):
+            result = kb_mod._load_implementation("testmod", "TestHandler")
+            assert result is mock_handler
 
 
 # Mock Serial