TestServer.test_server_method() - Code Metrics - Inspection of "Merge pull request #185 from FreeOpcUa/metrics" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — master (fc7bd2)

by Olivier

created 2016-05-15 07:35 UTC

TestServer.test_server_method() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	2

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	8
ccs	8
cts	8
cp	1
crap	2
rs	9.4285

import unittest
from tests_common import CommonTests, add_server_methods, MySubHandler
import time
from datetime import timedelta, datetime

import opcua
from opcua import Server
from opcua import Client
from opcua import ua
from opcua import uamethod


port_num = 485140
port_discovery = 48550


class TestServer(unittest.TestCase, CommonTests):

    '''
    Run common tests on server side
    Tests that can only be run on server side must be defined here
    '''
    @classmethod
    def setUpClass(self):
        self.srv = Server()
        self.srv.set_endpoint('opc.tcp://localhost:%d' % port_num)
        add_server_methods(self.srv)
        self.srv.start()
        self.opc = self.srv
        self.discovery = Server()
        self.discovery.set_application_uri("urn:freeopcua:python:discovery")
        self.discovery.set_endpoint('opc.tcp://localhost:%d' % port_discovery)
        self.discovery.start()

    @classmethod
    def tearDownClass(self):
        self.srv.stop()
        self.discovery.stop()

    def test_discovery(self):
        client = Client(self.discovery.endpoint.geturl())
        client.connect()
        try:
            servers = client.find_servers()
            new_app_uri = "urn:freeopcua:python:server:test_discovery"
            self.srv.application_uri = new_app_uri
            self.srv.register_to_discovery(self.discovery.endpoint.geturl(), 0)
            time.sleep(0.1) # let server register registration
            new_servers = client.find_servers()
            self.assertEqual(len(new_servers) - len(servers) , 1)
            self.assertFalse(new_app_uri in [s.ApplicationUri for s in servers])
            self.assertTrue(new_app_uri in [s.ApplicationUri for s in new_servers])
        finally:
            client.disconnect()

    def test_find_servers2(self):
        client = Client(self.discovery.endpoint.geturl())
        client.connect()
        try:
            servers = client.find_servers()
            new_app_uri1 = "urn:freeopcua:python:server:test_discovery1"
            self.srv.application_uri = new_app_uri1
            self.srv.register_to_discovery(self.discovery.endpoint.geturl(), period=0)
            new_app_uri2 = "urn:freeopcua:python:test_discovery2"
            self.srv.application_uri = new_app_uri2
            self.srv.register_to_discovery(self.discovery.endpoint.geturl(), period=0)
            time.sleep(0.1) # let server register registration
            new_servers = client.find_servers()
            self.assertEqual(len(new_servers) - len(servers) , 2)
            self.assertFalse(new_app_uri1 in [s.ApplicationUri for s in servers])
            self.assertFalse(new_app_uri2 in [s.ApplicationUri for s in servers])
            self.assertTrue(new_app_uri1 in [s.ApplicationUri for s in new_servers])
            self.assertTrue(new_app_uri2 in [s.ApplicationUri for s in new_servers])
            # now do a query with filer
            new_servers = client.find_servers(["urn:freeopcua:python:server"])
            self.assertEqual(len(new_servers) - len(servers) , 0)
            self.assertTrue(new_app_uri1 in [s.ApplicationUri for s in new_servers])
            self.assertFalse(new_app_uri2 in [s.ApplicationUri for s in new_servers])
            # now do a query with filer
            new_servers = client.find_servers(["urn:freeopcua:python"])
            self.assertEqual(len(new_servers) - len(servers) , 2)
            self.assertTrue(new_app_uri1 in [s.ApplicationUri for s in new_servers])
            self.assertTrue(new_app_uri2 in [s.ApplicationUri for s in new_servers])
        finally:
            client.disconnect()


    """
    # not sure if this test is necessary, and there is a lot repetition with previous test
    def test_discovery_server_side(self):
        servers = self.discovery.find_servers()
        self.assertEqual(len(servers), 1)
        self.srv.register_to_discovery(self.discovery.endpoint.geturl(), 1)
        time.sleep(1) # let server register registration
        servers = self.discovery.find_servers()
        print("SERVERS 2", servers)
        self.assertEqual(len(servers), 2)
    """
    #def test_register_server2(self):
        #servers = self.opc.register_server()

    def test_register_namespace(self):
        uri = 'http://mycustom.Namespace.com'
        idx1 = self.opc.register_namespace(uri)
        idx2 = self.opc.get_namespace_index(uri)
        self.assertEqual(idx1, idx2)

    def test_register_use_namespace(self):
        uri = 'http://my_very_custom.Namespace.com'
        idx = self.opc.register_namespace(uri)
        root = self.opc.get_root_node()
        myvar = root.add_variable(idx, 'var_in_custom_namespace', [5])
        myid = myvar.nodeid
        self.assertEqual(idx, myid.NamespaceIndex)

    def test_server_method(self):
        def func(parent, variant):
            variant.Value *= 2
            return [variant]
        o = self.opc.get_objects_node()
        v = o.add_method(3, 'Method1', func, [ua.VariantType.Int64], [ua.VariantType.Int64])
        result = o.call_method(v, ua.Variant(2.1))
        self.assertEqual(result, 4.2)

    def test_xml_import(self):
        self.srv.import_xml("tests/custom_nodes.xml")
        o = self.opc.get_objects_node()
        v = o.get_child(["MyXMLFolder", "MyXMLObject", "MyXMLVariable"])
        val = v.get_value()
        self.assertEqual(val, "StringValue")

    def test_historize_variable(self):
        o = self.opc.get_objects_node()
        var = o.add_variable(3, "test_hist", 1.0)
        self.srv.iserver.enable_history_data_change(var, timedelta(days=1))
        time.sleep(1)
        var.set_value(2.0)
        var.set_value(3.0)
        self.srv.iserver.disable_history_data_change(var)

    def test_historize_events(self):
        srv_node = self.srv.get_node(ua.ObjectIds.Server)
        srvevgen = self.srv.get_event_generator()
        self.srv.iserver.enable_history_event(srv_node, period=None)
        srvevgen.trigger(message="Message")
        self.srv.iserver.disable_history_event(srv_node)

    def test_references_for_added_nodes_method(self):
        objects = self.opc.get_objects_node()
        o = objects.add_object(3, 'MyObject')
        nodes = objects.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Forward, includesubtypes=False)
        self.assertTrue(o in nodes)
        nodes = o.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
        self.assertTrue(objects in nodes)
        self.assertEqual(o.get_parent(), objects)
        self.assertEqual(o.get_type_definition(), ua.ObjectIds.BaseObjectType)

        @uamethod
        def callback(parent):
            return

        m = o.add_method(3, 'MyMethod', callback)
        nodes = o.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Forward, includesubtypes=False)
        self.assertTrue(m in nodes)
        nodes = m.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
        self.assertTrue(o in nodes)
        self.assertEqual(m.get_parent(), o)


    # This should work for following BaseEvent tests to work (maybe to write it a bit differentlly since they are not independent)
    def test_get_event_from_type_node_BaseEvent(self):
        ev = opcua.server.event.get_event_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)))
        check_base_event(self, ev)

    def test_get_event_from_type_node_Inhereted_AuditEvent(self):
        ev = opcua.server.event.get_event_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.AuditEventType)))
        self.assertIsNot(ev, None)  # we did not receive event
        self.assertIsInstance(ev, ua.BaseEvent)
        self.assertIsInstance(ev, ua.AuditEvent)
        self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.AuditEventType))
        self.assertEqual(ev.Severity, ua.Variant(1, ua.VariantType.UInt16))
        self.assertEqual(ev.ActionTimeStamp, None)
        self.assertEqual(ev.Status, False)
        self.assertEqual(ev.ServerId, None)
        self.assertEqual(ev.ClientAuditEntryId, None)
        self.assertEqual(ev.ClientUserId, None)
        self.assertEqual(ev._freeze, True)

    def test_eventgenerator_default(self):
        evgen = self.opc.get_event_generator()
        check_eventgenerator_BaseEvent(self, evgen)
        check_eventgenerator_SourceServer(self, evgen)

    def test_eventgenerator_BaseEvent_object(self):
        evgen = self.opc.get_event_generator(ua.BaseEvent())
        check_eventgenerator_BaseEvent(self, evgen)
        check_eventgenerator_SourceServer(self, evgen)

    def test_eventgenerator_BaseEvent_Node(self):
        evgen = self.opc.get_event_generator(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)))
        check_eventgenerator_BaseEvent(self, evgen)
        check_eventgenerator_SourceServer(self, evgen)

    def test_eventgenerator_BaseEvent_NodeId(self):
        evgen = self.opc.get_event_generator(ua.NodeId(ua.ObjectIds.BaseEventType))
        check_eventgenerator_BaseEvent(self, evgen)
        check_eventgenerator_SourceServer(self, evgen)

    def test_eventgenerator_BaseEvent_ObjectIds(self):
        evgen = self.opc.get_event_generator(ua.ObjectIds.BaseEventType)
        check_eventgenerator_BaseEvent(self, evgen)
        check_eventgenerator_SourceServer(self, evgen)

    def test_eventgenerator_BaseEvent_Identifier(self):
        evgen = self.opc.get_event_generator(2041)
        check_eventgenerator_BaseEvent(self, evgen)
        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)))
        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))
        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)
        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)
        check_eventgenerator_BaseEvent(self, evgen)
        check_event_generator_object(self, evgen, o)

    def test_eventgenerator_source_collision(self):
        objects = self.opc.get_objects_node()

        o = objects.add_object(3, 'MyObject')
        event = ua.BaseEvent(sourcenode=o.nodeid)
        evgen = self.opc.get_event_generator(event, ua.ObjectIds.Server)
        check_eventgenerator_BaseEvent(self, evgen)
        check_event_generator_object(self, evgen, o)

    def test_eventgenerator_InheritedEvent(self):
        evgen = self.opc.get_event_generator(ua.ObjectIds.AuditEventType)
        check_eventgenerator_SourceServer(self, evgen)

        ev = evgen.event
        self.assertIsNot(ev, None)  # we did not receive event
        self.assertIsInstance(ev, ua.BaseEvent)
        self.assertIsInstance(ev, ua.AuditEvent)
        self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.AuditEventType))
        self.assertEqual(ev.Severity, ua.Variant(1, ua.VariantType.UInt16))
        self.assertEqual(ev.ActionTimeStamp, None)
        self.assertEqual(ev.Status, False)
        self.assertEqual(ev.ServerId, None)
        self.assertEqual(ev.ClientAuditEntryId, None)
        self.assertEqual(ev.ClientUserId, None)
        self.assertEqual(ev._freeze, True)

    def test_create_custom_event_type_ObjectId(self):
        etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
        check_custom_event_type(self, etype)

    def test_create_custom_event_type_NodeId(self):
        etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.NodeId(ua.ObjectIds.BaseEventType), [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
        check_custom_event_type(self, etype)

    def test_create_custom_event_type_Node(self):
        etype = self.opc.create_custom_event_type(2, 'MyEvent', opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)), [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
        check_custom_event_type(self, etype)

    def test_get_event_from_type_node_CustomEvent(self):
        etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])


        ev = opcua.server.event.get_event_from_type_node(etype)
        check_custom_event(self, ev, etype)
        self.assertEqual(ev.PropertyNum, None)
        self.assertEqual(ev.PropertyString, None)

    def test_eventgenerator_customEvent(self):
        etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])

        evgen = self.opc.get_event_generator(etype, ua.ObjectIds.Server)
        check_eventgenerator_CustomEvent(self, evgen, etype)
        check_eventgenerator_SourceServer(self, evgen)

        self.assertEqual(evgen.event.PropertyNum, None)
        self.assertEqual(evgen.event.PropertyString, None)

    def test_eventgenerator_double_customEvent(self):
        event1 = self.opc.create_custom_event_type(3, 'MyEvent1', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])

        event2 = self.opc.create_custom_event_type(4, 'MyEvent2', event1, [('PropertyBool', ua.VariantType.Boolean), ('PropertyInt', ua.VariantType.Int32)])

        evgen = self.opc.get_event_generator(event2, ua.ObjectIds.Server)
        check_eventgenerator_CustomEvent(self, evgen, event2)
        check_eventgenerator_SourceServer(self, evgen)

        # Properties from MyEvent1
        self.assertEqual(evgen.event.PropertyNum, None)

        self.assertEqual(evgen.event.PropertyString, None)

         # Properties from MyEvent2
        self.assertEqual(evgen.event.PropertyBool, None)
        self.assertEqual(evgen.event.PropertyInt, None)

    def test_eventgenerator_customEvent_MyObject(self):
        objects = self.opc.get_objects_node()
        o = objects.add_object(3, 'MyObject')
        etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])

        evgen = self.opc.get_event_generator(etype, o)
        check_eventgenerator_CustomEvent(self, evgen, etype)
        check_event_generator_object(self, evgen, o)

        self.assertEqual(evgen.event.PropertyNum, None)
        self.assertEqual(evgen.event.PropertyString, None)


