implement specific errors, catch some errors in __main__

cosmo/__main__.py

@@ -9,7 +9,7 @@
 import argparse
 
 from cosmo.graphqlclient import GraphqlClient
-from cosmo.serializer import RouterSerializer, SwitchSerializer
+from cosmo.serializer import RouterSerializer, SwitchSerializer, AbstractRecoverableError
 
 
 def main() -> int:
@@ -73,14 +73,15 @@ def noop(*args, **kwargs):
         print(f"[INFO] Generating {device_fqdn}")
 
         content = None
-        if device['name'] in cosmo_configuration['devices']['router']:
-            serializer = RouterSerializer(device, cosmo_data['l2vpn_list'], cosmo_data["vrf_list"])
-            content = serializer.serialize()
-        elif device['name'] in cosmo_configuration['devices']['switch']:
-            serializer = SwitchSerializer(device)
-            content = serializer.serialize()
-
-        if not content:
+        try:
+            if device['name'] in cosmo_configuration['devices']['router']:
+                serializer = RouterSerializer(device, cosmo_data['l2vpn_list'], cosmo_data["vrf_list"])
+                content = serializer.serialize()
+            elif device['name'] in cosmo_configuration['devices']['switch']:
+                serializer = SwitchSerializer(device)
+                content = serializer.serialize()
+        except AbstractRecoverableError as e:
+            warnings.warn(f"{device['name']} serialization error \"{e}\", skipping ...")
             continue
 
         match cosmo_configuration['output_format']:

cosmo/serializer.py

@@ -2,8 +2,19 @@
 import re
 import json
 import warnings
+import abc
 from collections import defaultdict
 
+class AbstractRecoverableError(Exception, abc.ABC):
+    pass
+
+class DeviceSerializationError(AbstractRecoverableError):
+    pass
+
+class InterfaceSerializationError(AbstractRecoverableError):
+    pass
+
+
 
 class Tags:
     def __init__(self, tags):
@@ -78,7 +89,7 @@ def __init__(self, device, l2vpn_list, vrfs):
                 self.bmc_interface = "bmc0"
                 self.lo_interface = "lo-0/0/0"
             case other:
-                raise Exception(f"unsupported platform vendor: {other}")
+                raise DeviceSerializationError(f"unsupported platform vendor: {other}")
                 return
 
         self.device = device
@@ -171,8 +182,7 @@ def _get_unit(self, iface):
             # abort if a private IP is used on a unit without a VRF
             # we use !is_global instead of is_private since the latter ignores 100.64/10
             if not iface["vrf"] and not ipa.is_global and not is_mgmt:
-                raise Exception(f"Private IP {ipa} used on interface {iface['name']} in default VRF. Did you forget to configure a VRF?"
-                f"Error while serializing device {self.device['name']}, aborting.")
+                raise InterfaceSerializationError(f"Private IP {ipa} used on interface {iface['name']} in default VRF for device {self.device['name']}. Did you forget to configure a VRF?")
 
             if ipa.version == 4:
                 ipv4Family[ipa.network].add_ip(ipa, is_secondary)

cosmo/tests/integration_test.py

@@ -48,3 +48,12 @@ def test_device_generation_nix(mocker):
     assert cosmoMain() == 0
     testEnv.stop()
     assert os.path.isfile('machines/test0001/generated-cosmo.json')
+
+def test_device_processing_error(mocker):
+    testEnv = utils.CommonSetup(mocker, cfgFile='cosmo/tests/cosmo.devgen_nix.yml')
+    with open(f"cosmo/tests/test_case_vendor_unknown.yaml") as f:
+        utils.RequestResponseMock.patchTool(
+            mocker,{'status_code': 200, 'text': '{"data": ' + json.dumps(yaml.safe_load(f)) + '}'})
+    with pytest.warns(UserWarning, match="unsupported platform vendor"):
+        assert cosmoMain() == 0
+    testEnv.stop()