ros_opcua_impl_python_opcua: tests

Go to the documentation of this file.
00001 import unittest
00002 import os
00003 import shelve
00004 import time
00005 
00006 from tests_common import CommonTests, add_server_methods
00007 from tests_xml import XmlTests
00008 from tests_subscriptions import SubscriptionTests
00009 from datetime import timedelta, datetime
00010 from tempfile import NamedTemporaryFile
00011 
00012 import opcua
00013 from opcua import Server
00014 from opcua import Client
00015 from opcua import ua
00016 from opcua import uamethod
00017 from opcua.common.event_objects import BaseEvent, AuditEvent
00018 from opcua.common import ua_utils
00019 
00020 
00021 port_num = 48540
00022 port_discovery = 48550
00023 
00024 
00025 class TestServer(unittest.TestCase, CommonTests, SubscriptionTests, XmlTests):
00026 
00027     '''
00028     Run common tests on server side
00029     Tests that can only be run on server side must be defined here
00030     '''
00031     @classmethod
00032     def setUpClass(cls):
00033         cls.srv = Server()
00034         cls.srv.set_endpoint('opc.tcp://localhost:{0:d}'.format(port_num))
00035         add_server_methods(cls.srv)
00036         cls.srv.start()
00037         cls.opc = cls.srv
00038         cls.discovery = Server()
00039         cls.discovery.set_application_uri("urn:freeopcua:python:discovery")
00040         cls.discovery.set_endpoint('opc.tcp://localhost:{0:d}'.format(port_discovery))
00041         cls.discovery.start()
00042 
00043     @classmethod
00044     def tearDownClass(cls):
00045         cls.srv.stop()
00046         cls.discovery.stop()
00047 
00048     def test_discovery(self):
00049         client = Client(self.discovery.endpoint.geturl())
00050         client.connect()
00051         try:
00052             servers = client.find_servers()
00053             new_app_uri = "urn:freeopcua:python:server:test_discovery"
00054             self.srv.application_uri = new_app_uri
00055             self.srv.register_to_discovery(self.discovery.endpoint.geturl(), 0)
00056             time.sleep(0.1)  # let server register registration
00057             new_servers = client.find_servers()
00058             self.assertEqual(len(new_servers) - len(servers) , 1)
00059             self.assertFalse(new_app_uri in [s.ApplicationUri for s in servers])
00060             self.assertTrue(new_app_uri in [s.ApplicationUri for s in new_servers])
00061         finally:
00062             client.disconnect()
00063 
00064     def test_find_servers2(self):
00065         client = Client(self.discovery.endpoint.geturl())
00066         client.connect()
00067         try:
00068             servers = client.find_servers()
00069             new_app_uri1 = "urn:freeopcua:python:server:test_discovery1"
00070             self.srv.application_uri = new_app_uri1
00071             self.srv.register_to_discovery(self.discovery.endpoint.geturl(), period=0)
00072             new_app_uri2 = "urn:freeopcua:python:test_discovery2"
00073             self.srv.application_uri = new_app_uri2
00074             self.srv.register_to_discovery(self.discovery.endpoint.geturl(), period=0)
00075             time.sleep(0.1)  # let server register registration
00076             new_servers = client.find_servers()
00077             self.assertEqual(len(new_servers) - len(servers) , 2)
00078             self.assertFalse(new_app_uri1 in [s.ApplicationUri for s in servers])
00079             self.assertFalse(new_app_uri2 in [s.ApplicationUri for s in servers])
00080             self.assertTrue(new_app_uri1 in [s.ApplicationUri for s in new_servers])
00081             self.assertTrue(new_app_uri2 in [s.ApplicationUri for s in new_servers])
00082             # now do a query with filer
00083             new_servers = client.find_servers(["urn:freeopcua:python:server"])
00084             self.assertEqual(len(new_servers) - len(servers) , 0)
00085             self.assertTrue(new_app_uri1 in [s.ApplicationUri for s in new_servers])
00086             self.assertFalse(new_app_uri2 in [s.ApplicationUri for s in new_servers])
00087             # now do a query with filer
00088             new_servers = client.find_servers(["urn:freeopcua:python"])
00089             self.assertEqual(len(new_servers) - len(servers) , 2)
00090             self.assertTrue(new_app_uri1 in [s.ApplicationUri for s in new_servers])
00091             self.assertTrue(new_app_uri2 in [s.ApplicationUri for s in new_servers])
00092         finally:
00093             client.disconnect()
00094 
00095 
00096     """
00097     # not sure if this test is necessary, and there is a lot repetition with previous test
00098     def test_discovery_server_side(self):
00099         servers = self.discovery.find_servers()
00100         self.assertEqual(len(servers), 1)
00101         self.srv.register_to_discovery(self.discovery.endpoint.geturl(), 1)
00102         time.sleep(1) # let server register registration
00103         servers = self.discovery.find_servers()
00104         print("SERVERS 2", servers)
00105         self.assertEqual(len(servers), 2)
00106     """
00107     # def test_register_server2(self):
00108         # servers = self.opc.register_server()
00109 
00110     def test_register_namespace(self):
00111         uri = 'http://mycustom.Namespace.com'
00112         idx1 = self.opc.register_namespace(uri)
00113         idx2 = self.opc.get_namespace_index(uri)
00114         self.assertEqual(idx1, idx2)
00115 
00116     def test_register_existing_namespace(self):
00117         uri = 'http://mycustom.Namespace.com'
00118         idx1 = self.opc.register_namespace(uri)
00119         idx2 = self.opc.register_namespace(uri)
00120         idx3 = self.opc.get_namespace_index(uri)
00121         self.assertEqual(idx1, idx2)
00122         self.assertEqual(idx1, idx3)
00123 
00124     def test_register_use_namespace(self):
00125         uri = 'http://my_very_custom.Namespace.com'
00126         idx = self.opc.register_namespace(uri)
00127         root = self.opc.get_root_node()
00128         myvar = root.add_variable(idx, 'var_in_custom_namespace', [5])
00129         myid = myvar.nodeid
00130         self.assertEqual(idx, myid.NamespaceIndex)
00131 
00132     def test_server_method(self):
00133         def func(parent, variant):
00134             variant.Value *= 2
00135             return [variant]
00136         o = self.opc.get_objects_node()
00137         v = o.add_method(3, 'Method1', func, [ua.VariantType.Int64], [ua.VariantType.Int64])
00138         result = o.call_method(v, ua.Variant(2.1))
00139         self.assertEqual(result, 4.2)
00140 
00141     def test_historize_variable(self):
00142         o = self.opc.get_objects_node()
00143         var = o.add_variable(3, "test_hist", 1.0)
00144         self.srv.iserver.enable_history_data_change(var, timedelta(days=1))
00145         time.sleep(1)
00146         var.set_value(2.0)
00147         var.set_value(3.0)
00148         self.srv.iserver.disable_history_data_change(var)
00149 
00150     def test_historize_events(self):
00151         srv_node = self.srv.get_node(ua.ObjectIds.Server)
00152         srvevgen = self.srv.get_event_generator()
00153         self.srv.iserver.enable_history_event(srv_node, period=None)
00154         srvevgen.trigger(message="Message")
00155         self.srv.iserver.disable_history_event(srv_node)
00156 
00157     def test_references_for_added_nodes_method(self):
00158         objects = self.opc.get_objects_node()
00159         o = objects.add_object(3, 'MyObject')
00160         nodes = objects.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Forward, includesubtypes=False)
00161         self.assertTrue(o in nodes)
00162         nodes = o.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
00163         self.assertTrue(objects in nodes)
00164         self.assertEqual(o.get_parent(), objects)
00165         self.assertEqual(o.get_type_definition().Identifier, ua.ObjectIds.BaseObjectType)
00166 
00167         @uamethod
00168         def callback(parent):
00169             return
00170 
00171         m = o.add_method(3, 'MyMethod', callback)
00172         nodes = o.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Forward, includesubtypes=False)
00173         self.assertTrue(m in nodes)
00174         nodes = m.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
00175         self.assertTrue(o in nodes)
00176         self.assertEqual(m.get_parent(), o)
00177 
00178     # This should work for following BaseEvent tests to work (maybe to write it a bit differentlly since they are not independent)
00179     def test_get_event_from_type_node_BaseEvent(self):
00180         ev = opcua.common.events.get_event_obj_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)))
00181         check_base_event(self, ev)
00182 
00183     def test_get_event_from_type_node_Inhereted_AuditEvent(self):
00184         ev = opcua.common.events.get_event_obj_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.AuditEventType)))
00185         self.assertIsNot(ev, None)  # we did not receive event
00186         self.assertIsInstance(ev, BaseEvent)
00187         self.assertIsInstance(ev, AuditEvent)
00188         self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.AuditEventType))
00189         self.assertEqual(ev.Severity, 1)
00190         self.assertEqual(ev.ActionTimeStamp, None)
00191         self.assertEqual(ev.Status, False)
00192         self.assertEqual(ev.ServerId, None)
00193         self.assertEqual(ev.ClientAuditEntryId, None)
00194         self.assertEqual(ev.ClientUserId, None)
00195 
00196     def test_eventgenerator_default(self):
00197         evgen = self.opc.get_event_generator()
00198         check_eventgenerator_BaseEvent(self, evgen)
00199         check_eventgenerator_SourceServer(self, evgen)
00200 
00201     def test_eventgenerator_BaseEvent_object(self):
00202         evgen = self.opc.get_event_generator(BaseEvent())
00203         check_eventgenerator_BaseEvent(self, evgen)
00204         check_eventgenerator_SourceServer(self, evgen)
00205 
00206     def test_eventgenerator_BaseEvent_Node(self):
00207         evgen = self.opc.get_event_generator(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)))
00208         check_eventgenerator_BaseEvent(self, evgen)
00209         check_eventgenerator_SourceServer(self, evgen)
00210 
00211     def test_eventgenerator_BaseEvent_NodeId(self):
00212         evgen = self.opc.get_event_generator(ua.NodeId(ua.ObjectIds.BaseEventType))
00213         check_eventgenerator_BaseEvent(self, evgen)
00214         check_eventgenerator_SourceServer(self, evgen)
00215 
00216     def test_eventgenerator_BaseEvent_ObjectIds(self):
00217         evgen = self.opc.get_event_generator(ua.ObjectIds.BaseEventType)
00218         check_eventgenerator_BaseEvent(self, evgen)
00219         check_eventgenerator_SourceServer(self, evgen)
00220 
00221     def test_eventgenerator_BaseEvent_Identifier(self):
00222         evgen = self.opc.get_event_generator(2041)
00223         check_eventgenerator_BaseEvent(self, evgen)
00224         check_eventgenerator_SourceServer(self, evgen)
00225 
00226     def test_eventgenerator_sourceServer_Node(self):
00227         evgen = self.opc.get_event_generator(source=opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.Server)))
00228         check_eventgenerator_BaseEvent(self, evgen)
00229         check_eventgenerator_SourceServer(self, evgen)
00230 
00231     def test_eventgenerator_sourceServer_NodeId(self):
00232         evgen = self.opc.get_event_generator(source=ua.NodeId(ua.ObjectIds.Server))
00233         check_eventgenerator_BaseEvent(self, evgen)
00234         check_eventgenerator_SourceServer(self, evgen)
00235 
00236     def test_eventgenerator_sourceServer_ObjectIds(self):
00237         evgen = self.opc.get_event_generator(source=ua.ObjectIds.Server)
00238         check_eventgenerator_BaseEvent(self, evgen)
00239         check_eventgenerator_SourceServer(self, evgen)
00240 
00241     def test_eventgenerator_sourceMyObject(self):
00242         objects = self.opc.get_objects_node()
00243         o = objects.add_object(3, 'MyObject')
00244         evgen = self.opc.get_event_generator(source=o)
00245         check_eventgenerator_BaseEvent(self, evgen)
00246         check_event_generator_object(self, evgen, o)
00247 
00248     def test_eventgenerator_source_collision(self):
00249         objects = self.opc.get_objects_node()
00250         o = objects.add_object(3, 'MyObject')
00251         event = BaseEvent(sourcenode=o.nodeid)
00252         evgen = self.opc.get_event_generator(event, ua.ObjectIds.Server)
00253         check_eventgenerator_BaseEvent(self, evgen)
00254         check_event_generator_object(self, evgen, o)
00255 
00256     def test_eventgenerator_InheritedEvent(self):
00257         evgen = self.opc.get_event_generator(ua.ObjectIds.AuditEventType)
00258         check_eventgenerator_SourceServer(self, evgen)
00259 
00260         ev = evgen.event
00261         self.assertIsNot(ev, None)  # we did not receive event
00262         self.assertIsInstance(ev, BaseEvent)
00263         self.assertIsInstance(ev, AuditEvent)
00264         self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.AuditEventType))
00265         self.assertEqual(ev.Severity, 1)
00266         self.assertEqual(ev.ActionTimeStamp, None)
00267         self.assertEqual(ev.Status, False)
00268         self.assertEqual(ev.ServerId, None)
00269         self.assertEqual(ev.ClientAuditEntryId, None)
00270         self.assertEqual(ev.ClientUserId, None)
00271 
00272     # For the custom events all posibilites are tested. For other custom types only one test case is done since they are using the same code
00273     def test_create_custom_data_type_ObjectId(self):
00274         type = self.opc.create_custom_data_type(2, 'MyDataType', ua.ObjectIds.BaseDataType, [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00275         check_custom_type(self, type, ua.ObjectIds.BaseDataType)
00276 
00277     def test_create_custom_event_type_ObjectId(self):
00278         type = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00279         check_custom_type(self, type, ua.ObjectIds.BaseEventType)
00280 
00281     def test_create_custom_object_type_ObjectId(self):
00282         def func(parent, variant):
00283             return [ua.Variant(ret, ua.VariantType.Boolean)]
00284 
00285         properties = [('PropertyNum', ua.VariantType.Int32),
00286                       ('PropertyString', ua.VariantType.String)]
00287         variables = [('VariableString', ua.VariantType.String),
00288                      ('MyEnumVar', ua.VariantType.Int32, ua.NodeId(ua.ObjectIds.ApplicationType))]
00289         methods = [('MyMethod', func, [ua.VariantType.Int64], [ua.VariantType.Boolean])]
00290 
00291         node_type = self.opc.create_custom_object_type(2, 'MyObjectType', ua.ObjectIds.BaseObjectType, properties, variables, methods)
00292 
00293         check_custom_type(self, node_type, ua.ObjectIds.BaseObjectType)
00294         variables = node_type.get_variables()
00295         self.assertTrue(node_type.get_child("2:VariableString") in variables)
00296         self.assertEqual(node_type.get_child("2:VariableString").get_data_value().Value.VariantType, ua.VariantType.String)
00297         self.assertTrue(node_type.get_child("2:MyEnumVar") in variables)
00298         self.assertEqual(node_type.get_child("2:MyEnumVar").get_data_value().Value.VariantType, ua.VariantType.Int32)
00299         self.assertEqual(node_type.get_child("2:MyEnumVar").get_data_type(), ua.NodeId(ua.ObjectIds.ApplicationType))
00300         methods = node_type.get_methods()
00301         self.assertTrue(node_type.get_child("2:MyMethod") in methods)
00302 
00303     # def test_create_custom_refrence_type_ObjectId(self):
00304         # type = self.opc.create_custom_reference_type(2, 'MyEvent', ua.ObjectIds.Base, [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00305         # check_custom_type(self, type, ua.ObjectIds.BaseObjectType)
00306 
00307     def test_create_custom_variable_type_ObjectId(self):
00308         type = self.opc.create_custom_variable_type(2, 'MyVariableType', ua.ObjectIds.BaseVariableType, [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00309         check_custom_type(self, type, ua.ObjectIds.BaseVariableType)
00310 
00311     def test_create_custom_event_type_NodeId(self):
00312         etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.NodeId(ua.ObjectIds.BaseEventType), [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00313         check_custom_type(self, etype, ua.ObjectIds.BaseEventType)
00314 
00315     def test_create_custom_event_type_Node(self):
00316         etype = self.opc.create_custom_event_type(2, 'MyEvent', opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)), [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00317         check_custom_type(self, etype, ua.ObjectIds.BaseEventType)
00318 
00319     def test_get_event_from_type_node_CustomEvent(self):
00320         etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00321 
00322         ev = opcua.common.events.get_event_obj_from_type_node(etype)
00323         check_custom_event(self, ev, etype)
00324         self.assertEqual(ev.PropertyNum, 0)
00325         self.assertEqual(ev.PropertyString, None)
00326 
00327     def test_eventgenerator_customEvent(self):
00328         etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00329 
00330         evgen = self.opc.get_event_generator(etype, ua.ObjectIds.Server)
00331         check_eventgenerator_CustomEvent(self, evgen, etype)
00332         check_eventgenerator_SourceServer(self, evgen)
00333 
00334         self.assertEqual(evgen.event.PropertyNum, 0)
00335         self.assertEqual(evgen.event.PropertyString, None)
00336 
00337     def test_eventgenerator_double_customEvent(self):
00338         event1 = self.opc.create_custom_event_type(3, 'MyEvent1', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00339 
00340         event2 = self.opc.create_custom_event_type(4, 'MyEvent2', event1, [('PropertyBool', ua.VariantType.Boolean), ('PropertyInt', ua.VariantType.Int32)])
00341 
00342         evgen = self.opc.get_event_generator(event2, ua.ObjectIds.Server)
00343         check_eventgenerator_CustomEvent(self, evgen, event2)
00344         check_eventgenerator_SourceServer(self, evgen)
00345 
00346         # Properties from MyEvent1
00347         self.assertEqual(evgen.event.PropertyNum, 0)
00348         self.assertEqual(evgen.event.PropertyString, None)
00349 
00350          # Properties from MyEvent2
00351         self.assertEqual(evgen.event.PropertyBool, False)
00352         self.assertEqual(evgen.event.PropertyInt, 0)
00353 
00354     def test_eventgenerator_customEvent_MyObject(self):
00355         objects = self.opc.get_objects_node()
00356         o = objects.add_object(3, 'MyObject')
00357         etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Int32), ('PropertyString', ua.VariantType.String)])
00358 
00359         evgen = self.opc.get_event_generator(etype, o)
00360         check_eventgenerator_CustomEvent(self, evgen, etype)
00361         check_event_generator_object(self, evgen, o)
00362 
00363         self.assertEqual(evgen.event.PropertyNum, 0)
00364         self.assertEqual(evgen.event.PropertyString, None)
00365 
00366     def test_context_manager(self):
00367         """ Context manager calls start() and stop()
00368         """
00369         state = [0]
00370         def increment_state(self, *args, **kwargs):
00371             state[0] += 1
00372 
00373         # create server and replace instance methods with dummy methods
00374         server = Server()
00375         server.start = increment_state.__get__(server)
00376         server.stop = increment_state.__get__(server)
00377 
00378         assert state[0] == 0
00379         with server:
00380             # test if server started
00381             self.assertEqual(state[0], 1)
00382         # test if server stopped
00383         self.assertEqual(state[0], 2)
00384 
00385     def test_get_node_by_ns(self):
00386 
00387         def get_ns_of_nodes(nodes):
00388             ns_list = set()
00389             for node in nodes:
00390                 ns_list.add(node.nodeid.NamespaceIndex)
00391             return ns_list
00392 
00393         # incase other testss created nodes  in unregistered namespace
00394         _idx_d = self.opc.register_namespace('dummy1')
00395         _idx_d = self.opc.register_namespace('dummy2')
00396         _idx_d = self.opc.register_namespace('dummy3')
00397 
00398         # create the test namespaces and vars
00399         idx_a = self.opc.register_namespace('a')
00400         idx_b = self.opc.register_namespace('b')
00401         idx_c = self.opc.register_namespace('c')
00402         o = self.opc.get_objects_node()
00403         _myvar2 = o.add_variable(idx_a, "MyBoolVar2", True)
00404         _myvar3 = o.add_variable(idx_b, "MyBoolVar3", True)
00405         _myvar4 = o.add_variable(idx_c, "MyBoolVar4", True)
00406 
00407         # the tests
00408         nodes = ua_utils.get_nodes_of_namespace(self.opc, namespaces=[idx_a, idx_b, idx_c])
00409         self.assertEqual(len(nodes), 3)
00410         self.assertEqual(get_ns_of_nodes(nodes), set([idx_a, idx_b, idx_c]))
00411 
00412         nodes = ua_utils.get_nodes_of_namespace(self.opc, namespaces=[idx_a])
00413         self.assertEqual(len(nodes), 1)
00414         self.assertEqual(get_ns_of_nodes(nodes), set([idx_a]))
00415 
00416         nodes = ua_utils.get_nodes_of_namespace(self.opc, namespaces=[idx_b])
00417         self.assertEqual(len(nodes), 1)
00418         self.assertEqual(get_ns_of_nodes(nodes), set([idx_b]))
00419 
00420         nodes = ua_utils.get_nodes_of_namespace(self.opc, namespaces=['a'])
00421         self.assertEqual(len(nodes), 1)
00422         self.assertEqual(get_ns_of_nodes(nodes), set([idx_a]))
00423 
00424         nodes = ua_utils.get_nodes_of_namespace(self.opc, namespaces=['a', 'c'])
00425         self.assertEqual(len(nodes), 2)
00426         self.assertEqual(get_ns_of_nodes(nodes), set([idx_a, idx_c]))
00427 
00428         nodes = ua_utils.get_nodes_of_namespace(self.opc, namespaces='b')
00429         self.assertEqual(len(nodes), 1)
00430         self.assertEqual(get_ns_of_nodes(nodes), set([idx_b]))
00431 
00432         nodes = ua_utils.get_nodes_of_namespace(self.opc, namespaces=idx_b)
00433         self.assertEqual(len(nodes), 1)
00434         self.assertEqual(get_ns_of_nodes(nodes), set([idx_b]))
00435 
00436         self.assertRaises(ValueError, ua_utils.get_nodes_of_namespace, self.opc, namespaces='non_existing_ns')
00437 
00438 def check_eventgenerator_SourceServer(test, evgen):
00439     server = test.opc.get_server_node()
00440     test.assertEqual(evgen.event.SourceName, server.get_browse_name().Name)
00441     test.assertEqual(evgen.event.SourceNode, ua.NodeId(ua.ObjectIds.Server))
00442 
00443     test.assertEqual(
00444         server.get_event_notifier(),
00445         {ua.EventNotifier.SubscribeToEvents, ua.EventNotifier.HistoryRead}
00446     )
00447 
00448     refs = server.get_referenced_nodes(ua.ObjectIds.GeneratesEvent, ua.BrowseDirection.Forward, ua.NodeClass.ObjectType, False)
00449     test.assertGreaterEqual(len(refs), 1)
00450 
00451 
00452 def check_event_generator_object(test, evgen, obj):
00453     test.assertEqual(evgen.event.SourceName, obj.get_browse_name().Name)
00454     test.assertEqual(evgen.event.SourceNode, obj.nodeid)
00455 
00456     test.assertEqual(
00457         obj.get_event_notifier(),
00458         {ua.EventNotifier.SubscribeToEvents, ua.EventNotifier.HistoryRead}
00459     )
00460 
00461     refs = obj.get_referenced_nodes(ua.ObjectIds.GeneratesEvent, ua.BrowseDirection.Forward, ua.NodeClass.ObjectType, False)
00462     test.assertEqual(len(refs), 1)
00463     test.assertEqual(refs[0].nodeid, evgen.event.EventType)
00464 
00465 
00466 def check_eventgenerator_BaseEvent(test, evgen):
00467     test.assertIsNot(evgen, None)  # we did not receive event generator
00468     test.assertIs(evgen.isession, test.opc.iserver.isession)
00469     check_base_event(test, evgen.event)
00470 
00471 
00472 def check_base_event(test, ev):
00473     test.assertIsNot(ev, None)  # we did not receive event
00474     test.assertIsInstance(ev, BaseEvent)
00475     test.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.BaseEventType))
00476     test.assertEqual(ev.Severity, 1)
00477 
00478 
00479 def check_eventgenerator_CustomEvent(test, evgen, etype):
00480     test.assertIsNot(evgen, None)  # we did not receive event generator
00481     test.assertIs(evgen.isession, test.opc.iserver.isession)
00482     check_custom_event(test, evgen.event, etype)
00483 
00484 
00485 def check_custom_event(test, ev, etype):
00486     test.assertIsNot(ev, None)  # we did not receive event
00487     test.assertIsInstance(ev, BaseEvent)
00488     test.assertEqual(ev.EventType, etype.nodeid)
00489     test.assertEqual(ev.Severity, 1)
00490 
00491 
00492 def check_custom_type(test, type, base_type):
00493     base = opcua.Node(test.opc.iserver.isession, ua.NodeId(base_type))
00494     test.assertTrue(type in base.get_children())
00495     nodes = type.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True)
00496     test.assertEqual(base, nodes[0])
00497     properties = type.get_properties()
00498     test.assertIsNot(properties, None)
00499     test.assertEqual(len(properties), 2)
00500     test.assertTrue(type.get_child("2:PropertyNum") in properties)
00501     test.assertEqual(type.get_child("2:PropertyNum").get_data_value().Value.VariantType, ua.VariantType.Int32)
00502     test.assertTrue(type.get_child("2:PropertyString") in properties)
00503     test.assertEqual(type.get_child("2:PropertyString").get_data_value().Value.VariantType, ua.VariantType.String)
00504 
00505 
00506 class TestServerCaching(unittest.TestCase):
00507     def runTest(self):
00508         return # FIXME broken
00509         tmpfile = NamedTemporaryFile()
00510         path = tmpfile.name
00511         tmpfile.close()
00512 
00513         # create cache file
00514         server = Server(shelffile=path)
00515 
00516         # modify cache content
00517         id = ua.NodeId(ua.ObjectIds.Server_ServerStatus_SecondsTillShutdown)
00518         s = shelve.open(path, "w", writeback=True)
00519         s[id.to_string()].attributes[ua.AttributeIds.Value].value = ua.DataValue(123)
00520         s.close()
00521 
00522         # ensure that we are actually loading from the cache
00523         server = Server(shelffile=path)
00524         self.assertEqual(server.get_node(id).get_value(), 123)
00525 
00526         os.remove(path)