def check_eventgenerator_SourceServer(test, evgen):

    server = test.opc.get_server_node()
    test.assertEqual(evgen.event.SourceName, server.get_display_name().Text)
    test.assertEqual(evgen.event.SourceNode, ua.NodeId(ua.ObjectIds.Server))
    test.assertEqual(server.get_attribute(ua.AttributeIds.EventNotifier).Value, ua.Variant(1, ua.VariantType.Byte))
    refs = server.get_referenced_nodes(ua.ObjectIds.GeneratesEvent, ua.BrowseDirection.Forward, ua.NodeClass.ObjectType, False)
    test.assertGreaterEqual(len(refs), 1)


def check_event_generator_object(test, evgen, obj):

    test.assertEqual(evgen.event.SourceName, obj.get_display_name().Text)
    test.assertEqual(evgen.event.SourceNode, obj.nodeid)
    test.assertEqual(obj.get_attribute(ua.AttributeIds.EventNotifier).Value, ua.Variant(1, ua.VariantType.Byte))
    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)


def check_eventgenerator_BaseEvent(test, evgen):
    test.assertIsNot(evgen, None)  # we did not receive event generator
    test.assertIs(evgen.isession, test.opc.iserver.isession)
    check_base_event(test, evgen.event)


def check_base_event(test, ev):
    test.assertIsNot(ev, None)  # we did not receive event
    test.assertIsInstance(ev, ua.BaseEvent)
    test.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.BaseEventType))
    test.assertEqual(ev.Severity, ua.Variant(1, ua.VariantType.UInt16))
    test.assertEqual(ev._freeze, True)


