import unittest
import logging
import xml.etree.ElementTree as Et
from opcua import ua, Server
import opcua.common.type_dictionary_buider
from opcua.common.type_dictionary_buider import OPCTypeDictionaryBuilder, DataTypeDictionaryBuilder
from opcua.common.type_dictionary_buider import get_ua_class, StructNode

port_num = 48540
idx_name = 'http://test.freeopcua.github.io'


def to_camel_case(name):
    func = getattr(opcua.common.type_dictionary_buider, '_to_camel_case')
    return func(name)


def reference_generator(source_id, target_id, reference_type, is_forward=True):
    func = getattr(opcua.common.type_dictionary_buider, '_reference_generator')
    return func(source_id, target_id, reference_type, is_forward)


def set_up_test_tree():
    ext_head_attributes = {'xmlns:xsi': 'http://www.w3.org/2001/XMLSchema-instance', 'xmlns:tns': idx_name,
                           'DefaultByteOrder': 'LittleEndian', 'xmlns:opc': 'http://opcfoundation.org/BinarySchema/',
                           'xmlns:ua': 'http://opcfoundation.org/UA/', 'TargetNamespace': idx_name}

    test_etree = Et.ElementTree(Et.Element('opc:TypeDictionary', ext_head_attributes))
    name_space = Et.SubElement(test_etree.getroot(), 'opc:Import')
    name_space.attrib['Namespace'] = 'http://opcfoundation.org/UA/'
    return test_etree


