FilesExpand file tree

 develop

/

test_bridge.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

126 lines (102 loc) · 4.09 KB

 develop

/

test_bridge.py

Copy path

Top

File metadata and controls

• Code
• Blame

126 lines (102 loc) · 4.09 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

#!/usr/bin/env python

"""

This module tests the functions inside of bridge.py

"""

import random

import string

import sys

import threading

import time

from time import sleep as real_sleep

import unittest.mock

import can

import can.bridge

from can.interfaces import virtual

from .message_helper import ComparingMessagesTestCase

class TestBridgeScriptModule(unittest.TestCase, ComparingMessagesTestCase):

TIMEOUT = 3.0

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

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

ComparingMessagesTestCase.__init__(

self,

allowed_timestamp_delta=None,

preserves_channel=False,

)

def setUp(self) -> None:

self.stop_event = threading.Event()

self.channel1 = "".join(random.choices(string.ascii_letters, k=8))

self.channel2 = "".join(random.choices(string.ascii_letters, k=8))

self.cli_args = [

"--bus1-interface",

"virtual",

"--bus1-channel",

self.channel1,

"--bus2-interface",

"virtual",

"--bus2-channel",

self.channel2,

]

self.testmsg = can.Message(

arbitration_id=0xC0FFEE, data=[0, 25, 0, 1, 3, 1, 4, 1], is_extended_id=True

)

def fake_sleep(self, duration):

"""A fake replacement for time.sleep that checks periodically

whether self.stop_event is set, and raises KeyboardInterrupt

if so.

This allows tests to simulate an interrupt (like Ctrl+C)

during long sleeps, in a controlled and responsive way.

"""

interval = 0.05 # Small interval for responsiveness

t_wakeup = time.perf_counter() + duration

while time.perf_counter() < t_wakeup:

if self.stop_event.is_set():

raise KeyboardInterrupt("Simulated interrupt from fake_sleep")

real_sleep(interval)

def test_bridge(self):

with (

unittest.mock.patch("can.bridge.time.sleep", new=self.fake_sleep),

unittest.mock.patch("can.bridge.sys.argv", [sys.argv[0], *self.cli_args]),

):

# start script

thread = threading.Thread(target=can.bridge.main)

thread.start()

# wait until script instantiates virtual buses

t0 = time.perf_counter()

while True:

with virtual.channels_lock:

if (

self.channel1 in virtual.channels

and self.channel2 in virtual.channels

):

break

if time.perf_counter() > t0 + 2.0:

raise TimeoutError("Bridge script did not create virtual buses")

real_sleep(0.2)

# create buses with the same channels as in scripts

with (

can.interfaces.virtual.VirtualBus(self.channel1) as bus1,

can.interfaces.virtual.VirtualBus(self.channel2) as bus2,

):

# send test message to bus1, it should be received on bus2

bus1.send(self.testmsg)

recv_msg = bus2.recv(self.TIMEOUT)

self.assertMessageEqual(self.testmsg, recv_msg)

# assert that both buses are empty

self.assertIsNone(bus1.recv(0))

self.assertIsNone(bus2.recv(0))

# send test message to bus2, it should be received on bus1

bus2.send(self.testmsg)

recv_msg = bus1.recv(self.TIMEOUT)

self.assertMessageEqual(self.testmsg, recv_msg)

# assert that both buses are empty

self.assertIsNone(bus1.recv(0))

self.assertIsNone(bus2.recv(0))

# stop the bridge script

self.stop_event.set()

thread.join()

# assert that the virtual buses were closed

with virtual.channels_lock:

self.assertNotIn(self.channel1, virtual.channels)

self.assertNotIn(self.channel2, virtual.channels)

if __name__ == "__main__":

unittest.main()