def check_eventgenerator_CustomEvent(test, evgen, etype):
    test.assertIsNot(evgen, None)  # we did not receive event generator
    test.assertIs(evgen.isession, test.opc.iserver.isession)
    check_custom_event(test, evgen.event, etype)


def check_custom_event(test, ev, etype):
    test.assertIsNot(ev, None)  # we did not receive event
    test.assertIsInstance(ev, ua.BaseEvent)
    test.assertEqual(ev.EventType, etype.nodeid)
    test.assertEqual(ev.Severity, ua.Variant(1, ua.VariantType.UInt16))
    test.assertEqual(ev._freeze, True)


def check_custom_event_type(test, ev):
    base = opcua.Node(test.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType))
    test.assertTrue(ev in base.get_children())
    nodes = ev.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
    test.assertEqual(base, nodes[0])
    properties = ev.get_properties()
    test.assertIsNot(properties, None)
    test.assertEqual(len(properties), 2)
    test.assertTrue(ev.get_child("2:PropertyNum") in properties)
    test.assertEqual(ev.get_child("2:PropertyNum").get_data_value().Value.VariantType, ua.VariantType.Float)
    test.assertTrue(ev.get_child("2:PropertyString") in properties)
    test.assertEqual(ev.get_child("2:PropertyString").get_data_value().Value.VariantType, ua.VariantType.String)


