load_enums: skip bad enum definitions with error message (#1384)

* load_enums: On unexpected errors, skip entry with error message

* fix test if not run in isolation

asyncua/common/structures104.py

@@ -535,6 +535,7 @@ class {name}({enum_type}):
 
 
 async def load_enums(server: Union["Server", "Client"], base_node: Node = None, option_set: bool = False) -> Dict:
+    typename = "OptionSet" if option_set else "Enum"
     if base_node is None:
         base_node = server.nodes.enum_data_type
     new_enums = {}
@@ -542,11 +543,15 @@ async def load_enums(server: Union["Server", "Client"], base_node: Node = None,
         name = clean_name(desc.BrowseName.Name)
         if hasattr(ua, name):
             continue
-        _logger.info("Registering Enum %s %s OptionSet=%s", desc.NodeId, name, option_set)
-        edef = await _read_data_type_definition(server, desc)
-        if not edef:
+        _logger.info("Registering %s %s %s", typename, desc.NodeId, name)
+        try:
+            edef = await _read_data_type_definition(server, desc)
+            if not edef:
+                continue
+            env = await _generate_object(name, edef, enum=True, option_set=option_set, log_fail=False)
+        except Exception:
+            _logger.exception("%s %s (NodeId: %s): Failed to generate class from UA datatype", typename, name, desc.NodeId)
             continue
-        env = await _generate_object(name, edef, enum=True, option_set=option_set)
         ua.register_enum(name, desc.NodeId, env[name])
         new_enums[name] = env[name]
     return new_enums

tests/test_common.py

@@ -1247,6 +1247,30 @@ async def test_custom_enum_x(opc):
     assert val == 1
 
 
+async def test_custom_enum_with_bad_dtd(opc, caplog):
+    """Enum with invalid type definition is skipped and an error is logged"""
+    caplog.set_level("ERROR")
+    idx = 4
+    dtype = await opc.opc.nodes.enum_data_type.add_data_type(idx, "MyEnum")
+    # Set some garbage as type definition
+    await dtype.write_data_type_definition(ua.StructureField())
+
+    # If leftover from previous test, it would be skipped, clear out
+    try:
+        del ua.MyEnum
+    except AttributeError:
+        pass
+    # Does not raise
+    await opc.opc.load_data_type_definitions()
+    # Check that error was logged
+    logs = [t for t in caplog.record_tuples if t[0].endswith(".structures104")]
+    assert len(logs) == 1
+    message = logs[0][2]
+    assert "MyEnum" in message
+    assert "Failed to generate class from UA datatype" in message
+
+
+
 async def test_custom_option_set(opc):
     idx = 4
     await new_enum(opc.opc, idx, "MyOptionSet", ["tata", "titi", "toto", "None"], True)