allow setting StatusCode

asyncua/server/address_space.py

@@ -780,8 +780,19 @@ class AddressSpace:
         attval = node.attributes.get(attr, None)
         if attval is None:
             return ua.StatusCode(ua.StatusCodes.BadAttributeIdInvalid)
-
-        if not self._is_expected_variant_type(value, attval, node):
+        if value.StatusCode is not None and not value.StatusCode.is_good():
+            # https://reference.opcfoundation.org/v104/Core/docs/Part4/7.7.1/
+            # If the StatusCode indicates an error then the value is to be ignored and the Server shall set it to null.
+            value = ua.DataValue(
+                ua.Variant(ua.Null, ua.VariantType.Null),
+                value.StatusCode,
+                value.ServerTimestamp,
+                value.SourceTimestamp,
+                value.ServerPicoseconds,
+                value.SourcePicoseconds
+            )
+        elif not self._is_expected_variant_type(value, attval, node):
+            # Only check datatype if no bad StatusCode is set
             return ua.StatusCode(ua.StatusCodes.BadTypeMismatch)
 
         old = attval.value

tests/test_common.py

@@ -642,6 +642,18 @@ async def test_write_value(opc):
     await opc.opc.delete_nodes([v])
 
 
+async def test_write_value_statuscode_bad(opc):
+    o = opc.opc.nodes.objects
+    var = ua.Variant('Some value that should not be set!')
+    dvar = ua.DataValue(var, StatusCode_=ua.StatusCode(ua.StatusCodes.BadDeviceFailure))
+    v = await o.add_variable(3, 'VariableValueBad', var)
+    await v.write_value(dvar)
+    with pytest.raises(ua.UaStatusCodeError) as error_read:
+        await v.read_data_value()
+    assert error_read.type.code == dvar.StatusCode.value
+    await opc.opc.delete_nodes([v])
+
+
 async def test_array_value(opc):
     o = opc.opc.nodes.objects
     v = await o.add_variable(3, 'VariableArrayValue', [1, 2, 3])
@@ -1014,7 +1026,7 @@ async def test_instantiate_abstract(opc):
     finit_statemachine_type = opc.opc.get_node("ns=0;i=2771")  # IsAbstract=True
     with pytest.raises(ua.UaError):
         node = await instantiate(opc.opc.nodes.objects, finit_statemachine_type, bname="2:TestFiniteStateMachine")
-        
+
 
 async def test_variable_with_datatype(opc):
     v1 = await opc.opc.nodes.objects.add_variable(