Updated from review

Author: sveinse

canopen/network.py

@@ -79,6 +79,9 @@ def unsubscribe(self, can_id, callback=None) -> None:
             del self.subscribers[can_id]
         else:
             self.subscribers[can_id].remove(callback)
+            # Remove empty list entry
+            if not self.subscribers[can_id]:
+                del self.subscribers[can_id]
 
     def connect(self, *args, **kwargs) -> Network:
         """Connect to CAN bus using python-can.

canopen/node/local.py

@@ -39,9 +39,8 @@ 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()
+        if self.has_network():
+            raise RuntimeError("Node is already associated with a network")
         self.network = network
         self.sdo.network = network
         self.tpdo.network = network
@@ -53,7 +52,7 @@ def associate_network(self, network: canopen.network.Network):
 
     def remove_network(self) -> None:
         # Make it safe to call this method multiple times
-        if self.network is canopen.network._UNINITIALIZED_NETWORK:
+        if not self.has_network():
             return
         self.network.unsubscribe(self.sdo.rx_cobid, self.sdo.on_request)
         self.network.unsubscribe(0, self.nmt.on_command)

canopen/node/remote.py

@@ -51,9 +51,8 @@ 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()
+        if self.has_network():
+            raise RuntimeError("Node is already associated with a network")
         self.network = network
         self.sdo.network = network
         self.pdo.network = network
@@ -68,7 +67,7 @@ def associate_network(self, network: canopen.network.Network):
 
     def remove_network(self) -> None:
         # Make it safe to call this method multiple times
-        if self.network is canopen.network._UNINITIALIZED_NETWORK:
+        if not self.has_network():
             return
         for sdo in self.sdo_channels:
             self.network.unsubscribe(sdo.tx_cobid, sdo.on_response)

test/test_node.py

@@ -2,6 +2,7 @@
 import unittest
 
 import canopen
+import canopen.network
 
 from .util import SAMPLE_EDS
 
@@ -14,15 +15,13 @@ def count_subscribers(network: canopen.Network) -> int:
 
 
 class TestLocalNode(unittest.TestCase):
-    """
-    Test local node.
-    """
+    """Unit tests for the LocalNode class."""
 
     @classmethod
     def setUpClass(cls):
         cls.network = canopen.Network()
         cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0
-        cls.network.connect("test", interface="virtual")
+        cls.network.connect(interface="virtual")
 
         cls.node = canopen.LocalNode(2, SAMPLE_EDS)
 
@@ -46,11 +45,10 @@ def test_associate_network(self):
         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 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
@@ -69,15 +67,13 @@ def test_associate_network(self):
 
 
 class TestRemoteNode(unittest.TestCase):
-    """
-    Test remote node.
-    """
+    """Unittests for the RemoteNode class."""
 
     @classmethod
     def setUpClass(cls):
         cls.network = canopen.Network()
         cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0
-        cls.network.connect("test", interface="virtual")
+        cls.network.connect(interface="virtual")
 
         cls.node = canopen.RemoteNode(2, SAMPLE_EDS)
 
@@ -100,11 +96,10 @@ def test_associate_network(self):
         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 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