Adapt tests to new event-emitting-node behavior / API

Author: Nebukadneza

tests/tests_server.py

@@ -250,24 +250,24 @@ def test_eventgenerator_BaseEvent_Identifier(self):
         check_eventgenerator_SourceServer(self, evgen)
 
     def test_eventgenerator_sourceServer_Node(self):
-        evgen = self.opc.get_event_generator(source=opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.Server)))
+        evgen = self.opc.get_event_generator(emitting_node=opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.Server)))
         check_eventgenerator_BaseEvent(self, evgen)
         check_eventgenerator_SourceServer(self, evgen)
 
     def test_eventgenerator_sourceServer_NodeId(self):
-        evgen = self.opc.get_event_generator(source=ua.NodeId(ua.ObjectIds.Server))
+        evgen = self.opc.get_event_generator(emitting_node=ua.NodeId(ua.ObjectIds.Server))
         check_eventgenerator_BaseEvent(self, evgen)
         check_eventgenerator_SourceServer(self, evgen)
 
     def test_eventgenerator_sourceServer_ObjectIds(self):
-        evgen = self.opc.get_event_generator(source=ua.ObjectIds.Server)
+        evgen = self.opc.get_event_generator(emitting_node=ua.ObjectIds.Server)
         check_eventgenerator_BaseEvent(self, evgen)
         check_eventgenerator_SourceServer(self, evgen)
 
     def test_eventgenerator_sourceMyObject(self):
         objects = self.opc.get_objects_node()
         o = objects.add_object(3, 'MyObject')
-        evgen = self.opc.get_event_generator(source=o)
+        evgen = self.opc.get_event_generator(emitting_node=o)
         check_eventgenerator_BaseEvent(self, evgen)
         check_event_generator_object(self, evgen, o)
 
@@ -276,8 +276,10 @@ def test_eventgenerator_source_collision(self):
         o = objects.add_object(3, 'MyObject')
         event = BaseEvent(sourcenode=o.nodeid)
         evgen = self.opc.get_event_generator(event, ua.ObjectIds.Server)
+        evgen.event.SourceNode = o.nodeid
+        evgen.event.SourceName = o.get_browse_name().Name
         check_eventgenerator_BaseEvent(self, evgen)
-        check_event_generator_object(self, evgen, o)
+        check_event_generator_object(self, evgen, o, emitting_node=opcua.Node(self.opc.iserver.isession, ua.ObjectIds.Server))
 
     def test_eventgenerator_InheritedEvent(self):
         evgen = self.opc.get_event_generator(ua.ObjectIds.AuditEventType)
@@ -523,14 +525,17 @@ def check_eventgenerator_SourceServer(test, evgen):
     test.assertGreaterEqual(len(refs), 1)
 
 
-def check_event_generator_object(test, evgen, obj):
+def check_event_generator_object(test, evgen, obj, emitting_node=None):
     test.assertEqual(evgen.event.SourceName, obj.get_browse_name().Name)
     test.assertEqual(evgen.event.SourceNode, obj.nodeid)
-    test.assertEqual(obj.get_event_notifier(), {ua.EventNotifier.SubscribeToEvents})
-
-    refs = obj.get_referenced_nodes(ua.ObjectIds.GeneratesEvent, ua.BrowseDirection.Forward, ua.NodeClass.ObjectType, False)
-    test.assertEqual(len(refs), 1)
-    test.assertEqual(refs[0].nodeid, evgen.event.EventType)
+    if not emitting_node:
+        test.assertEqual(obj.get_event_notifier(), {ua.EventNotifier.SubscribeToEvents})
+        refs = obj.get_referenced_nodes(ua.ObjectIds.GeneratesEvent, ua.BrowseDirection.Forward, ua.NodeClass.ObjectType, False)
+    else:
+        test.assertEqual(emitting_node.get_event_notifier(), {ua.EventNotifier.SubscribeToEvents})
+        refs = emitting_node.get_referenced_nodes(ua.ObjectIds.GeneratesEvent, ua.BrowseDirection.Forward, ua.NodeClass.ObjectType, False)
+
+    test.assertIn(evgen.event.EventType, [x.nodeid for x in refs])
 
 
 def check_eventgenerator_BaseEvent(test, evgen):

tests/tests_subscriptions.py

@@ -373,11 +373,13 @@ def test_events_default(self):
     def test_events_MyObject(self):
         objects = self.srv.get_objects_node()
         o = objects.add_object(3, 'MyObject')
-        evgen = self.srv.get_event_generator(source=o)
+        evgen = self.srv.get_event_generator()
+        evgen.event.SourceNode=o.nodeid
+        evgen.event.SourceName=o.get_browse_name().Name
 
         myhandler = MySubHandler()
         sub = self.opc.create_subscription(100, myhandler)
-        handle = sub.subscribe_events(o)
+        handle = sub.subscribe_events()
 
         tid = datetime.utcnow()
         msg = "this is my msg "
@@ -399,7 +401,7 @@ def test_events_MyObject(self):
     def test_events_wrong_source(self):
         objects = self.srv.get_objects_node()
         o = objects.add_object(3, 'MyObject')
-        evgen = self.srv.get_event_generator(source=o)
+        evgen = self.srv.get_event_generator(emitting_node=o)
 
         myhandler = MySubHandler()
         sub = self.opc.create_subscription(100, myhandler)
@@ -422,7 +424,7 @@ def test_events_CustomEvent(self):
 
         myhandler = MySubHandler()
         sub = self.opc.create_subscription(100, myhandler)
-        handle = sub.subscribe_events(evtypes=etype)
+        handle = sub.subscribe_events(sourcenode=ua.ObjectIds.Server, evtypes=etype)
 
         propertynum = 2
         propertystring = "This is my test"
@@ -453,16 +455,18 @@ def test_events_CustomEvent_MyObject(self):
         objects = self.srv.get_objects_node()
         o = objects.add_object(3, 'MyObject')
         etype = self.srv.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
-        evgen = self.srv.get_event_generator(etype, o)
+        evgen = self.srv.get_event_generator(etype, emitting_node=o)
+        evgen.event.SourceNode = o.nodeid
 
         myhandler = MySubHandler()
         sub = self.opc.create_subscription(100, myhandler)
-        handle = sub.subscribe_events(o, etype)
+        handle = sub.subscribe_events(sourcenode=o, evtypes=etype)
 
         propertynum = 2
         propertystring = "This is my test"
         evgen.event.PropertyNum = propertynum
         evgen.event.PropertyString = propertystring
+        evgen.event.SourceNode = o.nodeid
         tid = datetime.utcnow()
         msg = "this is my msg "
         evgen.trigger(tid, msg)