|
8
|
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()
|
