fix(typing): resolve mypy errors in api.routes.compute.iou_nodes

This commit is contained in:
Mark Paronyan 2026-09-28 05:50:35 +00:00
parent b9192438c9
commit 9218f3fc72
No known key found for this signature in database
GPG Key ID: 8DC7338A87FE22F8
2 changed files with 18 additions and 17 deletions

View File

@ -23,7 +23,7 @@ import os
from fastapi import APIRouter, WebSocket, Depends, Body, status, HTTPException from fastapi import APIRouter, WebSocket, Depends, Body, status, HTTPException
from fastapi.encoders import jsonable_encoder from fastapi.encoders import jsonable_encoder
from fastapi.responses import StreamingResponse from fastapi.responses import StreamingResponse
from typing import Union from typing import Any, Union
from uuid import UUID from uuid import UUID
from gns3server import schemas from gns3server import schemas
@ -32,7 +32,9 @@ from gns3server.compute.iou.iou_vm import IOUVM
from .dependencies.authentication import compute_authentication, ws_compute_authentication from .dependencies.authentication import compute_authentication, ws_compute_authentication
responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or IOU node"}} responses: dict[int | str, dict[str, Any]] = {
404: {"model": schemas.ErrorMessage, "description": "Could not find project or IOU node"}
}
router = APIRouter(responses=responses) router = APIRouter(responses=responses)
@ -60,18 +62,18 @@ async def create_iou_node(project_id: UUID, node_data: schemas.IOUCreate) -> sch
""" """
iou = IOU.instance() iou = IOU.instance()
node_data = jsonable_encoder(node_data, exclude_unset=True) data = jsonable_encoder(node_data, exclude_unset=True)
vm = await iou.create_node( vm = await iou.create_node(
node_data.pop("name"), data.pop("name"),
str(project_id), str(project_id),
node_data.get("node_id"), data.get("node_id"),
application_id=node_data.get("application_id"), application_id=data.get("application_id"),
path=node_data.get("path"), path=data.get("path"),
console=node_data.get("console"), console=data.get("console"),
console_type=node_data.get("console_type", "telnet"), console_type=data.get("console_type", "telnet"),
) )
for name, value in node_data.items(): for name, value in data.items():
if hasattr(vm, name) and getattr(vm, name) != value: if hasattr(vm, name) and getattr(vm, name) != value:
if name == "application_id": if name == "application_id":
continue # we must ignore this to avoid overwriting the application_id allocated by the controller continue # we must ignore this to avoid overwriting the application_id allocated by the controller
@ -79,7 +81,7 @@ async def create_iou_node(project_id: UUID, node_data: schemas.IOUCreate) -> sch
continue continue
if name == "private_config_content" and (vm.private_config_content and len(vm.private_config_content) > 0): if name == "private_config_content" and (vm.private_config_content and len(vm.private_config_content) > 0):
continue continue
if node_data.get("use_default_iou_values") and (name == "ram" or name == "nvram"): if data.get("use_default_iou_values") and (name == "ram" or name == "nvram"):
continue continue
setattr(vm, name, value) setattr(vm, name, value)
return vm.asdict() return vm.asdict()
@ -100,8 +102,8 @@ async def update_iou_node(node_data: schemas.IOUUpdate, node: IOUVM = Depends(de
Update an IOU node. Update an IOU node.
""" """
node_data = jsonable_encoder(node_data, exclude_unset=True) data = jsonable_encoder(node_data, exclude_unset=True)
for name, value in node_data.items(): for name, value in data.items():
if hasattr(node, name) and getattr(node, name) != value: if hasattr(node, name) and getattr(node, name) != value:
if name == "application_id": if name == "application_id":
continue # we must ignore this to avoid overwriting the application_id allocated by the IOU manager continue # we must ignore this to avoid overwriting the application_id allocated by the IOU manager
@ -147,8 +149,8 @@ async def start_iou_node(start_data: schemas.IOUStart, node: IOUVM = Depends(dep
Start an IOU node. Start an IOU node.
""" """
start_data = jsonable_encoder(start_data, exclude_unset=True) start_values = jsonable_encoder(start_data, exclude_unset=True)
for name, value in start_data.items(): for name, value in start_values.items():
if hasattr(node, name) and getattr(node, name) != value: if hasattr(node, name) and getattr(node, name) != value:
setattr(node, name, value) setattr(node, name, value)
@ -223,7 +225,7 @@ async def update_iou_node_nio(
nio = node.get_nio(adapter_number, port_number) nio = node.get_nio(adapter_number, port_number)
nio.filters.clear() nio.filters.clear()
if nio_data.filters: if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters:
nio.filters = nio_data.filters nio.filters = nio_data.filters
# NIO type is a Union (Ethernet/TAP/UDP); only UDPNIO carries markers. # NIO type is a Union (Ethernet/TAP/UDP); only UDPNIO carries markers.
nio.markers = getattr(nio_data, "markers", None) or {} nio.markers = getattr(nio_data, "markers", None) or {}

View File

@ -329,7 +329,6 @@ module = [
"gns3server.api.routes.compute.ethernet_hub_nodes", # 9 "gns3server.api.routes.compute.ethernet_hub_nodes", # 9
"gns3server.api.routes.compute.ethernet_switch_nodes", # 16 "gns3server.api.routes.compute.ethernet_switch_nodes", # 16
"gns3server.api.routes.compute.frame_relay_switch_nodes", # 9 "gns3server.api.routes.compute.frame_relay_switch_nodes", # 9
"gns3server.api.routes.compute.iou_nodes", # 15
"gns3server.api.routes.compute.nat_nodes", # 10 "gns3server.api.routes.compute.nat_nodes", # 10
"gns3server.api.routes.compute.notifications", # 4 "gns3server.api.routes.compute.notifications", # 4
"gns3server.api.routes.compute.projects", # 7 "gns3server.api.routes.compute.projects", # 7