1		import unittest
2		from tests_common import CommonTests, add_server_methods, MySubHandler
3		import time
4		from datetime import timedelta, datetime
5
6		import opcua
7		from opcua import Server
8		from opcua import Client
9		from opcua import ua
10		from opcua import uamethod
11
12
13		port_num = 485140
14		port_discovery = 48550
15
16
17		class TestServer(unittest.TestCase, CommonTests):
18
19		'''
20		Run common tests on server side
21		Tests that can only be run on server side must be defined here
22		'''
23		@classmethod
24		def setUpClass(self):
25		self.srv = Server()
26		self.srv.set_endpoint('opc.tcp://localhost:%d' % port_num)
27		add_server_methods(self.srv)
28		self.srv.start()
29		self.opc = self.srv
30		self.discovery = Server()
31		self.discovery.set_application_uri("urn:freeopcua:python:discovery")
32		self.discovery.set_endpoint('opc.tcp://localhost:%d' % port_discovery)
33		self.discovery.start()
34
35		@classmethod
36		def tearDownClass(self):
37		self.srv.stop()
38		self.discovery.stop()
39
40		def test_discovery(self):
41		client = Client(self.discovery.endpoint.geturl())
42		client.connect()
43		try:
44		servers = client.find_servers()
45		new_app_uri = "urn:freeopcua:python:server:test_discovery"
46		self.srv.application_uri = new_app_uri
47		self.srv.register_to_discovery(self.discovery.endpoint.geturl(), 0)
48		time.sleep(0.1) # let server register registration
49		new_servers = client.find_servers()
50		self.assertEqual(len(new_servers) - len(servers) , 1)
51		self.assertFalse(new_app_uri in [s.ApplicationUri for s in servers])
52		self.assertTrue(new_app_uri in [s.ApplicationUri for s in new_servers])
53		finally:
54		client.disconnect()
55
56		def test_find_servers2(self):
57		client = Client(self.discovery.endpoint.geturl())
58		client.connect()
59		try:
60		servers = client.find_servers()
61		new_app_uri1 = "urn:freeopcua:python:server:test_discovery1"
62		self.srv.application_uri = new_app_uri1
63		self.srv.register_to_discovery(self.discovery.endpoint.geturl(), period=0)
64		new_app_uri2 = "urn:freeopcua:python:test_discovery2"
65		self.srv.application_uri = new_app_uri2
66		self.srv.register_to_discovery(self.discovery.endpoint.geturl(), period=0)
67		time.sleep(0.1) # let server register registration
68		new_servers = client.find_servers()
69		self.assertEqual(len(new_servers) - len(servers) , 2)
70		self.assertFalse(new_app_uri1 in [s.ApplicationUri for s in servers])
71		self.assertFalse(new_app_uri2 in [s.ApplicationUri for s in servers])
72		self.assertTrue(new_app_uri1 in [s.ApplicationUri for s in new_servers])
73		self.assertTrue(new_app_uri2 in [s.ApplicationUri for s in new_servers])
74		# now do a query with filer
75		new_servers = client.find_servers(["urn:freeopcua:python:server"])
76		self.assertEqual(len(new_servers) - len(servers) , 0)
77		self.assertTrue(new_app_uri1 in [s.ApplicationUri for s in new_servers])
78		self.assertFalse(new_app_uri2 in [s.ApplicationUri for s in new_servers])
79		# now do a query with filer
80		new_servers = client.find_servers(["urn:freeopcua:python"])
81		self.assertEqual(len(new_servers) - len(servers) , 2)
82		self.assertTrue(new_app_uri1 in [s.ApplicationUri for s in new_servers])
83		self.assertTrue(new_app_uri2 in [s.ApplicationUri for s in new_servers])
84		finally:
85		client.disconnect()
86
87
88		"""
89		# not sure if this test is necessary, and there is a lot repetition with previous test
90		def test_discovery_server_side(self):
91		servers = self.discovery.find_servers()
92		self.assertEqual(len(servers), 1)
93		self.srv.register_to_discovery(self.discovery.endpoint.geturl(), 1)
94		time.sleep(1) # let server register registration
95		servers = self.discovery.find_servers()
96		print("SERVERS 2", servers)
97		self.assertEqual(len(servers), 2)
98		"""
99		#def test_register_server2(self):
100		#servers = self.opc.register_server()
101
102		def test_register_namespace(self):
103		uri = 'http://mycustom.Namespace.com'
104		idx1 = self.opc.register_namespace(uri)
105		idx2 = self.opc.get_namespace_index(uri)
106		self.assertEqual(idx1, idx2)
107
108		def test_register_use_namespace(self):
109		uri = 'http://my_very_custom.Namespace.com'
110		idx = self.opc.register_namespace(uri)
111		root = self.opc.get_root_node()
112		myvar = root.add_variable(idx, 'var_in_custom_namespace', [5])
113		myid = myvar.nodeid
114		self.assertEqual(idx, myid.NamespaceIndex)
115
116		def test_server_method(self):
117		def func(parent, variant):
118		variant.Value *= 2
119		return [variant]
120		o = self.opc.get_objects_node()
121		v = o.add_method(3, 'Method1', func, [ua.VariantType.Int64], [ua.VariantType.Int64])
122		result = o.call_method(v, ua.Variant(2.1))
123		self.assertEqual(result, 4.2)
124
125		def test_xml_import(self):
126		self.srv.import_xml("tests/custom_nodes.xml")
127		o = self.opc.get_objects_node()
128		v = o.get_child(["MyXMLFolder", "MyXMLObject", "MyXMLVariable"])
129		val = v.get_value()
130		self.assertEqual(val, "StringValue")
131
132		def test_historize_variable(self):
133		o = self.opc.get_objects_node()
134		var = o.add_variable(3, "test_hist", 1.0)
135		self.srv.iserver.enable_history_data_change(var, timedelta(days=1))
136		time.sleep(1)
137		var.set_value(2.0)
138		var.set_value(3.0)
139		self.srv.iserver.disable_history_data_change(var)
140
141		def test_historize_events(self):
142		srv_node = self.srv.get_node(ua.ObjectIds.Server)
143		srvevgen = self.srv.get_event_generator()
144		self.srv.iserver.enable_history_event(srv_node, period=None)
145		srvevgen.trigger(message="Message")
146		self.srv.iserver.disable_history_event(srv_node)
147
148		def test_references_for_added_nodes_method(self):
149		objects = self.opc.get_objects_node()
150		o = objects.add_object(3, 'MyObject')
151		nodes = objects.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Forward, includesubtypes=False)
152		self.assertTrue(o in nodes)
153		nodes = o.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
154		self.assertTrue(objects in nodes)
155		self.assertEqual(o.get_parent(), objects)
156		self.assertEqual(o.get_type_definition(), ua.ObjectIds.BaseObjectType)
157
158		@uamethod
159		def callback(parent):
160		return
161
162		m = o.add_method(3, 'MyMethod', callback)
163		nodes = o.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Forward, includesubtypes=False)
164		self.assertTrue(m in nodes)
165		nodes = m.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
166		self.assertTrue(o in nodes)
167	View Code Duplication	self.assertEqual(m.get_parent(), o)
		0 ignored issues – show Duplication introduced 2016-04-27 12:31 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
