FilesExpand file tree

 develop

/

test_message_class.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

129 lines (110 loc) · 4.37 KB

 develop

/

test_message_class.py

Copy path

Top

File metadata and controls

• Code
• Blame

129 lines (110 loc) · 4.37 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

#!/usr/bin/env python

# coding: utf-8

import unittest

import sys

from math import isinf, isnan

from copy import copy, deepcopy

import pickle

from datetime import timedelta

from hypothesis import given, settings

import hypothesis.strategies as st

from can import Message

from .message_helper import ComparingMessagesTestCase

from .config import IS_GITHUB_ACTIONS

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,

deadline=None if IS_GITHUB_ACTIONS else timedelta(milliseconds=500),

)

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("{}".format(message)), 0)

_ = "{}".format(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()