diff --git a/opcua/common/structures.py b/opcua/common/structures.py index 21424db80..775e96c14 100644 --- a/opcua/common/structures.py +++ b/opcua/common/structures.py @@ -9,9 +9,11 @@ import re import logging # The next two imports are for generated code +from collections import namedtuple from datetime import datetime import uuid from enum import Enum, IntEnum, EnumMeta +from typing import Any, Union from lxml import objectify @@ -21,13 +23,13 @@ logger = logging.getLogger(__name__) - +# TODO: use non-empty fields for testing. def get_default_value(uatype, enums): - if uatype == "String": - return "None" + if uatype in ("String", "CharArray"): + return "''" elif uatype == "Guid": return "uuid.uuid4()" - elif uatype in ("ByteString", "CharArray", "Char"): + elif uatype in ("ByteString", "Char"): return "b''" elif uatype == "Boolean": return "True" @@ -82,9 +84,17 @@ def __init__(self, name, value): class Struct(object): def __init__(self, name): self.name = _clean_name(name) + self.bit_mapping = {} self.fields = [] self.typeid = None + def get_ua_switches(self): + ua_switches = {} + for field in self.fields: + if field.is_switch(): + ua_switches[field.name] = self.bit_mapping.get(field.switch_name) + return ua_switches + def __str__(self): return "Struct(name={}, fields={}".format(self.name, self.fields) __repr__ = __str__ @@ -99,7 +109,12 @@ class {0}(object): ''' """.format(self.name) - + ua_switches = self.get_ua_switches() + if ua_switches: + code += " ua_switches = {\n" + for field_name, encode_tuple in self.get_ua_switches().items(): + code += " '{}': {}, \n".format(field_name, encode_tuple) + code += " }\n" code += " ua_types = [\n" for field in self.fields: prefix = "ListOf" if field.array else "" @@ -107,7 +122,6 @@ class {0}(object): if uatype == "ListOfChar": uatype = "String" code += " ('{}', '{}'),\n".format(field.name, uatype) - code += " ]" code += """ def __str__(self): @@ -128,15 +142,51 @@ def __init__(self): class Field(object): def __init__(self, name): self.name = name + self.switch_name = "" self.uatype = None - self.value = None + self.value = None # e.g: ua.Int32(0) self.array = False + def is_switch(self): + return len(self.switch_name) > 0 + def __str__(self): return "Field(name={}, uatype={}".format(self.name, self.uatype) __repr__ = __str__ +class BitFieldState: + def __init__(self): + self.encoding_field: Union[Field, None] = None + self.bit_size = 0 + self.bit_offset = 0 + self.encoding_field_counter = 0 + + def add_bit(self, length: int) -> Union[Field, None]: + """ Returns field if a new one was added. Else None """ + if not self.encoding_field: + return self.reset_encoding_field() + else: + if self.bit_size + length > 32: + return self.reset_encoding_field() + else: + self.bit_size += length + self.bit_offset += 1 + return None + + def reset_encoding_field(self) -> Field: + field = Field(f"BitEncoding{self.encoding_field_counter}") + field.uatype = "UInt32" + self.encoding_field = field + self.bit_offset = 0 + self.encoding_field_counter += 1 + return field + + def get_bit_info(self) -> (str, int): + """ With the field name and bit offset, we can extract the bit later.""" + return self.encoding_field.name, self.bit_offset + + class StructGenerator(object): def __init__(self): self.model = [] @@ -165,22 +215,33 @@ def _make_model(self, root): for child in root.iter("{*}StructuredType"): struct = Struct(child.get("Name")) array = False + bit_state = BitFieldState() for xmlfield in child.iter("{*}Field"): name = xmlfield.get("Name") + clean_name = _clean_name(name) if name.startswith("NoOf"): array = True continue - field = Field(_clean_name(name)) - field.uatype = xmlfield.get("TypeName") - if ":" in field.uatype: - field.uatype = field.uatype.split(":")[1] - field.uatype = _clean_name(field.uatype) - field.value = get_default_value(field.uatype, enums) - if array: - field.array = True - field.value = [] - array = False - struct.fields.append(field) + _type = xmlfield.get("TypeName") + if ":" in _type: + _type = _type.split(":")[1] + _type = _clean_name(_type) + if _type == 'Bit': + bit_length = int(xmlfield.get("Length", 1)) # Will longer bits be used? + field = bit_state.add_bit(bit_length) + # Whether or not a new encoding field was added, we want to store the current one. + struct.bit_mapping[name] = (bit_state.encoding_field.name, bit_state.bit_offset) + else: + field = Field(clean_name) + field.uatype = _type + field.switch_name = xmlfield.get('SwitchField', "") + if field: + field.value = get_default_value(field.uatype, enums) + if array: + field.array = True + field.value = [] + array = False + struct.fields.append(field) self.model.append(struct) def save_to_file(self, path, register=False): @@ -226,6 +287,7 @@ def _make_header(self, _file): ''' from datetime import datetime +from enum import IntEnum import uuid from opcua import ua diff --git a/opcua/ua/uatypes.py b/opcua/ua/uatypes.py index 911292d19..319f98df4 100644 --- a/opcua/ua/uatypes.py +++ b/opcua/ua/uatypes.py @@ -1010,7 +1010,7 @@ def get_default_value(vtype): raise RuntimeError("function take a uatype as argument, got:", vtype) -# These dictionnaries are used to register extensions classes for automatic +# These dictionaries are used to register extensions classes for automatic # decoding and encoding extension_object_classes = {} extension_object_ids = {} diff --git a/run-tests.sh b/run-tests.sh deleted file mode 100755 index 7b1457153..000000000 --- a/run-tests.sh +++ /dev/null @@ -1,2 +0,0 @@ -#!/bin/sh -python tests/tests.py diff --git a/tests/custom_extension_with_optional_fields.py b/tests/custom_extension_with_optional_fields.py new file mode 100644 index 000000000..ce0b89145 --- /dev/null +++ b/tests/custom_extension_with_optional_fields.py @@ -0,0 +1,256 @@ + +''' +THIS FILE IS AUTOGENERATED, DO NOT EDIT!!! +''' + +from datetime import datetime +from enum import IntEnum +import uuid + +from opcua import ua + + +class ChannelType(IntEnum): + + ''' + ChannelType EnumInt autogenerated from xml + ''' + + Universal = 0 + Pressure = 1 + Customizable = 2 + + +class InputSignalCategory(IntEnum): + + ''' + InputSignalCategory EnumInt autogenerated from xml + ''' + + Unassigned = 0 + MeasuringStart = 1 + MeasuringStop = 2 + CycleEnd = 3 + SwitchOverPoint = 4 + CycleEvent = 5 + + +class MoldTypeEnumeration(IntEnum): + + ''' + MoldTypeEnumeration EnumInt autogenerated from xml + ''' + + NormalMold = 0 + MultiComponentMold = 1 + RTMMold = 2 + + +class SetupChangeType(IntEnum): + + ''' + SetupChangeType EnumInt autogenerated from xml + ''' + + SetupLoaded = 0 + SetupChanged = 1 + SetupCreated = 2 + SetupUnloaded = 3 + SetupImported = 4 + + +class StopConditionType(IntEnum): + + ''' + StopConditionType EnumInt autogenerated from xml + ''' + + Time = 0 + StartSignal = 1 + StopSignal = 2 + + +class ChannelIdDataType(object): + + ''' + ChannelIdDataType structure autogenerated from xml + ''' + + ua_types = [ + ('Id', 'Int16'), + ('Type', 'ChannelType'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.Id = 0 + self.Type = ua.ChannelType(2) + + +class CurveDataType(object): + + ''' + CurveDataType structure autogenerated from xml + ''' + + ua_switches = { + 'Description': ('BitEncoding0', 0), + } + ua_types = [ + ('BitEncoding0', 'UInt32'), + ('ChannelId', 'ChannelIdDataType'), + ('Data', 'ListOfCurvePointDataType'), + ('Description', 'CharArray'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.BitEncoding0 = 0 + self.ChannelId = ua.ChannelIdDataType() + self.Data = [] + self.Description = '' + + +class CurvePointDataType(object): + + ''' + CurvePointDataType structure autogenerated from xml + ''' + + ua_types = [ + ('Time', 'Double'), + ('Data', 'Double'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.Time = 0 + self.Data = 0 + + +class DigitalSignalChangeDataType(object): + + ''' + DigitalSignalChangeDataType structure autogenerated from xml + ''' + + ua_types = [ + ('Id', 'InputSignalIdDataType'), + ('Time', 'Double'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.Id = ua.InputSignalIdDataType() + self.Time = 0 + + +class InputSignalIdDataType(object): + + ''' + InputSignalIdDataType structure autogenerated from xml + ''' + + ua_types = [ + ('Category', 'InputSignalCategory'), + ('SubId', 'SignalSubIdDataType'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.Category = ua.InputSignalCategory(5) + self.SubId = ua.SignalSubIdDataType() + + +class ProcessValueType(object): + + ''' + ProcessValueType structure autogenerated from xml + ''' + + ua_switches = { + 'cavityId': ('BitEncoding0', 0), + 'description': ('BitEncoding0', 1), + } + ua_types = [ + ('BitEncoding0', 'UInt32'), + ('name', 'CharArray'), + ('value', 'Double'), + ('assignment', 'UInt32'), + ('source', 'UInt32'), + ('cavityId', 'UInt32'), + ('id', 'CharArray'), + ('description', 'CharArray'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.BitEncoding0 = 0 + self.name = '' + self.value = 0 + self.assignment = 0 + self.source = 0 + self.cavityId = 0 + self.id = '' + self.description = '' + + +class SignalSubIdDataType(object): + + ''' + SignalSubIdDataType structure autogenerated from xml + ''' + + ua_types = [ + ('Number', 'Byte'), + ('Number2', 'Byte'), + ('Letter', 'Byte'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.Number = 0 + self.Number2 = 0 + self.Letter = 0 + + +ua.register_extension_object('ChannelType', ua.NodeId.from_string('ns=0;i=1'), ChannelType) +ua.register_extension_object('InputSignalCategory', ua.NodeId.from_string('ns=0;i=2'), InputSignalCategory) +ua.register_extension_object('MoldTypeEnumeration', ua.NodeId.from_string('ns=0;i=3'), MoldTypeEnumeration) +ua.register_extension_object('SetupChangeType', ua.NodeId.from_string('ns=0;i=4'), SetupChangeType) +ua.register_extension_object('StopConditionType', ua.NodeId.from_string('ns=0;i=5'), StopConditionType) +ua.register_extension_object('ChannelIdDataType', ua.NodeId.from_string('ns=0;i=6'), ChannelIdDataType) +ua.register_extension_object('CurveDataType', ua.NodeId.from_string('ns=0;i=7'), CurveDataType) +ua.register_extension_object('CurvePointDataType', ua.NodeId.from_string('ns=0;i=8'), CurvePointDataType) +ua.register_extension_object('DigitalSignalChangeDataType', ua.NodeId.from_string('ns=0;i=9'), DigitalSignalChangeDataType) +ua.register_extension_object('InputSignalIdDataType', ua.NodeId.from_string('ns=0;i=10'), InputSignalIdDataType) +ua.register_extension_object('ProcessValueType', ua.NodeId.from_string('ns=0;i=11'), ProcessValueType) +ua.register_extension_object('SignalSubIdDataType', ua.NodeId.from_string('ns=0;i=12'), SignalSubIdDataType) diff --git a/tests/custom_extension_with_optional_fields.xml b/tests/custom_extension_with_optional_fields.xml new file mode 100644 index 000000000..953dd2845 --- /dev/null +++ b/tests/custom_extension_with_optional_fields.xml @@ -0,0 +1,77 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + Mold type + + + + + + + + + + + + + Type of measurement stop condition + + + + + diff --git a/tests/structures.py b/tests/structures.py new file mode 100644 index 000000000..6e38330ce --- /dev/null +++ b/tests/structures.py @@ -0,0 +1,274 @@ + +''' +THIS FILE IS AUTOGENERATED, DO NOT EDIT!!! +''' + +from datetime import datetime +from enum import IntEnum +import uuid + +from opcua import ua + + +class ScalarValueDataType(object): + + ''' + ScalarValueDataType structure autogenerated from xml + ''' + + ua_types = [ + ('BooleanValue', 'Boolean'), + ('SByteValue', 'SByte'), + ('ByteValue', 'Byte'), + ('Int16Value', 'Int16'), + ('UInt16Value', 'UInt16'), + ('Int32Value', 'Int32'), + ('UInt32Value', 'UInt32'), + ('Int64Value', 'Int64'), + ('UInt64Value', 'UInt64'), + ('FloatValue', 'Float'), + ('DoubleValue', 'Double'), + ('StringValue', 'String'), + ('DateTimeValue', 'DateTime'), + ('GuidValue', 'Guid'), + ('ByteStringValue', 'ByteString'), + ('XmlElementValue', 'XmlElement'), + ('NodeIdValue', 'NodeId'), + ('ExpandedNodeIdValue', 'ExpandedNodeId'), + ('QualifiedNameValue', 'QualifiedName'), + ('LocalizedTextValue', 'LocalizedText'), + ('StatusCodeValue', 'StatusCode'), + ('VariantValue', 'Variant'), + ('EnumerationValue', 'Int32'), + ('StructureValue', 'ExtensionObject'), + ('Number', 'Variant'), + ('Integer', 'Variant'), + ('UInteger', 'Variant'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.BooleanValue = True + self.SByteValue = 0 + self.ByteValue = 0 + self.Int16Value = 0 + self.UInt16Value = 0 + self.Int32Value = 0 + self.UInt32Value = 0 + self.Int64Value = 0 + self.UInt64Value = 0 + self.FloatValue = 0 + self.DoubleValue = 0 + self.StringValue = '' + self.DateTimeValue = datetime.utcnow() + self.GuidValue = uuid.uuid4() + self.ByteStringValue = b'' + self.XmlElementValue = ua.XmlElement() + self.NodeIdValue = ua.NodeId() + self.ExpandedNodeIdValue = ua.ExpandedNodeId() + self.QualifiedNameValue = ua.QualifiedName() + self.LocalizedTextValue = ua.LocalizedText() + self.StatusCodeValue = ua.StatusCode() + self.VariantValue = ua.Variant() + self.EnumerationValue = 0 + self.StructureValue = ua.ExtensionObject() + self.Number = ua.Variant() + self.Integer = ua.Variant() + self.UInteger = ua.Variant() + + +class ArrayValueDataType(object): + + ''' + ArrayValueDataType structure autogenerated from xml + ''' + + ua_types = [ + ('BooleanValue', 'ListOfBoolean'), + ('SByteValue', 'ListOfSByte'), + ('ByteValue', 'ListOfByte'), + ('Int16Value', 'ListOfInt16'), + ('UInt16Value', 'ListOfUInt16'), + ('Int32Value', 'ListOfInt32'), + ('UInt32Value', 'ListOfUInt32'), + ('Int64Value', 'ListOfInt64'), + ('UInt64Value', 'ListOfUInt64'), + ('FloatValue', 'ListOfFloat'), + ('DoubleValue', 'ListOfDouble'), + ('StringValue', 'ListOfString'), + ('DateTimeValue', 'ListOfDateTime'), + ('GuidValue', 'ListOfGuid'), + ('ByteStringValue', 'ListOfByteString'), + ('XmlElementValue', 'ListOfXmlElement'), + ('NodeIdValue', 'ListOfNodeId'), + ('ExpandedNodeIdValue', 'ListOfExpandedNodeId'), + ('QualifiedNameValue', 'ListOfQualifiedName'), + ('LocalizedTextValue', 'ListOfLocalizedText'), + ('StatusCodeValue', 'ListOfStatusCode'), + ('VariantValue', 'ListOfVariant'), + ('EnumerationValue', 'ListOfInt32'), + ('StructureValue', 'ListOfExtensionObject'), + ('Number', 'ListOfVariant'), + ('Integer', 'ListOfVariant'), + ('UInteger', 'ListOfVariant'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.BooleanValue = [] + self.SByteValue = [] + self.ByteValue = [] + self.Int16Value = [] + self.UInt16Value = [] + self.Int32Value = [] + self.UInt32Value = [] + self.Int64Value = [] + self.UInt64Value = [] + self.FloatValue = [] + self.DoubleValue = [] + self.StringValue = [] + self.DateTimeValue = [] + self.GuidValue = [] + self.ByteStringValue = [] + self.XmlElementValue = [] + self.NodeIdValue = [] + self.ExpandedNodeIdValue = [] + self.QualifiedNameValue = [] + self.LocalizedTextValue = [] + self.StatusCodeValue = [] + self.VariantValue = [] + self.EnumerationValue = [] + self.StructureValue = [] + self.Number = [] + self.Integer = [] + self.UInteger = [] + + +class UserScalarValueDataType(object): + + ''' + UserScalarValueDataType structure autogenerated from xml + ''' + + ua_types = [ + ('BooleanDataType', 'Boolean'), + ('SByteDataType', 'SByte'), + ('ByteDataType', 'Byte'), + ('Int16DataType', 'Int16'), + ('UInt16DataType', 'UInt16'), + ('Int32DataType', 'Int32'), + ('UInt32DataType', 'UInt32'), + ('Int64DataType', 'Int64'), + ('UInt64DataType', 'UInt64'), + ('FloatDataType', 'Float'), + ('DoubleDataType', 'Double'), + ('StringDataType', 'String'), + ('DateTimeDataType', 'DateTime'), + ('GuidDataType', 'Guid'), + ('ByteStringDataType', 'ByteString'), + ('XmlElementDataType', 'XmlElement'), + ('NodeIdDataType', 'NodeId'), + ('ExpandedNodeIdDataType', 'ExpandedNodeId'), + ('QualifiedNameDataType', 'QualifiedName'), + ('LocalizedTextDataType', 'LocalizedText'), + ('StatusCodeDataType', 'StatusCode'), + ('VariantDataType', 'Variant'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.BooleanDataType = True + self.SByteDataType = 0 + self.ByteDataType = 0 + self.Int16DataType = 0 + self.UInt16DataType = 0 + self.Int32DataType = 0 + self.UInt32DataType = 0 + self.Int64DataType = 0 + self.UInt64DataType = 0 + self.FloatDataType = 0 + self.DoubleDataType = 0 + self.StringDataType = '' + self.DateTimeDataType = datetime.utcnow() + self.GuidDataType = uuid.uuid4() + self.ByteStringDataType = b'' + self.XmlElementDataType = ua.XmlElement() + self.NodeIdDataType = ua.NodeId() + self.ExpandedNodeIdDataType = ua.ExpandedNodeId() + self.QualifiedNameDataType = ua.QualifiedName() + self.LocalizedTextDataType = ua.LocalizedText() + self.StatusCodeDataType = ua.StatusCode() + self.VariantDataType = ua.Variant() + + +class UserArrayValueDataType(object): + + ''' + UserArrayValueDataType structure autogenerated from xml + ''' + + ua_types = [ + ('BooleanDataType', 'ListOfBoolean'), + ('SByteDataType', 'ListOfSByte'), + ('ByteDataType', 'ListOfByte'), + ('Int16DataType', 'ListOfInt16'), + ('UInt16DataType', 'ListOfUInt16'), + ('Int32DataType', 'ListOfInt32'), + ('UInt32DataType', 'ListOfUInt32'), + ('Int64DataType', 'ListOfInt64'), + ('UInt64DataType', 'ListOfUInt64'), + ('FloatDataType', 'ListOfFloat'), + ('DoubleDataType', 'ListOfDouble'), + ('StringDataType', 'ListOfString'), + ('DateTimeDataType', 'ListOfDateTime'), + ('GuidDataType', 'ListOfGuid'), + ('ByteStringDataType', 'ListOfByteString'), + ('XmlElementDataType', 'ListOfXmlElement'), + ('NodeIdDataType', 'ListOfNodeId'), + ('ExpandedNodeIdDataType', 'ListOfExpandedNodeId'), + ('QualifiedNameDataType', 'ListOfQualifiedName'), + ('LocalizedTextDataType', 'ListOfLocalizedText'), + ('StatusCodeDataType', 'ListOfStatusCode'), + ('VariantDataType', 'ListOfVariant'), + ] + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.BooleanDataType = [] + self.SByteDataType = [] + self.ByteDataType = [] + self.Int16DataType = [] + self.UInt16DataType = [] + self.Int32DataType = [] + self.UInt32DataType = [] + self.Int64DataType = [] + self.UInt64DataType = [] + self.FloatDataType = [] + self.DoubleDataType = [] + self.StringDataType = [] + self.DateTimeDataType = [] + self.GuidDataType = [] + self.ByteStringDataType = [] + self.XmlElementDataType = [] + self.NodeIdDataType = [] + self.ExpandedNodeIdDataType = [] + self.QualifiedNameDataType = [] + self.LocalizedTextDataType = [] + self.StatusCodeDataType = [] + self.VariantDataType = [] diff --git a/tests/tests_cmd_lines.py b/tests/tests_cmd_lines.py index 0bfacb1b9..7ee0f96aa 100644 --- a/tests/tests_cmd_lines.py +++ b/tests/tests_cmd_lines.py @@ -25,25 +25,25 @@ def setUpClass(cls): cls.srv.start() def test_uals(self): - s = subprocess.check_output(["python", "tools/uals", "--url", self.srv_url]) + s = subprocess.check_output(["python", "../tools/uals", "--url", self.srv_url]) self.assertIn(b"i=85", s) self.assertNotIn(b"i=89", s) self.assertNotIn(b"1.999", s) - s = subprocess.check_output(["python", "tools/uals", "--url", self.srv_url, "-d", "3"]) + s = subprocess.check_output(["python", "../tools/uals", "--url", self.srv_url, "-d", "3"]) self.assertIn(b"1.999", s) def test_uaread(self): - s = subprocess.check_output(["python", "tools/uaread", "--url", self.srv_url, "--path", "0:Objects,4:directory,4:variable"]) + s = subprocess.check_output(["python", "../tools/uaread", "--url", self.srv_url, "--path", "0:Objects,4:directory,4:variable"]) self.assertIn(b"1.999", s) def test_uawrite(self): - s = subprocess.check_output(["python", "tools/uawrite", "--url", self.srv_url, "--path", "0:Objects,4:directory,4:variable2", "1.789"]) - s = subprocess.check_output(["python", "tools/uaread", "--url", self.srv_url, "--path", "0:Objects,4:directory,4:variable2"]) + s = subprocess.check_output(["python", "../tools/uawrite", "--url", self.srv_url, "--path", "0:Objects,4:directory,4:variable2", "1.789"]) + s = subprocess.check_output(["python", "../tools/uaread", "--url", self.srv_url, "--path", "0:Objects,4:directory,4:variable2"]) self.assertIn(b"1.789", s) self.assertNotIn(b"1.999", s) def test_uadiscover(self): - s = subprocess.check_output(["python", "tools/uadiscover", "--url", self.srv_url]) + s = subprocess.check_output(["python", "../tools/uadiscover", "--url", self.srv_url]) self.assertIn(b"opc.tcp://127.0.0.1", s) self.assertIn(b"FreeOpcUa", s) self.assertIn(b"urn:freeopcua:python:server", s) diff --git a/tests/tests_crypto_connect.py b/tests/tests_crypto_connect.py index c2b85ba08..0c6eeeb4e 100644 --- a/tests/tests_crypto_connect.py +++ b/tests/tests_crypto_connect.py @@ -34,8 +34,8 @@ def setUpClass(cls): cls.srv_crypto.set_endpoint(cls.uri_crypto) # load server certificate and private key. This enables endpoints # with signing and encryption. - cls.srv_crypto.load_certificate("examples/certificate-example.der") - cls.srv_crypto.load_private_key("examples/private-key-example.pem") + cls.srv_crypto.load_certificate("../examples/certificate-example.der") + cls.srv_crypto.load_private_key("../examples/private-key-example.pem") cls.srv_crypto.start() # start a server without crypto @@ -48,8 +48,8 @@ def setUpClass(cls): cls.srv_crypto2 = Server() cls.uri_crypto2 = 'opc.tcp://127.0.0.1:{0:d}'.format(port_num3) cls.srv_crypto2.set_endpoint(cls.uri_crypto2) - cls.srv_crypto2.load_certificate("examples/certificate-3072-example.der") - cls.srv_crypto2.load_private_key("examples/private-key-3072-example.pem") + cls.srv_crypto2.load_certificate("../examples/certificate-3072-example.der") + cls.srv_crypto2.load_private_key("../examples/private-key-3072-example.pem") cls.srv_crypto2.start() cls.srv_crypto_no_anoymous = Server() @@ -78,12 +78,12 @@ def test_nocrypto(self): def test_nocrypto_fail(self): clt = Client(self.uri_no_crypto) with self.assertRaises(ua.UaError): - clt.set_security_string("Basic256Sha256,Sign,examples/certificate-example.der,examples/private-key-example.pem") + clt.set_security_string("Basic256Sha256,Sign,../examples/certificate-example.der,../examples/private-key-example.pem") def test_basic256sha256(self): clt = Client(self.uri_crypto) try: - clt.set_security_string("Basic256Sha256,Sign,examples/certificate-example.der,examples/private-key-example.pem") + clt.set_security_string("Basic256Sha256,Sign,../examples/certificate-example.der,../examples/private-key-example.pem") clt.connect() self.assertTrue(clt.get_objects_node().get_children()) finally: @@ -92,7 +92,7 @@ def test_basic256sha256(self): def test_basic256sha256_longkey(self): clt = Client(self.uri_crypto2) try: - clt.set_security_string("Basic256Sha256,Sign,examples/certificate-example.der,examples/private-key-example.pem") + clt.set_security_string("Basic256Sha256,Sign,../examples/certificate-example.der,../examples/private-key-example.pem") clt.connect() self.assertTrue(clt.get_objects_node().get_children()) finally: @@ -101,7 +101,7 @@ def test_basic256sha256_longkey(self): def test_basic256sha256_encrypt(self): clt = Client(self.uri_crypto) try: - clt.set_security_string("Basic256Sha256,SignAndEncrypt,examples/certificate-example.der,examples/private-key-example.pem") + clt.set_security_string("Basic256Sha256,SignAndEncrypt,../examples/certificate-example.der,../examples/private-key-example.pem") clt.connect() self.assertTrue(clt.get_objects_node().get_children()) finally: @@ -110,7 +110,7 @@ def test_basic256sha256_encrypt(self): def test_basic256sha256_encrypt_longkey(self): clt = Client(self.uri_crypto2) try: - clt.set_security_string("Basic256Sha256,SignAndEncrypt,examples/certificate-example.der,examples/private-key-example.pem") + clt.set_security_string("Basic256Sha256,SignAndEncrypt,../examples/certificate-example.der,../examples/private-key-example.pem") clt.connect() self.assertTrue(clt.get_objects_node().get_children()) finally: @@ -120,8 +120,8 @@ def test_basic256sha56_encrypt_success(self): clt = Client(self.uri_crypto) try: clt.set_security(security_policies.SecurityPolicyBasic256Sha256, - 'examples/certificate-example.der', - 'examples/private-key-example.pem', + '../examples/certificate-example.der', + '../examples/private-key-example.pem', None, ua.MessageSecurityMode.SignAndEncrypt ) @@ -135,8 +135,8 @@ def test_basic256sha56_encrypt_fail(self): clt = Client(self.uri_crypto) with self.assertRaises(ua.UaError): clt.set_security(security_policies.SecurityPolicyBasic256Sha256, - 'examples/certificate-example.der', - 'examples/private-key-example.pem', + '../examples/certificate-example.der', + '../examples/private-key-example.pem', None, ua.MessageSecurityMode.None_ ) diff --git a/tests/tests_custom_structures_optional_fields.py b/tests/tests_custom_structures_optional_fields.py new file mode 100644 index 000000000..6e5617bad --- /dev/null +++ b/tests/tests_custom_structures_optional_fields.py @@ -0,0 +1,107 @@ +import inspect +import sys +import unittest +from datetime import datetime +from enum import EnumMeta + +from opcua import ua +from opcua.common.structures import StructGenerator, Struct, EnumType +from opcua.ua.ua_binary import struct_to_binary, struct_from_binary + + +# An ExtensionObject that is decoded becomes a generated class of a custom name. +# The class below represents an example that includes OPC UA optional fields. +class ObjectWithOptionalFields(object): + ua_switches = { + 'cavityId': ('BitEncoding0', 0), + 'description': ('BitEncoding0', 1), + # This exists in xml but gets ignored because no matching switches. + # 'Reserved1': ('ByteEncoding0', 2) + } + ua_types = [ + # The bits fields are not necessary because 'ua_switches' provides us with + # the related bit we care about. + # ('cavityIdSpecified', 'Bit'), + # ('descriptionSpecified', 'Bit'), + # ('Reserved1', 'Bit'), + ('BitEncoding0', 'UInt32'), + ('name', 'CharArray'), + ('value', 'Double'), + ('assignment', 'UInt32'), + ('source', 'UInt32'), + ('cavityId', 'UInt32'), + ('id', 'CharArray'), + ('description', 'CharArray'), + ] + + def __str__(self): + vals = [name + ": " + str(val) for name, val in self.__dict__.items()] + return self.__class__.__name__ + "(" + ", ".join(vals) + ")" + + __repr__ = __str__ + + def __init__(self): + self.BitEncoding0 = 0x03 + self.name = 'SomeAmazingCycleParameter' + self.value = 8 + self.assignment = 8 + self.source = 8 + self.cavityId = 5 + self.id = 'abcdefgh' + self.description = 'BBA' + + +class CustomStructTestCase(unittest.TestCase): + + def setUp(self): + self.identifier_count = 0 + + def _generate_node_id(self): + self.identifier_count += 1 + return f"ns=0;i={self.identifier_count}" + + @staticmethod + def is_struct(obj): + # TODO: putting this definition in the generated class would be better + return hasattr(obj, 'ua_types') + + def assertCustomStructEqual(self, original, deserialized): + if hasattr(original, 'ua_switches'): + self.assertEqual(len(original.ua_switches), len(deserialized.ua_switches)) + self.assertEqual(len(original.ua_types), len(deserialized.ua_types)) + for field, _ in original.ua_types: + field_obj = getattr(original, field) + deserialized_obj = getattr(deserialized, field) + if self.is_struct(field_obj): + self.assertCustomStructEqual(field_obj, deserialized_obj) + else: + self.assertEqual(getattr(original, field), getattr(deserialized, field)) + + def test_binary_struct_example(self): + # Example test so that we can manually control the object that gets + # generated and see how it gets serialized/deserialized + original = ObjectWithOptionalFields() + serialized = struct_to_binary(original) + deserialized = struct_from_binary(ObjectWithOptionalFields, ua.utils.Buffer(serialized)) + self.assertEqual(len(original.ua_switches), len(deserialized.ua_switches)) + self.assertEqual(len(original.ua_types), len(deserialized.ua_types)) + for field, _ in original.ua_types: + self.assertEqual(getattr(original, field), getattr(deserialized, field)) + + def test_custom_struct_with_optional_fields(self): + xmlpath = "custom_extension_with_optional_fields.xml" + c = StructGenerator() + c.make_model_from_file(xmlpath) + for m in c.model: + if type(m) in (Struct, EnumType): + m.typeid = self._generate_node_id() + c.save_to_file("custom_extension_with_optional_fields.py", register=True) + import como_structures as s + for name, obj in inspect.getmembers(sys.modules[s.__name__], predicate=inspect.isclass): + if name.startswith('__') or obj in (datetime,) or isinstance(obj, EnumMeta): + continue + with self.subTest(name=name): + original = obj() + serialized = struct_to_binary(original) + deserialized = struct_from_binary(obj, ua.utils.Buffer(serialized)) + self.assertCustomStructEqual(original, deserialized) diff --git a/tests/tests_enum_struct.py b/tests/tests_enum_struct.py index ef9e5247d..09a3d289e 100644 --- a/tests/tests_enum_struct.py +++ b/tests/tests_enum_struct.py @@ -30,7 +30,7 @@ def __str__(self): def add_server_custom_enum_struct(server): # import some nodes from xml - server.import_xml("tests/enum_struct_test_nodes.xml") + server.import_xml("enum_struct_test_nodes.xml") ns = server.get_namespace_index('http://yourorganisation.org/struct_enum_example/') uatypes.register_extension_object('ExampleStruct', ua.NodeId(5001, ns), ExampleStruct) val = ua.ExampleStruct() diff --git a/tests/tests_unit.py b/tests/tests_unit.py index cf7c247b7..ac017275c 100755 --- a/tests/tests_unit.py +++ b/tests/tests_unit.py @@ -50,7 +50,7 @@ def test_variant_empty_list(self): self.assertTrue(v2.is_array) def test_structs_save_and_import(self): - xmlpath = "tests/example.bsd" + xmlpath = "example.bsd" c = StructGenerator() c.make_model_from_file(xmlpath) struct_dict = c.save_and_import("structures.py") @@ -59,10 +59,10 @@ def test_structs_save_and_import(self): self.assertEqual(k, a.__class__.__name__) def test_custom_structs(self): - xmlpath = "tests/example.bsd" + xmlpath = "example.bsd" c = StructGenerator() c.make_model_from_file(xmlpath) - c.save_to_file("tests/structures.py") + c.save_to_file("structures.py") import structures as s # test with default values @@ -106,10 +106,10 @@ def test_custom_structs(self): self.assertEqual(v.NodeIdValue, v2.NodeIdValue) def test_custom_structs_array(self): - xmlpath = "tests/example.bsd" + xmlpath = "example.bsd" c = StructGenerator() c.make_model_from_file(xmlpath) - c.save_to_file("tests/structures.py") + c.save_to_file("structures.py") import structures as s # test with default values diff --git a/tests/tests_xml.py b/tests/tests_xml.py index 6020e1c87..687e12d16 100644 --- a/tests/tests_xml.py +++ b/tests/tests_xml.py @@ -24,7 +24,7 @@ class XmlTests(object): assertEqual = dir def test_xml_import(self): - self.opc.import_xml("tests/custom_nodes.xml") + self.opc.import_xml("custom_nodes.xml") o = self.opc.get_objects_node() v = o.get_child(["1:MyXMLFolder", "1:MyXMLObject", "1:MyXMLVariable"]) val = v.get_value() @@ -52,7 +52,7 @@ def test_xml_import_additional_ns(self): self.srv.register_namespace('http://placeholder.toincrease.nsindex') # if not already shift the new namespaces # "tests/custom_nodes.xml" isn't created with namespaces in mind, provide new test file - self.opc.import_xml("tests/custom_nodesns.xml") # the ns=1 in to file now should be mapped to ns=2 + self.opc.import_xml("custom_nodesns.xml") # the ns=1 in to file now should be mapped to ns=2 ns = self.opc.get_namespace_index("http://examples.freeopcua.github.io/") o = self.opc.get_objects_node()