replace assertions with a dedicated UaError

asyncua/common/connection.py

@@ -5,6 +5,7 @@
 import copy
 
 from asyncua import ua
+from asyncua.ua.uaerrors import UaInvalidParameterError
 from ..ua.ua_binary import struct_from_binary, struct_to_binary, header_from_binary, header_to_binary
 
 try:
@@ -340,7 +341,8 @@ def _check_sym_header(self, security_hdr):
         Validates the symmetric header of the message chunk and revolves the
         security token if needed.
         """
-        assert isinstance(security_hdr, ua.SymmetricAlgorithmHeader), f"Expected SymAlgHeader, got: {type(security_hdr)}"
+        if not isinstance(security_hdr, ua.SymmetricAlgorithmHeader):
+            raise UaInvalidParameterError(f"Expected SymAlgHeader, got: {type(security_hdr)}")
 
         if security_hdr.TokenId == self.security_token.TokenId:
             return

asyncua/common/copy_node_util.py

@@ -6,6 +6,7 @@
 import asyncua
 from asyncua import ua
 from asyncua.common.session_interface import AbstractSession
+from asyncua.ua.uaerrors import UaInvalidParameterError
 from .node_factory import make_node
 
 
@@ -55,7 +56,8 @@ async def _rdesc_from_node(parent: asyncua.Node, node: asyncua.Node) -> ua.Refer
     variants: List[ua.Variant] = []
     for res in results:
         res.StatusCode.check()
-        assert res.Value is not None, "Value must not be None if the result is in Good status"
+        if res.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         variants.append(res.Value)
     nclass, qname, dname = [v.Value for v in variants]
     rdesc = ua.ReferenceDescription()
@@ -82,7 +84,8 @@ async def _read_and_copy_attrs(node_type: asyncua.Node, struct: Any, addnode: ua
     for idx, name in enumerate(names):
         if results[idx].StatusCode.is_good():
             variant = results[idx].Value
-            assert variant is not None, "Value must not be None if the result is in Good status"
+            if variant is None:
+                raise UaInvalidParameterError("Value must not be None if the result is in Good status")
             if name == "Value":
                 setattr(struct, name, variant)
             else:

asyncua/common/node.py

@@ -15,6 +15,7 @@
 
 from asyncua import ua
 from asyncua.common.session_interface import AbstractSession
+from asyncua.ua.uaerrors import UaInvalidParameterError
 from .ua_utils import value_to_datavalue
 
 from .events import Event, get_filter_from_event_type
@@ -91,23 +92,26 @@ async def read_browse_name(self) -> ua.QualifiedName:
         composed of a string(name) and a namespace index.
         """
         result = await self.read_attribute(ua.AttributeIds.BrowseName)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return result.Value.Value
 
     async def read_display_name(self) -> ua.LocalizedText:
         """
         get DisplayName attribute of node
         """
         result = await self.read_attribute(ua.AttributeIds.DisplayName)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return result.Value.Value
 
     async def read_data_type(self) -> ua.NodeId:
         """
         get data type of node as NodeId
         """
         result = await self.read_attribute(ua.AttributeIds.DataType)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return result.Value.Value
 
     async def read_data_type_as_variant_type(self) -> ua.VariantType:
@@ -117,31 +121,35 @@ async def read_data_type_as_variant_type(self) -> ua.VariantType:
         may not be convertible to VariantType
         """
         result = await self.read_attribute(ua.AttributeIds.DataType)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return await data_type_to_variant_type(Node(self.session, result.Value.Value))
 
     async def get_access_level(self) -> Set[ua.AccessLevel]:
         """
         Get the access level attribute of the node as a set of AccessLevel enum values.
         """
         result = await self.read_attribute(ua.AttributeIds.AccessLevel)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return ua.AccessLevel.parse_bitfield(result.Value.Value)
 
     async def get_user_access_level(self) -> Set[ua.AccessLevel]:
         """
         Get the user access level attribute of the node as a set of AccessLevel enum values.
         """
         result = await self.read_attribute(ua.AttributeIds.UserAccessLevel)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return ua.AccessLevel.parse_bitfield(result.Value.Value)
 
     async def read_event_notifier(self) -> Set[ua.EventNotifier]:
         """
         Get the event notifier attribute of the node as a set of EventNotifier enum values.
         """
         result = await self.read_attribute(ua.AttributeIds.EventNotifier)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return ua.EventNotifier.parse_bitfield(result.Value.Value)
 
     async def set_event_notifier(self, values) -> None:
@@ -158,7 +166,8 @@ async def read_node_class(self) -> ua.NodeClass:
         get node class attribute of node
         """
         result = await self.read_attribute(ua.AttributeIds.NodeClass)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return ua.NodeClass(result.Value.Value)
 
     async def read_data_type_definition(self) -> ua.DataTypeDefinition:
