refactor(gns3server,tests): StrEnum, ASYNC109 and others

gns3server/compute/docker/__init__.py

@@ -283,7 +283,7 @@ class Docker(BaseManager):
         log.debug("Query Docker %s %s params=%s data=%s Response: %s", method, path, params, data, body)
         return body
 
-    async def http_query(self, method, path, data={}, params={}, timeout=300):
+    async def http_query(self, method, path, data={}, params={}, timeout=300):  # noqa: ASYNC109 - operation-specific timeout is intentional
         """
         Makes a query to the docker daemon
 

gns3server/compute/docker/docker_vm.py

@@ -1711,7 +1711,15 @@ class DockerVM(DockerKernelDatapathMixin, BaseNode):
             status_line = headers.split(b"\r\n", 1)[0]
             if b" 101 " not in status_line and b" 200 " not in status_line:
                 raise DockerError(f"Docker interface monitor was rejected: {status_line.decode(errors='ignore')}")
-        except (TimeoutError, DockerError, OSError, KeyError, TypeError, RuntimeError, asyncio.IncompleteReadError) as e:
+        except (
+            TimeoutError,
+            DockerError,
+            OSError,
+            KeyError,
+            TypeError,
+            RuntimeError,
+            asyncio.IncompleteReadError,
+        ) as e:
             if "writer" in locals():
                 await self._close_interface_monitor_writer(writer)
             log.warning("Could not monitor interfaces for Docker container '%s': %s", self.name, e)

gns3server/compute/dynamips/dynamips_hypervisor.py

@@ -57,7 +57,7 @@ class DynamipsHypervisor:
         self._writer = None
         self._io_lock = asyncio.Lock()
 
-    async def connect(self, timeout=10):
+    async def connect(self, timeout=10):  # noqa: ASYNC109 - operation-specific timeout is intentional
         """
         Connects to the hypervisor.
         """

gns3server/compute/qemu/qemu_vm.py

@@ -1307,7 +1307,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
                 # already gone and sweeps nothing.
                 await self._stop_ubridge()
 
-    async def _open_qemu_monitor_connection_vm(self, timeout=10):
+    async def _open_qemu_monitor_connection_vm(self, timeout=10):  # noqa: ASYNC109 - operation-specific timeout is intentional
         """
         Opens a connection to the QEMU monitor.
 

gns3server/compute/ubridge/tc_probe.py

@@ -62,7 +62,7 @@ _bridge_tap_cache: dict[tuple[str, int, int, int], Optional[bool]] = {}
 _lock = asyncio.Lock()
 
 
-async def probe_tc_capabilities(timeout: float = 15.0):
+async def probe_tc_capabilities(timeout: float = 15.0):  # noqa: ASYNC109 - operation-specific timeout is intentional
     """
     Probe the configured uBridge's tc module once per binary identity.
 
@@ -88,7 +88,7 @@ async def probe_tc_capabilities(timeout: float = 15.0):
         return caps
 
 
-async def _probe(path, config, timeout):
+async def _probe(path, config, timeout):  # noqa: ASYNC109 - operation-specific timeout is intentional
     """
     Spawn a throwaway uBridge, ask ``tc capabilities``, tear it down.
     """
@@ -124,7 +124,7 @@ async def _probe(path, config, timeout):
         shutil.rmtree(working_dir, ignore_errors=True)
 
 
-async def probe_tap_support(timeout: float = 15.0):
+async def probe_tap_support(timeout: float = 15.0):  # noqa: ASYNC109 - operation-specific timeout is intentional
     """
     Whether this host's uBridge can create persistent TAPs (the ``tap``
     module), probed the same way as the tc report — a throwaway uBridge, the
@@ -151,7 +151,7 @@ async def probe_tap_support(timeout: float = 15.0):
         return supported
 
 
