test(dbus): Add support for testing driver reset notification
Introduce `FakeResetDriverWatchdog` to simulate driver reset triggers. Update existing tests and add a new test, `test_notify_reset_driver`, to verify reset notifications using the event signaling mechanism.
This commit is contained in:
@@ -2,11 +2,13 @@
|
||||
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
from ctypes import c_int
|
||||
from queue import Queue
|
||||
from queue import Empty, Queue
|
||||
from tempfile import NamedTemporaryFile
|
||||
from threading import Event, Thread
|
||||
from unittest import TestCase
|
||||
|
||||
IOCTL_QUEUE = Queue()
|
||||
RESET_DRIVER_EVENT = Event()
|
||||
|
||||
|
||||
class FakePiControlDevice:
|
||||
@@ -48,14 +50,67 @@ class FakePiControlDevice:
|
||||
return self._fh.write(buffer)
|
||||
|
||||
|
||||
class FakeResetDriverWatchdog(Thread):
|
||||
|
||||
def __init__(self, picontrol_device: str):
|
||||
super().__init__()
|
||||
self.daemon = True
|
||||
self._calls = []
|
||||
self._exit = False
|
||||
self.not_implemented = True
|
||||
self._triggered = False
|
||||
self.start()
|
||||
|
||||
def run(self):
|
||||
while not self._exit:
|
||||
if RESET_DRIVER_EVENT.wait(0.1):
|
||||
RESET_DRIVER_EVENT.clear()
|
||||
self._triggered = True
|
||||
for func in self._calls:
|
||||
func()
|
||||
|
||||
def register_call(self, function):
|
||||
"""Register a function, if watchdog triggers."""
|
||||
if not callable(function):
|
||||
raise ValueError("Function is not callable.")
|
||||
if function not in self._calls:
|
||||
self._calls.append(function)
|
||||
|
||||
def stop(self):
|
||||
"""Stop watchdog for reset_driver."""
|
||||
self._exit = True
|
||||
|
||||
def unregister_call(self, function=None):
|
||||
"""Remove a function from the watchdog trigger."""
|
||||
if function is None:
|
||||
self._calls.clear()
|
||||
elif function in self._calls:
|
||||
self._calls.remove(function)
|
||||
|
||||
@property
|
||||
def triggered(self):
|
||||
"""Will return True one time after watchdog was triggered."""
|
||||
rc = self._triggered
|
||||
self._triggered = False
|
||||
return rc
|
||||
|
||||
|
||||
class PiControlDeviceMockup(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
# Empty the queue
|
||||
while True:
|
||||
try:
|
||||
IOCTL_QUEUE.get_nowait()
|
||||
except Empty:
|
||||
break
|
||||
|
||||
# Replace classes with mockup classes
|
||||
import revpi_middleware.dbus_middleware1.process_image.interface_picontrol as test_helpers
|
||||
test_helpers.PiControlIoctl = FakePiControlDevice
|
||||
test_helpers.ResetDriverWatchdog = FakeResetDriverWatchdog
|
||||
|
||||
# Create a fake picontrol0 device
|
||||
self.picontrol = FakePiControlDevice(picontrol_device="/dev/fake_device_0")
|
||||
|
||||
Reference in New Issue
Block a user