168
169		# This should work for following BaseEvent tests to work (maybe to write it a bit differentlly since they are not independent)
170		def test_get_event_from_type_node_BaseEvent(self):
171		ev = opcua.server.event.get_event_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)))
172		check_base_event(self, ev)
173
174		def test_get_event_from_type_node_Inhereted_AuditEvent(self):
175		ev = opcua.server.event.get_event_from_type_node(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.AuditEventType)))
176		self.assertIsNot(ev, None) # we did not receive event
177		self.assertIsInstance(ev, ua.BaseEvent)
178		self.assertIsInstance(ev, ua.AuditEvent)
179		self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.AuditEventType))
180		self.assertEqual(ev.Severity, ua.Variant(1, ua.VariantType.UInt16))
181		self.assertEqual(ev.ActionTimeStamp, None)
182		self.assertEqual(ev.Status, False)
183		self.assertEqual(ev.ServerId, None)
184		self.assertEqual(ev.ClientAuditEntryId, None)
185		self.assertEqual(ev.ClientUserId, None)
186		self.assertEqual(ev._freeze, True)
187
188		def test_eventgenerator_default(self):
189		evgen = self.opc.get_event_generator()
190		check_eventgenerator_BaseEvent(self, evgen)
191		check_eventgenerator_SourceServer(self, evgen)
192
193		def test_eventgenerator_BaseEvent_object(self):
194		evgen = self.opc.get_event_generator(ua.BaseEvent())
195		check_eventgenerator_BaseEvent(self, evgen)
196		check_eventgenerator_SourceServer(self, evgen)
197
198		def test_eventgenerator_BaseEvent_Node(self):
199		evgen = self.opc.get_event_generator(opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)))
200		check_eventgenerator_BaseEvent(self, evgen)
201		check_eventgenerator_SourceServer(self, evgen)
202
203		def test_eventgenerator_BaseEvent_NodeId(self):
204		evgen = self.opc.get_event_generator(ua.NodeId(ua.ObjectIds.BaseEventType))
205		check_eventgenerator_BaseEvent(self, evgen)
206		check_eventgenerator_SourceServer(self, evgen)
207
208		def test_eventgenerator_BaseEvent_ObjectIds(self):
209		evgen = self.opc.get_event_generator(ua.ObjectIds.BaseEventType)
210		check_eventgenerator_BaseEvent(self, evgen)
211		check_eventgenerator_SourceServer(self, evgen)
212
213		def test_eventgenerator_BaseEvent_Identifier(self):
214		evgen = self.opc.get_event_generator(2041)
215		check_eventgenerator_BaseEvent(self, evgen)
216		check_eventgenerator_SourceServer(self, evgen)
217
218		def test_eventgenerator_sourceServer_Node(self):
219		evgen = self.opc.get_event_generator(source=opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.Server)))
220		check_eventgenerator_BaseEvent(self, evgen)
221		check_eventgenerator_SourceServer(self, evgen)
222
223		def test_eventgenerator_sourceServer_NodeId(self):
224		evgen = self.opc.get_event_generator(source=ua.NodeId(ua.ObjectIds.Server))
225		check_eventgenerator_BaseEvent(self, evgen)
226		check_eventgenerator_SourceServer(self, evgen)
227
228		def test_eventgenerator_sourceServer_ObjectIds(self):
229		evgen = self.opc.get_event_generator(source=ua.ObjectIds.Server)
230		check_eventgenerator_BaseEvent(self, evgen)
231		check_eventgenerator_SourceServer(self, evgen)
232
233		def test_eventgenerator_sourceMyObject(self):
234		objects = self.opc.get_objects_node()
235		o = objects.add_object(3, 'MyObject')
236		evgen = self.opc.get_event_generator(source=o)
237		check_eventgenerator_BaseEvent(self, evgen)
238		check_event_generator_object(self, evgen, o)
239
240		def test_eventgenerator_source_collision(self):
241	View Code Duplication	objects = self.opc.get_objects_node()
		0 ignored issues – show Duplication introduced 2016-05-03 15:34 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
