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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
#!/usr/bin/env python
import pickle
import sys
import unittest
from copy import copy, deepcopy
from datetime import timedelta
from math import isinf, isnan
import hypothesis.errors
import hypothesis.strategies as st
import pytest
from hypothesis import HealthCheck, given, settings
from can import Message
from .config import IS_GITHUB_ACTIONS, IS_PYPY, IS_WINDOWS
from .message_helper import ComparingMessagesTestCase
class TestMessageClass(unittest.TestCase):
"""
This test tries many inputs to the message class constructor and then sanity checks
all methods and ensures that nothing crashes. It also checks whether Message._check()
allows all valid can frames.
"""
@given(
timestamp=st.floats(min_value=0.0),
arbitration_id=st.integers(),
is_extended_id=st.booleans(),
is_remote_frame=st.booleans(),
is_error_frame=st.booleans(),
channel=st.one_of(st.text(), st.integers()),
dlc=st.integers(min_value=0, max_value=8),
data=st.one_of(st.binary(min_size=0, max_size=8), st.none()),
is_fd=st.booleans(),
bitrate_switch=st.booleans(),
error_state_indicator=st.booleans(),
)
# The first run may take a second on CI runners and will hit the deadline
@settings(
max_examples=2000,
suppress_health_check=[HealthCheck.too_slow],
deadline=None if IS_GITHUB_ACTIONS else timedelta(milliseconds=500),
)
@pytest.mark.xfail(
IS_WINDOWS and IS_PYPY,
raises=hypothesis.errors.Flaky,
reason="Hypothesis generates inconsistent timestamp floats on Windows+PyPy-3.7",
)
def test_methods(self, **kwargs):
is_valid = not (
(
not kwargs["is_remote_frame"]
and (len(kwargs["data"] or []) != kwargs["dlc"])
)
or (kwargs["arbitration_id"] >= 0x800 and not kwargs["is_extended_id"])
or kwargs["arbitration_id"] >= 0x20000000
or kwargs["arbitration_id"] < 0
or (
kwargs["is_remote_frame"]
and (kwargs["is_fd"] or kwargs["is_error_frame"])
)
or (kwargs["is_remote_frame"] and len(kwargs["data"] or []) > 0)
or (
(kwargs["bitrate_switch"] or kwargs["error_state_indicator"])
and not kwargs["is_fd"]
)
or isnan(kwargs["timestamp"])
or isinf(kwargs["timestamp"])
)
# this should return normally and not throw an exception
message = Message(check=is_valid, **kwargs)
if kwargs["data"] is None or kwargs["is_remote_frame"]:
kwargs["data"] = bytearray()
if not is_valid and not kwargs["is_remote_frame"]:
with self.assertRaises(ValueError):
Message(check=True, **kwargs)
self.assertGreater(len(str(message)), 0)
self.assertGreater(len(message.__repr__()), 0)
if is_valid:
self.assertEqual(len(message), kwargs["dlc"])
self.assertTrue(bool(message))
self.assertGreater(len(f"{message}"), 0)
_ = f"{message}"
with self.assertRaises(Exception):
_ = "{somespec}".format(
message
) # pylint: disable=missing-format-argument-key
if sys.version_info.major > 2:
self.assertEqual(bytearray(bytes(message)), kwargs["data"])
# check copies and equalities
if is_valid:
self.assertEqual(message, message)
normal_copy = copy(message)
deep_copy = deepcopy(message)
for other in (normal_copy, deep_copy, message):
self.assertTrue(message.equals(other, timestamp_delta=None))
self.assertTrue(message.equals(other))
self.assertTrue(message.equals(other, timestamp_delta=0))
class MessageSerialization(unittest.TestCase, ComparingMessagesTestCase):
def __init__(self, *args, **kwargs):
unittest.TestCase.__init__(self, *args, **kwargs)
ComparingMessagesTestCase.__init__(
self, allowed_timestamp_delta=0.016, preserves_channel=True
)
def test_serialization(self):
message = Message(
timestamp=1.0,
arbitration_id=0x401,
is_extended_id=False,
is_remote_frame=False,
is_error_frame=False,
channel=1,
dlc=6,
data=bytearray([0x01, 0x02, 0x03, 0x04, 0x05, 0x06]),
is_fd=False,
)
serialized = pickle.dumps(message, -1)
deserialized = pickle.loads(serialized)
self.assertMessageEqual(message, deserialized)
if __name__ == "__main__":
unittest.main()
|
