attempt to assess type of nodes having a Null value deep in address space

asyncua/server/address_space.py

@@ -706,11 +706,7 @@ class AddressSpace:
         if attval is None:
             return ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid)
 
-        if value.Value.VariantType != attval.value.Value.VariantType:
-            if value.Value.VariantType == ua.VariantType.Null or  attval.value.Value.VariantType == ua.VariantType.Null:
-                pass
-            else:
-                _logger.critical("Write refused: Variant: %s with type %s does not have expected type: %s", value.Value, value.Value.VariantType, attval.value.Value.VariantType)
+        if not self._is_expected_variant_type(value, attval, node):
             return ua.StatusCode(ua.StatusCodes.BadTypeMismatch)
 
         old = attval.value
@@ -728,6 +724,27 @@ class AddressSpace:
 
         return ua.StatusCode()
 
+    def _is_expected_variant_type(self, value, attval, node):
+        if value.Value.VariantType == ua.VariantType.Null:
+            # we accept overwrite with Null, not sure if this is OK in spec...
+            return True
+        vtype = attval.value.Value.VariantType
+        if vtype == ua.VariantType.Null:
+            # Node had a null value, many nodes are initialized with that value
+            # we should check what the real type is
+            dtype = node.attributes[ua.AttributeIds.DataType].value.Value.Value
+            if dtype.NamespaceIndex == 0 and dtype.Identifier <= 25:
+                vtype = ua.VariantType(dtype.Identifier)
+            else:
+                # FIXME: should find the correct variant type given data type but
+                # this is a bit complicaed so trusting the first write
+                return True
+        if value.Value.VariantType == vtype:
+            return True
+        _logger.critical("Write refused: Variant: %s with type %s does not have expected type: %s",
+                value.Value, value.Value.VariantType, attval.value.Value.VariantType)
+        return False
+
     def add_datachange_callback(self, nodeid, attr, callback):
         self.logger.debug("set attr callback: %s %s %s", nodeid, attr, callback)
         if nodeid not in self._nodes: