-
Notifications
You must be signed in to change notification settings - Fork 46
Expand file tree
/
Copy pathstreamer_test_utils.py
More file actions
117 lines (102 loc) · 4.47 KB
/
streamer_test_utils.py
File metadata and controls
117 lines (102 loc) · 4.47 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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
from nose.tools import eq_, ok_
class BaseStreamerTests(object):
def test_receive_complete(self):
serialized_message = self.streamer.serialize_for_stream({'name': "foo", 'value': 42})
ok_(len(serialized_message) > 1)
self.streamer.receive(serialized_message)
message = self.streamer.parse_next_message()
ok_(message is not None)
eq_(message['name'], "foo")
eq_(message['value'], 42)
def test_receive_partial(self):
serialized_message = self.streamer.serialize_for_stream({'name': "foo", 'value': 42})
self.streamer.receive(serialized_message[:-5])
eq_(None, self.streamer.parse_next_message())
self.streamer.receive(serialized_message[-5:])
message = self.streamer.parse_next_message()
ok_(message is not None)
eq_(message['name'], "foo")
eq_(message['value'], 42)
def test_receive_two(self):
serialized_messages = self.streamer.serialize_for_stream({'name': "foo", 'value': 42})
serialized_messages += self.streamer.serialize_for_stream({'name': "bar", 'value': 24})
self.streamer.receive(serialized_messages)
message = self.streamer.parse_next_message()
ok_(message is not None)
eq_(message['name'], "foo")
eq_(message['value'], 42)
message = self.streamer.parse_next_message()
ok_(message is not None)
eq_(message['name'], "bar")
eq_(message['value'], 24)
eq_(None, self.streamer.parse_next_message())
def test_serialize_command(self):
serialized_message = self.streamer.serialize_for_stream(
{'command': "version"})
self.streamer.receive(serialized_message)
message = self.streamer.parse_next_message()
ok_(message is not None)
eq_(message['command'], "version")
class BaseFormatterTests(object):
"""A test for every format defined in the OpenXC Message Format
spec: https://github.com/openxc/openxc-message-format
"""
def _check_serialized_deserialize_equal(self, deserialized):
serialized = self.formatter.serialize(deserialized)
eq_(deserialized, self.formatter.deserialize(serialized))
def test_simple_vehicle_message(self):
self._check_serialized_deserialize_equal({'name': "foo", 'event': False, 'value': 42})
def test_command(self):
self._check_serialized_deserialize_equal({'command': "version"})
self._check_serialized_deserialize_equal({'command': "device_id"})
def test_command_response(self):
self._check_serialized_deserialize_equal({ "command_response":
"version", "message": "v6.0-dev (default)", "status": True})
self._check_serialized_deserialize_equal({ "command_response":
"device_id", "message": "v6.0-dev (default)", "status": True})
self._check_serialized_deserialize_equal({ "command_response":
"passthrough", "status": False})
def test_passthrough_command(self):
self._check_serialized_deserialize_equal({ "command": "passthrough",
"bus": 1,
"enabled": True
})
def test_evented(self):
self._check_serialized_deserialize_equal({"name": "button_event",
"value": "up", "event": "pressed"})
def test_can_message(self):
self._check_serialized_deserialize_equal({"bus": 1, "id": 1234,
"data": "0x12345678"})
def test_diagnostic_request(self):
self._check_serialized_deserialize_equal(
{ "command": "diagnostic_request",
"action": "add",
"request": {
"bus": 1,
"id": 1234,
"mode": 1,
"pid": 5,
"payload": "0x1234",
"multiple_responses": False,
"frequency": 1,
"name": "my_pid"
}
})
def test_diagnostic_response(self):
self._check_serialized_deserialize_equal(
{"bus": 1,
"id": 1234,
"mode": 1,
"pid": 5,
"frame": 0,
"success": True,
"payload": "0x1234"
})
def test_negative_diagnostic_response(self):
self._check_serialized_deserialize_equal({"bus": 1,
"id": 1234,
"pid": 0,
"frame": 0,
"mode": 1,
"success": False,
"negative_response_code": 17})