Minor optimizations to logic

gns3server/api/routes/controller/nodes.py

@@ -105,6 +105,16 @@ async def dep_node(node_id: UUID, project: Project = Depends(dep_project)) -> No
     return node
 
 
+def _check_node_type(node: Node, *required_types: str) -> None:
+    """
+    Raise ControllerBadRequestError if node is not one of the required types.
+    """
+
+    if node.node_type not in required_types:
+        type_str = "/".join(required_types)
+        raise ControllerBadRequestError(f"This endpoint is only supported on a {type_str} node")
+
+
 @router.post(
     "",
     status_code=status.HTTP_201_CREATED,
@@ -407,8 +417,7 @@ async def auto_idlepc(node: Node = Depends(dep_node)) -> dict:
     Required privilege: Node.Audit
     """
 
-    if node.node_type != "dynamips":
+    _check_node_type(node, "dynamips")
-        raise ControllerBadRequestError("Auto Idle-PC is only supported on a Dynamips node")
     return await node.dynamips_auto_idlepc()
 
 
@@ -420,8 +429,7 @@ async def idlepc_proposals(node: Node = Depends(dep_node)) -> List[str]:
     Required privilege: Node.Audit
     """
 
-    if node.node_type != "dynamips":
+    _check_node_type(node, "dynamips")
-        raise ControllerBadRequestError("Idle-PC proposals is only supported on a Dynamips node")
     return await node.dynamips_idlepc_proposals()
 
 
@@ -441,8 +449,7 @@ async def create_disk_image(
     Required privilege: Node.Allocate
     """
 
-    if node.node_type != "qemu":
+    _check_node_type(node, "qemu")
-        raise ControllerBadRequestError("Creating a disk image is only supported on a Qemu node")
     await node.post(f"/disk_image/{disk_name}", data=disk_data.model_dump(exclude_unset=True))
 
 
@@ -462,8 +469,7 @@ async def update_disk_image(
     Required privilege: Node.Allocate
     """
 
-    if node.node_type != "qemu":
+    _check_node_type(node, "qemu")
-        raise ControllerBadRequestError("Updating a disk image is only supported on a Qemu node")
     await node.put(f"/disk_image/{disk_name}", data=disk_data.model_dump(exclude_unset=True))
 
 
@@ -482,8 +488,7 @@ async def delete_disk_image(
     Required privilege: Node.Allocate
     """
 
-    if node.node_type != "qemu":
+    _check_node_type(node, "qemu")
-        raise ControllerBadRequestError("Deleting a disk image is only supported on a Qemu node")
     await node.delete(f"/disk_image/{disk_name}")
 
 

gns3server/controller/export_project.py

@@ -289,7 +289,7 @@ def _export_local_image(image, zstream):
             # Some modules don't have images
             continue
 
-        directory = os.path.split(images_directory)[-1:][0]
+        directory = os.path.basename(images_directory)
         if os.path.exists(image):
             path = image
         else:
@@ -309,7 +309,7 @@ async def _export_remote_images(project, compute_id, image_type, image, project_
 
     log.debug(f"Downloading image '{image}' from compute '{compute_id}'")
     try:
-        compute = [compute for compute in project.computes if compute.id == compute_id][0]
+        compute = next(c for c in project.computes if c.id == compute_id)
     except IndexError:
         raise ControllerNotFoundError(f"Cannot export image from '{compute_id}' compute. Compute doesn't exist.")
 

gns3server/controller/topology.py

@@ -175,7 +175,8 @@ def load_topology(path):
     # Version GNS3 2.2 dev (for project created with 2.2dev).
     # Appliance ID has been replaced by Template ID
     if topo["revision"] == 9:
-        for node in topo.get("topology", {}).get("nodes", []):
+        nodes = topo.get("topology", {}).get("nodes", [])
+        for node in nodes:
             if "appliance_id" in node:
                 node["template_id"] = node["appliance_id"]
                 del node["appliance_id"]
@@ -218,7 +219,8 @@ def _convert_2_2_0(topo, topo_path):
 
     topo["revision"] = 10
 
-    for node in topo.get("topology", {}).get("nodes", []):
+    nodes = topo.get("topology", {}).get("nodes", [])
+    for node in nodes:
         if "properties" in node:
             if node["node_type"] in ("qemu", "docker") and not is_rfc1123_hostname_valid(node["name"]):
                 new_name = to_rfc1123_hostname(node["name"])
@@ -246,7 +248,8 @@ def _convert_2_1_0(topo, topo_path):
         # to avoid overlapping grids
         topo["drawing_grid_size"] = topo["grid_size"]
 
-    for node in topo.get("topology", {}).get("nodes", []):
+    nodes = topo.get("topology", {}).get("nodes", [])
+    for node in nodes:
         # make sure console_type is not None but "none" string
         if "console_type" in node and node["console_type"] is None:
             node["console_type"] = "none"
@@ -272,7 +275,8 @@ def _convert_2_0_0(topo, topo_path):
     """
     topo["revision"] = 8
 
-    for node in topo.get("topology", {}).get("nodes", []):
+    nodes = topo.get("topology", {}).get("nodes", [])
+    for node in nodes:
         if "properties" in node:
             if node["node_type"] == "vpcs":
                 if "startup_script_path" in node["properties"]:
@@ -301,7 +305,8 @@ def _convert_2_0_0_beta_2(topo, topo_path):
     topo_dir = os.path.dirname(topo_path)
     topo["revision"] = 7
 
-    for node in topo.get("topology", {}).get("nodes", []):
+    nodes = topo.get("topology", {}).get("nodes", [])
+    for node in nodes:
         if node["node_type"] == "dynamips":
             node_id = node["node_id"]
             dynamips_id = node["properties"]["dynamips_id"]
@@ -328,7 +333,8 @@ def _convert_2_0_0_alpha(topo, topo_path):
      * No more option for VMware / VirtualBox remote console (always use telnet)
     """
     topo["revision"] = 6
-    for node in topo.get("topology", {}).get("nodes", []):
+    nodes = topo.get("topology", {}).get("nodes", [])
+    for node in nodes:
         if node.get("console_type") == "serial":
             node["console_type"] = "telnet"
         if node["node_type"] in ("vmware", "virtualbox"):