How to use opcua - 10 common examples

def encrypted_size(self, plain_size):
        size = plain_size + self.security_policy.signature_size()
        pbs = self.security_policy.plain_block_size()
        if size % pbs != 0:
            print("ENC", plain_size, size, pbs)
            raise ua.UaError("Encryption error")
        return size // pbs * self.security_policy.encrypted_block_size()

def setUpClass(self):
        #start server in its own process
        self.srv = ServerProcess()
        self.srv.start()
        self.srv.started.wait() # let it initialize

        #start client
        self.clt = opcua.Client();
        self.clt.set_endpoint("opc.tcp://localhost:4841")
        self.clt.connect()
        self.opc = self.clt

def test_xml_ext_obj(self):
        arg = ua.Argument()
        arg.DataType = ua.NodeId(ua.ObjectIds.Float)
        arg.Description = ua.LocalizedText("Nice description")
        arg.ArrayDimensions = [1, 2, 3]
        arg.Name = "MyArg"

        node = self.opc.nodes.objects.add_variable(2, "xmlexportobj2", arg)
        node2 = self._test_xml_var_type(node, "ext_obj", test_equality=False)
        arg2 = node2.get_value()

        self.assertEqual(arg.Name, arg2.Name)
        self.assertEqual(arg.ArrayDimensions, arg2.ArrayDimensions)
        self.assertEqual(arg.Description, arg2.Description)
        self.assertEqual(arg.DataType, arg2.DataType)

def test_nodeid_string(self):
        nid0 = ua.NodeId(45)
        self.assertEqual(nid0, ua.NodeId.from_string("i=45"))
        self.assertEqual(nid0, ua.NodeId.from_string("ns=0;i=45"))
        nid = ua.NodeId(45, 10)
        self.assertEqual(nid, ua.NodeId.from_string("i=45; ns=10"))
        self.assertNotEqual(nid, ua.NodeId.from_string("i=45; ns=11"))
        self.assertNotEqual(nid, ua.NodeId.from_string("i=5; ns=10"))
        # not sure the next one is correct...
        self.assertEqual(nid, ua.NodeId.from_string("i=45; ns=10; srv=serverid"))

        nid1 = ua.NodeId("myid.mynodeid", 7)
        self.assertEqual(nid1, ua.NodeId.from_string("ns=7; s=myid.mynodeid"))
        #with self.assertRaises(ua.UaError):

def test_nodeid_ordering(self):
        a = ua.NodeId(2000, 1)
        b = ua.NodeId(3000, 1)
        c = ua.NodeId(20, 0)
        d = ua.NodeId("tititu", 1)
        e = ua.NodeId("aaaaa", 1)
        f = ua.NodeId("aaaaa", 2)
        g = ua.NodeId(uuid.uuid4(), 1)
        h = ua.TwoByteNodeId(2001)
        i = ua.NodeId(b"lkjkl", 1, ua.NodeIdType.ByteString)
        j = ua.NodeId(b"aaa", 5, ua.NodeIdType.ByteString)

        mylist = [a, b, c, d, e, f, g, h, i, j]
        mylist.sort()
        expected = [h, c, a, b, e, d, f, g, i, j]
        self.assertEqual(mylist, expected)

def test_reference_generator_2(self):
        id1 = ua.NodeId(1, namespaceidx=2, nodeidtype=ua.NodeIdType.Numeric)
        id2 = ua.NodeId(2, namespaceidx=2, nodeidtype=ua.NodeIdType.Numeric)
        ref = ua.NodeId(ua.ObjectIds.HasEncoding, 0)
        result = reference_generator(id1, id2, ref, False)
        self.assertFalse(result.IsForward)
        self.assertEqual(result.ReferenceTypeId, ref)
        self.assertEqual(result.SourceNodeId, id1)
        self.assertEqual(result.TargetNodeClass, ua.NodeClass.DataType)
        self.assertEqual(result.TargetNodeId, id2)

def test_unicode_string_nodeid(self):
        nid = ua.NodeId('hëllò', 1)
        self.assertEqual(nid.NamespaceIndex, 1)
        self.assertEqual(nid.Identifier, 'hëllò')
        self.assertEqual(nid.NodeIdType, ua.NodeIdType.String)
        d = nid.to_binary()
        new_nid = nid.from_binary(io.BytesIO(d))
        self.assertEqual(new_nid, nid)
        self.assertEqual(new_nid.Identifier, 'hëllò')
        self.assertEqual(new_nid.NodeIdType, ua.NodeIdType.String)

def check_custom_type(test, type, base_type, node_class=None):
    base = opcua.Node(test.opc.iserver.isession, ua.NodeId(base_type))
    test.assertTrue(type in base.get_children())
    nodes = type.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse, includesubtypes=True)
    test.assertEqual(base, nodes[0])
    properties = type.get_properties()

    if node_class:
        test.assertEqual(node_class, type.get_node_class())

    test.assertIsNot(properties, None)
    test.assertEqual(len(properties), 2)
    test.assertTrue(type.get_child("2:PropertyNum") in properties)
    test.assertEqual(type.get_child("2:PropertyNum").get_data_value().Value.VariantType, ua.VariantType.Int32)
    test.assertTrue(type.get_child("2:PropertyString") in properties)
    test.assertEqual(type.get_child("2:PropertyString").get_data_value().Value.VariantType, ua.VariantType.String)

def test_data_type_to_variant_type(self):
        test_data = {
            ua.ObjectIds.Boolean: ua.VariantType.Boolean,
            ua.ObjectIds.Byte: ua.VariantType.Byte,
            ua.ObjectIds.String: ua.VariantType.String,
            ua.ObjectIds.Int32: ua.VariantType.Int32,
            ua.ObjectIds.UInt32: ua.VariantType.UInt32,
            ua.ObjectIds.NodeId: ua.VariantType.NodeId,
            ua.ObjectIds.LocalizedText: ua.VariantType.LocalizedText,
            ua.ObjectIds.Structure: ua.VariantType.ExtensionObject,
            ua.ObjectIds.EnumValueType: ua.VariantType.ExtensionObject,
            ua.ObjectIds.Enumeration: ua.VariantType.Int32,  # enumeration
            ua.ObjectIds.AttributeWriteMask: ua.VariantType.UInt32,
            ua.ObjectIds.AxisScaleEnumeration: ua.VariantType.Int32  # enumeration
            }
        for dt, vdt in test_data.items():
            self.assertEqual(ua_utils.data_type_to_variant_type(self.opc.get_node(ua.NodeId(dt))), vdt)

def runTest(self):
        return # FIXME broken
        tmpfile = NamedTemporaryFile()
        path = tmpfile.name
        tmpfile.close()

        # create cache file
        server = Server(shelffile=path)

        # modify cache content
        id = ua.NodeId(ua.ObjectIds.Server_ServerStatus_SecondsTillShutdown)
        s = shelve.open(path, "w", writeback=True)
        s[id.to_string()].attributes[ua.AttributeIds.Value].value = ua.DataValue(123)
        s.close()

        # ensure that we are actually loading from the cache
        server = Server(shelffile=path)
        self.assertEqual(server.get_node(id).get_value(), 123)

        os.remove(path)