class TypeDictionaryBuilderTest(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.srv = Server()
        cls.srv.set_endpoint('opc.tcp://127.0.0.1:{0:d}'.format(port_num))
        cls.idx = cls.srv.register_namespace(idx_name)
        cls.srv.start()

    @classmethod
    def tearDownClass(cls):
        cls.srv.stop()

    def setUp(self):
        self.test_etree = set_up_test_tree()
        self.opc_type_builder = OPCTypeDictionaryBuilder(idx_name)
        self.dict_builder = DataTypeDictionaryBuilder(self.srv, self.idx, idx_name, 'TestDict')

    def tearDown(self):
        return

    def test_camel_case_1(self):
        case = 'TurtleActionlibShapeActionFeedback'
        result = to_camel_case('turtle_actionlib/ShapeActionFeedback')
        self.assertEquals(result, case)

    def test_camel_case_2(self):
        case = 'HelloWorldFffD'
        result = to_camel_case('Hello#world+fff_**?&&d')
        self.assertEquals(result, case)

    def test_opc_type_dict_process_type_opc(self):
        case = 'opc:Boolean'
        result = getattr(self.opc_type_builder, '_process_type')('Boolean')
        self.assertEquals(result, case)

    def test_opc_type_dict_process_type_tns(self):
        case = 'tns:CustomizedStruct'
        result = getattr(self.opc_type_builder, '_process_type')('CustomizedStruct')
        self.assertEquals(result, case)

    def test_opc_type_dict_append_struct_1(self):
        case = {'BaseType': 'ua:ExtensionObject',
                'Name': 'CustomizedStruct'}
        result = self.opc_type_builder.append_struct('CustomizedStruct')
        self.assertEquals(result.attrib, case)

    def DISABLED_test_opc_type_dict_append_struct_2(self):
        case = {'BaseType': 'ua:ExtensionObject',
                'Name': 'CustomizedStruct'}
        result = self.opc_type_builder.append_struct('customized_#?+`struct')
        self.assertEquals(result.attrib, case)

    def test_opc_type_dict_add_field_1(self):
        structure_name = 'CustomizedStruct'
        self.opc_type_builder.append_struct(structure_name)
        self.opc_type_builder.add_field('id', ua.VariantType.Boolean, structure_name)
        case = {'TypeName': 'opc:Boolean',
                'Name': 'id'}
        struct_dict = getattr(self.opc_type_builder, '_structs_dict')
        result = list(struct_dict[structure_name])[0]
        self.assertEquals(result.attrib, case)

    def test_opc_type_dict_add_field_2(self):
        structure_name = 'CustomizedStruct'
        self.opc_type_builder.append_struct(structure_name)
        self.opc_type_builder.add_field('id', 'Boolean', structure_name)
        case = {'TypeName': 'opc:Boolean',
                'Name': 'id'}
        struct_dict = getattr(self.opc_type_builder, '_structs_dict')
        result = list(struct_dict[structure_name])[0]
        self.assertEquals(result.attrib, case)

    def test_opc_type_dict_add_field_3(self):
        structure_name = 'CustomizedStruct'
        self.opc_type_builder.append_struct(structure_name)
        self.opc_type_builder.add_field('id', ua.VariantType.Boolean, structure_name, is_array=True)
        case = [{'TypeName': 'opc:Int32',
                'Name': 'NoOfid'},
                {'TypeName': 'opc:Boolean',
                 'LengthField': 'NoOfid',
                 'Name': 'id'}]
        struct_dict = getattr(self.opc_type_builder, '_structs_dict')
        result = [item.attrib for item in list(struct_dict[structure_name])]
        self.assertTrue(len(case), len(result))
        for item in case:
            self.assertTrue(item in result)

    def test_opc_type_dict_get_dict_value(self):
        structure_name = 'CustomizedStruct'
        self.opc_type_builder.append_struct(structure_name)
        # external tree operation
        appended_struct = Et.SubElement(self.test_etree.getroot(), 'opc:StructuredType')
        appended_struct.attrib['BaseType'] = 'ua:ExtensionObject'
        appended_struct.attrib['Name'] = to_camel_case(structure_name)

        self.opc_type_builder.add_field('id', ua.VariantType.Boolean, structure_name)
        # external tree operation
        field = Et.SubElement(appended_struct, 'opc:Field')
        field.attrib['Name'] = 'id'
        field.attrib['TypeName'] = 'opc:Boolean'
        case = Et.tostring(self.test_etree.getroot(), encoding='utf-8').decode("utf-8").replace(' ', '')
        result = self.opc_type_builder.get_dict_value().decode("utf-8").replace(' ', '').replace('\n', '')
        self.assertEqual(result, case)

    def test_reference_generator_1(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)
        self.assertTrue(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_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_data_type_dict_general(self):
        self.assertIsNotNone(self.dict_builder.dict_id)
        self.assertIsNotNone(getattr(self.dict_builder, '_type_dictionary'))

    def test_data_type_dict_add_dictionary(self):
        add_dictionary = getattr(self.dict_builder, '_add_dictionary')
        dict_name = 'TestDict'
        dict_node = self.srv.get_node(add_dictionary(dict_name))
        self.assertEqual(dict_node.get_browse_name(), ua.QualifiedName(dict_name, self.idx))
        self.assertEqual(dict_node.get_node_class(), ua.NodeClass.Variable)
        self.assertEqual(dict_node.get_parent().nodeid, ua.NodeId(ua.ObjectIds.OPCBinarySchema_TypeSystem, 0))
        self.assertEqual(ua.NodeId(ua.ObjectIds.HasComponent, 0),
                         dict_node.get_references(refs=ua.ObjectIds.HasComponent)[0].ReferenceTypeId)
        self.assertEqual(dict_node.get_type_definition(), ua.NodeId(ua.ObjectIds.DataTypeDictionaryType, 0))
        self.assertEqual(dict_node.get_display_name(), ua.LocalizedText(dict_name))
        self.assertEqual(dict_node.get_data_type(), ua.NodeId(ua.ObjectIds.ByteString))
        self.assertEqual(dict_node.get_value_rank(), -1)

    def test_data_type_dict_create_data_type(self):
        type_name = 'CustomizedStruct2'
        created_type = self.dict_builder.create_data_type(type_name)
        self.assertTrue(isinstance(created_type, StructNode))
        # Test data type node
        type_node = self.srv.get_node(created_type.data_type)
        self.assertEqual(type_node.get_browse_name(), ua.QualifiedName(type_name, self.idx))
        self.assertEqual(type_node.get_node_class(), ua.NodeClass.DataType)
        self.assertEqual(type_node.get_parent().nodeid, ua.NodeId(ua.ObjectIds.Structure, 0))
        self.assertEqual(ua.NodeId(ua.ObjectIds.HasSubtype, 0),
                         type_node.get_references(refs=ua.ObjectIds.HasSubtype)[0].ReferenceTypeId)
        self.assertEqual(type_node.get_display_name(), ua.LocalizedText(type_name))

        # Test description node
        desc_node = self.srv.get_node(self.dict_builder.dict_id).get_child("2:{}".format(type_name))
        self.assertEqual(desc_node.get_browse_name(), ua.QualifiedName(type_name, self.idx))
        self.assertEqual(desc_node.get_node_class(), ua.NodeClass.Variable)
        self.assertEqual(desc_node.get_parent().nodeid, self.dict_builder.dict_id)
        self.assertEqual(ua.NodeId(ua.ObjectIds.HasComponent, 0),
                         desc_node.get_references(refs=ua.ObjectIds.HasComponent)[0].ReferenceTypeId)
        self.assertEqual(desc_node.get_type_definition(), ua.NodeId(ua.ObjectIds.DataTypeDescriptionType, 0))

        self.assertEqual(desc_node.get_display_name(), ua.LocalizedText(type_name))
        self.assertEqual(desc_node.get_data_type(), ua.NodeId(ua.ObjectIds.String))
        self.assertEqual(desc_node.get_value(), type_name)
        self.assertEqual(desc_node.get_value_rank(), -1)

        # Test object node
        obj_node = type_node.get_children(refs=ua.ObjectIds.HasEncoding)[0]
        self.assertEqual(obj_node.get_browse_name(), ua.QualifiedName('Default Binary', 0))
        self.assertEqual(obj_node.get_node_class(), ua.NodeClass.Object)
        self.assertEqual(obj_node.get_references(refs=ua.ObjectIds.HasEncoding)[0].NodeId, type_node.nodeid)
        self.assertEqual(ua.NodeId(ua.ObjectIds.HasEncoding, 0),
                         obj_node.get_references(refs=ua.ObjectIds.HasEncoding)[0].ReferenceTypeId)
        self.assertEqual(obj_node.get_type_definition(), ua.NodeId(ua.ObjectIds.DataTypeEncodingType, 0))
        self.assertEqual(obj_node.get_display_name(), ua.LocalizedText('Default Binary'))
        self.assertEqual(len(obj_node.get_event_notifier()), 0)

        # Test links, three were tested above
        struct_node = self.srv.get_node(ua.NodeId(ua.ObjectIds.Structure, 0))
        struct_children = struct_node.get_children(refs=ua.ObjectIds.HasSubtype)
        self.assertTrue(type_node in struct_children)
        dict_node = self.srv.get_node(self.dict_builder.dict_id)
        dict_children = dict_node.get_children(refs=ua.ObjectIds.HasComponent)
        self.assertTrue(desc_node in dict_children)
        self.assertTrue(obj_node in type_node.get_children(ua.ObjectIds.HasEncoding))
        self.assertTrue(desc_node in obj_node.get_children(refs=ua.ObjectIds.HasDescription))
        self.assertEqual(obj_node.nodeid, desc_node.get_references(refs=ua.ObjectIds.HasDescription,
                                                                   direction=ua.BrowseDirection.Inverse)[0].NodeId)

    def test_data_type_dict_set_dict_byte_string(self):
        structure_name = 'CustomizedStruct'
        self.dict_builder.create_data_type(structure_name)
        self.dict_builder.add_field('id', ua.VariantType.Int32, structure_name)
        self.dict_builder.set_dict_byte_string()
        # external tree operation
        appended_struct = Et.SubElement(self.test_etree.getroot(), 'opc:StructuredType')
        appended_struct.attrib['BaseType'] = 'ua:ExtensionObject'
        appended_struct.attrib['Name'] = to_camel_case(structure_name)

        # external tree operation
        field = Et.SubElement(appended_struct, 'opc:Field')
        field.attrib['Name'] = 'id'
        field.attrib['TypeName'] = 'opc:Int32'
        case = Et.tostring(self.test_etree.getroot(), encoding='utf-8').decode("utf-8").replace(' ', '')
        result_string = self.srv.get_node(self.dict_builder.dict_id).get_value().decode("utf-8")
        result = result_string.replace(' ', '').replace('\n', '')
        self.assertEqual(result, case)

    def test_data_type_dict_add_field_1(self):
        struct_name = 'CustomizedStruct'
        self.dict_builder.create_data_type(struct_name)
        self.dict_builder.add_field('id', ua.VariantType.Int32, struct_name)
        self.dict_builder.set_dict_byte_string()
        self.srv.load_type_definitions()
        struct = get_ua_class(struct_name)
        self.assertEqual(struct.ua_types[0][0], 'id')
        self.assertEqual(struct.ua_types[0][1], 'Int32')
        struct_instance = struct()
        self.assertEqual(struct_instance.id, 0)

    def test_data_type_dict_add_field_2(self):
        struct_name = 'AnotherCustomizedStruct'
        self.dict_builder.create_data_type(struct_name)
        self.dict_builder.add_field('id', ua.VariantType.Int32, struct_name, is_array=True)
        self.dict_builder.set_dict_byte_string()
        self.srv.load_type_definitions()
        struct = get_ua_class(struct_name)
        self.assertEqual(struct.ua_types[0][0], 'id')
        self.assertEqual(struct.ua_types[0][1], 'ListOfInt32')
        struct_instance = struct()
        self.assertTrue(isinstance(struct_instance.id, list))

    def test_struct_node_general(self):
        struct_name = 'CustomizedStruct'
        struct_node = self.dict_builder.create_data_type(struct_name)
        self.assertEqual(getattr(struct_node, '_type_dict'), self.dict_builder)
        self.assertTrue(isinstance(struct_node.data_type, ua.NodeId))
        self.assertEqual(struct_node.name, struct_name)

    def test_struct_node_add_field(self):
        struct_name = 'CustomizedStruct'
        struct_node = self.dict_builder.create_data_type(struct_name)
        struct_node.add_field('id', ua.VariantType.Int32)
        self.dict_builder.set_dict_byte_string()
        self.srv.load_type_definitions()
        struct = get_ua_class(struct_name)
        self.assertEqual(struct.ua_types[0][0], 'id')
        self.assertEqual(struct.ua_types[0][1], 'Int32')
        struct_instance = struct()
        self.assertEqual(struct_instance.id, 0)

    def test_get_ua_class_1(self):
        struct_name = 'CustomizedStruct'
        struct_node = self.dict_builder.create_data_type(struct_name)
        struct_node.add_field('id', ua.VariantType.Int32)
        self.dict_builder.set_dict_byte_string()
        self.srv.load_type_definitions()
        try:
            self.assertIsNotNone(get_ua_class(struct_name))
        except AttributeError:
            pass

    def DISABLED_test_get_ua_class_2(self):
        struct_name = '*c*u_stom-ized&Stru#ct'
        struct_node = self.dict_builder.create_data_type(struct_name)
        struct_node.add_field('id', ua.VariantType.Int32)
        self.dict_builder.set_dict_byte_string()
        self.srv.load_type_definitions()
        try:
            self.assertIsNotNone(get_ua_class(struct_name))
        except AttributeError:
            pass

    def test_functional_basic(self):
        basic_struct_name = 'basic_structure'
        basic_struct = self.dict_builder.create_data_type(basic_struct_name)
        basic_struct.add_field('ID', ua.VariantType.Int32)
        basic_struct.add_field('Gender', ua.VariantType.Boolean)
        basic_struct.add_field('Comments', ua.VariantType.String)

        self.dict_builder.set_dict_byte_string()
        self.srv.load_type_definitions()

        basic_var = self.srv.nodes.objects.add_variable(ua.NodeId(namespaceidx=self.idx), 'BasicStruct',
                                                        ua.Variant(None, ua.VariantType.Null),
                                                        datatype=basic_struct.data_type)

        basic_msg = get_ua_class(basic_struct_name)()
        basic_msg.ID = 3
        basic_msg.Gender = True
        basic_msg.Comments = 'Test string'
        basic_var.set_value(basic_msg)

        basic_result = basic_var.get_value()
        self.assertEqual(basic_result, basic_msg)

    def test_functional_advance(self):
        basic_struct_name = 'base_structure'
        basic_struct = self.dict_builder.create_data_type(basic_struct_name)
        basic_struct.add_field('ID', ua.VariantType.Int32)
        basic_struct.add_field('Gender', ua.VariantType.Boolean)
        basic_struct.add_field('Comments', ua.VariantType.String)

        nested_struct_name = 'nested_structure'
        nested_struct = self.dict_builder.create_data_type(nested_struct_name)
        nested_struct.add_field('Name', ua.VariantType.String)
        nested_struct.add_field('Surname', ua.VariantType.String)
        nested_struct.add_field('Stuff', basic_struct)

        self.dict_builder.set_dict_byte_string()
        self.srv.load_type_definitions()

        basic_var = self.srv.nodes.objects.add_variable(ua.NodeId(namespaceidx=self.idx), 'BaseStruct',
                                                        ua.Variant(None, ua.VariantType.Null),
                                                        datatype=basic_struct.data_type)

        basic_msg = get_ua_class(basic_struct_name)()
        basic_msg.ID = 3
        basic_msg.Gender = True
        basic_msg.Comments = 'Test string'
        basic_var.set_value(basic_msg)

        nested_var = self.srv.nodes.objects.add_variable(ua.NodeId(namespaceidx=self.idx), 'NestedStruct',
                                                         ua.Variant(None, ua.VariantType.Null),
                                                         datatype=nested_struct.data_type)

        nested_msg = get_ua_class(nested_struct_name)()
        nested_msg.Stuff = basic_msg
        nested_msg.Name = 'Max'
        nested_msg.Surname = 'Karl'
        nested_var.set_value(nested_msg)

        basic_result = basic_var.get_value()
        self.assertEqual(basic_result, basic_msg)
        nested_result = nested_var.get_value()
        self.assertEqual(nested_result, nested_msg)

    def test_siemens_flavored_struct_name(self):
        struct_name = '"Siemens"."OPC-UA"."Identifiers_LookLikeThis"'
        struct_node = self.dict_builder.create_data_type(struct_name)
        struct_node.add_field('"a"."field"', ua.VariantType.Int32)
        self.dict_builder.set_dict_byte_string()
        self.srv.load_type_definitions()

if __name__ == '__main__':
    logging.basicConfig(level=logging.WARNING)
    unittest.main(verbosity=3)
