-
Notifications
You must be signed in to change notification settings - Fork 46
Expand file tree
/
Copy pathtest_vehicle.py
More file actions
79 lines (61 loc) · 2.25 KB
/
test_vehicle.py
File metadata and controls
79 lines (61 loc) · 2.25 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
from nose.tools import eq_, ok_
import unittest
import time
from openxc.sources import DataSource
from openxc.measurements import Measurement, NamedMeasurement, \
UnrecognizedMeasurementError
from openxc.vehicle import Vehicle
class VehicleTests(unittest.TestCase):
def setUp(self):
super(VehicleTests, self).setUp()
self.vehicle = Vehicle()
self.messages_received = []
def _callback(self, message):
self.messages_received.append(message)
def test_get(self):
measurement = self.vehicle.get(TestMeasurement)
ok_(measurement is None)
def test_add_listener(self):
source = TestDataSource()
self.vehicle.add_source(source)
self.vehicle.listen(TestMeasurement, self._callback)
data = {'name': TestMeasurement.name, 'value': 100}
source.inject(data)
ok_(len(self.messages_received) > 0)
def test_remove_listener(self):
source = TestDataSource()
self.vehicle.add_source(source)
self.vehicle.listen(TestMeasurement, self._callback)
self.vehicle.unlisten(TestMeasurement, self._callback)
data = {'name': TestMeasurement.name, 'value': 100}
source.inject(data)
eq_(len(self.messages_received), 0)
def test_get_one(self):
source = TestDataSource()
self.vehicle.add_source(source)
measurement = self.vehicle.get(TestMeasurement)
ok_(measurement is None)
data = {'name': TestMeasurement.name, 'value': 100}
source.inject(data)
measurement = self.vehicle.get(TestMeasurement)
ok_(measurement is not None)
eq_(measurement.name, data['name'])
eq_(measurement.value.num, data['value'])
def test_bad_measurement_type(self):
class NotAMeasurement(object):
pass
try:
self.vehicle.get(NotAMeasurement)
except UnrecognizedMeasurementError:
pass
else:
self.fail("Expected an %s" %
str(UnrecognizedMeasurementError.__name__))
class TestMeasurement(NamedMeasurement):
name = "test"
class TestDataSource(DataSource):
def inject(self, message):
self.callback(message)
time.sleep(0.001)
def run(self):
pass