-async def _probe_tap(path, config, timeout):
+async def _probe_tap(path, config, timeout):  # noqa: ASYNC109 - operation-specific timeout is intentional
     """
     Spawn a throwaway uBridge and try to create (then delete) one persistent
     TAP. The probe device is named after a uuid so a concurrent probe on the
@@ -200,7 +200,7 @@ async def _probe_tap(path, config, timeout):
 _IOL_PROBE_BRIDGE_ID = 1050
 
 
-async def probe_iol_tap_support(timeout: float = 15.0):
+async def probe_iol_tap_support(timeout: float = 15.0):  # noqa: ASYNC109 - operation-specific timeout is intentional
     """
     Whether this host's uBridge can terminate an IOL port on a persistent
     TAP (``iol_bridge add_nio_tap``) — the anchor IOU's Ethernet ports need
@@ -229,7 +229,7 @@ async def probe_iol_tap_support(timeout: float = 15.0):
         return supported
 
 
-async def _probe_iol_tap(path, config, timeout):
+async def _probe_iol_tap(path, config, timeout):  # noqa: ASYNC109 - operation-specific timeout is intentional
     """
     Spawn a throwaway uBridge and run one add_nio_tap/delete_nio_tap cycle on
     a scratch bridge and scratch TAP — attaching works on a stopped bridge,
@@ -280,7 +280,7 @@ async def _probe_iol_tap(path, config, timeout):
         shutil.rmtree(working_dir, ignore_errors=True)
 
 
-async def probe_bridge_tap_support(timeout: float = 15.0):
+async def probe_bridge_tap_support(timeout: float = 15.0):  # noqa: ASYNC109 - operation-specific timeout is intentional
     """
     Whether this host's uBridge has ``bridge delete_nio_tap`` — the command
     that makes the bridge module's TAP leg swappable (stop -> delete -> add ->
@@ -311,7 +311,7 @@ async def probe_bridge_tap_support(timeout: float = 15.0):
         return supported
 
 
-async def _probe_bridge_tap(path, config, timeout):
+async def _probe_bridge_tap(path, config, timeout):  # noqa: ASYNC109 - operation-specific timeout is intentional
     """
     Spawn a throwaway uBridge and issue the probe command against a bridge
     name that cannot exist (uuid). The reply splits the builds: "bridge ...

gns3server/compute/ubridge/ubridge_hypervisor.py

@@ -58,7 +58,7 @@ class UBridgeHypervisor:
         self._reader = None
         self._writer = None
 
-    async def connect(self, timeout=10):
+    async def connect(self, timeout=10):  # noqa: ASYNC109 - operation-specific timeout is intentional
         """
         Connects to the hypervisor.
         """

gns3server/compute/virtualbox/__init__.py

@@ -85,7 +85,7 @@ class VirtualBox(BaseManager):
         self._vboxmanage_path = vboxmanage_path
         return vboxmanage_path
 
-    async def execute(self, subcommand, args, timeout=60):
+    async def execute(self, subcommand, args, timeout=60):  # noqa: ASYNC109 - operation-specific timeout is intentional
 
         # We use a lock prevent parallel execution due to strange errors
         # reported by a user and reproduced by us.

gns3server/compute/vmware/__init__.py

@@ -344,7 +344,7 @@ class VMware(BaseManager):
 
         return self._host_type
 
-    async def execute(self, subcommand, args, timeout=120, log_level=logging.INFO):
+    async def execute(self, subcommand, args, timeout=120, log_level=logging.INFO):  # noqa: ASYNC109 - operation-specific timeout is intentional
         trial = 2
 
         while True:
@@ -361,7 +361,7 @@ class VMware(BaseManager):
                     trial -= 1
                     await asyncio.sleep(0.5)
 
-    async def _execute(self, subcommand, args, timeout=120, log_level=logging.INFO):
+    async def _execute(self, subcommand, args, timeout=120, log_level=logging.INFO):  # noqa: ASYNC109 - operation-specific timeout is intentional
         if self.host_type is None:
             await self.check_vmware_version()
 

gns3server/compute/vpcs/vpcs_probe.py

@@ -57,7 +57,7 @@ _lock = asyncio.Lock()
 _NO_HOSTCONF_NEEDLE = "do not auto-configure"
 
 
-async def probe_vpcs_tap(timeout: float = 5.0) -> bool | None:
+async def probe_vpcs_tap(timeout: float = 5.0) -> bool | None:  # noqa: ASYNC109 - operation-specific timeout is intentional
     """
     Whether this host's VPCS binary can anchor a kernel link (tap mode with
     ``-n``), probed once per binary identity.