Merge pull request #2916 from markparonyan/mypy-agent-gns3_copilot-plus

fix(typing): resolve mypy errors in agent.gns3_copilot + agent.mcp + agent.web_wireshark + api.routes

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)

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

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()

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()
 

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 {}

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