FilesExpand file tree

 develop

/

test_message_class.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

137 lines (117 loc) · 4.62 KB

 develop

/

test_message_class.py

Copy path

Top

File metadata and controls

• Code
• Blame

137 lines (117 loc) · 4.62 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

130

131

132

133

134

135

136

137

#!/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.errors

import hypothesis.strategies as st

from can import Message

from .message_helper import ComparingMessagesTestCase

from .config import IS_GITHUB_ACTIONS, IS_WINDOWS, IS_PYPY

import pytest

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),

)

@pytest.mark.xfail(

IS_WINDOWS and IS_PYPY,

raises=hypothesis.errors.Flaky,

reason="Hypothesis generates inconistent 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("{}".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()