Mercurial > ~darius > hgwebdir.cgi > epro

comparison velib_python/test/test_vedbus.py @ 8:9c0435a617db

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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 #!/usr/bin/env python3
2 # -*- coding: utf-8 -*-
3
4 # Python
5 import logging
6 import os
7 import sys
8 import unittest
9 import subprocess
10 import time
11 import dbus
12 import threading
13 import fcntl
14 from dbus.mainloop.glib import DBusGMainLoop
15
16 # Local
17 sys.path.insert(1, os.path.join(os.path.dirname(__file__), '../'))
18 from vedbus import VeDbusService, VeDbusItemImport
19
20 logger = logging.getLogger(__file__)
21 """
22 class VeDbusServiceTests(unittest.TestCase):
23 def incrementcallback(self, path, value):
24 self.calledback += 1
25 return True if value < 50 else False
26
27 def setUp(self):
28 self.calledback = 0
29
30
31 self.service = VeDbusService('com.victronenergy.testservice')
32 self.service.add_path(path='/Int', value=10, description="int", writeable=True,
33 onchangecallback=self.incrementcallback, gettextcallback=None)
34
35 self.thread = threading.Thread(target=self.mainloop.run)
36 self.thread.start()
37
38 def test_callback(self):
39 a = subprocess.check_output('dbus', '-y com.victronenergy.testservice')
40 print(a)
41
42 def tearDown(self):
43 self.thread.kill()
44 self.thread = None
45 """
46
47
48 class VeDbusItemExportTests(unittest.TestCase):
49 # The actual code calling VeDbusItemExport is in fixture_vedbus.py, which is ran as a subprocess. That
50 # code exports several values to the dbus. And then below test cases check if the exported values are
51 # what the should be, by using the bare dbus import objects and functions.
52
53 def setUp(self):
54 self.sp = subprocess.Popen([sys.executable, "fixture_vedbus.py"], stdout=subprocess.PIPE)
55 self.dbusConn = dbus.SessionBus() if 'DBUS_SESSION_BUS_ADDRESS' in os.environ else dbus.SystemBus()
56
57 # Wait for fixture to be up and running. 'b' prefix is for python3,
58 # it works in both python versions.
59 while (self.sp.stdout.readline().rstrip() != b'up and running'):
60 pass
61
62 def tearDown(self):
63 self.sp.kill()
64 self.sp.wait()
65 self.sp.stdout.close()
66
67 def test_get_value_invalid(self):
68 v = self.dbusConn.get_object('com.victronenergy.dbusexample', '/Invalid').GetValue()
69 self.assertEqual(v, dbus.Array([], signature=dbus.Signature('i'), variant_level=1))
70 self.assertIs(type(v), dbus.Array)
71 self.assertEqual(self.dbusConn.get_object('com.victronenergy.dbusexample', '/Invalid').GetText(), '---')
72
73 def test_get_value_string(self):
74 v = self.dbusConn.get_object('com.victronenergy.dbusexample', '/String').GetValue()
75 self.assertEqual(v, 'this is a string')
76 self.assertIs(type(v), dbus.String)
77 self.assertEqual(self.dbusConn.get_object('com.victronenergy.dbusexample', '/String').GetText(), 'this is a string')
78
79 def test_get_value_int(self):
80 v = self.dbusConn.get_object('com.victronenergy.dbusexample', '/Int').GetValue()
81 self.assertEqual(v, 40000)
82 self.assertIs(type(v), dbus.Int32)
83 self.assertEqual(self.dbusConn.get_object('com.victronenergy.dbusexample', '/Int').GetText(), '40000')
84
85 def test_get_value_negativeint(self):
86 v = self.dbusConn.get_object('com.victronenergy.dbusexample', '/NegativeInt').GetValue()
87 self.assertEqual(v, -10)
88 self.assertIs(type(v), dbus.Int32)
89 self.assertEqual(self.dbusConn.get_object('com.victronenergy.dbusexample', '/NegativeInt').GetText(), '-10')
90
91 def test_get_value_float(self):
92 v = self.dbusConn.get_object('com.victronenergy.dbusexample', '/Float').GetValue()
93 self.assertEqual(v, 1.5)
94 self.assertIs(type(v), dbus.Double)
95 self.assertEqual(self.dbusConn.get_object('com.victronenergy.dbusexample', '/Float').GetText(), '1.5')
96
97 def test_get_text_byte(self):
98 v = self.dbusConn.get_object('com.victronenergy.dbusexample', '/Byte').GetText()
99 self.assertEqual('84', v)
100
101 def test_get_value_byte(self):
102 v = self.dbusConn.get_object('com.victronenergy.dbusexample', '/Byte').GetValue()
103 self.assertEqual(84, v)
104
105 def test_set_value(self):
106 self.assertNotEqual(0, self.dbusConn.get_object('com.victronenergy.dbusexample', '/NotWriteable').SetValue(12))
107 self.assertEqual('original', self.dbusConn.get_object('com.victronenergy.dbusexample', '/NotWriteable').GetValue())
108
109 self.assertEqual(0, self.dbusConn.get_object('com.victronenergy.dbusexample', '/Writeable').SetValue(12))
110 self.assertEqual(12, self.dbusConn.get_object('com.victronenergy.dbusexample', '/Writeable').GetValue())
111
112 self.assertNotEqual(0, self.dbusConn.get_object('com.victronenergy.dbusexample', '/WriteableUpTo100').SetValue(102))
113 self.assertEqual('original', self.dbusConn.get_object('com.victronenergy.dbusexample', '/WriteableUpTo100').GetValue())
114
115 self.assertEqual(0, self.dbusConn.get_object('com.victronenergy.dbusexample', '/WriteableUpTo100').SetValue(50))
116 self.assertEqual(50, self.dbusConn.get_object('com.victronenergy.dbusexample', '/WriteableUpTo100').GetValue())
117
118 def test_gettextcallback(self):
119 self.assertEqual('gettexted /Gettextcallback 10', self.dbusConn.get_object('com.victronenergy.dbusexample', '/Gettextcallback').GetText())
120
121 def waitandkill(self, seconds=5):
122 time.sleep(seconds)
123 self.process.kill()
124 self.process.wait()
125
126 def test_changedsignal(self):
127 self.process = subprocess.Popen(['dbus-monitor', "type='signal',sender='com.victronenergy.dbusexample',interface='com.victronenergy.BusItem'"], stdout=subprocess.PIPE)
128
129 #wait for dbus-monitor to start up
130 time.sleep(0.5)
131
132 #set timeout
133 thread = threading.Thread(target=self.waitandkill)
134 thread.start()
135
136 self.dbusConn.get_object('com.victronenergy.dbusexample', '/Gettextcallback').SetValue(60)
137
138 fcntl.fcntl(self.process.stdout.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)
139
140 time.sleep(0.5)
141
142 t = bytes()
143 while self.process.returncode is None:
144 try:
145 t += self.process.stdout.readline()
146 except IOError:
147 break
148 self.process.stdout.close()
149
150 text = b" dict entry(\n"
151 text += b" string \"Text\"\n"
152 text += b" variant string \"gettexted /Gettextcallback 60\"\n"
153 text += b" )\n"
154
155 value = b" dict entry(\n"
156 value += b" string \"Value\"\n"
157 value += b" variant int32 60\n"
158 value += b" )\n"
159
160 self.assertNotEqual(-1, t.find(text))
161 self.assertNotEqual(-1, t.find(value))
162
163 thread.join()
164
165 """
166 MVA 2014-08-30: this test of VEDbusItemImport doesn't work, since there is no gobject-mainloop.
167 Probably making some automated functional test, using bash and some scripts, will work much
168 simpler and better
169 class VeDbusItemImportTests(unittest.TestCase):
170 # VeDbusItemImport class is tested against dbus objects exported by fixture_vedbus.py, which is ran as a
171 # subprocess.
172
173 def setUp(self):
174 self.sp = subprocess.Popen([sys.executable, "fixture_vedbus.py"], stdout=subprocess.PIPE)
175 self.dbusConn = dbus.SessionBus() if 'DBUS_SESSION_BUS_ADDRESS' in os.environ else dbus.SystemBus()
176
177 #wait for fixture to be up and running
178 while (self.sp.stdout.readline().rstrip() != 'up and running'):
179 pass
180
181 def tearDown(self):
182 self.sp.kill()
183 self.sp.wait()
184
185 def test_get_invalid(self):
186 self.assertIs(None, VeDbusItemImport(self.dbusConn, 'com.victronenergy.dbusexample', '/Invalid').get_value())
187 self.assertEqual('---', VeDbusItemImport(self.dbusConn, 'com.victronenergy.dbusexample', '/Invalid').get_text())
188
189 def test_get_string(self):
190 v = VeDbusItemImport(self.dbusConn, 'com.victronenergy.dbusexample', '/String')
191 self.assertEqual('this is a string', v.get_value())
192 self.assertIs(dbus.String, type(v.get_value()))
193 self.assertEqual('this is a string', v.get_text())
194
195 def test_get_int(self):
196 v = VeDbusItemImport(self.dbusConn, 'com.victronenergy.dbusexample', '/Int')
197 self.assertEqual(40000, v.get_value())
198 self.assertIs(dbus.Int32, type(v.get_value()))
199 self.assertEqual('40000', v.get_text())
200
201 def test_get_byte(self):
202 v = VeDbusItemImport(self.dbusConn, 'com.victronenergy.dbusexample', '/Byte')
203 self.assertEqual(84, v.get_value())
204 self.assertEqual('84', v.get_text())
205
206 def test_set_value(self):
207 nw = VeDbusItemImport(self.dbusConn, 'com.victronenergy.dbusexample', '/NotWriteable')
208 wr = VeDbusItemImport(self.dbusConn, 'com.victronenergy.dbusexample', '/Writeable')
209 wc = VeDbusItemImport(self.dbusConn, 'com.victronenergy.dbusexample', '/WriteableUpTo100')
210
211 self.assertNotEqual(0, nw.set_value(12))
212 self.assertEqual('original', nw.get_value())
213
214 self.assertEqual(0, wr.set_value(12))
215 self.assertEqual(12, wr.get_value())
216
217 self.assertNotEqual(0, wc.set_value(102))
218 self.assertEqual('original', wc.get_value())
219
220 self.assertEqual(0, wc.set_value(50))
221 self.assertEqual(50, wc.get_value())
222 """
223
224 if __name__ == "__main__":
225 logging.basicConfig(stream=sys.stderr)
226 logging.getLogger('').setLevel(logging.WARNING)
227 unittest.main()