FilesExpand file tree

 3.3.4

/

network_test.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

119 lines (92 loc) · 3.64 KB

 3.3.4

/

network_test.py

Copy path

Top

File metadata and controls

• Code
• Blame

119 lines (92 loc) · 3.64 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

#!/usr/bin/env python

# coding: utf-8

from __future__ import print_function

import unittest

import threading

try:

import queue

except ImportError:

import Queue as queue

import random

import logging

logging.getLogger(__file__).setLevel(logging.WARNING)

# make a random bool:

rbool = lambda: bool(round(random.random()))

import can

channel = 'vcan0'

can.rc['interface'] = 'virtual'

@unittest.skipIf('interface' not in can.rc, "Need a CAN interface")

class ControllerAreaNetworkTestCase(unittest.TestCase):

"""

This test ensures that what messages go in to the bus is what comes out.

Requires a can interface.

To ensure that hardware and/or software message priority queues don't

effect the test, messages are sent one at a time.

"""

num_messages = 512

# TODO check if error flags are working (don't currently appear on bus)

error_flags = [False for _ in range(num_messages)]

remote_flags = [rbool() for _ in range(num_messages)]

extended_flags = [rbool() for _ in range(num_messages)]

ids = list(range(num_messages))

data = list(bytearray([random.randrange(0, 2 ** 8 - 1)

for a in range(random.randrange(9))])

for b in range(num_messages))

def producer(self, ready_event, msg_read):

self.client_bus = can.interface.Bus(channel=channel)

ready_event.wait()

for i in range(self.num_messages):

m = can.Message(

arbitration_id=self.ids[i],

is_remote_frame=self.remote_flags[i],

is_error_frame=self.error_flags[i],

is_extended_id=self.extended_flags[i],

data=self.data[i]

)

#logging.debug("writing message: {}".format(m))

if msg_read is not None:

# Don't send until the other thread is ready

msg_read.wait()

msg_read.clear()

self.client_bus.send(m)

def testProducer(self):

"""Verify that we can send arbitrary messages on the bus"""

logging.debug("testing producer alone")

ready = threading.Event()

ready.set()

self.producer(ready, None)

logging.debug("producer test complete")

def testProducerConsumer(self):

logging.debug("testing producer/consumer")

ready = threading.Event()

msg_read = threading.Event()

self.server_bus = can.interface.Bus(channel=channel)

t = threading.Thread(target=self.producer, args=(ready, msg_read))

t.start()

# Ensure there are no messages on the bus

while True:

m = self.server_bus.recv(timeout=0.5)

if m is None:

print("No messages... lets go")

break

else:

self.fail("received messages before the test has started ...")

ready.set()

i = 0

while i < self.num_messages:

msg_read.set()

msg = self.server_bus.recv(timeout=0.5)

self.assertIsNotNone(msg, "Didn't receive a message")

#logging.debug("Received message {} with data: {}".format(i, msg.data))

self.assertEqual(msg.is_extended_id, self.extended_flags[i])

if not msg.is_remote_frame:

self.assertEqual(msg.data, self.data[i])

self.assertEqual(msg.arbitration_id, self.ids[i])

self.assertEqual(msg.is_error_frame, self.error_flags[i])

self.assertEqual(msg.is_remote_frame, self.remote_flags[i])

i += 1

t.join()

self.server_bus.flush_tx_buffer()

self.server_bus.shutdown()

if __name__ == '__main__':

unittest.main()