|
@@ 371-395 (lines=25) @@
|
| 368 |
|
sub.unsubscribe(handle) |
| 369 |
|
sub.delete() |
| 370 |
|
|
| 371 |
|
def test_events_MyObject(self): |
| 372 |
|
objects = self.srv.get_objects_node() |
| 373 |
|
o = objects.add_object(3, 'MyObject') |
| 374 |
|
evgen = self.srv.get_event_generator(source=o) |
| 375 |
|
|
| 376 |
|
myhandler = MySubHandler() |
| 377 |
|
sub = self.opc.create_subscription(100, myhandler) |
| 378 |
|
handle = sub.subscribe_events(o) |
| 379 |
|
|
| 380 |
|
tid = datetime.utcnow() |
| 381 |
|
msg = b"this is my msg " |
| 382 |
|
evgen.trigger(tid, msg) |
| 383 |
|
|
| 384 |
|
ev = myhandler.future.result(10) |
| 385 |
|
self.assertIsNot(ev, None) # we did not receive event |
| 386 |
|
self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.BaseEventType)) |
| 387 |
|
self.assertEqual(ev.Severity, 1) |
| 388 |
|
self.assertEqual(ev.SourceName, 'MyObject') |
| 389 |
|
self.assertEqual(ev.SourceNode, o.nodeid) |
| 390 |
|
self.assertEqual(ev.Message.Text, msg) |
| 391 |
|
self.assertEqual(ev.Time, tid) |
| 392 |
|
|
| 393 |
|
# time.sleep(0.1) |
| 394 |
|
sub.unsubscribe(handle) |
| 395 |
|
sub.delete() |
| 396 |
|
|
| 397 |
|
def test_events_wrong_source(self): |
| 398 |
|
objects = self.srv.get_objects_node() |
|
@@ 347-369 (lines=23) @@
|
| 344 |
|
self.assertTrue(etype.get_child("2:PropertyNum") in properties) |
| 345 |
|
self.assertTrue(etype.get_child("2:PropertyString") in properties) |
| 346 |
|
|
| 347 |
|
def test_events_default(self): |
| 348 |
|
evgen = self.srv.get_event_generator() |
| 349 |
|
|
| 350 |
|
myhandler = MySubHandler() |
| 351 |
|
sub = self.opc.create_subscription(100, myhandler) |
| 352 |
|
handle = sub.subscribe_events() |
| 353 |
|
|
| 354 |
|
tid = datetime.utcnow() |
| 355 |
|
msg = b"this is my msg " |
| 356 |
|
evgen.trigger(tid, msg) |
| 357 |
|
|
| 358 |
|
ev = myhandler.future.result() |
| 359 |
|
self.assertIsNot(ev, None) # we did not receive event |
| 360 |
|
self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.BaseEventType)) |
| 361 |
|
self.assertEqual(ev.Severity, 1) |
| 362 |
|
self.assertEqual(ev.SourceName, self.opc.get_server_node().get_browse_name().Name) |
| 363 |
|
self.assertEqual(ev.SourceNode, self.opc.get_server_node().nodeid) |
| 364 |
|
self.assertEqual(ev.Message.Text, msg) |
| 365 |
|
self.assertEqual(ev.Time, tid) |
| 366 |
|
|
| 367 |
|
# time.sleep(0.1) |
| 368 |
|
sub.unsubscribe(handle) |
| 369 |
|
sub.delete() |
| 370 |
|
|
| 371 |
|
def test_events_MyObject(self): |
| 372 |
|
objects = self.srv.get_objects_node() |
FreeOpcUa /
python-opcua
