Fix custom data type creation from amonsch and other cleanups

asyncua/server/server.py

@@ -28,6 +28,14 @@ from ..crypto import security_policies, uacrypto
 _logger = logging.getLogger(__name__)
 
 
+def _get_node(isession, whatever):
+    if isinstance(whatever, Node):
+        return whatever
+    if isinstance(whatever, ua.NodeId):
+        return Node(isession, whatever)
+    return Node(isession, ua.NodeId(whatever))
+
+
 class Server:
     """
     High level Server class
@@ -412,17 +420,25 @@ class Server:
         await ev_gen.init(etype, emitting_node=emitting_node)
         return ev_gen
 
-    def create_custom_data_type(self, idx, name, basetype=ua.ObjectIds.BaseDataType, properties=None) -> Coroutine:
+    async def create_custom_data_type(self, idx, name, basetype=ua.ObjectIds.BaseDataType, properties=None, description=None) -> Coroutine:
         if properties is None:
             properties = []
-        return self._create_custom_type(idx, name, basetype, properties, [], [])
+        base_t = _get_node(self.iserver.isession, basetype)
 
-    def create_custom_event_type(self, idx, name, basetype=ua.ObjectIds.BaseEventType, properties=None) -> Coroutine:
+        custom_t = await base_t.add_data_type(idx, name, description)
+        for prop in properties:
+            datatype = None
+            if len(prop) > 2:
+                datatype = prop[2]
+            await custom_t.add_property(idx, prop[0], ua.get_default_value(prop[1]), varianttype=prop[1], datatype=datatype)
+        return custom_t
+
+    async def create_custom_event_type(self, idx, name, basetype=ua.ObjectIds.BaseEventType, properties=None) -> Coroutine:
         if properties is None:
             properties = []
-        return self._create_custom_type(idx, name, basetype, properties, [], [])
+        return await self._create_custom_type(idx, name, basetype, properties, [], [])
 
-    def create_custom_object_type(self,
+    async def create_custom_object_type(self,
                                   idx,
                                   name,
                                   basetype=ua.ObjectIds.BaseObjectType,
@@ -435,12 +451,12 @@ class Server:
             variables = []
         if methods is None:
             methods = []
-        return self._create_custom_type(idx, name, basetype, properties, variables, methods)
+        return await self._create_custom_type(idx, name, basetype, properties, variables, methods)
 
     # def create_custom_reference_type(self, idx, name, basetype=ua.ObjectIds.BaseReferenceType, properties=[]):
     # return self._create_custom_type(idx, name, basetype, properties)
 
-    def create_custom_variable_type(self,
+    async def create_custom_variable_type(self,
                                     idx,
                                     name,
                                     basetype=ua.ObjectIds.BaseVariableType,
@@ -453,15 +469,10 @@ class Server:
             variables = []
         if methods is None:
             methods = []
-        return self._create_custom_type(idx, name, basetype, properties, variables, methods)
+        return await self._create_custom_type(idx, name, basetype, properties, variables, methods)
 
     async def _create_custom_type(self, idx, name, basetype, properties, variables, methods):
-        if isinstance(basetype, Node):
-            base_t = basetype
-        elif isinstance(basetype, ua.NodeId):
-            base_t = Node(self.iserver.isession, basetype)
-        else:
-            base_t = Node(self.iserver.isession, ua.NodeId(basetype))
+        base_t = _get_node(self.iserver.isession, basetype)
         custom_t = await base_t.add_object_type(idx, name)
         for prop in properties:
             datatype = None

tests/test_server.py

@@ -173,7 +173,7 @@ async def test_references_for_added_nodes_method(server):
 async def test_get_event_from_type_node_BaseEvent(server):
     """
     This should work for following BaseEvent tests to work
-    (maybe to write it a bit differentlly since they are not independent) 
+    (maybe to write it a bit differentlly since they are not independent)
     """
     ev = await asyncua.common.events.get_event_obj_from_type_node(
         asyncua.Node(server.iserver.isession, ua.NodeId(ua.ObjectIds.BaseEventType))
@@ -333,7 +333,7 @@ async def test_create_custom_data_type_object_id(server):
     type = await server.create_custom_data_type(2, 'MyDataType', ua.ObjectIds.BaseDataType,
                                                 [('PropertyNum', ua.VariantType.Int32),
                                                  ('PropertyString', ua.VariantType.String)])
-    await check_custom_type(type, ua.ObjectIds.BaseDataType, server)
+    await check_custom_type(type, ua.ObjectIds.BaseDataType, server, ua.NodeClass.DataType)
 
 
 async def test_create_custom_event_type_object_id(server):
@@ -596,19 +596,21 @@ def check_custom_event(ev, etype):
     assert ev.Severity == 1
 
 
-async def check_custom_type(type, base_type, server: Server):
+async def check_custom_type(ntype, base_type, server: Server, node_class=None):
     base = asyncua.Node(server.iserver.isession, ua.NodeId(base_type))
-    assert type in await base.get_children()
-    nodes = await type.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse,
+    assert ntype in await base.get_children()
+    nodes = await ntype.get_referenced_nodes(refs=ua.ObjectIds.HasSubtype, direction=ua.BrowseDirection.Inverse,
                                             includesubtypes=True)
     assert base == nodes[0]
-    properties = await type.get_properties()
+    if node_class:
+        assert node_class == await ntype.get_node_class()
+    properties = await ntype.get_properties()
     assert properties is not None
     assert len(properties) == 2
-    assert await type.get_child("2:PropertyNum") in properties
-    assert (await(await type.get_child("2:PropertyNum")).get_data_value()).Value.VariantType == ua.VariantType.Int32
-    assert await type.get_child("2:PropertyString") in properties
-    assert (await(await type.get_child("2:PropertyString")).get_data_value()).Value.VariantType == ua.VariantType.String
+    assert await ntype.get_child("2:PropertyNum") in properties
+    assert (await(await ntype.get_child("2:PropertyNum")).get_data_value()).Value.VariantType == ua.VariantType.Int32
+    assert await ntype.get_child("2:PropertyString") in properties
+    assert (await(await ntype.get_child("2:PropertyString")).get_data_value()).Value.VariantType == ua.VariantType.String
 
 
 """