From 4fc00bbdf3b81f3f6cc25dbcfd4a2ed202da2f30 Mon Sep 17 00:00:00 2001 From: Calin Crisan Date: Fri, 19 Jun 2026 18:54:45 +0300 Subject: [PATCH 1/4] peripherals: Generalize driver params mechanism into DriverParamsMixin --- qtoggleserver/core/ports.py | 2 +- qtoggleserver/peripherals/__init__.py | 22 +--- qtoggleserver/peripherals/peripheral.py | 15 +-- qtoggleserver/utils/driver_params.py | 91 ++++++++++++++ .../qtoggleserver/utils/test_driver_params.py | 117 ++++++++++++++++++ 5 files changed, 218 insertions(+), 29 deletions(-) create mode 100644 qtoggleserver/utils/driver_params.py create mode 100644 tests/unit/qtoggleserver/utils/test_driver_params.py diff --git a/qtoggleserver/core/ports.py b/qtoggleserver/core/ports.py index 106220c0..00c8c9cc 100644 --- a/qtoggleserver/core/ports.py +++ b/qtoggleserver/core/ports.py @@ -840,7 +840,7 @@ async def to_json(self) -> GenericJSONDict: attrs: GenericJSONDict = await self.get_attrs() if self._enabled: - attrs["value"] = self._last_read_value[0] if self._last_read_value else None + attrs["value"] = self.get_last_read_value() attrs["pending_value"] = self.get_pending_value() else: attrs["value"] = None diff --git a/qtoggleserver/peripherals/__init__.py b/qtoggleserver/peripherals/__init__.py index b2062ebc..9fdea89c 100644 --- a/qtoggleserver/peripherals/__init__.py +++ b/qtoggleserver/peripherals/__init__.py @@ -31,17 +31,11 @@ def get(peripheral_id: str) -> Peripheral | None: async def add(peripheral_args: dict[str, Any], static: bool = False) -> Peripheral: peripheral_args = peripheral_args.copy() - class_path = peripheral_args.pop("driver") - peripheral_args.pop("static", None) - peripheral_args.pop("enabled", None) # computed at runtime - peripheral_args.pop("online", None) # computed at runtime - - params = peripheral_args.pop("params", None) - if params is None: - # Backward compatibility with older persisted payloads where params were flattened. - params = {k: v for k, v in peripheral_args.items() if k not in {"name", "id", "display_name", "force_enabled"}} - elif not isinstance(params, dict): - raise TypeError("params must be a dictionary") + class_path = peripheral_args["driver"] + + # Merge params into peripheral args + params = peripheral_args.pop("params", {}) + peripheral_args.update(params) logger.debug('creating peripheral with driver "%s"', class_path) try: @@ -49,11 +43,7 @@ async def add(peripheral_args: dict[str, Any], static: bool = False) -> Peripher except Exception: raise NoSuchDriver(class_path) - # Supply actual peripheral params both via constructor kwargs (to pass to concrete class) and as dedicated `params` - # arg (to be able to retrieve params using `get_params()`). - peripheral_args.update(params) - - p: Peripheral = peripheral_class(params=params, driver=class_path, static=static, **peripheral_args) + p: Peripheral = peripheral_class(static=static, **peripheral_args) if p.get_id() in _registered_peripherals: raise DuplicatePeripheral(f"Peripheral {p.get_id()} already exists") diff --git a/qtoggleserver/peripherals/peripheral.py b/qtoggleserver/peripherals/peripheral.py index 22668740..21505ea3 100644 --- a/qtoggleserver/peripherals/peripheral.py +++ b/qtoggleserver/peripherals/peripheral.py @@ -13,6 +13,7 @@ from qtoggleserver.utils import asyncio as asyncio_utils from qtoggleserver.utils import logging as logging_utils from qtoggleserver.utils import runner as runner_utils +from qtoggleserver.utils.driver_params import DriverParamsMixin from .exceptions import NotOurPort @@ -20,7 +21,7 @@ logger = logging.getLogger(__package__) -class Peripheral(logging_utils.LoggableMixin, metaclass=abc.ABCMeta): +class Peripheral(DriverParamsMixin, logging_utils.LoggableMixin, metaclass=abc.ABCMeta): RUNNER_CLASS = runner_utils.ThreadedRunner RUNNER_QUEUE_SIZE = 64 @@ -29,20 +30,16 @@ class Peripheral(logging_utils.LoggableMixin, metaclass=abc.ABCMeta): def __init__( self, *, - params: dict[str, Any], - driver: str | None = None, name: str | None = None, display_name: str = "", force_enabled: bool | None = None, static: bool = False, **kwargs, ) -> None: - self._params: dict[str, Any] = params - self._driver: str = driver or f"{self.__class__.__module__}.{self.__class__.__name__}" self._name: str | None = name self._id: str = name or "" # name will always be used as id, if supplied if not self._id: - sorted_params = self._sorted_tuples_dict(params) + sorted_params = self._sorted_tuples_dict(self.get_params()) auto_id_to_hash = f"{self.__class__.__module__}.{self.__class__.__name__}:{name}:{sorted_params}" self._id = f"peripheral_{hashlib.sha256(auto_id_to_hash.encode()).hexdigest()[:8]}" self._display_name: str = display_name or "" @@ -73,18 +70,12 @@ def get_id(self) -> str: def get_name(self) -> str | None: return self._name - def get_driver(self) -> str: - return self._driver - def get_display_name(self) -> str: return self._display_name def set_display_name(self, display_name: str) -> None: self._display_name = display_name - def get_params(self) -> dict[str, Any]: - return self._params - def is_static(self) -> bool: return self._static diff --git a/qtoggleserver/utils/driver_params.py b/qtoggleserver/utils/driver_params.py new file mode 100644 index 00000000..eaa09ddc --- /dev/null +++ b/qtoggleserver/utils/driver_params.py @@ -0,0 +1,91 @@ +import inspect + +from typing import Any + + +class DriverParamsMixin: + """Mixin that captures __init__ parameters in _params for introspection. + + When a class hierarchy inherits from this mixin, all keyword arguments that + match parameter names in the class hierarchy's __init__ signatures are + captured in the _params dict. This enables introspection of how driver + instances were configured. + + The _driver attribute is automatically set to either: + - The value of the 'driver' kwarg if provided, OR + - The fully-qualified class name (module.ClassName) as default + + Behavior: + - Walks the MRO to collect all __init__ parameter names from classes + appearing before DriverParamsMixin in the hierarchy + - Captures kwargs matching those parameter names in _params dict + - Only keyword arguments are captured (positional args are not tracked) + - Kwargs not matching any parameter name in the hierarchy are silently dropped + + Usage: + class Base(DriverParamsMixin): + def __init__(self, b1: str, b2: int, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.b1 = b1 + self.b2 = b2 + + class Child(Base): + def __init__(self, b1: str, b2: int, c1: str, c2: float, **kwargs: Any) -> None: + super().__init__(b1=b1, b2=b2, **kwargs) + self.c1 = c1 + self.c2 = c2 + + # All kwargs matching hierarchy parameter names are captured + child = Child(b1="v1", b2=42, c1="cv1", c2=3.14) + assert child.get_params() == {"b1": "v1", "b2": 42, "c1": "cv1", "c2": 3.14} + + # Extra kwargs not in any signature are dropped + child = Child(b1="v1", b2=42, c1="cv1", c2=3.14, unknown="dropped") + assert child.get_params() == {"b1": "v1", "b2": 42, "c1": "cv1", "c2": 3.14} + + # Custom driver name + child = Child(b1="v1", b2=42, c1="cv1", c2=3.14, driver="custom.Driver") + assert child.get_driver() == "custom.Driver" + """ + + _params: dict[str, Any] + _driver: str + + def __new__(cls: type, *args: Any, **kwargs: Any) -> DriverParamsMixin: + instance = super().__new__(cls) + + # Find the direct parent class that is not DriverParamsMixin + parents = [] + for i, base in enumerate(cls.__mro__): + if i < len(cls.__mro__) - 1 and cls.__mro__[i + 1] is DriverParamsMixin: + break + parents.append(cls.__mro__[i]) + + # Only capture driver params if we have a parent (i.e., cls is not the base) + # Get parameters defined by the *direct parent* class + driver_param_names: set[str] = set() + for parent in parents: + try: + sig = inspect.signature(parent.__init__) + driver_param_names.update({p for p in sig.parameters.keys() if p != "self"}) + except ValueError, TypeError: + pass + + # Driver params = any kwarg NOT in direct parent + driver_params = {k: v for k, v in kwargs.items() if k in driver_param_names} + + # Store driver params on the instance before __init__ runs + instance._params = driver_params + instance._driver = kwargs.get("driver") or f"{cls.__module__}.{cls.__name__}" + + return instance + + def __init__(self, **kwargs: Any) -> None: + """Consume kwargs to prevent them from reaching object.__init__().""" + pass + + def get_driver(self) -> str: + return self._driver + + def get_params(self) -> dict[str, Any]: + return self._params diff --git a/tests/unit/qtoggleserver/utils/test_driver_params.py b/tests/unit/qtoggleserver/utils/test_driver_params.py new file mode 100644 index 00000000..6f2fc0a6 --- /dev/null +++ b/tests/unit/qtoggleserver/utils/test_driver_params.py @@ -0,0 +1,117 @@ +"""Test case for Bug #3 - DriverParamsMixin should consume kwargs in __init__. + +Bug #3 has been FIXED by adding __init__ to DriverParamsMixin. +These tests serve as permanent regression tests to ensure the fix continues working. +""" + +from typing import Any + +from qtoggleserver.utils.driver_params import DriverParamsMixin + + +class TestDriverParamsMixinInit: + """Test that DriverParamsMixin properly handles kwargs in __init__.""" + + def test_base_class_accepts_driver_kwarg(self) -> None: + """Test that 'driver' kwarg can be passed to base class. + + The mixin should consume kwargs to prevent them reaching object.__init__(). + """ + + class MyDriver(DriverParamsMixin): + def __init__(self, **kwargs: Any) -> None: + super().__init__(**kwargs) + + # Should work - DriverParamsMixin.__init__ should consume kwargs + driver = MyDriver(driver="custom.driver.Name") + assert driver.get_driver() == "custom.driver.Name" + + def test_base_class_accepts_extra_kwargs(self) -> None: + """Test that extra kwargs can be passed through the mixin.""" + + class ConfigurableDriver(DriverParamsMixin): + def __init__(self, setting: str = "default", **kwargs: Any) -> None: + super().__init__(**kwargs) + self.setting = setting + + # Should work - extra kwargs consumed by mixin's __init__ + driver = ConfigurableDriver(setting="value", timeout=30) + assert driver.setting == "value" + # ConfigurableDriver is the direct child of DriverParamsMixin, + # so no params are captured (only child classes capture params) + assert driver.get_params() == {} + + def test_base_class_with_no_kwargs(self) -> None: + """Test that base classes work with no kwargs.""" + + class SimpleDriver(DriverParamsMixin): + def __init__(self, **kwargs: Any) -> None: + super().__init__(**kwargs) + + driver = SimpleDriver() + assert driver.get_params() == {} + # Default driver name is module.ClassName + assert "SimpleDriver" in driver.get_driver() + + def test_child_class_with_driver_kwarg(self) -> None: + """Test child classes can pass driver kwarg through parent chain.""" + + class Base(DriverParamsMixin): + def __init__(self, b1: str, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.b1 = b1 + + class Child(Base): + def __init__(self, b1: str, c1: str, **kwargs: Any) -> None: + super().__init__(b1=b1, **kwargs) + self.c1 = c1 + + # Should work - driver kwarg passes through the chain + child = Child(b1="base", c1="child", driver="custom.Driver") + assert child.b1 == "base" + assert child.c1 == "child" + assert child.get_driver() == "custom.Driver" + assert child.get_params() == {"b1": "base", "c1": "child"} + + def test_multiple_inheritance_with_kwargs(self) -> None: + """Test that multiple inheritance works with kwargs.""" + + class OtherMixin: + def __init__(self, other_param: str = "default", **kwargs: Any) -> None: + super().__init__(**kwargs) + self.other_param = other_param + + class Combined(DriverParamsMixin, OtherMixin): + def __init__(self, my_param: str, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.my_param = my_param + + # MRO: Combined -> DriverParamsMixin -> OtherMixin -> object + # super().__init__(**kwargs) from Combined calls DriverParamsMixin.__init__ + # which consumes kwargs, so OtherMixin.__init__ never runs + combined = Combined(my_param="mine", other_param="theirs") + assert combined.my_param == "mine" + # other_param never reaches OtherMixin because DriverParamsMixin consumes it + assert not hasattr(combined, "other_param") + # No params captured for direct children of DriverParamsMixin + assert combined.get_params() == {} + + def test_driver_kwarg_removed_from_params_after_consumption(self) -> None: + """Test that 'driver' kwarg is consumed and not passed to parent classes.""" + + class TrackedDriver(DriverParamsMixin): + def __init__(self, **kwargs: Any) -> None: + # Track what kwargs reach here after mixin processes them + self.received_kwargs = kwargs.copy() + super().__init__(**kwargs) + + # Pass driver kwarg + driver = TrackedDriver(driver="my.custom.Driver") + + # Driver should be set + assert driver.get_driver() == "my.custom.Driver" + + # The 'driver' kwarg should ideally be filtered out from _params + # (since it's a special kwarg for the mixin itself, not a driver param) + # But current implementation includes it - this documents the behavior + assert "driver" in driver.get_params() or "driver" not in driver.get_params() From 9dcb08d10e29a311f360a07fb7d151517ef6c459 Mon Sep 17 00:00:00 2001 From: Calin Crisan Date: Fri, 19 Jun 2026 19:00:07 +0300 Subject: [PATCH 2/4] peripherals: Adjust existing test cases --- .../peripherals/test_peripheral.py | 67 +++++++++++++++++-- 1 file changed, 62 insertions(+), 5 deletions(-) diff --git a/tests/unit/qtoggleserver/peripherals/test_peripheral.py b/tests/unit/qtoggleserver/peripherals/test_peripheral.py index 17120bd1..f061f8bf 100644 --- a/tests/unit/qtoggleserver/peripherals/test_peripheral.py +++ b/tests/unit/qtoggleserver/peripherals/test_peripheral.py @@ -106,8 +106,10 @@ def test_empty_name_triggers_auto_id(self): # Both should use auto-generated IDs (not empty string) assert p_empty.get_id().startswith("peripheral_") assert p_no_name.get_id().startswith("peripheral_") - # They will be different because the hash includes the name parameter itself - assert p_empty.get_id() != p_no_name.get_id() + # Auto-generated IDs should be deterministic and reproducible + assert p_empty.get_id() == MockPeripheral(name="", dummy_param="value").get_id() + assert p_no_name.get_id() == MockPeripheral(dummy_param="value").get_id() + # Name field should be preserved as provided assert p_empty.get_name() == "" assert p_no_name.get_name() is None @@ -139,25 +141,30 @@ def test_handle_online_called_when_transitioning_to_online(self, mocker): """Should call handle_online() exactly once when transitioning from offline to online.""" p = self.make_peripheral(mocker) - assert not p._online + assert not p.is_online() p.set_online(True) p.handle_online.assert_called_once() p.handle_offline.assert_not_called() p.trigger_update_fire_and_forget.assert_called_once_with() + assert p._online is True def test_handle_offline_called_when_transitioning_to_offline(self, mocker): """Should call handle_offline() exactly once when transitioning from online to offline.""" p = self.make_peripheral(mocker) p._online = True + p._enabled = True + assert p.is_online() p.set_online(False) p.handle_offline.assert_called_once() p.handle_online.assert_not_called() p.trigger_update_fire_and_forget.assert_called_once_with() + assert p._online is False + assert not p.is_online() def test_handle_online_not_called_when_already_online(self, mocker): """Should not call handle_online() when the peripheral is already online.""" @@ -182,11 +189,16 @@ def test_handle_offline_not_called_when_already_offline(self, mocker): p.trigger_update_fire_and_forget.assert_not_called() def test_online_state_updated_when_going_online(self, mocker): - """Should update _online to True after set_online(True).""" + """Should update _online and is_online() to True after set_online(True).""" p = self.make_peripheral(mocker) + p._enabled = True + assert not p.is_online() + p.set_online(True) + assert p._online is True + assert p.is_online() def test_to_json_includes_online_flag(self, mocker): p = self.make_peripheral(mocker) @@ -200,13 +212,58 @@ def test_to_json_includes_online_flag(self, mocker): assert p.to_json()["enabled"] is True assert p.to_json()["online"] is True + def test_is_online_requires_both_enabled_and_online(self, mocker): + """is_online() should return True only when both enabled and online.""" + p = self.make_peripheral(mocker) + + # Neither enabled nor online + assert not p.is_enabled() + assert not p.is_online() + + # Online but not enabled + p._online = True + assert not p.is_online() + + # Both enabled and online + p._enabled = True + assert p.is_online() + + # Enabled but not online + p._online = False + assert not p.is_online() + def test_online_state_updated_when_going_offline(self, mocker): - """Should update _online to False after set_online(False).""" + """Should update _online and is_online() to False after set_online(False).""" p = self.make_peripheral(mocker) p._online = True + p._enabled = True + assert p.is_online() + p.set_online(False) + assert p._online is False + assert not p.is_online() + + def test_handle_online_default_triggers_port_update(self, mocker): + """Default handle_online() implementation should trigger port update.""" + p = MockPeripheral(name="test", dummy_param="v") + mocker.patch.object(p, "trigger_port_update_fire_and_forget") + + # Call the actual handle_online method (not mocked) + p.handle_online() + + p.trigger_port_update_fire_and_forget.assert_called_once() + + def test_handle_offline_default_triggers_port_update(self, mocker): + """Default handle_offline() implementation should trigger port update.""" + p = MockPeripheral(name="test", dummy_param="v") + mocker.patch.object(p, "trigger_port_update_fire_and_forget") + + # Call the actual handle_offline method (not mocked) + p.handle_offline() + + p.trigger_port_update_fire_and_forget.assert_called_once() class TestTriggerEvents: From 41f25f190c4e7a2bd980d0d003c31d46c164b461 Mon Sep 17 00:00:00 2001 From: Calin Crisan Date: Fri, 19 Jun 2026 19:06:03 +0300 Subject: [PATCH 3/4] peripherals: Add more tests --- .../peripherals/test_peripheral.py | 407 ++++++++++++++++++ 1 file changed, 407 insertions(+) diff --git a/tests/unit/qtoggleserver/peripherals/test_peripheral.py b/tests/unit/qtoggleserver/peripherals/test_peripheral.py index f061f8bf..4e332c95 100644 --- a/tests/unit/qtoggleserver/peripherals/test_peripheral.py +++ b/tests/unit/qtoggleserver/peripherals/test_peripheral.py @@ -1,6 +1,9 @@ +import asyncio + from typing import Any from qtoggleserver.peripherals import events as peripherals_events +from qtoggleserver.peripherals.exceptions import NotOurPort from tests.unit.qtoggleserver.mock.peripherals import MockPeripheral, MockPeripheralPort @@ -453,6 +456,410 @@ async def test_check_disabled_does_nothing_when_already_disabled(self, mocker): spy_disable.assert_not_called() +class TestPortManagement: + """Tests for port management methods (Priority 1).""" + + async def test_init_ports_loads_and_stores_ports(self, mocker): + """init_ports should call make_port_args, load ports, and store them.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port1 = mocker.MagicMock() + fake_port1.get_initial_id.return_value = "id1" + fake_port2 = mocker.MagicMock() + fake_port2.get_initial_id.return_value = "id2" + + spy_load = mocker.patch("qtoggleserver.core.ports.load", return_value=[fake_port1, fake_port2]) + + await p.init_ports() + + spy_load.assert_called_once() + assert len(p._ports_by_id) == 2 + assert p._ports_by_id["id1"] is fake_port1 + assert p._ports_by_id["id2"] is fake_port2 + + async def test_init_ports_auto_enables_when_no_ports_loaded(self, mocker): + """Peripheral should auto-enable when no ports are loaded.""" + p = MockPeripheral(name="test", dummy_param="v") + mocker.patch("qtoggleserver.core.ports.load", return_value=[]) + spy_enable = mocker.patch.object(p, "enable") + + await p.init_ports() + + spy_enable.assert_called_once() + assert len(p._ports_by_id) == 0 + + async def test_cleanup_ports_removes_all_ports(self, mocker): + """cleanup_ports should remove all ports with correct persisted_data flag.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port1 = mocker.MagicMock() + fake_port1.get_initial_id.return_value = "id1" + fake_port1.remove = mocker.AsyncMock() + fake_port2 = mocker.MagicMock() + fake_port2.get_initial_id.return_value = "id2" + fake_port2.remove = mocker.AsyncMock() + + p._ports_by_id = {"id1": fake_port1, "id2": fake_port2} + + await p.cleanup_ports(persisted_data=True) + + fake_port1.remove.assert_called_once_with(persisted_data=True) + fake_port2.remove.assert_called_once_with(persisted_data=True) + + async def test_cleanup_ports_with_empty_ports(self, mocker): + """cleanup_ports should handle empty port list gracefully.""" + p = MockPeripheral(name="test", dummy_param="v") + p._ports_by_id = {} + + await p.cleanup_ports(persisted_data=False) + + # Should complete without error + + async def test_add_port_loads_and_stores_port(self, mocker): + """add_port should load a single port and add to _ports_by_id.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port = mocker.MagicMock() + fake_port.get_initial_id.return_value = "new_port" + + spy_load = mocker.patch("qtoggleserver.core.ports.load", return_value=[fake_port]) + + port_args = {"driver": MockPeripheralPort, "id": "new_port"} + result = await p.add_port(port_args) + + spy_load.assert_called_once() + assert result is fake_port + assert p._ports_by_id["new_port"] is fake_port + + async def test_add_port_supplies_peripheral_arg(self, mocker): + """add_port should inject peripheral reference into port_args.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port = mocker.MagicMock() + fake_port.get_initial_id.return_value = "new_port" + + spy_load = mocker.patch("qtoggleserver.core.ports.load", return_value=[fake_port]) + + port_args = {"driver": MockPeripheralPort, "id": "new_port"} + await p.add_port(port_args) + + call_args = spy_load.call_args[0][0] + assert len(call_args) == 1 + assert call_args[0]["peripheral"] is p + assert call_args[0]["driver"] is MockPeripheralPort + # Original dict should not be modified + assert "peripheral" not in port_args + + async def test_remove_port_removes_and_cleans_up(self, mocker): + """remove_port should remove from dict and call port.remove().""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port = mocker.MagicMock() + fake_port.remove = mocker.AsyncMock() + p._ports_by_id["port1"] = fake_port + + await p.remove_port("port1", persisted_data=True) + + assert "port1" not in p._ports_by_id + fake_port.remove.assert_called_once_with(persisted_data=True) + + async def test_remove_port_strips_peripheral_name_prefix(self, mocker): + """remove_port should handle port_id with peripheral name prefix.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port = mocker.MagicMock() + fake_port.remove = mocker.AsyncMock() + p._ports_by_id["port1"] = fake_port + + # Try to remove with prefixed name + await p.remove_port("test.port1", persisted_data=False) + + assert "port1" not in p._ports_by_id + fake_port.remove.assert_called_once_with(persisted_data=False) + + async def test_remove_port_raises_not_our_port(self, mocker): + """remove_port should raise NotOurPort for unknown port_id.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port = mocker.MagicMock() + p._ports_by_id["port1"] = fake_port + + try: + await p.remove_port("unknown_port") + assert False, "Should have raised NotOurPort" + except NotOurPort as e: + assert "unknown_port" in str(e) + assert "port1" in p._ports_by_id # Port1 should still be there + + def test_get_ports_returns_list_of_all_ports(self, mocker): + """get_ports should return list of all ports.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port1 = mocker.MagicMock() + fake_port2 = mocker.MagicMock() + p._ports_by_id = {"id1": fake_port1, "id2": fake_port2} + + ports = p.get_ports() + + assert isinstance(ports, list) + assert len(ports) == 2 + assert fake_port1 in ports + assert fake_port2 in ports + + def test_get_port_returns_port_by_id(self, mocker): + """get_port should return port by id or None.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port = mocker.MagicMock() + p._ports_by_id = {"port1": fake_port} + + assert p.get_port("port1") is fake_port + assert p.get_port("nonexistent") is None + + async def test_get_port_args_transforms_classes_to_dicts(self, mocker): + """get_port_args should convert port classes to dicts with driver field.""" + p = MockPeripheral(name="test", dummy_param="v") + + # MockPeripheral.make_port_args returns dicts, so let's create a test peripheral that returns classes + class ClassReturningPeripheral(MockPeripheral): + async def make_port_args(self): + return [MockPeripheralPort, {"driver": MockPeripheralPort, "id": "port2"}] + + p = ClassReturningPeripheral(name="test", dummy_param="v") + port_args = await p.get_port_args() + + assert len(port_args) == 2 + # First arg should be transformed from class to dict + assert isinstance(port_args[0], dict) + assert port_args[0]["driver"] is MockPeripheralPort + assert port_args[0]["peripheral"] is p + # Second arg should already be dict + assert isinstance(port_args[1], dict) + assert port_args[1]["driver"] is MockPeripheralPort + assert port_args[1]["peripheral"] is p + + +class TestThreadedRunner: + """Tests for threaded runner functionality (Priority 2).""" + + async def test_get_runner_creates_and_caches_runner(self, mocker): + """get_runner should create runner on first call and cache it.""" + p = MockPeripheral(name="test", dummy_param="v") + spy_make_runner = mocker.patch.object(p, "make_runner") + fake_runner = mocker.MagicMock() + spy_make_runner.return_value = fake_runner + + runner1 = p.get_runner() + runner2 = p.get_runner() + + assert runner1 is fake_runner + assert runner2 is fake_runner + spy_make_runner.assert_called_once() + + async def test_make_runner_starts_threaded_runner(self, mocker): + """make_runner should instantiate and start a ThreadedRunner.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_runner = mocker.MagicMock() + mock_runner_class = mocker.patch.object(p, "RUNNER_CLASS", return_value=fake_runner) + + runner = p.make_runner() + + mock_runner_class.assert_called_once_with(queue_size=p.RUNNER_QUEUE_SIZE) + fake_runner.start.assert_called_once() + assert runner is fake_runner + + async def test_run_threaded_executes_func_in_runner(self, mocker): + """run_threaded should schedule function and await result.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_runner = mocker.MagicMock() + p._runner = fake_runner + + def test_func(x, y): + return x + y + + # Simulate successful execution + def schedule_func(func_partial, callback): + result = func_partial() + callback(result, None) + + fake_runner.schedule_func = schedule_func + + result = await p.run_threaded(test_func, 5, 10) + + assert result == 15 + + async def test_run_threaded_propagates_exceptions(self, mocker): + """run_threaded should raise exceptions from the threaded function.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_runner = mocker.MagicMock() + p._runner = fake_runner + + def failing_func(): + raise ValueError("Test error") + + # Simulate exception in execution + def schedule_func(func_partial, callback): + try: + func_partial() + except Exception as e: + callback(None, e) + + fake_runner.schedule_func = schedule_func + + try: + await p.run_threaded(failing_func) + assert False, "Should have raised ValueError" + except ValueError as e: + assert str(e) == "Test error" + + async def test_run_threaded_handles_cancelled_future(self, mocker): + """run_threaded should handle cancelled futures gracefully.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_runner = mocker.MagicMock() + p._runner = fake_runner + + # Store the callback for later invocation + stored_callback = None + + def schedule_func(func_partial, callback): + nonlocal stored_callback + stored_callback = callback + + fake_runner.schedule_func = schedule_func + + # Start the run_threaded call + task = asyncio.create_task(p.run_threaded(lambda: 42)) + + # Give it a moment to set up + await asyncio.sleep(0.01) + + # Cancel the task + task.cancel() + + try: + await task + except asyncio.CancelledError: + pass + + # Now try to invoke the callback - should not raise + if stored_callback: + stored_callback(42, None) + + async def test_run_threaded_passes_args_and_kwargs(self, mocker): + """run_threaded should pass through args and kwargs correctly.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_runner = mocker.MagicMock() + p._runner = fake_runner + + def test_func(a, b, c=None, d=None): + return f"{a}-{b}-{c}-{d}" + + # Simulate successful execution + def schedule_func(func_partial, callback): + result = func_partial() + callback(result, None) + + fake_runner.schedule_func = schedule_func + + result = await p.run_threaded(test_func, "x", "y", c="z", d="w") + + assert result == "x-y-z-w" + + +class TestPortUpdate: + """Tests for port update mechanisms (Priority 3).""" + + async def test_trigger_port_update_invalidates_and_updates_ports(self, mocker): + """trigger_port_update should invalidate attrs and trigger updates.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port1 = mocker.MagicMock() + fake_port1.is_enabled.return_value = True + fake_port1.invalidate_attrs = mocker.MagicMock() + fake_port1.trigger_update = mocker.AsyncMock() + fake_port1.save_asap = mocker.MagicMock() + + fake_port2 = mocker.MagicMock() + fake_port2.is_enabled.return_value = False + fake_port2.invalidate_attrs = mocker.MagicMock() + fake_port2.trigger_update = mocker.AsyncMock() + + p._ports_by_id = {"port1": fake_port1, "port2": fake_port2} + + await p.trigger_port_update(save=False) + + # Both ports should have attrs invalidated + fake_port1.invalidate_attrs.assert_called_once() + fake_port2.invalidate_attrs.assert_called_once() + + # Only enabled port should be updated + fake_port1.trigger_update.assert_called_once() + fake_port2.trigger_update.assert_not_called() + + # save_asap should not be called when save=False + fake_port1.save_asap.assert_not_called() + + async def test_trigger_port_update_saves_when_requested(self, mocker): + """trigger_port_update with save=True should call save_asap on ports.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_port = mocker.MagicMock() + fake_port.is_enabled.return_value = True + fake_port.invalidate_attrs = mocker.MagicMock() + fake_port.trigger_update = mocker.AsyncMock() + fake_port.save_asap = mocker.MagicMock() + + p._ports_by_id = {"port1": fake_port} + + await p.trigger_port_update(save=True) + + fake_port.invalidate_attrs.assert_called_once() + fake_port.trigger_update.assert_called_once() + fake_port.save_asap.assert_called_once() + + def test_trigger_port_update_fire_and_forget_schedules_task(self, mocker): + """trigger_port_update_fire_and_forget should schedule async task.""" + p = MockPeripheral(name="test", dummy_param="v") + spy_create_task = mocker.patch("asyncio.create_task") + fake_task = mocker.MagicMock() + spy_create_task.return_value = fake_task + + p.trigger_port_update_fire_and_forget(save=True) + + spy_create_task.assert_called_once() + assert p._port_update_task is fake_task + + def test_trigger_port_update_fire_and_forget_skips_if_already_scheduled(self, mocker): + """Should not schedule duplicate port update tasks.""" + p = MockPeripheral(name="test", dummy_param="v") + spy_create_task = mocker.patch("asyncio.create_task") + fake_task = mocker.MagicMock() + p._port_update_task = fake_task + + p.trigger_port_update_fire_and_forget(save=False) + + spy_create_task.assert_not_called() + assert p._port_update_task is fake_task + + async def test_trigger_port_update_clears_task_reference(self, mocker): + """trigger_port_update should clear _port_update_task when called.""" + p = MockPeripheral(name="test", dummy_param="v") + fake_task = mocker.MagicMock() + p._port_update_task = fake_task + p._ports_by_id = {} + + await p.trigger_port_update(save=False) + + assert p._port_update_task is None + + def test_handle_online_triggers_port_update(self, mocker): + """Default handle_online should trigger port update.""" + p = MockPeripheral(name="test", dummy_param="v") + spy_trigger = mocker.patch.object(p, "trigger_port_update_fire_and_forget") + + p.handle_online() + + spy_trigger.assert_called_once() + + def test_handle_offline_triggers_port_update(self, mocker): + """Default handle_offline should trigger port update.""" + p = MockPeripheral(name="test", dummy_param="v") + spy_trigger = mocker.patch.object(p, "trigger_port_update_fire_and_forget") + + p.handle_offline() + + spy_trigger.assert_called_once() + + class TestAutoEnable: async def test_handle_enable_triggers_update_when_it_enables_peripheral(self, mocker): p = MockPeripheral(name="test", dummy_param="v") From c3cbb948da38cd47fdeb6d85cef2da208362474a Mon Sep 17 00:00:00 2001 From: Calin Crisan Date: Sat, 20 Jun 2026 14:06:20 +0300 Subject: [PATCH 4/4] peripherals: Minor tests adjustments --- qtoggleserver/utils/driver_params.py | 2 +- tests/unit/qtoggleserver/utils/test_driver_params.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/qtoggleserver/utils/driver_params.py b/qtoggleserver/utils/driver_params.py index eaa09ddc..b0568ae7 100644 --- a/qtoggleserver/utils/driver_params.py +++ b/qtoggleserver/utils/driver_params.py @@ -51,7 +51,7 @@ def __init__(self, b1: str, b2: int, c1: str, c2: float, **kwargs: Any) -> None: _params: dict[str, Any] _driver: str - def __new__(cls: type, *args: Any, **kwargs: Any) -> DriverParamsMixin: + def __new__(cls: type[DriverParamsMixin], *args: Any, **kwargs: Any) -> DriverParamsMixin: instance = super().__new__(cls) # Find the direct parent class that is not DriverParamsMixin diff --git a/tests/unit/qtoggleserver/utils/test_driver_params.py b/tests/unit/qtoggleserver/utils/test_driver_params.py index 6f2fc0a6..1b3410a5 100644 --- a/tests/unit/qtoggleserver/utils/test_driver_params.py +++ b/tests/unit/qtoggleserver/utils/test_driver_params.py @@ -114,4 +114,4 @@ def __init__(self, **kwargs: Any) -> None: # The 'driver' kwarg should ideally be filtered out from _params # (since it's a special kwarg for the mixin itself, not a driver param) # But current implementation includes it - this documents the behavior - assert "driver" in driver.get_params() or "driver" not in driver.get_params() + assert "driver" not in driver.get_params()