242		o = objects.add_object(3, 'MyObject')
243		event = ua.BaseEvent(sourcenode=o.nodeid)
244		evgen = self.opc.get_event_generator(event, ua.ObjectIds.Server)
245		check_eventgenerator_BaseEvent(self, evgen)
246		check_event_generator_object(self, evgen, o)
247
248		def test_eventgenerator_InheritedEvent(self):
249		evgen = self.opc.get_event_generator(ua.ObjectIds.AuditEventType)
250		check_eventgenerator_SourceServer(self, evgen)
251
252		ev = evgen.event
253		self.assertIsNot(ev, None) # we did not receive event
254		self.assertIsInstance(ev, ua.BaseEvent)
255		self.assertIsInstance(ev, ua.AuditEvent)
256		self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.AuditEventType))
257		self.assertEqual(ev.Severity, ua.Variant(1, ua.VariantType.UInt16))
258		self.assertEqual(ev.ActionTimeStamp, None)
259		self.assertEqual(ev.Status, False)
260		self.assertEqual(ev.ServerId, None)
261		self.assertEqual(ev.ClientAuditEntryId, None)
262		self.assertEqual(ev.ClientUserId, None)
263		self.assertEqual(ev._freeze, True)
264
265		def test_create_custom_event_type_ObjectId(self):
266		etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
267		check_custom_event_type(self, etype)
268
269		def test_create_custom_event_type_NodeId(self):
270		etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.NodeId(ua.ObjectIds.BaseEventType), [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
271		check_custom_event_type(self, etype)
272
273		def test_create_custom_event_type_Node(self):
274		etype = self.opc.create_custom_event_type(2, 'MyEvent', opcua.Node(self.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType)), [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
275		check_custom_event_type(self, etype)
276
277		def test_get_event_from_type_node_CustomEvent(self):
278	View Code Duplication	etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
		0 ignored issues – show Duplication introduced 2016-04-29 18:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
279
280		ev = opcua.server.event.get_event_from_type_node(etype)
281		check_custom_event(self, ev, etype)
282		self.assertEqual(ev.PropertyNum, None)
283		self.assertEqual(ev.PropertyString, None)
284
285		def test_eventgenerator_customEvent(self):
286		etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
287
288		evgen = self.opc.get_event_generator(etype, ua.ObjectIds.Server)
289		check_eventgenerator_CustomEvent(self, evgen, etype)
290		check_eventgenerator_SourceServer(self, evgen)
291
292		self.assertEqual(evgen.event.PropertyNum, None)
293		self.assertEqual(evgen.event.PropertyString, None)
294
295		def test_eventgenerator_double_customEvent(self):
296		event1 = self.opc.create_custom_event_type(3, 'MyEvent1', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
297
298		event2 = self.opc.create_custom_event_type(4, 'MyEvent2', event1, [('PropertyBool', ua.VariantType.Boolean), ('PropertyInt', ua.VariantType.Int32)])
299
300		evgen = self.opc.get_event_generator(event2, ua.ObjectIds.Server)
301		check_eventgenerator_CustomEvent(self, evgen, event2)
302		check_eventgenerator_SourceServer(self, evgen)
303
304		# Properties from MyEvent1
305	View Code Duplication	self.assertEqual(evgen.event.PropertyNum, None)
		0 ignored issues – show Duplication introduced 2016-04-29 18:49 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
306		self.assertEqual(evgen.event.PropertyString, None)
307
308		# Properties from MyEvent2
309		self.assertEqual(evgen.event.PropertyBool, None)
310		self.assertEqual(evgen.event.PropertyInt, None)
311
312		def test_eventgenerator_customEvent_MyObject(self):
313		objects = self.opc.get_objects_node()
314		o = objects.add_object(3, 'MyObject')
315		etype = self.opc.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
316
317		evgen = self.opc.get_event_generator(etype, o)
318		check_eventgenerator_CustomEvent(self, evgen, etype)
319		check_event_generator_object(self, evgen, o)
320
321		self.assertEqual(evgen.event.PropertyNum, None)
322		self.assertEqual(evgen.event.PropertyString, None)
323
324
325	View Code Duplication	def check_eventgenerator_SourceServer(test, evgen):
		0 ignored issues – show Duplication introduced 2016-05-10 00:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
326		server = test.opc.get_server_node()
327		test.assertEqual(evgen.event.SourceName, server.get_display_name().Text)
328		test.assertEqual(evgen.event.SourceNode, ua.NodeId(ua.ObjectIds.Server))
329		test.assertEqual(server.get_attribute(ua.AttributeIds.EventNotifier).Value, ua.Variant(1, ua.VariantType.Byte))
330		refs = server.get_referenced_nodes(ua.ObjectIds.GeneratesEvent, ua.BrowseDirection.Forward, ua.NodeClass.ObjectType, False)
331		test.assertGreaterEqual(len(refs), 1)
332
333
334	View Code Duplication	def check_event_generator_object(test, evgen, obj):
		0 ignored issues – show Duplication introduced 2016-05-10 00:52 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
335		test.assertEqual(evgen.event.SourceName, obj.get_display_name().Text)
336		test.assertEqual(evgen.event.SourceNode, obj.nodeid)
337		test.assertEqual(obj.get_attribute(ua.AttributeIds.EventNotifier).Value, ua.Variant(1, ua.VariantType.Byte))
338		refs = obj.get_referenced_nodes(ua.ObjectIds.GeneratesEvent, ua.BrowseDirection.Forward, ua.NodeClass.ObjectType, False)
339		test.assertEqual(len(refs), 1)
340		test.assertEqual(refs[0].nodeid, evgen.event.EventType)
341
342
343		def check_eventgenerator_BaseEvent(test, evgen):
344		test.assertIsNot(evgen, None) # we did not receive event generator
345		test.assertIs(evgen.isession, test.opc.iserver.isession)
346		check_base_event(test, evgen.event)
347
348
349		def check_base_event(test, ev):
350		test.assertIsNot(ev, None) # we did not receive event
351		test.assertIsInstance(ev, ua.BaseEvent)
352		test.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.BaseEventType))
353		test.assertEqual(ev.Severity, ua.Variant(1, ua.VariantType.UInt16))
354		test.assertEqual(ev._freeze, True)
355
356
357		def check_eventgenerator_CustomEvent(test, evgen, etype):
358		test.assertIsNot(evgen, None) # we did not receive event generator
359		test.assertIs(evgen.isession, test.opc.iserver.isession)
360		check_custom_event(test, evgen.event, etype)
361
362
363		def check_custom_event(test, ev, etype):
364		test.assertIsNot(ev, None) # we did not receive event
365		test.assertIsInstance(ev, ua.BaseEvent)
366		test.assertEqual(ev.EventType, etype.nodeid)
367		test.assertEqual(ev.Severity, ua.Variant(1, ua.VariantType.UInt16))
368		test.assertEqual(ev._freeze, True)
369
370
371		def check_custom_event_type(test, ev):
372		base = opcua.Node(test.opc.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType))
373		test.assertTrue(ev in base.get_children())
374		nodes = ev.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=False)
375		test.assertEqual(base, nodes[0])
376		properties = ev.get_properties()
377		test.assertIsNot(properties, None)
378		test.assertEqual(len(properties), 2)
379		test.assertTrue(ev.get_child("2:PropertyNum") in properties)
380		test.assertEqual(ev.get_child("2:PropertyNum").get_data_value().Value.VariantType, ua.VariantType.Float)
381		test.assertTrue(ev.get_child("2:PropertyString") in properties)
382		test.assertEqual(ev.get_child("2:PropertyString").get_data_value().Value.VariantType, ua.VariantType.String)
383

FreeOpcUa / python-opcua

Branch — master (fc7bd2)

TestServer.test_server_method() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like