FilesExpand file tree

 3.3.4

/

test_message_sync.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

125 lines (95 loc) · 3.67 KB

 3.3.4

/

test_message_sync.py

Copy path

Top

File metadata and controls

• Code
• Blame

125 lines (95 loc) · 3.67 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

#!/usr/bin/env python

# coding: utf-8

"""

This module tests :class:`can.MessageSync`.

"""

from __future__ import absolute_import

from copy import copy

from time import time

import gc

import unittest

import pytest

from can import MessageSync, Message

from .config import IS_CI, IS_APPVEYOR, IS_TRAVIS, IS_OSX

from .message_helper import ComparingMessagesTestCase

from .data.example_data import TEST_MESSAGES_BASE

TEST_FEWER_MESSAGES = TEST_MESSAGES_BASE[::2]

def inc(value):

"""Makes the test boundaries give some more space when run on the CI server."""

if IS_CI:

return value * 1.5

else:

return value

@unittest.skipIf(IS_APPVEYOR or (IS_TRAVIS and IS_OSX),

"this environment's timings are too unpredictable")

class TestMessageSync(unittest.TestCase, ComparingMessagesTestCase):

def __init__(self, *args, **kwargs):

unittest.TestCase.__init__(self, *args, **kwargs)

ComparingMessagesTestCase.__init__(self)

def setup_method(self, _):

# disabling the garbage collector makes the time readings more reliable

gc.disable()

def teardown_method(self, _):

# we need to reenable the garbage collector again

gc.enable()

@pytest.mark.timeout(inc(0.2))

def test_general(self):

messages = [

Message(timestamp=50.0),

Message(timestamp=50.0),

Message(timestamp=50.0 + 0.05),

Message(timestamp=50.0 + 0.05 + 0.08),

Message(timestamp=50.0) # back in time

]

sync = MessageSync(messages, gap=0.0)

start = time()

collected = []

timings = []

for message in sync:

collected.append(message)

now = time()

timings.append(now - start)

start = now

self.assertMessagesEqual(messages, collected)

self.assertEqual(len(timings), len(messages), "programming error in test code")

self.assertTrue(0.0 <= timings[0] < inc(0.005), str(timings[0]))

self.assertTrue(0.0 <= timings[1] < inc(0.005), str(timings[1]))

self.assertTrue(0.045 <= timings[2] < inc(0.055), str(timings[2]))

self.assertTrue(0.075 <= timings[3] < inc(0.085), str(timings[3]))

self.assertTrue(0.0 <= timings[4] < inc(0.005), str(timings[4]))

@pytest.mark.timeout(inc(0.1) * len(TEST_FEWER_MESSAGES)) # very conservative

def test_skip(self):

messages = copy(TEST_FEWER_MESSAGES)

sync = MessageSync(messages, skip=0.005, gap=0.0)

before = time()

collected = list(sync)

after = time()

took = after - before

# the handling of the messages itself also takes some time:

# ~0.001 s/message on a ThinkPad T560 laptop (Ubuntu 18.04, i5-6200U)

assert 0 < took < inc(len(messages) * (0.005 + 0.003)), "took: {}s".format(took)

self.assertMessagesEqual(messages, collected)

if not IS_APPVEYOR: # this environment's timings are too unpredictable

@pytest.mark.timeout(inc(0.3))

@pytest.mark.parametrize("timestamp_1,timestamp_2", [

(0.0, 0.0),

(0.0, 0.01),

(0.01, 0.0),

])

def test_gap(timestamp_1, timestamp_2):

"""This method is alone so it can be parameterized."""

messages = [

Message(arbitration_id=0x1, timestamp=timestamp_1),

Message(arbitration_id=0x2, timestamp=timestamp_2)

]

sync = MessageSync(messages, gap=0.1)

gc.disable()

before = time()

collected = list(sync)

after = time()

gc.enable()

took = after - before

assert 0.1 <= took < inc(0.3)

assert messages == collected

if __name__ == '__main__':

unittest.main()