@@ -167,7 +176,8 @@ async def read_data_type_definition(self) -> ua.DataTypeDefinition:
         only DataType nodes following spec >= 1.04 have that attribute
         """
         result = await self.read_attribute(ua.AttributeIds.DataTypeDefinition)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return result.Value.Value
 
     async def write_data_type_definition(self, sdef: ua.DataTypeDefinition) -> None:
@@ -183,7 +193,8 @@ async def read_description(self) -> ua.LocalizedText:
         get description attribute class of node
         """
         result = await self.read_attribute(ua.AttributeIds.Description)
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return result.Value.Value
 
     async def read_value(self) -> Any:
@@ -194,7 +205,8 @@ async def read_value(self) -> Any:
         Do not modify it if it is a mutable object unless you know what you are doing
         """
         result = await self.read_data_value()
-        assert result.Value is not None, "Value must not be None if the result is in Good status"
+        if result.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return result.Value.Value
 
     get_value = read_value  # legacy compatibility
@@ -220,7 +232,8 @@ async def read_array_dimensions(self) -> int:
         Read and return ArrayDimensions attribute of node
         """
         res = await self.read_attribute(ua.AttributeIds.ArrayDimensions)
-        assert res.Value is not None, "Value must not be None if the result is in Good status"
+        if res.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return res.Value.Value
 
     async def write_value_rank(self, value: int) -> None:
@@ -235,7 +248,8 @@ async def read_value_rank(self) -> int:
         Read and return ValueRank attribute of node
         """
         res = await self.read_attribute(ua.AttributeIds.ValueRank)
-        assert res.Value is not None, "Value must not be None if the result is in Good status"
+        if res.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         return ua.ValueRank(res.Value.Value)
 
     async def write_value(self, value: Any, varianttype: Optional[ua.VariantType] = None) -> None:
@@ -271,13 +285,15 @@ async def set_writable(self, writable: bool = True) -> None:
 
     async def set_attr_bit(self, attr: ua.AttributeIds, bit: int) -> None:
         dv = await self.read_attribute(attr)
-        assert dv.Value is not None, "Value must not be None if the result is in Good status"
+        if dv.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         val = ua.ua_binary.set_bit(dv.Value.Value, bit)
         await self.write_attribute(attr, ua.DataValue(ua.Variant(val, dv.Value.VariantType)))
 
     async def unset_attr_bit(self, attr: ua.AttributeIds, bit: int) -> None:
         dv = await self.read_attribute(attr)
-        assert dv.Value is not None, "Value must not be None if the result is in Good status"
+        if dv.Value is None:
+            raise UaInvalidParameterError("Value must not be None if the result is in Good status")
         val = ua.ua_binary.unset_bit(dv.Value.Value, bit)
         await self.write_attribute(attr, ua.DataValue(ua.Variant(val, dv.Value.VariantType)))
 

asyncua/common/structures104.py

@@ -13,6 +13,7 @@
 from asyncua import ua
 from asyncua import Node
 from asyncua.common.manage_nodes import create_encoding, create_data_type
+from asyncua.ua.uaerrors import UaInvalidParameterError
 if TYPE_CHECKING:
     from asyncua import Client, Server
 
@@ -374,7 +375,8 @@ async def _get_parent_types(node: Node):
 
 async def load_custom_struct(node: Node) -> Any:
     sdef = await node.read_data_type_definition()
-    assert isinstance(sdef, ua.StructureDefinition), f"Expected StructureDefinition, got: {type(sdef)}"
+    if not isinstance(sdef, ua.StructureDefinition):
+        raise UaInvalidParameterError(f"Expected StructureDefinition, got: {type(sdef)}")
     name = (await node.read_browse_name()).Name
     for parent in await _get_parent_types(node):
         parent_sdef = await parent.read_data_type_definition()

asyncua/common/xmlexporter.py

@@ -17,7 +17,7 @@
 import asyncua
 from asyncua import ua
 from asyncua.ua.uatypes import type_string_from_type
-from asyncua.ua.uaerrors import UaError
+from asyncua.ua.uaerrors import UaError, UaInvalidParameterError
 from .. import Node
 from ..ua import object_ids as o_ids
 from .ua_utils import get_base_data_type
@@ -451,7 +451,8 @@ async def _value_to_etree(self, el: Et.Element, type_name: str, dtype: ua.NodeId
             return
 
         if isinstance(val, (list, tuple)):
-            assert isinstance(dtype.Identifier, int), f"Expected int, got {type(dtype.Identifier)}"
+            if not isinstance(dtype.Identifier, int):
+                raise UaInvalidParameterError(f"Expected int, got {type(dtype.Identifier)}")
             if dtype.NamespaceIndex == 0 and dtype.Identifier <= 21:
                 elname = "uax:ListOf" + type_name
             else:  # this is an extensionObject:

asyncua/ua/uaerrors/_base.py

@@ -103,3 +103,7 @@ class UaStringParsingError(UaError):
 
 class UaStructParsingError(UaError):
     pass
+
+
+class UaInvalidParameterError(UaError):
+    pass