SIMPLE-7461-consolidate-exceptions-in-virl2-client (#144)

* SIMPLE-7461: initial consolidation of .*NotFound and ElementAlreadyExists exceptions

* SIMPLE-7461: implemented fixes based on review

* minor cleanup

---------

Co-authored-by: Tomas Mikuska <[email protected]>

Author: BregaladTaran

tests/test_image_upload.py

@@ -91,36 +91,26 @@ def windows_path(path: str):
 
 @pytest.mark.parametrize(
     "test_path",
-    [
-        pytest.param(""),
-        pytest.param("/"),
-        pytest.param("./"),
-        pytest.param("./../"),
-        pytest.param("test/test/"),
-        pytest.param("/test/test/"),
-        pytest.param("\\"),
-        pytest.param("..\\..\\"),
-        pytest.param("\\test\\"),
-    ],
+    ["", "/", "./", "./../", "test/test/", "/test/test/", "\\", "..\\..\\", "\\test\\"],
 )
-@pytest.mark.parametrize("usage", [pytest.param("name"), pytest.param("rename")])
+@pytest.mark.parametrize("rename", [None, "rename"])
 @pytest.mark.parametrize(
-    "test_string, message",
-    [pytest.param(test_str, "wrong format") for test_str in WRONG_FORMAT_LIST]
-    + [pytest.param(test_str, "not supported") for test_str in NOT_SUPPORTED_LIST]
-    + [pytest.param(test_str, "") for test_str in EXPECTED_PASS_LIST],
+    "test_string",
+    WRONG_FORMAT_LIST + NOT_SUPPORTED_LIST + EXPECTED_PASS_LIST,
 )
-def test_image_upload_file(usage: str, test_string: str, message: str, test_path: str):
+def test_image_upload_file(rename: str, test_string: str, test_path: str):
     session = MagicMock()
     nid = NodeImageDefinitions(session)
-    rename = None
     filename = test_path + test_string
 
-    if message in ("wrong format", "not supported"):
-        with pytest.raises(InvalidImageFile, match=message):
+    if test_string in WRONG_FORMAT_LIST:
+        with pytest.raises(InvalidImageFile, match="wrong format"):
+            with windows_path(filename):
+                nid.upload_image_file(filename, rename)
+    elif test_string in NOT_SUPPORTED_LIST:
+        with pytest.raises(InvalidImageFile, match="unsupported extension"):
             with windows_path(filename):
                 nid.upload_image_file(filename, rename)
-
     elif test_path == "test_data/":
         with windows_path(filename):
             nid.upload_image_file(filename, rename)
@@ -132,7 +122,6 @@ def test_image_upload_file(usage: str, test_string: str, message: str, test_path
         assert isinstance(file, BufferedReader)
         assert pathlib.Path(file.name).resolve() == pathlib.Path(filename).resolve()
         file.close()
-
     else:
         with pytest.raises(FileNotFoundError):
             with windows_path(filename):

virl2_client/exceptions.py

@@ -60,10 +60,6 @@ class LabNotFound(ElementNotFound):
     pass
 
 
-class DesynchronizedError(VirlException):
-    pass
-
-
 class InvalidContentType(VirlException):
     pass
 

virl2_client/models/cl_pyats.py

@@ -91,7 +91,7 @@ def _load_pyats_testbed(self, testbed_yaml: str) -> Testbed:
         Load a PyATS testbed instance from YAML representation.
 
         Disable all templating features of PyATS markup processor.
-        Also disable extensions loading (which still uses all of the templating)
+        Also disable extensions loading (which still uses all the templating)
         https://pubhub.devnetcloud.com/media/pyats/docs/utilities/yaml_markup.html
         """
         processor = _PyatsTMProcessor(

virl2_client/models/lab.py

@@ -566,7 +566,7 @@ def get_smart_annotation_by_tag(self, tag: str) -> SmartAnnotation:
         for annotation in self._smart_annotations.values():
             if tag == annotation.tag:
                 return annotation
-        raise SmartAnnotationNotFound
+        raise SmartAnnotationNotFound(tag)
 
     def find_nodes_by_tag(self, tag: str) -> list[Node]:
         """
@@ -979,7 +979,7 @@ def create_interface(
         if desired_interface is None:
             # Shouldn't happen, but type checkers complain about desired_interface
             # possibly being None otherwise
-            raise InterfaceNotFound
+            raise InterfaceNotFound(node.id)
 
         return desired_interface
 
@@ -1385,7 +1385,7 @@ def _sync_topology(self, exclude_configurations: bool | None = False) -> None:
                 and f"Lab not found: {self._id}" in exc.response.text
             ):
                 self._stale = True
-                raise LabNotFound(f"Error syncing lab: {error_msg}")
+                raise LabNotFound(self._id)
             raise
 
         topology = result.json()
@@ -1456,7 +1456,7 @@ def _handle_import_nodes(self, topology: dict) -> None:
             node_id = node["id"]
 
             if node_id in self._nodes:
-                raise ElementAlreadyExists("Node already exists")
+                raise ElementAlreadyExists(node_id)
 
             self._import_node(node_id, node)
 
@@ -1468,7 +1468,7 @@ def _handle_import_nodes(self, topology: dict) -> None:
                 iface_id = iface["id"]
 
                 if iface_id in self._interfaces:
-                    raise ElementAlreadyExists("Interface already exists")
+                    raise ElementAlreadyExists(iface_id)
 
                 self._import_interface(iface_id, node_id, iface)
 
@@ -1487,7 +1487,7 @@ def _handle_import_interfaces(self, topology: dict) -> None:
             node_id = iface["node"]
 
             if iface_id in self._interfaces:
-                raise ElementAlreadyExists("Interface already exists")
+                raise ElementAlreadyExists(iface_id)
 
             self._import_interface(iface_id, node_id, iface)
 
@@ -1502,7 +1502,7 @@ def _handle_import_links(self, topology: dict) -> None:
             link_id = link["id"]
 
             if link_id in self._links:
-                raise ElementAlreadyExists("Link already exists")
+                raise ElementAlreadyExists(link_id)
 
             iface_a_id = link["interface_a"]
             iface_b_id = link["interface_b"]
@@ -1524,7 +1524,7 @@ def _handle_import_annotations(self, topology: dict) -> None:
             annotation_id = annotation["id"]
 
             if annotation_id in self._annotations:
-                raise ElementAlreadyExists("Annotation already exists")
+                raise ElementAlreadyExists(annotation_id)
 
             self._import_annotation(annotation_id, annotation)
 
@@ -1535,7 +1535,7 @@ def _handle_import_annotations(self, topology: dict) -> None:
             smart_annotation_id = smart_annotation["id"]
 
             if smart_annotation_id in self._smart_annotations:
-                raise ElementAlreadyExists("Smart annotation already exists")
+                raise ElementAlreadyExists(smart_annotation_id)
 
             self._import_smart_annotation(smart_annotation_id, smart_annotation)
 
@@ -1970,7 +1970,7 @@ def _find_link_in_topology(link_id: str, topology: dict) -> dict:
             if link["id"] == link_id:
                 return link
         # if it cannot be found, it is an internal structure error
-        raise LinkNotFound
+        raise LinkNotFound(link_id)
 
     @staticmethod
     def _find_interface_in_topology(interface_id: str, topology: dict) -> dict:
@@ -1990,7 +1990,7 @@ def _find_interface_in_topology(interface_id: str, topology: dict) -> dict:
                 if interface["id"] == interface_id:
                     return interface
         # if it cannot be found, it is an internal structure error
-        raise InterfaceNotFound
+        raise InterfaceNotFound(interface_id)
 
     @staticmethod
     def _find_node_in_topology(node_id: str, topology: dict) -> dict:
@@ -2006,7 +2006,7 @@ def _find_node_in_topology(node_id: str, topology: dict) -> dict:
             if node["id"] == node_id:
                 return node
         # if it cannot be found, it is an internal structure error
-        raise NodeNotFound
+        raise NodeNotFound(node_id)
 
     @staticmethod
     def _find_annotation_in_topology(
@@ -2025,7 +2025,7 @@ def _find_annotation_in_topology(
             if annotation["id"] == annotation_id:
                 return annotation
         # if it cannot be found, it is an internal structure error
-        raise AnnotationNotFound
+        raise AnnotationNotFound(annotation_id)
 
     @staticmethod
     def _find_smart_annotation_in_topology(
@@ -2044,7 +2044,7 @@ def _find_smart_annotation_in_topology(
             if annotation["id"] == annotation_id:
                 return annotation
         # if it cannot be found, it is an internal structure error
-        raise SmartAnnotationNotFound
+        raise SmartAnnotationNotFound(annotation_id)
 
     @check_stale
     def get_pyats_testbed(self, hostname: str | None = None) -> str:

virl2_client/models/node.py

@@ -578,7 +578,7 @@ def get_interface_by_label(self, label: str) -> Interface:
         for iface in self.interfaces():
             if iface.label == label:
                 return iface
-        raise InterfaceNotFound(f"{label}:{self}")
+        raise InterfaceNotFound(label)
 
     @locked
     def get_interface_by_slot(self, slot: int) -> Interface:
@@ -592,7 +592,7 @@ def get_interface_by_slot(self, slot: int) -> Interface:
         for iface in self.interfaces():
             if iface.slot == slot:
                 return iface
-        raise InterfaceNotFound(f"{slot}:{self}")
+        raise InterfaceNotFound(slot)
 
     def get_links_to(self, other_node: Node) -> list[Link]:
         """

virl2_client/models/node_image_definition.py

@@ -226,21 +226,20 @@ def upload_image_file(
 
         if extension == "" or name == "":
             message = (
-                f"Name specified ({name}) is in wrong format "
-                f"(correct: filename.({'|'.join(extension_list)}) )."
+                f"Specified filename ({name}) has wrong format "
+                f"(correct format is filename.({'|'.join(extension_list)}) )."
             )
             raise InvalidImageFile(message)
 
         if extension not in extension_list and last_ext not in extension_list:
             message = (
-                f"Extension in {name} not supported. "
+                f"Specified filename ({name}) has unsupported extension ({extension}) "
                 f"(supported extensions are {', '.join(extension_list)})."
             )
             raise InvalidImageFile(message)
 
         if not os.path.exists(filename):
-            message = f"File with specified name ({filename}) does not exist."
-            raise FileNotFoundError(message)
+            raise FileNotFoundError(filename)
 
         print(f"Uploading {name}")
         headers = {"X-Original-File-Name": name}

virl2_client/models/resource_pool.py

@@ -112,7 +112,7 @@ def get_resource_pools_by_ids(
         :param resource_pool_ids: A resource pool ID or an iterable of IDs.
         :returns: A ResourcePool object when one ID is passed, or a dictionary of IDs to
             either ResourcePools when a resource pool exists or None when it doesn't.
-        :raises KeyError: When one ID is passed and it doesn't exist.
+        :raises KeyError: When one ID is passed, and it doesn't exist.
         """
         self.sync_resource_pools_if_outdated()
         if isinstance(resource_pool_ids, str):

virl2_client/utils.py

@@ -60,17 +60,6 @@ def _make_not_found(instance: Element) -> ElementNotFound:
     class_name = type(instance).__name__
     if class_name.startswith("Annotation"):
         class_name = "Annotation"
-    instance_id = instance._id
-    if class_name == "Lab":
-        instance_label = instance._title
-    elif class_name.startswith("Annotation"):
-        instance_label = instance._type
-    else:
-        instance_label = instance._label
-
-    error_text = (
-        f"{class_name} {instance_label} ({instance_id}) no longer exists on the server."
-    )
     error: Type[ElementNotFound] = {
         "Lab": LabNotFound,
         "Node": NodeNotFound,
@@ -79,15 +68,10 @@ def _make_not_found(instance: Element) -> ElementNotFound:
         "Annotation": AnnotationNotFound,
         "SmartAnnotation": SmartAnnotationNotFound,
     }[class_name]
-    return error(error_text)
+    return error(instance._id)
 
 
-def _check_and_mark_stale(
-    func: Callable,
-    instance: Element,
-    *args,
-    **kwargs,
-):
+def _check_and_mark_stale(func: Callable, instance: Element, *args, **kwargs):
     """
     Check staleness before and after calling `func`
     and updates staleness if a 404 is raised.
@@ -107,7 +91,6 @@ def _check_and_mark_stale(
         if instance._stale:
             raise _make_not_found(instance)
         return ret
-
     except httpx.HTTPStatusError as exc:
         resp = exc.response
         class_name = type(instance).__name__

virl2_client/virl2_client.py

@@ -596,8 +596,7 @@ def import_lab_from_path(self, path: str, title: str | None = None) -> Lab:
         """
         topology_path = Path(path)
         if not topology_path.exists():
-            message = f"{path} can not be found"
-            raise FileNotFoundError(message)
+            raise FileNotFoundError(path)
 
         topology = topology_path.read_text()
         return self.import_lab(
@@ -789,7 +788,7 @@ def join_existing_lab(self, lab_id: str, sync_lab: bool = True) -> Lab:
                 topology = self._session.get(url).json()
             except httpx.HTTPStatusError as exc:
                 if exc.response.status_code == 404:
-                    raise LabNotFound("No lab with the given ID exists on the host.")
+                    raise LabNotFound(lab_id)
                 raise
             title = topology.get("lab", {}).get("title")
         else: