Mercurial > ~darius > hgwebdir.cgi > epro
comparison velib_python/test/mock_dbus_service.py @ 8:9c0435a617db
Import velib_python
author | Daniel O'Connor <darius@dons.net.au> |
---|---|
date | Sun, 05 Dec 2021 14:35:36 +1030 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
5:982eeffe9d95 | 8:9c0435a617db |
---|---|
1 # Simulates the busService object without using the D-Bus (intended for unit tests). Data usually stored in | |
2 # D-Bus items is now stored in memory. | |
3 class MockDbusService(object): | |
4 def __init__(self, servicename): | |
5 self._dbusobjects = {} | |
6 self._callbacks = {} | |
7 self._service_name = servicename | |
8 | |
9 def add_path(self, path, value, description="", writeable=False, onchangecallback=None, | |
10 gettextcallback=None): | |
11 self._dbusobjects[path] = value | |
12 if onchangecallback is not None: | |
13 self._callbacks[path] = onchangecallback | |
14 | |
15 # Add the mandatory paths, as per victron dbus api doc | |
16 def add_mandatory_paths(self, processname, processversion, connection, | |
17 deviceinstance, productid, productname, firmwareversion, hardwareversion, connected): | |
18 self.add_path('/Management/ProcessName', processname) | |
19 self.add_path('/Management/ProcessVersion', processversion) | |
20 self.add_path('/Management/Connection', connection) | |
21 | |
22 # Create rest of the mandatory objects | |
23 self.add_path('/DeviceInstance', deviceinstance) | |
24 self.add_path('/ProductId', productid) | |
25 self.add_path('/ProductName', productname) | |
26 self.add_path('/FirmwareVersion', firmwareversion) | |
27 self.add_path('/HardwareVersion', hardwareversion) | |
28 self.add_path('/Connected', connected) | |
29 | |
30 # Simulates a SetValue from the D-Bus, if avaible the onchangecallback associated with the path will | |
31 # be called before the data is changed. | |
32 def set_value(self, path, newvalue): | |
33 callback = self._callbacks.get(path) | |
34 if callback is None or callback(path, newvalue): | |
35 self._dbusobjects[path] = newvalue | |
36 | |
37 def __getitem__(self, path): | |
38 return self._dbusobjects[path] | |
39 | |
40 def __setitem__(self, path, newvalue): | |
41 if path not in self._dbusobjects: | |
42 raise Exception('Path not registered in service: {}{} (use add_path to register)'.\ | |
43 format(self._service_name, path)) | |
44 self._dbusobjects[path] = newvalue | |
45 | |
46 def __delitem__(self, path): | |
47 del self._dbusobjects[path] | |
48 | |
49 def __contains__(self, path): | |
50 return path in self._dbusobjects | |
51 | |
52 def __enter__(self): | |
53 # No batching done in mock object, and we already | |
54 # support the required dict interface. | |
55 return self | |
56 | |
57 def __exit__(self, *exc): | |
58 pass |