Fix registering multiple notifications on the AdsTestServer

CHANGELOG.md

@@ -12,6 +12,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 ### Changed
 * [#400](https://github.com/stlehmann/pyads/issues/400) Full support for pyproject.toml
 
+### Fixed
+* [#471](https://github.com/stlehmann/pyads/issues/471) Fixed registering of multiple notifications on AdsTestServer
+
 ## 3.5.0
 
 ### Added

src/pyads/testserver/advanced_handler.py

@@ -164,9 +164,9 @@ def write(self, value: bytes, request: AmsPacket = None):
     def register_notification(self) -> int:
         """Register a new notification."""
 
-        handle = self.notification_count
+        handle = PLCVariable.notification_count
         self.notifications.append(handle)
-        self.notification_count += 1
+        PLCVariable.notification_count += 1
         return handle
 
     def unregister_notification(self, handle: int = None):

tests/test_testserver.py

@@ -10,7 +10,7 @@
 import time
 import unittest
 import pyads
-from pyads.testserver import AdsTestServer, BasicHandler
+from pyads.testserver import AdsTestServer, BasicHandler, AdvancedHandler, PLCVariable
 
 # These are pretty arbitrary
 TEST_SERVER_AMS_NET_ID = "127.0.0.1.1.1"
@@ -85,5 +85,95 @@ def test_server_disconnect_then_del_device_notification(self):
             self.fail(f"Closing server connection raised: {e}")
 
 
+    def test_advanced_handler_register_multiple_notifications(self):
+        """Test registering multiple notifications on the same variable.
+
+        Tests fix of issue [#471](https://github.com/stlehmann/pyads/issues/471), original pull request: [#474](https://github.com/stlehmann/pyads/pull/474)
+        """
+        handler = AdvancedHandler()
+        test_server = AdsTestServer(handler=handler, logging=False, ip_address=TEST_SERVER_IP_ADDRESS)
+
+        var1 = "MAIN.var1"
+        var2 = "MAIN.var2"
+
+        # Create two WORD variables in the handler, with explicit indices
+        handler.add_variable(
+            PLCVariable(
+                var1,
+                0,
+                symbol_type="WORD",
+                ads_type=pyads.constants.ADST_UINT16
+            )
+        )
+
+        handler.add_variable(
+            PLCVariable(
+                var2,
+                0,
+                symbol_type="WORD",
+                ads_type=pyads.constants.ADST_UINT16,
+            )
+        )
+
+        test_server.start()
+        time.sleep(0.1)
+
+        plc = pyads.Connection(TEST_SERVER_AMS_NET_ID, TEST_SERVER_AMS_PORT, TEST_SERVER_IP_ADDRESS)
+        plc.open()
+
+        try:
+            # Collect notifications
+            events = []
+
+            sym1 = plc.get_symbol(var1)
+            sym2 = plc.get_symbol(var2)
+
+            @plc.notification(pyads.PLCTYPE_UINT)
+            def on_change(handle, name, timestamp, value):
+                events.append((name, value))
+
+            attrib = pyads.NotificationAttrib(
+                length=2,
+                trans_mode=pyads.ADSTRANS_SERVERONCHA,
+                max_delay=0,
+                cycle_time=0,
+            )
+
+            # Adding notifications using index_group and index_offset from the symbols
+            # as using name "add_variable" without explicit indices would create indices (12345, 10000 + some offset) 
+            # and add_device_notification expects (61445, 10000 + some offset) in this case.
+            handles1 = plc.add_device_notification((sym1.index_group, sym1.index_offset), attrib, on_change)
+            handles2 = plc.add_device_notification((sym2.index_group, sym2.index_offset), attrib, on_change)
+
+            # Trigger notifications
+            triggers = [
+                (var1, 456),
+                (var2, 4321),
+                (var1, 789),
+            ]
+            for var, value in triggers:
+                plc.write_by_name(var, value, pyads.PLCTYPE_UINT)
+
+            time.sleep(0.2)
+
+            # Cleanup notifications
+            if handles1 is not None:
+                plc.del_device_notification(*handles1)
+            if handles2 is not None:
+                plc.del_device_notification(*handles2)
+
+            # Assert we observed expected changes
+            values_by_name = {}
+            for name, value in events:
+                values_by_name.setdefault(name, []).append(value)
+
+            for var, value in triggers:
+                self.assertIn(value, values_by_name[var])
+        finally:
+            plc.close()
+            test_server.stop()
+            time.sleep(0.1)
+
+
 if __name__ == "__main__":
     unittest.main()