Merge pull request #2904 from markparonyan/mypy-agent-web-wireshark

fix(typing): resolve mypy errors in agent.web_wireshark

gns3server/agent/web_wireshark/manage_wireshark.py

@@ -81,10 +81,7 @@ from gns3server.utils.uuid_validator import validate_uuid
 if __name__ == "__main__":
     sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
 
-try:
-    from gns3server.agent.web_wireshark.manager import WebWiresharkManager
-except ImportError:
-    from manager import WebWiresharkManager
+from gns3server.agent.web_wireshark.manager import WebWiresharkManager
 
 logger = logging.getLogger(__name__)
 

gns3server/agent/web_wireshark/manager.py

@@ -84,7 +84,7 @@ class WebWiresharkManager:
             logger.warning(f"Container {container_id[:12]} health check failed: {e}")
             return False
 
-    async def _exec_in_container(self, container_id: str, command: str, timeout: int = None) -> tuple:
+    async def _exec_in_container(self, container_id: str, command: str, timeout: Optional[int] = None) -> tuple:
         """Execute command in container with timeout.
 
         Args:
@@ -197,7 +197,7 @@ class WebWiresharkManager:
             stdout, _ = await proc.communicate()
 
             # Build parent->children mapping
-            children_map = {}
+            children_map: dict[int, list[int]] = {}
             for line in stdout.decode().strip().split("\n"):
                 parts = line.split(None, 2)
                 if len(parts) < 3:
@@ -326,7 +326,7 @@ class WebWiresharkManager:
                 return
 
             # Build parent->children mapping and collect process info
-            children_map = {}  # ppid -> [pid]
+            children_map: dict[int, list[int]] = {}  # ppid -> [pid]
             process_info = {}  # pid -> (ppid, command)
 
             for line in stdout.decode().strip().split("\n"):
@@ -480,7 +480,7 @@ class WebWiresharkManager:
             logger.debug(f"Cannot get URL from Config: {e}")
         return None
 
-    async def _get_container_gateway_ip(self, container_id: str = None) -> Optional[str]:
+    async def _get_container_gateway_ip(self, container_id: Optional[str] = None) -> Optional[str]:
         """Get the Docker bridge gateway IP for container to access host.
 
         Args:
@@ -519,7 +519,7 @@ class WebWiresharkManager:
 
         return None
 
-    async def get_container_ip(self, container_name: str, container_id: str = None) -> Optional[str]:
+    async def get_container_ip(self, container_name: str, container_id: Optional[str] = None) -> Optional[str]:
         """Get the container IP address in the wireshark network.
 
         Args:
@@ -626,7 +626,7 @@ class WebWiresharkManager:
         project_id: str,
         image: str = "gns3/web-wireshark:latest",
         memory: str = "2g",
-        memory_swap: str = None,
+        memory_swap: Optional[str] = None,
         cpus: float = 1.0,
         pids_limit: int = 1000,
     ) -> str:
@@ -935,10 +935,10 @@ class WebWiresharkManager:
         project_id: str,
         link_id: str,
         jwt_token: str,
-        capture_stream_url: str = None,
+        capture_stream_url: Optional[str] = None,
         image: str = "gns3/web-wireshark:latest",
         memory: str = "2g",
-        memory_swap: str = None,
+        memory_swap: Optional[str] = None,
         cpus: float = 1.0,
         pids_limit: int = 1000,
     ):

gns3server/agent/web_wireshark/stats.py

@@ -7,7 +7,7 @@ about Web Wireshark containers and sessions.
 
 import logging
 import subprocess
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 logger = logging.getLogger(__name__)
 
@@ -30,7 +30,7 @@ async def collect_webwireshark_stats(projects: List) -> Dict:
     """
     from .manager import WebWiresharkManager
 
-    stats = {"total_containers": 0, "running_containers": 0, "active_sessions": 0, "containers": []}
+    stats: Dict[str, Any] = {"total_containers": 0, "running_containers": 0, "active_sessions": 0, "containers": []}
 
     # Create a single manager instance and reuse it
     manager = WebWiresharkManager()

pyproject.toml

@@ -312,9 +312,6 @@ module = [
   "gns3server.agent.gns3_copilot.utils.parse_tool_content",  # 1
   "gns3server.agent.mcp.appliances",  # 1
   "gns3server.agent.web_wireshark.docker_client",  # 5
-  "gns3server.agent.web_wireshark.manage_wireshark",  # 1
-  "gns3server.agent.web_wireshark.manager",  # 9
-  "gns3server.agent.web_wireshark.stats",  # 4
   "gns3server.api.routes.compute.atm_switch_nodes",  # 9
   "gns3server.api.routes.compute.capabilities",  # 1
   "gns3server.api.routes.compute.cloud_nodes",  # 14