-
Notifications
You must be signed in to change notification settings - Fork 358
Expand file tree
/
Copy pathtest_node.py
More file actions
123 lines (95 loc) · 4.86 KB
/
Copy pathtest_node.py
File metadata and controls
123 lines (95 loc) · 4.86 KB
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
import unittest
import canopen
def count_subscribers(network: canopen.Network) -> int:
"""Count the number of subscribers in the network."""
return sum(len(n) for n in network.subscribers.values())
class TestBaseNode(unittest.TestCase):
def test_valid_node_id(self):
node = canopen.node.base.BaseNode(1, canopen.ObjectDictionary())
self.assertEqual(node.id, 1)
def test_valid_node_id_from_od(self):
od = canopen.ObjectDictionary()
od.node_id = 2
node = canopen.node.base.BaseNode(0, od)
self.assertEqual(node.id, 2)
def test_invalid_node_id(self):
with self.assertRaises(ValueError):
_ = canopen.node.base.BaseNode(0, canopen.ObjectDictionary())
with self.assertRaises(ValueError):
_ = canopen.node.base.BaseNode(128, canopen.ObjectDictionary())
class TestLocalNode(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.network = canopen.Network()
cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0
cls.network.connect(interface="virtual")
cls.node = canopen.LocalNode(2, canopen.objectdictionary.ObjectDictionary())
@classmethod
def tearDownClass(cls):
cls.network.disconnect()
def test_associate_network(self):
# Need to store the number of subscribers before associating because the
# network implementation automatically adds subscribers to the list
n_subscribers = count_subscribers(self.network)
# Associating the network with the local node
self.node.associate_network(self.network)
self.assertIs(self.node.network, self.network)
self.assertIs(self.node.sdo.network, self.network)
self.assertIs(self.node.tpdo.network, self.network)
self.assertIs(self.node.rpdo.network, self.network)
self.assertIs(self.node.nmt.network, self.network)
self.assertIs(self.node.emcy.network, self.network)
# Test that its not possible to associate the network multiple times
with self.assertRaises(RuntimeError) as cm:
self.node.associate_network(self.network)
self.assertIn("already associated with a network", str(cm.exception))
# Test removal of the network. The count of subscribers should
# be the same as before the association
self.node.remove_network()
uninitalized = canopen.network._UNINITIALIZED_NETWORK
self.assertIs(self.node.network, uninitalized)
self.assertIs(self.node.sdo.network, uninitalized)
self.assertIs(self.node.tpdo.network, uninitalized)
self.assertIs(self.node.rpdo.network, uninitalized)
self.assertIs(self.node.nmt.network, uninitalized)
self.assertIs(self.node.emcy.network, uninitalized)
self.assertEqual(count_subscribers(self.network), n_subscribers)
# Test that its possible to deassociate the network multiple times
self.node.remove_network()
class TestRemoteNode(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.network = canopen.Network()
cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0
cls.network.connect(interface="virtual")
cls.node = canopen.RemoteNode(2, canopen.objectdictionary.ObjectDictionary())
@classmethod
def tearDownClass(cls):
cls.network.disconnect()
def test_associate_network(self):
# Need to store the number of subscribers before associating because the
# network implementation automatically adds subscribers to the list
n_subscribers = count_subscribers(self.network)
# Associating the network with the local node
self.node.associate_network(self.network)
self.assertIs(self.node.network, self.network)
self.assertIs(self.node.sdo.network, self.network)
self.assertIs(self.node.tpdo.network, self.network)
self.assertIs(self.node.rpdo.network, self.network)
self.assertIs(self.node.nmt.network, self.network)
# Test that its not possible to associate the network multiple times
with self.assertRaises(RuntimeError) as cm:
self.node.associate_network(self.network)
self.assertIn("already associated with a network", str(cm.exception))
# Test removal of the network. The count of subscribers should
# be the same as before the association
self.node.remove_network()
uninitalized = canopen.network._UNINITIALIZED_NETWORK
self.assertIs(self.node.network, uninitalized)
self.assertIs(self.node.sdo.network, uninitalized)
self.assertIs(self.node.tpdo.network, uninitalized)
self.assertIs(self.node.rpdo.network, uninitalized)
self.assertIs(self.node.nmt.network, uninitalized)
self.assertEqual(count_subscribers(self.network), n_subscribers)
# Test that its possible to deassociate the network multiple times
self.node.remove_network()