diff --git a/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py b/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py index 3e951f10a..d4f7b1a38 100644 --- a/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py +++ b/gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py @@ -169,7 +169,7 @@ def group_devices_by_brand(device_types: List[str]) -> Dict[str, List[str]]: Returns: Dictionary mapping brand names to device type lists """ - brands = {} + brands: dict[str, list[str]] = {} for device_type in device_types: brand = extract_brand_name(device_type) diff --git a/gns3server/agent/mcp/appliances.py b/gns3server/agent/mcp/appliances.py index 45a9b472a..6e3a3ecf5 100644 --- a/gns3server/agent/mcp/appliances.py +++ b/gns3server/agent/mcp/appliances.py @@ -108,7 +108,7 @@ def install_appliance_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) if version: request_params["version"] = version response = conn.http_call("post", url, params=request_params) - result = {"message": f"Appliance {appliance_id} installed"} + result: dict[str, Any] = {"message": f"Appliance {appliance_id} installed"} if response.content: # the install endpoint returns the created template (201); tolerate an # empty body in case an older server still replies with 204 diff --git a/gns3server/agent/web_wireshark/docker_client.py b/gns3server/agent/web_wireshark/docker_client.py index d27873fa5..00f62edd2 100644 --- a/gns3server/agent/web_wireshark/docker_client.py +++ b/gns3server/agent/web_wireshark/docker_client.py @@ -23,9 +23,16 @@ This module handles Docker API operations via aiohttp Unix socket. import asyncio import logging +import sys +from typing import Any import aiohttp +if sys.version_info >= (3, 11): + from asyncio import timeout as async_timeout +else: + from async_timeout import timeout as async_timeout + from gns3server.utils import parse_version logger = logging.getLogger(__name__) @@ -132,7 +139,7 @@ class DockerHTTPClient: url = f"http://docker/v{self._api_version}/{endpoint}" try: - async with asyncio.timeout(self.REQUEST_TIMEOUT): + async with async_timeout(self.REQUEST_TIMEOUT): async with session.request(method, url, **kwargs) as response: if response.status >= 300: error_text = await response.text() @@ -157,9 +164,9 @@ class DockerHTTPClient: return await self._request(method, endpoint, check_connection=False, **kwargs) raise - async def create_network(self, name: str, driver: str = "bridge", subnet: str = None): + async def create_network(self, name: str, driver: str = "bridge", subnet: str | None = None): """Create Docker network.""" - data = {"Name": name, "Driver": driver} + data: dict[str, Any] = {"Name": name, "Driver": driver} if subnet: data["IPAM"] = {"Config": [{"Subnet": subnet}]} await self._request("POST", "networks/create", json=data) @@ -175,7 +182,7 @@ class DockerHTTPClient: async def create_container(self, name: str, image: str, **kwargs): """Create container.""" - data = {"Image": image, "name": name, "HostConfig": {}, "NetworkingConfig": {}} + data: dict[str, Any] = {"Image": image, "name": name, "HostConfig": {}, "NetworkingConfig": {}} # Handle network config if "network" in kwargs: @@ -248,7 +255,7 @@ class DockerHTTPClient: url = f"http://docker/v{self._api_version}/containers/{container_name}/top?ps_args=aux" try: - async with asyncio.timeout(self.REQUEST_TIMEOUT): + async with async_timeout(self.REQUEST_TIMEOUT): async with session.get(url) as response: if response.status >= 300: error_text = await response.text() diff --git a/gns3server/api/routes/compute/atm_switch_nodes.py b/gns3server/api/routes/compute/atm_switch_nodes.py index 65bc1157d..1282a162d 100644 --- a/gns3server/api/routes/compute/atm_switch_nodes.py +++ b/gns3server/api/routes/compute/atm_switch_nodes.py @@ -18,6 +18,7 @@ API routes for ATM switch nodes. """ +from typing import Any import os from fastapi import APIRouter, Depends, Body, Path, status, HTTPException @@ -29,7 +30,9 @@ from gns3server import schemas from gns3server.compute.dynamips import Dynamips from gns3server.compute.dynamips.nodes.atm_switch import ATMSwitch -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or ATM switch node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or ATM switch node"} +} router = APIRouter(responses=responses) @@ -57,13 +60,13 @@ async def create_atm_switch(project_id: UUID, node_data: schemas.ATMSwitchCreate # Use the Dynamips ATM switch to simulate this node dynamips_manager = Dynamips.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) node = await dynamips_manager.create_node( - node_data.get("name"), + data.get("name"), str(project_id), - node_data.get("node_id"), + data.get("node_id"), node_type="atm_switch", - mappings=node_data.get("mappings"), + mappings=data.get("mappings"), ) return node.asdict() @@ -97,11 +100,11 @@ async def update_atm_switch( Update an ATM switch node. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - if "name" in node_data and node.name != node_data["name"]: - await node.set_name(node_data["name"]) - if "mappings" in node_data: - node.mappings = node_data["mappings"] + data = jsonable_encoder(node_data, exclude_unset=True) + if "name" in data and node.name != data["name"]: + await node.set_name(data["name"]) + if "mappings" in data: + node.mappings = data["mappings"] node.updated() return node.asdict() diff --git a/gns3server/api/routes/compute/cloud_nodes.py b/gns3server/api/routes/compute/cloud_nodes.py index 7baf8ca17..d0a4ead9e 100644 --- a/gns3server/api/routes/compute/cloud_nodes.py +++ b/gns3server/api/routes/compute/cloud_nodes.py @@ -23,14 +23,16 @@ import os from fastapi import APIRouter, Depends, Path, status, HTTPException from fastapi.encoders import jsonable_encoder from fastapi.responses import StreamingResponse -from typing import Union +from typing import Any, Union from uuid import UUID from gns3server import schemas from gns3server.compute.builtin import Builtin from gns3server.compute.builtin.nodes.cloud import Cloud -responses = {404: {"model": schemas.ErrorMessage, "description": "Could not find project or cloud node"}} +responses: dict[int | str, dict[str, Any]] = { + 404: {"model": schemas.ErrorMessage, "description": "Could not find project or cloud node"} +} router = APIRouter(responses=responses) @@ -57,21 +59,21 @@ async def create_cloud(project_id: UUID, node_data: schemas.CloudCreate) -> sche """ builtin_manager = Builtin.instance() - node_data = jsonable_encoder(node_data, exclude_unset=True) + data = jsonable_encoder(node_data, exclude_unset=True) node = await builtin_manager.create_node( - node_data.pop("name"), + data.pop("name"), str(project_id), - node_data.get("node_id"), + data.get("node_id"), node_type="cloud", - ports=node_data.get("ports_mapping"), + ports=data.get("ports_mapping"), ) # add the remote console settings - node.remote_console_host = node_data.get("remote_console_host", node.remote_console_host) - node.remote_console_port = node_data.get("remote_console_port", node.remote_console_port) - node.remote_console_type = node_data.get("remote_console_type", node.remote_console_type) - node.remote_console_http_path = node_data.get("remote_console_http_path", node.remote_console_http_path) - node.usage = node_data.get("usage", "") + node.remote_console_host = data.get("remote_console_host", node.remote_console_host) + node.remote_console_port = data.get("remote_console_port", node.remote_console_port) + node.remote_console_type = data.get("remote_console_type", node.remote_console_type) + node.remote_console_http_path = data.get("remote_console_http_path", node.remote_console_http_path) + node.usage = data.get("usage", "") return node.asdict() @@ -90,8 +92,8 @@ async def update_cloud(node_data: schemas.CloudUpdate, node: Cloud = Depends(dep Update a cloud node. """ - node_data = jsonable_encoder(node_data, exclude_unset=True) - for name, value in node_data.items(): + data = jsonable_encoder(node_data, exclude_unset=True) + for name, value in data.items(): if hasattr(node, name) and getattr(node, name) != value: setattr(node, name, value) node.updated() @@ -178,7 +180,7 @@ async def update_cloud_nio( nio = node.get_nio(port_number) nio.filters.clear() - if nio_data.filters: + if isinstance(nio_data, schemas.UDPNIO) and nio_data.filters: nio.filters = nio_data.filters # NIO type is a Union (Ethernet/TAP/UDP); only UDPNIO carries markers. nio.markers = getattr(nio_data, "markers", None) or {} diff --git a/pyproject.toml b/pyproject.toml index 8dd7473ac..6ccf18973 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -294,11 +294,6 @@ enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"] # Baseline: modules with existing type errors (error count at baseline time). # Remove modules from this list one small PR at a time. Never add new ones. module = [ - "gns3server.agent.gns3_copilot.utils.custom_netmiko.scripts.list_netmiko_telnet_devices", # 1 - "gns3server.agent.mcp.appliances", # 1 - "gns3server.agent.web_wireshark.docker_client", # 5 - "gns3server.api.routes.compute.atm_switch_nodes", # 9 - "gns3server.api.routes.compute.cloud_nodes", # 14 "gns3server.api.routes.compute.ethernet_switch_nodes", # 16 "gns3server.api.routes.compute.frame_relay_switch_nodes", # 9 "gns3server.api.routes.compute.iou_nodes", # 15