Fix multiple assosciations and removals of networks

Author: sveinse

canopen/node/local.py

@@ -39,6 +39,9 @@ def __init__(
         self.emcy = EmcyProducer(0x80 + self.id)
 
     def associate_network(self, network: canopen.network.Network):
+        if self.network is not canopen.network._UNINITIALIZED_NETWORK:
+            # Unsubscribe from old network (to prevent double subscription)
+            self.remove_network()
         self.network = network
         self.sdo.network = network
         self.tpdo.network = network
@@ -49,6 +52,9 @@ def associate_network(self, network: canopen.network.Network):
         network.subscribe(0, self.nmt.on_command)
 
     def remove_network(self) -> None:
+        # Make it safe to call this method multiple times
+        if self.network is canopen.network._UNINITIALIZED_NETWORK:
+            return
         self.network.unsubscribe(self.sdo.rx_cobid, self.sdo.on_request)
         self.network.unsubscribe(0, self.nmt.on_command)
         self.network = canopen.network._UNINITIALIZED_NETWORK

canopen/node/remote.py

@@ -51,6 +51,9 @@ def __init__(
             self.load_configuration()
 
     def associate_network(self, network: canopen.network.Network):
+        if self.network is not canopen.network._UNINITIALIZED_NETWORK:
+            # Unsubscribe from old network (to prevent double subscriptions)
+            self.remove_network()
         self.network = network
         self.sdo.network = network
         self.pdo.network = network
@@ -64,6 +67,9 @@ def associate_network(self, network: canopen.network.Network):
         network.subscribe(0, self.nmt.on_command)
 
     def remove_network(self) -> None:
+        # Make it safe to call this method multiple times
+        if self.network is canopen.network._UNINITIALIZED_NETWORK:
+            return
         for sdo in self.sdo_channels:
             self.network.unsubscribe(sdo.tx_cobid, sdo.on_response)
         self.network.unsubscribe(0x700 + self.id, self.nmt.on_heartbeat)

test/test_node.py

@@ -0,0 +1,121 @@
+"""Unit tests for the RemoteNode and LocalNode classes."""
+import unittest
+
+import canopen
+
+from .util import SAMPLE_EDS
+
+
+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 TestLocalNode(unittest.TestCase):
+    """
+    Test local node.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls.network = canopen.Network()
+        cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0
+        cls.network.connect("test", interface="virtual")
+
+        cls.node = canopen.LocalNode(2, SAMPLE_EDS)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.network.disconnect()
+
+    def test_associate_network(self):
+
+        # Need to store the number of subscribers before associating because
+        # the current 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 possible to associate the network multiple times
+        # by checking that the number of subscribers remains the same
+        count = count_subscribers(self.network)
+        self.node.associate_network(self.network)
+        self.assertEqual(count_subscribers(self.network), count)
+
+        # 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):
+    """
+    Test remote node.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls.network = canopen.Network()
+        cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0
+        cls.network.connect("test", interface="virtual")
+
+        cls.node = canopen.RemoteNode(2, SAMPLE_EDS)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.network.disconnect()
+
+    def test_associate_network(self):
+
+        # Need to store the number of subscribers before associating because
+        # the current 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 possible to associate the network multiple times
+        # by checking that the number of subscribers remains the same
+        count = count_subscribers(self.network)
+        self.node.associate_network(self.network)
+        self.assertEqual(count_subscribers(self.network), count)
+
+        # 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()