diff --git a/gns3server/agent/gns3_copilot/agent_service.py b/gns3server/agent/gns3_copilot/agent_service.py
index be6ed201d..5c58cd4e1 100644
--- a/gns3server/agent/gns3_copilot/agent_service.py
+++ b/gns3server/agent/gns3_copilot/agent_service.py
@@ -685,7 +685,7 @@ class AgentService:
"Checkpointer connection closed for: %s",
self.project_path,
)
- except asyncio.TimeoutError:
+ except TimeoutError:
log.warning(
"Checkpointer connection close timeout for: %s (forcing cleanup)",
self.project_path,
diff --git a/gns3server/agent/mcp/__init__.py b/gns3server/agent/mcp/__init__.py
index 57b45e5ea..2a39bdb5f 100644
--- a/gns3server/agent/mcp/__init__.py
+++ b/gns3server/agent/mcp/__init__.py
@@ -215,7 +215,7 @@ async def wait_for_mcp_ready() -> bool:
await asyncio.wait_for(_mcp_ready_event.wait(), timeout=5.0)
log.debug("MCP server is now ready, proceeding with connection")
return True
- except asyncio.TimeoutError:
+ except TimeoutError:
log.warning("MCP server ready check timed out after 5 seconds - GNS3 server initialization may have issues")
return False
diff --git a/gns3server/agent/web_wireshark/docker_client.py b/gns3server/agent/web_wireshark/docker_client.py
index e030eab35..0ee7fa339 100644
--- a/gns3server/agent/web_wireshark/docker_client.py
+++ b/gns3server/agent/web_wireshark/docker_client.py
@@ -21,7 +21,6 @@ Web Wireshark - Docker HTTP API client
This module handles Docker API operations via aiohttp Unix socket.
"""
-import asyncio
import logging
from asyncio import timeout as async_timeout
from typing import Any
@@ -142,7 +141,7 @@ class DockerHTTPClient:
if response.status == 204: # No Content
return None
return await response.json()
- except asyncio.TimeoutError:
+ except TimeoutError:
raise RuntimeError(f"Docker API timeout after {self.REQUEST_TIMEOUT}s for {endpoint}")
except aiohttp.ClientError as e:
raise RuntimeError(f"Docker connection error: {e}") from e
@@ -225,7 +224,7 @@ class DockerHTTPClient:
return None
raise
- async def stop_container(self, container_id: str, timeout: int = 0):
+ async def stop_container(self, container_id: str, timeout: int = 0): # noqa: ASYNC109 - operation-specific timeout is intentional
"""Stop container immediately (force kill, no graceful shutdown)."""
await self._request("POST", f"containers/{container_id}/stop", params={"t": timeout})
@@ -256,7 +255,7 @@ class DockerHTTPClient:
error_text = await response.text()
raise RuntimeError(f"Docker API error {response.status}: {error_text}")
result = await response.json()
- except asyncio.TimeoutError:
+ except TimeoutError:
raise RuntimeError(f"Docker API timeout after {self.REQUEST_TIMEOUT}s for containers/{container_name}/top")
except aiohttp.ClientError as e:
raise RuntimeError(f"Docker connection error: {e}") from e
diff --git a/gns3server/agent/web_wireshark/manager.py b/gns3server/agent/web_wireshark/manager.py
index 0791c0668..ec638ee4a 100644
--- a/gns3server/agent/web_wireshark/manager.py
+++ b/gns3server/agent/web_wireshark/manager.py
@@ -75,7 +75,7 @@ class WebWiresharkManager:
)
stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=self.CONTAINER_EXEC_TIMEOUT)
return proc.returncode == 0 and b"ping" in stdout
- except asyncio.TimeoutError:
+ except TimeoutError:
logger.warning(f"Container {container_id[:12]} health check timeout")
return False
except FileNotFoundError:
@@ -85,7 +85,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: Optional[int] = None) -> tuple:
+ async def _exec_in_container(self, container_id: str, command: str, timeout: Optional[int] = None) -> tuple: # noqa: ASYNC109 - operation-specific timeout is intentional
"""Execute command in container with timeout.
Args:
@@ -113,12 +113,12 @@ class WebWiresharkManager:
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
return (proc.returncode, stdout.decode().strip(), stderr.decode().strip())
- except asyncio.TimeoutError:
+ except TimeoutError:
logger.error(f"Command timeout after {timeout}s: {command}")
proc.terminate()
try:
await asyncio.wait_for(proc.wait(), timeout=2)
- except asyncio.TimeoutError:
+ except TimeoutError:
proc.kill()
return (-1, "", f"timeout after {timeout}s")
except Exception as e:
diff --git a/gns3server/api/routes/controller/nodes.py b/gns3server/api/routes/controller/nodes.py
index 988524835..9a5ff928b 100644
--- a/gns3server/api/routes/controller/nodes.py
+++ b/gns3server/api/routes/controller/nodes.py
@@ -418,7 +418,12 @@ LIFECYCLE_RESPONSES: dict[int | str, dict[str, Any]] = {
async def _run_lifecycle(
- node: Node, action: Callable, target: str, wait: bool, timeout: int, tolerate_unsupported: bool = True
+ node: Node,
+ action: Callable,
+ target: str,
+ wait: bool,
+ timeout: int, # noqa: ASYNC109 - operation-specific timeout is intentional
+ tolerate_unsupported: bool = True,
) -> dict:
async def _run():
try:
@@ -435,7 +440,7 @@ async def _run_lifecycle(
return node.asdict()
try:
await asyncio.wait_for(_run(), timeout)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ControllerTimeoutError(f"Timeout when waiting for {node.name} to be {target}")
return node.asdict()
@@ -449,7 +454,7 @@ async def _run_lifecycle(
async def start_node(
start_data: Optional[dict] = None,
wait: bool = Query(False, description="Return only when the node status is 'started'"),
- timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
+ timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"), # noqa: ASYNC109 - operation-specific timeout is intentional
node: Node = Depends(dep_node),
) -> dict:
"""
@@ -469,7 +474,7 @@ async def start_node(
)
async def stop_node(
wait: bool = Query(False, description="Return only when the node status is 'stopped'"),
- timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
+ timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"), # noqa: ASYNC109 - operation-specific timeout is intentional
node: Node = Depends(dep_node),
) -> dict:
"""
@@ -491,7 +496,7 @@ async def stop_node(
)
async def suspend_node(
wait: bool = Query(False, description="Return only when the node status is 'suspended'"),
- timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
+ timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"), # noqa: ASYNC109 - operation-specific timeout is intentional
node: Node = Depends(dep_node),
) -> dict:
"""
@@ -514,7 +519,7 @@ async def suspend_node(
)
async def reload_node(
wait: bool = Query(False, description="Return only when the node status is 'started'"),
- timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"),
+ timeout: int = Query(240, ge=1, le=3600, description="Seconds to wait when 'wait' is true"), # noqa: ASYNC109 - operation-specific timeout is intentional
node: Node = Depends(dep_node),
) -> dict:
"""
@@ -711,7 +716,7 @@ async def get_file(file_path: str, node: Node = Depends(dep_node)) -> Response:
try:
async for chunk in compute_resp.content.iter_chunked(65536):
yield chunk
- except (OSError, asyncio.TimeoutError) as e:
+ except (TimeoutError, OSError) as e:
log.error(f"Error streaming file '{path}' from compute: {e}")
raise
finally:
diff --git a/gns3server/compute/docker/__init__.py b/gns3server/compute/docker/__init__.py
index 18d180e2b..fdb3eaf51 100644
--- a/gns3server/compute/docker/__init__.py
+++ b/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
@@ -323,7 +323,7 @@ class Docker(BaseManager):
)
except aiohttp.ClientError as e:
raise DockerError(f"Docker has returned an error: {e}")
- except asyncio.TimeoutError:
+ except TimeoutError:
raise DockerError("Docker timeout " + method + " " + path)
if response.status >= 300:
body = await response.read()
@@ -391,7 +391,7 @@ class Docker(BaseManager):
raise DockerError(
f"Disconnected while pulling Docker image '{image}' from Docker repository"
) from e
- except asyncio.TimeoutError as e:
+ except TimeoutError as e:
raise DockerError(f"Timeout while pulling Docker image '{image}' from Docker repository") from e
if not chunk:
break
@@ -448,7 +448,7 @@ class Docker(BaseManager):
chunk = await response.content.read(CHUNK_SIZE)
except aiohttp.ServerDisconnectedError as e:
raise DockerError("Disconnected while loading Docker image") from e
- except asyncio.TimeoutError as e:
+ except TimeoutError as e:
raise DockerError("Timeout while loading Docker image") from e
if not chunk:
break
diff --git a/gns3server/compute/docker/docker_vm.py b/gns3server/compute/docker/docker_vm.py
index 0612ff235..1bd732872 100644
--- a/gns3server/compute/docker/docker_vm.py
+++ b/gns3server/compute/docker/docker_vm.py
@@ -948,7 +948,7 @@ class DockerVM(DockerKernelDatapathMixin, BaseNode):
await self._start_vnc_process(restart=True)
try:
await wait_for_file_creation(x11_socket)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise DockerError(f'x11 socket file "{x11_socket}" does not exist')
monitor_process(self._vnc_process, self._vnc_callback)
@@ -1190,7 +1190,7 @@ class DockerVM(DockerKernelDatapathMixin, BaseNode):
return False
try:
_, stderr = await asyncio.wait_for(process.communicate(), timeout=60)
- except asyncio.TimeoutError:
+ except TimeoutError:
process.kill()
await process.communicate()
log.warning(f"Docker container '{self._name}': reclaiming '{directory}' timed out")
@@ -1254,7 +1254,7 @@ class DockerVM(DockerKernelDatapathMixin, BaseNode):
x11_socket = os.path.join("/tmp/.X11-unix/", f"X{self._display}")
try:
await wait_for_file_creation(x11_socket)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise DockerError(f'x11 socket file "{x11_socket}" does not exist')
if not hasattr(sys, "_called_from_test") or not sys._called_from_test:
@@ -1712,13 +1712,13 @@ class DockerVM(DockerKernelDatapathMixin, BaseNode):
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,
- asyncio.TimeoutError,
) as e:
if "writer" in locals():
await self._close_interface_monitor_writer(writer)
diff --git a/gns3server/compute/docker/vendor_docker_vm.py b/gns3server/compute/docker/vendor_docker_vm.py
index 3daa935e5..9601a4d06 100644
--- a/gns3server/compute/docker/vendor_docker_vm.py
+++ b/gns3server/compute/docker/vendor_docker_vm.py
@@ -459,7 +459,7 @@ class VendorDockerVM(DockerVM):
# setup; wait instead of silently blackholing the adapter.
try:
await wait_for_file_creation(remote_sock, timeout=30)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise DockerError(
f"Socket '{remote_sock}' for adapter {adapter_number} of container "
f"'{self._name}' did not appear within 30 seconds. Check that the "
@@ -806,7 +806,7 @@ class _LazyExecTelnetServer(AsyncioTelnetServer):
await writer.drain()
try:
headers = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), timeout=5)
- except (asyncio.IncompleteReadError, asyncio.TimeoutError) as e:
+ except (TimeoutError, asyncio.IncompleteReadError) as e:
writer.close()
raise DockerError(f"Docker exec start failed: {e}")
status_line = headers.split(b"\r\n", 1)[0]
diff --git a/gns3server/compute/dynamips/dynamips_hypervisor.py b/gns3server/compute/dynamips/dynamips_hypervisor.py
index 11f80221d..6c4bb0bbb 100644
--- a/gns3server/compute/dynamips/dynamips_hypervisor.py
+++ b/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.
"""
@@ -80,7 +80,7 @@ class DynamipsHypervisor:
self._reader, self._writer = await asyncio.wait_for(
asyncio.open_connection(host, self._port), timeout=1
)
- except (asyncio.TimeoutError, OSError) as e:
+ except (TimeoutError, OSError) as e:
last_exception = e
continue
connection_success = True
diff --git a/gns3server/compute/dynamips/hypervisor.py b/gns3server/compute/dynamips/hypervisor.py
index c1d3f5980..d56c81678 100644
--- a/gns3server/compute/dynamips/hypervisor.py
+++ b/gns3server/compute/dynamips/hypervisor.py
@@ -144,7 +144,7 @@ class Hypervisor(DynamipsHypervisor):
await asyncio.sleep(0.01)
try:
await wait_for_process_termination(self._process, timeout=3)
- except asyncio.TimeoutError:
+ except TimeoutError:
if self._process.returncode is None:
log.warning(f"Dynamips process {self._process.pid} is still running... killing it")
try:
diff --git a/gns3server/compute/iou/iou_vm.py b/gns3server/compute/iou/iou_vm.py
index 679050f6f..34b44bb09 100644
--- a/gns3server/compute/iou/iou_vm.py
+++ b/gns3server/compute/iou/iou_vm.py
@@ -1034,7 +1034,7 @@ class IOUVM(KernelDatapathMixin, BaseNode):
if self._iou_process.returncode is None:
try:
await gns3server.utils.asyncio.wait_for_process_termination(self._iou_process, timeout=3)
- except asyncio.TimeoutError:
+ except TimeoutError:
if self._iou_process.returncode is None:
log.warning(f"IOU process {self._iou_process.pid} is still running... killing it")
try:
diff --git a/gns3server/compute/qemu/qemu_vm.py b/gns3server/compute/qemu/qemu_vm.py
index e8ee33140..8b97b0c14 100644
--- a/gns3server/compute/qemu/qemu_vm.py
+++ b/gns3server/compute/qemu/qemu_vm.py
@@ -618,7 +618,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
else:
await command(f"eject -f {device}")
log.debug('QEMU VM "%s" has changed %s to %s', self._name, name, image)
- except (asyncio.TimeoutError, asyncio.IncompleteReadError, OSError) as e:
+ except (TimeoutError, asyncio.IncompleteReadError, OSError) as e:
raise QemuError(f"Could not change CD/DVD media: {e}") from e
finally:
writer.close()
@@ -1343,7 +1343,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
await gns3server.utils.asyncio.wait_for_process_termination(self._process, timeout=3)
except ProcessLookupError:
pass
- except asyncio.TimeoutError:
+ except TimeoutError:
if self._process:
try:
self._process.kill()
@@ -1378,7 +1378,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.
@@ -1395,7 +1395,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
try:
log.debug(f"Connecting to Qemu monitor on {self._monitor_host}:{self._monitor}")
reader, writer = await asyncio.open_connection(self._monitor_host, self._monitor)
- except (asyncio.TimeoutError, OSError) as e:
+ except (TimeoutError, OSError) as e:
last_exception = e
continue
connection_success = True
@@ -1434,7 +1434,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
line = await asyncio.wait_for(reader.readline(), timeout=3) # echo of the command
if not line or cmd_byte in line:
break
- except asyncio.TimeoutError:
+ except TimeoutError:
log.warning(f"Missing echo of command '{command}'")
except OSError as e:
log.warning(f"Could not write to QEMU monitor: {e}")
@@ -1450,7 +1450,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
if expect in line:
result = line.decode("utf-8").strip()
break
- except asyncio.TimeoutError:
+ except TimeoutError:
log.warning(f"Timeout while waiting for result of command '{command}'")
except (ConnectionError, EOFError) as e:
log.warning(f"Could not read from QEMU monitor: {e}")
@@ -1478,7 +1478,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
line = await asyncio.wait_for(reader.readline(), timeout=3) # echo of the command
if not line or cmd_byte in line:
break
- except asyncio.TimeoutError:
+ except TimeoutError:
log.warning(f"Missing echo of command '{command}'")
except OSError as e:
log.warning(f"Could not write to QEMU monitor: {e}")
diff --git a/gns3server/compute/ubridge/hypervisor.py b/gns3server/compute/ubridge/hypervisor.py
index 66dddd99a..8a7b78090 100644
--- a/gns3server/compute/ubridge/hypervisor.py
+++ b/gns3server/compute/ubridge/hypervisor.py
@@ -230,7 +230,7 @@ class Hypervisor(UBridgeHypervisor):
await UBridgeHypervisor.stop(self)
try:
await wait_for_process_termination(self._process, timeout=3)
- except asyncio.TimeoutError:
+ except TimeoutError:
if self._process and self._process.returncode is None:
log.warning(f"uBridge process {self._process.pid} is still running... killing it")
try:
diff --git a/gns3server/compute/ubridge/tc_probe.py b/gns3server/compute/ubridge/tc_probe.py
index 0f9bc31ef..3da66c42e 100644
--- a/gns3server/compute/ubridge/tc_probe.py
+++ b/gns3server/compute/ubridge/tc_probe.py
@@ -65,7 +65,7 @@ _vxlan_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.
@@ -91,7 +91,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.
"""
@@ -115,7 +115,7 @@ async def _probe(path, config, timeout):
try:
caps = await asyncio.wait_for(ask(), timeout=timeout)
- except (UbridgeError, OSError, asyncio.TimeoutError, ValueError) as e:
+ except (TimeoutError, UbridgeError, OSError, ValueError) as e:
# Old uBridge builds answer "Unknown command"/"Unknown module";
# anything else here is a broken install — either way: unknown.
log.debug("uBridge tc capabilities probe failed: %s", e)
@@ -127,7 +127,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
@@ -154,7 +154,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
@@ -186,7 +186,7 @@ async def _probe_tap(path, config, timeout):
try:
return await asyncio.wait_for(ask(), timeout=timeout)
- except (UbridgeError, OSError, asyncio.TimeoutError, ValueError) as e:
+ except (TimeoutError, UbridgeError, OSError, ValueError) as e:
# Missing module, old build, or no CAP_NET_ADMIN: unknown.
log.debug("uBridge tap-module probe failed: %s", e)
return None
@@ -203,7 +203,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
@@ -232,7 +232,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,
@@ -272,7 +272,7 @@ async def _probe_iol_tap(path, config, timeout):
try:
return await asyncio.wait_for(ask(), timeout=timeout)
- except (UbridgeError, OSError, asyncio.TimeoutError, ValueError) as e:
+ except (TimeoutError, UbridgeError, OSError, ValueError) as e:
# Old build ("Unknown command"), missing module, or no
# CAP_NET_ADMIN: unknown.
log.debug("uBridge iol-tap probe failed: %s", e)
@@ -283,7 +283,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 ->
@@ -314,7 +314,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 ...
@@ -351,7 +351,7 @@ async def _probe_bridge_tap(path, config, timeout):
try:
return await asyncio.wait_for(ask(), timeout=timeout)
- except (UbridgeError, OSError, asyncio.TimeoutError, ValueError) as e:
+ except (TimeoutError, UbridgeError, OSError, ValueError) as e:
# UbridgeError also covers hypervisor.start()/connect() failing
# (unlike the siblings, ask() re-raises those): an unusable
# binary means unknown, never a failed /capabilities response.
@@ -363,7 +363,7 @@ async def _probe_bridge_tap(path, config, timeout):
shutil.rmtree(working_dir, ignore_errors=True)
-async def probe_vxlan_support(timeout: float = 15.0):
+async def probe_vxlan_support(timeout: float = 15.0): # noqa: ASYNC109 - operation-specific timeout is intentional
"""
Whether this host's uBridge can create a VXLAN device (the ``vxlan``
module) — the carrier a cross-compute kernel link tunnels through.
@@ -395,7 +395,7 @@ async def probe_vxlan_support(timeout: float = 15.0):
return supported
-async def _probe_vxlan(path, config, timeout):
+async def _probe_vxlan(path, config, timeout): # noqa: ASYNC109 - operation-specific timeout is intentional
"""
Spawn a throwaway uBridge and run one ``vxlan create``/``delete`` cycle.
The scratch name (uuid suffix) cannot collide with a real link's
@@ -442,7 +442,7 @@ async def _probe_vxlan(path, config, timeout):
try:
return await asyncio.wait_for(ask(), timeout=timeout)
- except (UbridgeError, OSError, asyncio.TimeoutError, ValueError) as e:
+ except (TimeoutError, UbridgeError, OSError, ValueError) as e:
# Old build ("Unknown command"), missing module, or no
# CAP_NET_ADMIN: unknown.
log.debug("uBridge vxlan-module probe failed: %s", e)
diff --git a/gns3server/compute/ubridge/ubridge_hypervisor.py b/gns3server/compute/ubridge/ubridge_hypervisor.py
index 5b26e6db1..af819d1cf 100644
--- a/gns3server/compute/ubridge/ubridge_hypervisor.py
+++ b/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.
"""
diff --git a/gns3server/compute/virtualbox/__init__.py b/gns3server/compute/virtualbox/__init__.py
index 27313318f..1d0007dfb 100644
--- a/gns3server/compute/virtualbox/__init__.py
+++ b/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.
@@ -112,7 +112,7 @@ class VirtualBox(BaseManager):
try:
stdout_data, stderr_data = await asyncio.wait_for(process.communicate(), timeout=timeout)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise VirtualBoxError(f"VBoxManage has timed out after {timeout} seconds!")
if process.returncode:
diff --git a/gns3server/compute/vmware/__init__.py b/gns3server/compute/vmware/__init__.py
index cc480688e..6ac4fb495 100644
--- a/gns3server/compute/vmware/__init__.py
+++ b/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()
@@ -382,7 +382,7 @@ class VMware(BaseManager):
try:
stdout_data, _ = await asyncio.wait_for(process.communicate(), timeout=timeout)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise VMwareError(
f"vmrun has timed out after {timeout} seconds!\nTry to run {command_string} in a terminal to see more details.\n\nMake sure GNS3 and VMware run under the same user and whitelist vmrun.exe in your antivirus."
)
diff --git a/gns3server/compute/vpcs/vpcs_probe.py b/gns3server/compute/vpcs/vpcs_probe.py
index 695a6ff63..799211df8 100644
--- a/gns3server/compute/vpcs/vpcs_probe.py
+++ b/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.
@@ -83,7 +83,7 @@ async def probe_vpcs_tap(timeout: float = 5.0) -> bool | None:
return _cache[key]
try:
output = await asyncio.wait_for(subprocess_check_output(vpcs_path, "-h"), timeout=timeout)
- except (OSError, asyncio.TimeoutError, ValueError) as e:
+ except (TimeoutError, OSError, ValueError) as e:
log.debug("VPCS tap-mode probe failed: %s", e)
supported = None
else:
diff --git a/gns3server/compute/vpcs/vpcs_vm.py b/gns3server/compute/vpcs/vpcs_vm.py
index df9c31ab1..536aef70e 100644
--- a/gns3server/compute/vpcs/vpcs_vm.py
+++ b/gns3server/compute/vpcs/vpcs_vm.py
@@ -333,7 +333,7 @@ class VPCSVM(KernelDatapathMixin, BaseNode):
if self._process.returncode is None:
try:
await wait_for_process_termination(self._process, timeout=3)
- except asyncio.TimeoutError:
+ except TimeoutError:
if self._process.returncode is None:
try:
self._process.kill()
diff --git a/gns3server/controller/appliance_manager.py b/gns3server/controller/appliance_manager.py
index f4ecdded5..338bc2615 100644
--- a/gns3server/controller/appliance_manager.py
+++ b/gns3server/controller/appliance_manager.py
@@ -15,7 +15,6 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-import asyncio
import copy
import json
import logging
@@ -221,7 +220,7 @@ class ApplianceManager:
raise ControllerError(f"Could not save {image_type} image '{image_path}': {e}")
except ClientError as e:
raise ControllerError(f"Could not connect to download '{image_name}': {e}")
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ControllerError(f"Timeout while downloading '{image_name}' from '{image_url}'")
async def _find_appliance_version_images(
@@ -551,7 +550,7 @@ class ApplianceManager:
log.info(f"Saving {symbol} symbol to {destination_path}")
with open(destination_path, "wb") as f:
f.write(symbol_data)
- except asyncio.TimeoutError:
+ except TimeoutError:
log.warning(f"Timeout while downloading '{symbol_url}'")
except OSError as e:
log.warning(f"Could not write appliance symbol '{destination_path}': {e}")
@@ -601,7 +600,7 @@ class ApplianceManager:
continue
try:
appliance_data = await response.read()
- except asyncio.TimeoutError:
+ except TimeoutError:
log.warning("Timeout while downloading '{}'".format(appliance["download_url"]))
continue
path = os.path.join(appliances_dir, appliance_name)
diff --git a/gns3server/controller/compute.py b/gns3server/controller/compute.py
index aa2095eca..28adc2708 100644
--- a/gns3server/controller/compute.py
+++ b/gns3server/controller/compute.py
@@ -596,7 +596,7 @@ class Compute:
"""Returns URL for specific path at Compute"""
return self._getUrl(path)
- async def _run_http_query(self, method, path, data=None, timeout=120, raw=False, stream=False, params=None):
+ async def _run_http_query(self, method, path, data=None, timeout=120, raw=False, stream=False, params=None): # noqa: ASYNC109 - operation-specific timeout is intentional
async with asynctimeout(delay=timeout):
url = self._getUrl(path)
headers = {"content-type": "application/json"}
@@ -635,7 +635,7 @@ class Compute:
chunked=chunked,
timeout=timeout,
)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ComputeError(f"Timeout error for {method} call to {url} after {timeout}s")
except (
aiohttp.ClientError,
diff --git a/gns3server/controller/export_project.py b/gns3server/controller/export_project.py
index 0c34d2ed4..cb4c4f2fd 100644
--- a/gns3server/controller/export_project.py
+++ b/gns3server/controller/export_project.py
@@ -15,7 +15,6 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-import asyncio
import json
import logging
import os
@@ -134,7 +133,7 @@ async def export_project(
while True:
try:
data = await response.content.read(CHUNK_SIZE)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ControllerTimeoutError(
"Timeout when downloading file '{}' from remote compute {}:{}".format(
compute_file["path"], compute.host, compute.port
@@ -353,7 +352,7 @@ async def _export_remote_images(
while True:
try:
data = await response.content.read(CHUNK_SIZE)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ControllerTimeoutError(
f"Timeout when downloading image '{image}' from remote compute {compute.host}:{compute.port}"
)
diff --git a/gns3server/controller/gns3vm/virtualbox_gns3_vm.py b/gns3server/controller/gns3vm/virtualbox_gns3_vm.py
index fe39b3dd3..425a3ee5c 100644
--- a/gns3server/controller/gns3vm/virtualbox_gns3_vm.py
+++ b/gns3server/controller/gns3vm/virtualbox_gns3_vm.py
@@ -43,7 +43,7 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
self._system_properties = {}
self._virtualbox_manager = VirtualBox()
- async def _execute(self, subcommand, args, timeout=60):
+ async def _execute(self, subcommand, args, timeout=60): # noqa: ASYNC109 - operation-specific timeout is intentional
try:
result = await self._virtualbox_manager.execute(subcommand, args, timeout)
@@ -434,7 +434,7 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
return interface["ip_address"]
except ValueError:
pass
- except (OSError, aiohttp.ClientError, TimeoutError, asyncio.TimeoutError):
+ except (OSError, aiohttp.ClientError, TimeoutError):
pass
remaining_try -= 1
await asyncio.sleep(1)
diff --git a/gns3server/controller/gns3vm/vmware_gns3_vm.py b/gns3server/controller/gns3vm/vmware_gns3_vm.py
index ff0451463..f1d7f7c14 100644
--- a/gns3server/controller/gns3vm/vmware_gns3_vm.py
+++ b/gns3server/controller/gns3vm/vmware_gns3_vm.py
@@ -41,7 +41,7 @@ class VMwareGNS3VM(BaseGNS3VM):
def vmx_path(self):
return self._vmx_path
- async def _execute(self, subcommand, args, timeout=60, log_level=logging.INFO):
+ async def _execute(self, subcommand, args, timeout=60, log_level=logging.INFO): # noqa: ASYNC109 - operation-specific timeout is intentional
try:
result = await self._vmware_manager.execute(subcommand, args, timeout, log_level=log_level)
diff --git a/gns3server/controller/marker_replay.py b/gns3server/controller/marker_replay.py
index 6254d445a..2bd75ea92 100644
--- a/gns3server/controller/marker_replay.py
+++ b/gns3server/controller/marker_replay.py
@@ -526,7 +526,7 @@ class _SharkdSession:
self.proc.stdin.write((json.dumps(request) + "\n").encode())
await self.proc.stdin.drain()
raw = await asyncio.wait_for(self.proc.stdout.readline(), RPC_TIMEOUT_SECONDS)
- except asyncio.TimeoutError:
+ except TimeoutError:
await self._abort(f"sharkd timed out after {RPC_TIMEOUT_SECONDS:.0f}s on {method!r}")
except (OSError, ValueError) as e:
# ValueError: a reply line over the StreamReader limit — the
@@ -564,7 +564,7 @@ class _SharkdSession:
# the caller — leak the process with a log instead.
try:
await asyncio.wait_for(self.proc.wait(), timeout=2.0)
- except asyncio.TimeoutError:
+ except TimeoutError:
log.warning("sharkd session for %s did not exit after kill", self.pcap)
except ProcessLookupError:
pass
diff --git a/gns3server/controller/node.py b/gns3server/controller/node.py
index 5e7d7566c..419950839 100644
--- a/gns3server/controller/node.py
+++ b/gns3server/controller/node.py
@@ -976,7 +976,7 @@ class Node:
try:
await self._start_on_compute(data)
break
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ControllerTimeoutError(f"Timeout when starting {self._name}")
except ComputeError as e:
conflicted = self._vxlan_conflicted_link(e)
@@ -1057,7 +1057,7 @@ class Node:
# We don't care if a node is down at this step
if strict:
raise
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ControllerTimeoutError(f"Timeout when stopping {self._name}")
if stopped:
self._confirm_status("stopped")
@@ -1082,7 +1082,7 @@ class Node:
"""
try:
await self.post("/suspend", timeout=240)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ControllerTimeoutError(f"Timeout when suspending {self._name}")
self._confirm_status("suspended")
@@ -1092,7 +1092,7 @@ class Node:
"""
try:
await self.post("/reload", timeout=240)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ControllerTimeoutError(f"Timeout when reloading {self._name}")
self._confirm_status("started")
@@ -1104,7 +1104,7 @@ class Node:
if self._console and self._console_type in ("telnet", "ssh"):
try:
await self.post("/console/reset", timeout=240)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise ControllerTimeoutError(f"Timeout when reset console {self._name}")
async def get(self, path="", **kwargs):
diff --git a/gns3server/controller/snapshot.py b/gns3server/controller/snapshot.py
index 811db0fde..b7d615d76 100644
--- a/gns3server/controller/snapshot.py
+++ b/gns3server/controller/snapshot.py
@@ -23,7 +23,7 @@ import tempfile
import time
import uuid
import zipfile
-from datetime import datetime, timezone
+from datetime import UTC, datetime
import aiofiles
@@ -63,19 +63,19 @@ class Snapshot:
if created_at:
self._created_at = created_at
else:
- self._created_at = int(datetime.now(timezone.utc).timestamp())
+ self._created_at = int(datetime.now(UTC).timestamp())
if not filename:
filename = self._name + ".gns3snapshot"
else:
self._name = filename.rsplit("_", 2)[0]
datestring = filename.replace(self._name + "_", "").split(".")[0]
self._created_at = int(
- datetime.strptime(datestring, FILENAME_DATETIME_FORMAT).replace(tzinfo=timezone.utc).timestamp()
+ datetime.strptime(datestring, FILENAME_DATETIME_FORMAT).replace(tzinfo=UTC).timestamp()
)
if not description:
date = (
- datetime.fromtimestamp(self._created_at, tz=timezone.utc)
+ datetime.fromtimestamp(self._created_at, tz=UTC)
.replace(tzinfo=None)
.strftime(DESCRIPTION_DATETIME_FORMAT)
)
diff --git a/gns3server/db/repositories/images.py b/gns3server/db/repositories/images.py
index 90d5f601b..639c8f2e6 100644
--- a/gns3server/db/repositories/images.py
+++ b/gns3server/db/repositories/images.py
@@ -18,7 +18,7 @@
import asyncio
import logging
import os
-from datetime import datetime, timezone
+from datetime import UTC, datetime
from typing import Callable, List, Optional, cast
from sqlalchemy import delete, select, update
@@ -166,8 +166,8 @@ class ImagesRepository(BaseRepository):
values.update(
availability="available",
last_error=None,
- last_seen_at=datetime.now(timezone.utc).replace(tzinfo=None),
- last_verified_at=datetime.now(timezone.utc).replace(tzinfo=None),
+ last_seen_at=datetime.now(UTC).replace(tzinfo=None),
+ last_verified_at=datetime.now(UTC).replace(tzinfo=None),
)
for attempt in range(2):
try:
diff --git a/gns3server/schemas/common.py b/gns3server/schemas/common.py
index 8631e7bab..33577209b 100644
--- a/gns3server/schemas/common.py
+++ b/gns3server/schemas/common.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import Any, Optional
from pydantic import BaseModel, Field, field_validator
@@ -33,7 +33,7 @@ class ErrorMessage(BaseModel):
details: Optional[dict[str, Any]] = Field(None, description="Optional structured data about the error")
-class NodeStatus(str, Enum):
+class NodeStatus(StrEnum):
"""
Supported node statuses.
"""
@@ -85,7 +85,7 @@ class ExtraConfig(BaseModel):
return v
-class ConsoleType(str, Enum):
+class ConsoleType(StrEnum):
"""
Supported console types.
"""
@@ -101,7 +101,7 @@ class ConsoleType(str, Enum):
docker_exec = "docker_exec"
-class AuxType(str, Enum):
+class AuxType(StrEnum):
"""
Supported auxiliary console types.
"""
diff --git a/gns3server/schemas/compute/cloud_nodes.py b/gns3server/schemas/compute/cloud_nodes.py
index 274bf23ee..4adfbee35 100644
--- a/gns3server/schemas/compute/cloud_nodes.py
+++ b/gns3server/schemas/compute/cloud_nodes.py
@@ -15,7 +15,7 @@
# along with this program. If not, see .
import re
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional, Union
from uuid import UUID
@@ -27,17 +27,17 @@ from ..update import PartialUpdateModel
LINUX_TAP_NAME_PATTERN = r"^[a-zA-Z0-9_.:-]{1,15}$"
-class HostInterfaceType(str, Enum):
+class HostInterfaceType(StrEnum):
ethernet = "ethernet"
tap = "tap"
-class IPAddressFamily(str, Enum):
+class IPAddressFamily(StrEnum):
ipv4 = "ipv4"
ipv6 = "ipv6"
-class InterfaceStatus(str, Enum):
+class InterfaceStatus(StrEnum):
up = "up"
down = "down"
@@ -70,7 +70,7 @@ class HostInterface(BaseModel):
flags: List[str] = Field(default_factory=list, description="Interface flags")
-class EthernetType(str, Enum):
+class EthernetType(StrEnum):
ethernet = "ethernet"
@@ -85,7 +85,7 @@ class EthernetPort(BaseModel):
interface: str
-class TAPType(str, Enum):
+class TAPType(StrEnum):
tap = "tap"
@@ -113,7 +113,7 @@ class TAPPort(BaseModel):
return self
-class UDPType(str, Enum):
+class UDPType(StrEnum):
udp = "udp"
@@ -130,7 +130,7 @@ class UDPPort(BaseModel):
rport: int = Field(..., gt=0, le=65535, description="Remote port")
-class CloudConsoleType(str, Enum):
+class CloudConsoleType(StrEnum):
telnet = "telnet"
ssh = "ssh"
vnc = "vnc"
diff --git a/gns3server/schemas/compute/dynamips_nodes.py b/gns3server/schemas/compute/dynamips_nodes.py
index 8d7aaed98..e8a94ca70 100644
--- a/gns3server/schemas/compute/dynamips_nodes.py
+++ b/gns3server/schemas/compute/dynamips_nodes.py
@@ -15,7 +15,7 @@
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from uuid import UUID
@@ -25,7 +25,7 @@ from ..common import NodeStatus
from ..update import PartialUpdateModel
-class DynamipsPlatform(str, Enum):
+class DynamipsPlatform(StrEnum):
"""
Supported Dynamips Platforms.
"""
@@ -39,7 +39,7 @@ class DynamipsPlatform(str, Enum):
c1700 = "c1700"
-class DynamipsAdapters(str, Enum):
+class DynamipsAdapters(StrEnum):
"""
Supported Dynamips Network Modules.
"""
@@ -72,7 +72,7 @@ class DynamipsAdapters(str, Enum):
_ = ""
-class DynamipsWics(str, Enum):
+class DynamipsWics(StrEnum):
"""
Supported Dynamips WICs.
"""
@@ -83,7 +83,7 @@ class DynamipsWics(str, Enum):
_ = ""
-class DynamipsConsoleType(str, Enum):
+class DynamipsConsoleType(StrEnum):
"""
Supported Dynamips console types.
"""
@@ -93,7 +93,7 @@ class DynamipsConsoleType(str, Enum):
none = "none"
-class DynamipsNPE(str, Enum):
+class DynamipsNPE(StrEnum):
"""
Supported Dynamips NPE models.
"""
@@ -108,7 +108,7 @@ class DynamipsNPE(str, Enum):
npe_g2 = "npe-g2"
-class DynamipsMidplane(str, Enum):
+class DynamipsMidplane(StrEnum):
"""
Supported Dynamips Midplane models.
"""
diff --git a/gns3server/schemas/compute/ethernet_switch_nodes.py b/gns3server/schemas/compute/ethernet_switch_nodes.py
index 41328ca2f..e742ab5a2 100644
--- a/gns3server/schemas/compute/ethernet_switch_nodes.py
+++ b/gns3server/schemas/compute/ethernet_switch_nodes.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from uuid import UUID
@@ -24,13 +24,13 @@ from ..common import NodeStatus
from ..update import PartialUpdateModel
-class EthernetSwitchPortType(str, Enum):
+class EthernetSwitchPortType(StrEnum):
access = "access"
dot1q = "dot1q"
qinq = "qinq"
-class EthernetSwitchEtherType(str, Enum):
+class EthernetSwitchEtherType(StrEnum):
ethertype_8021q = "0x8100"
ethertype_qinq = "0x88A8"
ethertype_8021q9100 = "0x9100"
@@ -54,7 +54,7 @@ class EthernetSwitchPort(BaseModel):
return self
-class TelnetConsoleType(str, Enum):
+class TelnetConsoleType(StrEnum):
"""
Supported console types.
"""
diff --git a/gns3server/schemas/compute/nat_nodes.py b/gns3server/schemas/compute/nat_nodes.py
index 27cdcfc0e..042d2d58d 100644
--- a/gns3server/schemas/compute/nat_nodes.py
+++ b/gns3server/schemas/compute/nat_nodes.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional, Union
from uuid import UUID
@@ -24,7 +24,7 @@ from ..common import NodeStatus
from ..update import PartialUpdateModel
-class HostInterfaceType(str, Enum):
+class HostInterfaceType(StrEnum):
ethernet = "ethernet"
tap = "tap"
@@ -39,7 +39,7 @@ class HostInterface(BaseModel):
special: bool = Field(..., description="Whether the interface is non standard")
-class EthernetType(str, Enum):
+class EthernetType(StrEnum):
ethernet = "ethernet"
@@ -54,7 +54,7 @@ class EthernetPort(BaseModel):
interface: str
-class TAPType(str, Enum):
+class TAPType(StrEnum):
tap = "tap"
@@ -69,7 +69,7 @@ class TAPPort(BaseModel):
interface: str
-class UDPType(str, Enum):
+class UDPType(StrEnum):
udp = "udp"
diff --git a/gns3server/schemas/compute/nios.py b/gns3server/schemas/compute/nios.py
index 634b27dc9..9e8c0e325 100644
--- a/gns3server/schemas/compute/nios.py
+++ b/gns3server/schemas/compute/nios.py
@@ -15,13 +15,13 @@
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional, Union
from pydantic import BaseModel, Field
-class UDPNIOType(str, Enum):
+class UDPNIOType(StrEnum):
udp = "nio_udp"
@@ -43,7 +43,7 @@ class UDPNIO(BaseModel):
markers: Optional[dict] = Field(None, description="Traffic-insight markers")
-class EthernetNIOType(str, Enum):
+class EthernetNIOType(StrEnum):
ethernet = "nio_ethernet"
@@ -56,7 +56,7 @@ class EthernetNIO(BaseModel):
ethernet_device: str = Field(..., description="Ethernet device name e.g. eth0")
-class TAPNIOType(str, Enum):
+class TAPNIOType(StrEnum):
tap = "nio_tap"
@@ -69,7 +69,7 @@ class TAPNIO(BaseModel):
tap_device: str = Field(..., description="TAP device name e.g. tap0")
-class BridgeNIOType(str, Enum):
+class BridgeNIOType(StrEnum):
bridge = "nio_bridge"
@@ -133,7 +133,7 @@ class BridgeNIO(BaseModel):
markers: Optional[dict] = Field(None, description="Traffic-insight markers (attached to the veth host end)")
-class AnchorNIOType(str, Enum):
+class AnchorNIOType(StrEnum):
anchor = "nio_anchor"
diff --git a/gns3server/schemas/compute/qemu_nodes.py b/gns3server/schemas/compute/qemu_nodes.py
index 30c3b0f3d..0fef97597 100644
--- a/gns3server/schemas/compute/qemu_nodes.py
+++ b/gns3server/schemas/compute/qemu_nodes.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Literal, Optional
from uuid import UUID
@@ -24,7 +24,7 @@ from ..common import CustomAdapter, NodeStatus
from ..update import PartialUpdateModel
-class QemuPlatform(str, Enum):
+class QemuPlatform(StrEnum):
aarch64 = "aarch64"
alpha = "alpha"
arm = "arm"
@@ -55,7 +55,7 @@ class QemuPlatform(str, Enum):
xtensaeb = "xtensaeb"
-class QemuConsoleType(str, Enum):
+class QemuConsoleType(StrEnum):
"""
Supported console types.
"""
@@ -68,7 +68,7 @@ class QemuConsoleType(str, Enum):
none = "none"
-class QemuBootPriority(str, Enum):
+class QemuBootPriority(StrEnum):
"""
Supported boot priority types.
"""
@@ -84,7 +84,7 @@ class QemuBootPriority(str, Enum):
nd = "nd"
-class QemuOnCloseAction(str, Enum):
+class QemuOnCloseAction(StrEnum):
"""
Supported actions when closing Qemu VM.
"""
@@ -94,7 +94,7 @@ class QemuOnCloseAction(str, Enum):
save_vm_state = "save_vm_state"
-class QemuProcessPriority(str, Enum):
+class QemuProcessPriority(StrEnum):
realtime = "realtime"
very_high = "very high"
high = "high"
@@ -103,7 +103,7 @@ class QemuProcessPriority(str, Enum):
very_low = "very low"
-class QemuAdapterType(str, Enum):
+class QemuAdapterType(StrEnum):
"""
Supported Qemu VM adapter types.
"""
@@ -135,7 +135,7 @@ class QemuAdapterType(str, Enum):
vmxnet3 = "vmxnet3"
-class QemuDiskInterfaceType(str, Enum):
+class QemuDiskInterfaceType(StrEnum):
"""
Supported Qemu VM disk interface types.
"""
diff --git a/gns3server/schemas/compute/virtualbox_nodes.py b/gns3server/schemas/compute/virtualbox_nodes.py
index cd6b6b208..ca52364d0 100644
--- a/gns3server/schemas/compute/virtualbox_nodes.py
+++ b/gns3server/schemas/compute/virtualbox_nodes.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from uuid import UUID
@@ -24,7 +24,7 @@ from ..common import CustomAdapter, NodeStatus
from ..update import PartialUpdateModel
-class VirtualBoxConsoleType(str, Enum):
+class VirtualBoxConsoleType(StrEnum):
"""
Supported console types.
"""
@@ -34,7 +34,7 @@ class VirtualBoxConsoleType(str, Enum):
none = "none"
-class VirtualBoxOnCloseAction(str, Enum):
+class VirtualBoxOnCloseAction(StrEnum):
"""
Supported actions when closing VirtualBox VM.
"""
@@ -44,7 +44,7 @@ class VirtualBoxOnCloseAction(str, Enum):
save_vm_state = "save_vm_state"
-class VirtualBoxAdapterType(str, Enum):
+class VirtualBoxAdapterType(StrEnum):
pcnet_pci_ii = ("PCnet-PCI II (Am79C970A)",)
pcnet_fast_iii = ("PCNet-FAST III (Am79C973)",)
intel_pro_1000_mt_desktop = ("Intel PRO/1000 MT Desktop (82540EM)",)
diff --git a/gns3server/schemas/compute/vmware_nodes.py b/gns3server/schemas/compute/vmware_nodes.py
index 35186e17e..e46dc5c0b 100644
--- a/gns3server/schemas/compute/vmware_nodes.py
+++ b/gns3server/schemas/compute/vmware_nodes.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from uuid import UUID
@@ -24,7 +24,7 @@ from ..common import CustomAdapter, NodeStatus
from ..update import PartialUpdateModel
-class VMwareConsoleType(str, Enum):
+class VMwareConsoleType(StrEnum):
"""
Supported console types.
"""
@@ -34,7 +34,7 @@ class VMwareConsoleType(str, Enum):
none = "none"
-class VMwareOnCloseAction(str, Enum):
+class VMwareOnCloseAction(StrEnum):
"""
Supported actions when closing VMware VM.
"""
@@ -44,7 +44,7 @@ class VMwareOnCloseAction(str, Enum):
save_vm_state = "save_vm_state"
-class VMwareAdapterType(str, Enum):
+class VMwareAdapterType(StrEnum):
"""
Supported VMware VM adapter types.
"""
diff --git a/gns3server/schemas/compute/vpcs_nodes.py b/gns3server/schemas/compute/vpcs_nodes.py
index bc1249a9e..00eb72c4f 100644
--- a/gns3server/schemas/compute/vpcs_nodes.py
+++ b/gns3server/schemas/compute/vpcs_nodes.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import Optional
from uuid import UUID
@@ -24,7 +24,7 @@ from ..common import NodeStatus
from ..update import PartialUpdateModel
-class ConsoleType(str, Enum):
+class ConsoleType(StrEnum):
"""
Supported console types.
"""
diff --git a/gns3server/schemas/config.py b/gns3server/schemas/config.py
index f54725b67..74dba2dde 100644
--- a/gns3server/schemas/config.py
+++ b/gns3server/schemas/config.py
@@ -15,7 +15,7 @@
# along with this program. If not, see .
import socket
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from pydantic import BaseModel, ConfigDict, DirectoryPath, Field, FilePath, SecretStr, field_validator, model_validator
@@ -126,12 +126,12 @@ class WebWiresharkSettings(BaseModel):
model_config = ConfigDict(validate_assignment=True, str_strip_whitespace=True)
-class ServerProtocol(str, Enum):
+class ServerProtocol(StrEnum):
http = "http"
https = "https"
-class UbridgeControlTransport(str, Enum):
+class UbridgeControlTransport(StrEnum):
# TCP control channel: -H host:port. ubridge now binds loopback by default,
# so this is reachable only locally. Retained for backward compatibility.
tcp = "tcp"
@@ -140,7 +140,7 @@ class UbridgeControlTransport(str, Enum):
unix = "unix"
-class BuiltinSymbolTheme(str, Enum):
+class BuiltinSymbolTheme(StrEnum):
classic = "Classic"
affinity_square_blue = "Affinity-square-blue"
affinity_square_red = "Affinity-square-red"
diff --git a/gns3server/schemas/controller/appliances.py b/gns3server/schemas/controller/appliances.py
index a62617398..33799b4b9 100644
--- a/gns3server/schemas/controller/appliances.py
+++ b/gns3server/schemas/controller/appliances.py
@@ -16,7 +16,7 @@
# Unified Pydantic model supporting both appliance registry versions using discriminated unions
-from enum import Enum
+from enum import StrEnum
from typing import Annotated, List, Literal, Optional, Union
from uuid import UUID
@@ -30,7 +30,7 @@ from ..compute.qemu_nodes import QemuCdromInterface
# ============================================================================
-class Category(str, Enum):
+class Category(StrEnum):
"""Appliance category enum"""
router = "router"
@@ -40,7 +40,7 @@ class Category(str, Enum):
guest = "guest"
-class Status(str, Enum):
+class Status(StrEnum):
"""Appliance status enum"""
stable = "stable"
@@ -48,7 +48,7 @@ class Status(str, Enum):
broken = "broken"
-class Availability(str, Enum):
+class Availability(StrEnum):
"""Image availability enum"""
free = "free"
@@ -57,7 +57,7 @@ class Availability(str, Enum):
service_contract = "service-contract"
-class Compression(str, Enum):
+class Compression(StrEnum):
"""Compression type enum"""
bzip2 = "bzip2"
@@ -69,7 +69,7 @@ class Compression(str, Enum):
field_7z = "7z"
-class DynamipsPlatform(str, Enum):
+class DynamipsPlatform(StrEnum):
"""Dynamips platform enum"""
c1700 = "c1700"
@@ -81,7 +81,7 @@ class DynamipsPlatform(str, Enum):
c7200 = "c7200"
-class DynamipsChassis(str, Enum):
+class DynamipsChassis(StrEnum):
"""Dynamips chassis enum"""
chassis_1720 = "1720"
@@ -104,7 +104,7 @@ class DynamipsChassis(str, Enum):
_ = ""
-class DynamipsSlot(str, Enum):
+class DynamipsSlot(StrEnum):
"""Dynamips slot enum"""
C7200_IO_2FE = "C7200-IO-2FE"
@@ -135,7 +135,7 @@ class DynamipsSlot(str, Enum):
_ = ""
-class DynamipsWic(str, Enum):
+class DynamipsWic(StrEnum):
"""Dynamips WIC enum"""
WIC_1ENET = "WIC-1ENET"
@@ -144,14 +144,14 @@ class DynamipsWic(str, Enum):
_ = ""
-class DynamipsMidplane(str, Enum):
+class DynamipsMidplane(StrEnum):
"""Dynamips midplane enum"""
std = "std"
vxr = "vxr"
-class DynamipsNpe(str, Enum):
+class DynamipsNpe(StrEnum):
"""Dynamips NPE enum"""
npe_100 = "npe-100"
@@ -164,7 +164,7 @@ class DynamipsNpe(str, Enum):
npe_g2 = "npe-g2"
-class QemuConsoleType(str, Enum):
+class QemuConsoleType(StrEnum):
"""Qemu console type enum"""
telnet = "telnet"
@@ -175,7 +175,7 @@ class QemuConsoleType(str, Enum):
none = "none"
-class QemuAdapterType(str, Enum):
+class QemuAdapterType(StrEnum):
"""Qemu adapter type enum"""
e1000 = "e1000"
@@ -201,7 +201,7 @@ class QemuAdapterType(str, Enum):
vmxnet3 = "vmxnet3"
-class QemuPlatform(str, Enum):
+class QemuPlatform(StrEnum):
"""Qemu platform enum"""
aarch64 = "aarch64"
@@ -234,7 +234,7 @@ class QemuPlatform(str, Enum):
xtensaeb = "xtensaeb"
-class QemuDiskInterface(str, Enum):
+class QemuDiskInterface(StrEnum):
"""Disk interface enum"""
ide = "ide"
@@ -249,7 +249,7 @@ class QemuDiskInterface(str, Enum):
none = "none"
-class QemuBootPriority(str, Enum):
+class QemuBootPriority(StrEnum):
"""Boot priority enum"""
c = "c"
@@ -263,7 +263,7 @@ class QemuBootPriority(str, Enum):
nd = "nd"
-class QemuOnClose(str, Enum):
+class QemuOnClose(StrEnum):
"""Qemu on_close action enum"""
power_off = "power_off"
@@ -271,7 +271,7 @@ class QemuOnClose(str, Enum):
save_vm_state = "save_vm_state"
-class QemuProcessPriority(str, Enum):
+class QemuProcessPriority(StrEnum):
"""Qemu process priority enum"""
realtime = "realtime"
@@ -282,7 +282,7 @@ class QemuProcessPriority(str, Enum):
very_low = "very low"
-class DockerConsoleType(str, Enum):
+class DockerConsoleType(StrEnum):
"""Docker console type enum"""
telnet = "telnet"
@@ -294,13 +294,13 @@ class DockerConsoleType(str, Enum):
docker_exec = "docker_exec"
-class ChecksumType(str, Enum):
+class ChecksumType(StrEnum):
"""Checksum type enum"""
md5 = "md5"
-class TemplateType(str, Enum):
+class TemplateType(StrEnum):
"""Template type enum"""
docker = "docker"
@@ -314,7 +314,7 @@ class TemplateType(str, Enum):
# ============================================================================
-class Kvm(str, Enum):
+class Kvm(StrEnum):
"""KVM requirements enum"""
require = "require"
diff --git a/gns3server/schemas/controller/computes.py b/gns3server/schemas/controller/computes.py
index f7c1fa791..873228a70 100644
--- a/gns3server/schemas/controller/computes.py
+++ b/gns3server/schemas/controller/computes.py
@@ -15,7 +15,7 @@
# along with this program. If not, see .
import uuid
-from enum import Enum
+from enum import StrEnum
from typing import Any, Optional, Union
from pydantic import BaseModel, ConfigDict, Field, SecretStr, model_validator
@@ -29,7 +29,7 @@ from .base import DateTimeModelMixin
from .capabilities import Capabilities
-class Protocol(str, Enum):
+class Protocol(StrEnum):
"""
Protocol supported to communicate with a compute.
"""
diff --git a/gns3server/schemas/controller/gns3vm.py b/gns3server/schemas/controller/gns3vm.py
index 2c39a5657..1006b4ea1 100644
--- a/gns3server/schemas/controller/gns3vm.py
+++ b/gns3server/schemas/controller/gns3vm.py
@@ -14,13 +14,13 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import Optional
from pydantic import BaseModel, Field
-class WhenExit(str, Enum):
+class WhenExit(StrEnum):
"""
What to do with the VM when GNS3 VM exits.
"""
@@ -30,7 +30,7 @@ class WhenExit(str, Enum):
keep = "keep"
-class Engine(str, Enum):
+class Engine(StrEnum):
"""
"The engine to use for the GNS3 VM.
"""
diff --git a/gns3server/schemas/controller/images.py b/gns3server/schemas/controller/images.py
index 5d99c296c..9a45ec525 100644
--- a/gns3server/schemas/controller/images.py
+++ b/gns3server/schemas/controller/images.py
@@ -15,7 +15,7 @@
# along with this program. If not, see .
from datetime import datetime
-from enum import Enum
+from enum import StrEnum
from typing import Annotated, List, Literal, Optional
from uuid import UUID
@@ -24,7 +24,7 @@ from pydantic import BaseModel, ConfigDict, Field
from .base import DateTimeModelMixin
-class ImageType(str, Enum):
+class ImageType(StrEnum):
qemu = "qemu"
ios = "ios"
iou = "iou"
diff --git a/gns3server/schemas/controller/links.py b/gns3server/schemas/controller/links.py
index e1ac4993c..5f07350f9 100644
--- a/gns3server/schemas/controller/links.py
+++ b/gns3server/schemas/controller/links.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import Annotated, List, Literal, Optional, Tuple, Union
from uuid import UUID, uuid4
@@ -62,7 +62,7 @@ class LinkNodeCreate(BaseModel):
return self
-class LinkType(str, Enum):
+class LinkType(StrEnum):
"""
Link type.
"""
@@ -81,7 +81,7 @@ class LinkStyle(BaseModel):
control_offset: Optional[Tuple[float, float]] = None
-class LinkFilterType(str, Enum):
+class LinkFilterType(StrEnum):
"""
Packet filter type. The first five run on relay and kernel-datapath links
alike; the rest are kernel-datapath only (tc netem / cls_bpf / the eBPF
diff --git a/gns3server/schemas/controller/nodes.py b/gns3server/schemas/controller/nodes.py
index ff1d90d2c..0ca9d710c 100644
--- a/gns3server/schemas/controller/nodes.py
+++ b/gns3server/schemas/controller/nodes.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional, Union
from uuid import UUID, uuid4
@@ -25,7 +25,7 @@ from ..update import PartialUpdateModel
from .labels import Label
-class NodeType(str, Enum):
+class NodeType(StrEnum):
"""
Supported node types.
"""
@@ -56,7 +56,7 @@ class Image(BaseModel):
filesize: Optional[int] = None
-class LinkType(str, Enum):
+class LinkType(StrEnum):
"""
Supported link types.
"""
@@ -65,7 +65,7 @@ class LinkType(str, Enum):
serial = "serial"
-class DataLinkType(str, Enum):
+class DataLinkType(StrEnum):
"""
Supported data link types.
"""
diff --git a/gns3server/schemas/controller/notifications.py b/gns3server/schemas/controller/notifications.py
index 11ed85e4c..f2789d240 100644
--- a/gns3server/schemas/controller/notifications.py
+++ b/gns3server/schemas/controller/notifications.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import Any, Dict, List, Optional, Union
from uuid import UUID
@@ -28,7 +28,7 @@ from .projects import Project
from .snapshots import Snapshot
-class NotificationAction(str, Enum):
+class NotificationAction(StrEnum):
"""
Values of the "action" field of a notification.
"""
diff --git a/gns3server/schemas/controller/pools.py b/gns3server/schemas/controller/pools.py
index f72c488d4..a550837db 100644
--- a/gns3server/schemas/controller/pools.py
+++ b/gns3server/schemas/controller/pools.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import Optional
from uuid import UUID
@@ -23,7 +23,7 @@ from pydantic import BaseModel, ConfigDict, Field
from .base import DateTimeModelMixin
-class ResourceType(str, Enum):
+class ResourceType(StrEnum):
project = "project"
diff --git a/gns3server/schemas/controller/projects.py b/gns3server/schemas/controller/projects.py
index 138fd71ea..b0c20addc 100644
--- a/gns3server/schemas/controller/projects.py
+++ b/gns3server/schemas/controller/projects.py
@@ -16,7 +16,7 @@
from datetime import datetime
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from uuid import UUID
@@ -25,7 +25,7 @@ from pydantic import BaseModel, Field, HttpUrl
from ..update import PartialUpdateModel
-class ProjectStatus(str, Enum):
+class ProjectStatus(StrEnum):
"""
Supported project statuses.
"""
@@ -117,7 +117,7 @@ class NodeFile(BaseModel):
file_type: str = Field(..., description="File type determined by the file command")
-class ProjectCompression(str, Enum):
+class ProjectCompression(StrEnum):
"""
Supported project compression.
"""
diff --git a/gns3server/schemas/controller/rbac.py b/gns3server/schemas/controller/rbac.py
index 60ff78e96..73d36d655 100644
--- a/gns3server/schemas/controller/rbac.py
+++ b/gns3server/schemas/controller/rbac.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from uuid import UUID
@@ -37,7 +37,7 @@ class Privilege(DateTimeModelMixin, PrivilegeBase):
model_config = ConfigDict(from_attributes=True)
-class ACEType(str, Enum):
+class ACEType(StrEnum):
user = "user"
group = "group"
diff --git a/gns3server/schemas/controller/snapshots.py b/gns3server/schemas/controller/snapshots.py
index d86fd21aa..adbb7b726 100644
--- a/gns3server/schemas/controller/snapshots.py
+++ b/gns3server/schemas/controller/snapshots.py
@@ -15,7 +15,7 @@
# along with this program. If not, see .
-from datetime import datetime, timezone
+from datetime import UTC, datetime
from typing import Optional
from uuid import UUID
@@ -54,5 +54,5 @@ class Snapshot(SnapshotBase):
@classmethod
def _derive_created(cls, data):
if isinstance(data, dict) and "created" not in data and data.get("created_at") is not None:
- data = {**data, "created": datetime.fromtimestamp(data["created_at"], tz=timezone.utc)}
+ data = {**data, "created": datetime.fromtimestamp(data["created_at"], tz=UTC)}
return data
diff --git a/gns3server/schemas/controller/templates/__init__.py b/gns3server/schemas/controller/templates/__init__.py
index 59519318a..942de8b16 100644
--- a/gns3server/schemas/controller/templates/__init__.py
+++ b/gns3server/schemas/controller/templates/__init__.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from uuid import UUID
@@ -25,7 +25,7 @@ from ..base import DateTimeModelMixin
from ..nodes import NodeType
-class Category(str, Enum):
+class Category(StrEnum):
"""
Supported categories
"""
diff --git a/gns3server/schemas/controller/templates/dynamips_templates.py b/gns3server/schemas/controller/templates/dynamips_templates.py
index ac43ee13f..492742b80 100644
--- a/gns3server/schemas/controller/templates/dynamips_templates.py
+++ b/gns3server/schemas/controller/templates/dynamips_templates.py
@@ -14,7 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import Optional
from pydantic import Field
@@ -123,7 +123,7 @@ class C3745DynamipsTemplate(C3745DynamipsTemplateBase):
image: str = Field(..., description="Path to the IOS image")
-class C3600ChassisType(str, Enum):
+class C3600ChassisType(StrEnum):
chassis_3620 = "3620"
chassis_3640 = "3640"
chassis_3660 = "3660"
@@ -162,7 +162,7 @@ class C2691DynamipsTemplate(C2691DynamipsTemplateBase):
image: str = Field(..., description="Path to the IOS image")
-class C2600ChassisType(str, Enum):
+class C2600ChassisType(StrEnum):
chassis_2610 = "2610"
chassis_2620 = "2620"
chassis_2610xm = "2610XM"
@@ -191,7 +191,7 @@ class C2600DynamipsTemplate(C2600DynamipsTemplateBase):
image: str = Field(..., description="Path to the IOS image")
-class C1700ChassisType(str, Enum):
+class C1700ChassisType(StrEnum):
chassis_1720 = "1720"
chassis_1721 = "1721"
chassis_1750 = "1750"
diff --git a/gns3server/schemas/controller/templates/ethernet_switch_templates.py b/gns3server/schemas/controller/templates/ethernet_switch_templates.py
index 03d72e64f..a50b407e7 100644
--- a/gns3server/schemas/controller/templates/ethernet_switch_templates.py
+++ b/gns3server/schemas/controller/templates/ethernet_switch_templates.py
@@ -15,7 +15,7 @@
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from pydantic import Field
@@ -37,7 +37,7 @@ DEFAULT_PORTS = [
]
-class ConsoleType(str, Enum):
+class ConsoleType(StrEnum):
"""
Supported console types for Ethernet switch nodes.
"""
diff --git a/gns3server/schemas/controller/topology.py b/gns3server/schemas/controller/topology.py
index 365ca5983..f13083fd2 100644
--- a/gns3server/schemas/controller/topology.py
+++ b/gns3server/schemas/controller/topology.py
@@ -19,7 +19,7 @@
# This file contains the validation for checking a .gns3 file
#
-from enum import Enum
+from enum import StrEnum
from typing import List, Optional
from uuid import UUID
@@ -37,7 +37,7 @@ class TopologyLink(Link):
filters: Optional[dict] = None
-class TopologyType(str, Enum):
+class TopologyType(StrEnum):
topology = "topology"
diff --git a/gns3server/schemas/qemu_disk_image.py b/gns3server/schemas/qemu_disk_image.py
index 9a3b591b9..60efddaff 100644
--- a/gns3server/schemas/qemu_disk_image.py
+++ b/gns3server/schemas/qemu_disk_image.py
@@ -14,13 +14,13 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
-from enum import Enum
+from enum import StrEnum
from typing import Optional
from pydantic import BaseModel, Field
-class QemuDiskImageFormat(str, Enum):
+class QemuDiskImageFormat(StrEnum):
"""
Supported Qemu disk image formats.
"""
@@ -33,7 +33,7 @@ class QemuDiskImageFormat(str, Enum):
raw = "raw"
-class QemuDiskImagePreallocation(str, Enum):
+class QemuDiskImagePreallocation(StrEnum):
"""
Supported Qemu disk image pre-allocation options.
"""
@@ -44,7 +44,7 @@ class QemuDiskImagePreallocation(str, Enum):
full = "full"
-class QemuDiskImageOnOff(str, Enum):
+class QemuDiskImageOnOff(StrEnum):
"""
Supported Qemu image on/off options.
"""
@@ -53,7 +53,7 @@ class QemuDiskImageOnOff(str, Enum):
off = "off"
-class QemuDiskImageSubformat(str, Enum):
+class QemuDiskImageSubformat(StrEnum):
"""
Supported Qemu disk image sub-format options.
"""
@@ -67,7 +67,7 @@ class QemuDiskImageSubformat(str, Enum):
monolithic_flat = "monolithicFlat"
-class QemuDiskImageAdapterType(str, Enum):
+class QemuDiskImageAdapterType(StrEnum):
"""
Supported Qemu disk image on/off options.
"""
diff --git a/gns3server/services/authentication.py b/gns3server/services/authentication.py
index d4d6f6b79..6652439cb 100644
--- a/gns3server/services/authentication.py
+++ b/gns3server/services/authentication.py
@@ -18,7 +18,7 @@ import base64
import json
import logging
import time
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
from typing import Optional
import bcrypt
@@ -64,7 +64,7 @@ class AuthService:
def _create_token(self, username, token_version, token_type, expires_in, secret_key=None) -> str:
"""Shared helper to create any kind of signed JWT token."""
- issued_at = datetime.now(timezone.utc)
+ issued_at = datetime.now(UTC)
expire = issued_at + timedelta(minutes=expires_in)
to_encode = {"sub": username, "iat": issued_at, "exp": expire, "ver": token_version, "type": token_type}
if secret_key is None:
diff --git a/gns3server/services/image_reconciliation.py b/gns3server/services/image_reconciliation.py
index 6fac6bb3f..5411d6992 100644
--- a/gns3server/services/image_reconciliation.py
+++ b/gns3server/services/image_reconciliation.py
@@ -6,7 +6,7 @@ import os
import stat
import threading
import uuid
-from datetime import datetime, timezone
+from datetime import UTC, datetime
from sqlalchemy import delete, select, update
from sqlalchemy.ext.asyncio import AsyncSession
@@ -44,7 +44,7 @@ RAW_EXTENSIONS = {".raw", ".img", ".iso", ".fd", ".vhd", ".vdi", ".bin"}
def utcnow():
- return datetime.now(timezone.utc).replace(tzinfo=None)
+ return datetime.now(UTC).replace(tzinfo=None)
def ignored(name):
@@ -476,7 +476,7 @@ class ImageReconciliationService:
try:
await asyncio.wait_for(self.dirty.wait(), Config.instance().settings.Server.image_sync_interval)
await asyncio.sleep(1) # debounce bursts of file/directory events
- except asyncio.TimeoutError:
+ except TimeoutError:
pass
except asyncio.CancelledError:
break
diff --git a/gns3server/utils/asyncio/__init__.py b/gns3server/utils/asyncio/__init__.py
index 98b72c66b..d6f17f99e 100644
--- a/gns3server/utils/asyncio/__init__.py
+++ b/gns3server/utils/asyncio/__init__.py
@@ -85,7 +85,7 @@ async def subprocess_check_output(*args, cwd=None, env=None, stderr=False):
return output.decode("utf-8", errors="ignore")
-async def wait_for_process_termination(process, timeout=10):
+async def wait_for_process_termination(process, timeout=10): # noqa: ASYNC109 - operation-specific timeout is intentional
"""
Wait for a process terminate, and raise asyncio.TimeoutError in case of
timeout.
@@ -118,14 +118,14 @@ def monitor_process(process, termination_callback):
asyncio.ensure_future(_check_process(process, termination_callback))
-async def wait_for_file_creation(path, timeout=60):
+async def wait_for_file_creation(path, timeout=60): # noqa: ASYNC109 - operation-specific timeout is intentional
while timeout > 0:
if os.path.exists(path):
return
await asyncio.sleep(0.5)
timeout -= 0.5
- raise asyncio.TimeoutError()
+ raise TimeoutError()
def locking(f):
diff --git a/gns3server/utils/asyncio/serial.py b/gns3server/utils/asyncio/serial.py
index 8a20a0799..22c2e5690 100644
--- a/gns3server/utils/asyncio/serial.py
+++ b/gns3server/utils/asyncio/serial.py
@@ -66,7 +66,7 @@ async def _asyncio_open_serial_unix(path):
try:
# wait for VM to create the pipe file.
await wait_for_file_creation(path)
- except asyncio.TimeoutError:
+ except TimeoutError:
raise NodeError(f'Pipe file "{path}" is missing')
output = SerialReaderWriterProtocol()
diff --git a/gns3server/utils/asyncio/ssh_server.py b/gns3server/utils/asyncio/ssh_server.py
index 3064b729d..b67252a99 100644
--- a/gns3server/utils/asyncio/ssh_server.py
+++ b/gns3server/utils/asyncio/ssh_server.py
@@ -147,7 +147,7 @@ class AsyncioSSHServer:
try:
process.stdout.write(data)
await asyncio.wait_for(process.stdout.drain(), timeout=BROADCAST_DRAIN_TIMEOUT)
- except (OSError, ConnectionError, asyncio.TimeoutError, asyncssh.Error):
+ except (TimeoutError, OSError, ConnectionError, asyncssh.Error):
await self._disconnect_client(process)
except asyncio.CancelledError:
raise
diff --git a/gns3server/utils/asyncio/telnet_server.py b/gns3server/utils/asyncio/telnet_server.py
index 8cc248687..20d72e755 100644
--- a/gns3server/utils/asyncio/telnet_server.py
+++ b/gns3server/utils/asyncio/telnet_server.py
@@ -324,7 +324,7 @@ class AsyncioTelnetServer:
try:
connection.writer.write(data)
await asyncio.wait_for(connection.writer.drain(), timeout=BROADCAST_DRAIN_TIMEOUT)
- except (OSError, ConnectionError, asyncio.TimeoutError) as e:
+ except (TimeoutError, OSError, ConnectionError) as e:
client_info = self._get_peername(network_writer)
log.debug(
"Error sending data to client %s: %s, closing and removing from connection table.",
@@ -352,7 +352,7 @@ class AsyncioTelnetServer:
log.debug("Sending keepalive to client %s", client_info)
connection.writer.send_iac(IAC + NOP)
await asyncio.wait_for(connection.writer.drain(), timeout=BROADCAST_DRAIN_TIMEOUT)
- except (OSError, ConnectionError, asyncio.TimeoutError) as e:
+ except (TimeoutError, OSError, ConnectionError) as e:
log.debug(
"Keepalive failed for client %s: %s, closing connection.",
client_info,
diff --git a/gns3server/utils/notification_queue.py b/gns3server/utils/notification_queue.py
index 4c0e0d6a4..d4950a79c 100644
--- a/gns3server/utils/notification_queue.py
+++ b/gns3server/utils/notification_queue.py
@@ -38,7 +38,7 @@ class NotificationQueue(asyncio.Queue):
self._first = True
self._last_ping = None
- async def get(self, timeout):
+ async def get(self, timeout): # noqa: ASYNC109 - operation-specific timeout is intentional
"""
Return a notification, or a ping notification with server information
at least every `timeout` seconds. The ping used to be generated only
@@ -59,7 +59,7 @@ class NotificationQueue(asyncio.Queue):
try:
(action, msg, kwargs) = await asyncio.wait_for(super().get(), timeout - (now - self._last_ping))
return (action, msg, kwargs)
- except asyncio.TimeoutError:
+ except TimeoutError:
continue # the ping deadline has been reached
def _ping(self):
@@ -84,7 +84,7 @@ class NotificationQueue(asyncio.Queue):
log.warning(f"Could not get CPU and memory usage from psutil: {e}")
return msg
- async def get_json(self, timeout):
+ async def get_json(self, timeout): # noqa: ASYNC109 - operation-specific timeout is intentional
"""
Get a message as a JSON
"""
diff --git a/gns3server/utils/websocket_to_websocket.py b/gns3server/utils/websocket_to_websocket.py
index 12e410ce2..51e49ba5c 100644
--- a/gns3server/utils/websocket_to_websocket.py
+++ b/gns3server/utils/websocket_to_websocket.py
@@ -40,7 +40,7 @@ async def websocket_proxy(
target_url: str,
requested_protocols: Optional[list] = None,
buffer_size: int = 65536,
- timeout: Optional[float] = None,
+ timeout: Optional[float] = None, # noqa: ASYNC109 - operation-specific timeout is intentional
) -> None:
"""
Proxy binary WebSocket data between client and target WebSocket server.
@@ -142,7 +142,7 @@ async def websocket_proxy_with_manual_accept(
target_url: str,
requested_protocols: Optional[list] = None,
buffer_size: int = 65536,
- timeout: Optional[float] = None,
+ timeout: Optional[float] = None, # noqa: ASYNC109 - operation-specific timeout is intentional
) -> None:
"""
Proxy binary WebSocket data between client and target WebSocket server.
diff --git a/pyproject.toml b/pyproject.toml
index 138342e11..bf8cc0dc1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -76,7 +76,7 @@ asyncio_default_fixture_loop_scope = "function"
[tool.ruff]
-target-version = "py310"
+target-version = "py311"
line-length = 120
extend-exclude = [
"*.md",
diff --git a/tests/api/routes/controller/test_snapshots.py b/tests/api/routes/controller/test_snapshots.py
index aeff6d9f9..d4180a562 100644
--- a/tests/api/routes/controller/test_snapshots.py
+++ b/tests/api/routes/controller/test_snapshots.py
@@ -17,7 +17,7 @@
import os
import uuid
-from datetime import datetime, timezone
+from datetime import UTC, datetime
import pytest
import pytest_asyncio
@@ -57,7 +57,7 @@ class TestSnapshotRoutes:
assert len(response.json()) == 1
data = response.json()[0]
assert isinstance(data["created_at"], int)
- assert datetime.fromisoformat(data["created"]) == datetime.fromtimestamp(data["created_at"], tz=timezone.utc)
+ assert datetime.fromisoformat(data["created"]) == datetime.fromtimestamp(data["created_at"], tz=UTC)
async def test_delete_snapshot(
self, app: FastAPI, client: AsyncClient, project: Project, snapshot: Snapshot
@@ -86,7 +86,7 @@ class TestSnapshotRoutes:
assert len(os.listdir(os.path.join(project.path, "snapshots"))) == 2
data = response.json()
assert isinstance(data["created_at"], int)
- assert datetime.fromisoformat(data["created"]) == datetime.fromtimestamp(data["created_at"], tz=timezone.utc)
+ assert datetime.fromisoformat(data["created"]) == datetime.fromtimestamp(data["created_at"], tz=UTC)
async def test_snapshot_schema_timestamps(self, app: FastAPI) -> None:
diff --git a/tests/compute/docker/test_docker.py b/tests/compute/docker/test_docker.py
index 5905c30ea..7fe316f67 100644
--- a/tests/compute/docker/test_docker.py
+++ b/tests/compute/docker/test_docker.py
@@ -239,7 +239,7 @@ async def test_pull_image_propagates_timeout():
class Content:
async def read(self, size):
- raise asyncio.TimeoutError
+ raise TimeoutError
response = MagicMock()
response.content = Content()
diff --git a/tests/compute/docker/test_iol_docker_kernel_datapath.py b/tests/compute/docker/test_iol_docker_kernel_datapath.py
index 3491c128e..751b95d4d 100644
--- a/tests/compute/docker/test_iol_docker_kernel_datapath.py
+++ b/tests/compute/docker/test_iol_docker_kernel_datapath.py
@@ -64,7 +64,7 @@ async def vm(compute_project, manager):
def _capable(vm, monkeypatch):
"""Make both anchor-lifecycle probes answer True."""
- async def probe(timeout=15.0):
+ async def probe(timeout=15.0): # noqa: ASYNC109 - mock signature must match patched API
return True
monkeypatch.setattr("gns3server.compute.docker.iol_docker_vm.probe_bridge_tap_support", probe)
@@ -110,7 +110,7 @@ async def test_prepare_tap_datapath_falls_back_without_either_probe(vm, monkeypa
cannot carry kernel links.
"""
- async def probe(timeout=15.0):
+ async def probe(timeout=15.0): # noqa: ASYNC109 - mock signature must match patched API
return None
monkeypatch.setattr("gns3server.compute.docker.iol_docker_vm.probe_bridge_tap_support", probe)
diff --git a/tests/compute/docker/test_iol_docker_vm.py b/tests/compute/docker/test_iol_docker_vm.py
index 49b1952b5..9d6e5daa9 100644
--- a/tests/compute/docker/test_iol_docker_vm.py
+++ b/tests/compute/docker/test_iol_docker_vm.py
@@ -22,7 +22,6 @@ Image-free: everything is asserted against generated files, parsed knobs and
the uBridge command stream.
"""
-import asyncio
import glob
import json
import os
@@ -453,8 +452,8 @@ async def test_add_ubridge_connection_timeout_is_actionable(compute_project, man
_mock_wiring(vm)
vm._stop_ubridge = AsyncioMagicMock()
- async def raise_timeout(path, timeout=60):
- raise asyncio.TimeoutError()
+ async def raise_timeout(path, timeout=60): # noqa: ASYNC109 - mock signature must match patched API
+ raise TimeoutError()
with patch("gns3server.compute.docker.vendor_docker_vm.wait_for_file_creation", side_effect=raise_timeout):
with pytest.raises(DockerError) as excinfo:
diff --git a/tests/compute/iou/test_iou_kernel_datapath.py b/tests/compute/iou/test_iou_kernel_datapath.py
index 7121800e3..b3d947efe 100644
--- a/tests/compute/iou/test_iou_kernel_datapath.py
+++ b/tests/compute/iou/test_iou_kernel_datapath.py
@@ -60,7 +60,7 @@ async def vm(compute_project, manager):
def _capable(vm, monkeypatch):
"""Make the iol-tap capability probe answer True."""
- async def probe(timeout=15.0):
+ async def probe(timeout=15.0): # noqa: ASYNC109 - mock signature must match patched API
return True
monkeypatch.setattr("gns3server.compute.iou.iou_vm.probe_iol_tap_support", probe)
@@ -99,7 +99,7 @@ async def test_prepare_tap_datapath_falls_back_without_the_command(vm, monkeypat
datapath — it still starts, it just cannot carry kernel links.
"""
- async def probe(timeout=15.0):
+ async def probe(timeout=15.0): # noqa: ASYNC109 - mock signature must match patched API
return None
monkeypatch.setattr("gns3server.compute.iou.iou_vm.probe_iol_tap_support", probe)
diff --git a/tests/compute/qemu/test_qemu_kernel_datapath.py b/tests/compute/qemu/test_qemu_kernel_datapath.py
index a7226c08b..a2d57592a 100644
--- a/tests/compute/qemu/test_qemu_kernel_datapath.py
+++ b/tests/compute/qemu/test_qemu_kernel_datapath.py
@@ -358,7 +358,7 @@ async def test_stop_stops_the_process_before_deleting_the_taps(vm):
_running(vm)
order = []
- async def fake_termination(process, timeout=None):
+ async def fake_termination(process, timeout=None): # noqa: ASYNC109 - mock signature must match patched API
order.append("process")
async def fake_stop_ubridge():
@@ -476,7 +476,7 @@ async def test_concurrent_stops_sweep_every_tap_exactly_once(vm):
vm._ubridge_send = AsyncioMagicMock(side_effect=send)
- async def fake_termination(process, timeout=None):
+ async def fake_termination(process, timeout=None): # noqa: ASYNC109 - mock signature must match patched API
pass
with (
diff --git a/tests/compute/ubridge/test_tc_probe.py b/tests/compute/ubridge/test_tc_probe.py
index 289ce08a8..cab22077a 100644
--- a/tests/compute/ubridge/test_tc_probe.py
+++ b/tests/compute/ubridge/test_tc_probe.py
@@ -19,6 +19,7 @@ The standalone `tc capabilities` probe behind the compute /capabilities
payload: spawn a throwaway uBridge, ask once, cache by binary identity.
"""
+import asyncio
import os
import pytest
@@ -143,10 +144,14 @@ async def test_probe_missing_binary_means_unknown(monkeypatch):
async def test_probe_reprobes_when_the_binary_attributes_change(probe_env):
+ # Ensure following chmod changes mode even under restrictive test-process umasks.
+ os.chmod(probe_env, 0o700)
assert await probe_tc_capabilities() is not None
# chmod touches ctime but neither mtime nor size: a setcap/chmod remedy
# after a permission-related probe failure must be able to flip the
- # cached verdict without a server restart.
+ # cached verdict without a server restart. Wait for this filesystem's
+ # timestamp clock to advance before changing mode.
+ await asyncio.sleep(0.02)
os.chmod(probe_env, 0o600)
assert await probe_tc_capabilities() is not None
assert FakeHypervisor.spawned == 2
diff --git a/tests/services/test_image_reconciliation.py b/tests/services/test_image_reconciliation.py
index 47beef1b6..57611b66b 100644
--- a/tests/services/test_image_reconciliation.py
+++ b/tests/services/test_image_reconciliation.py
@@ -586,7 +586,7 @@ async def test_periodic_scan_recovers_without_watcher_events(inventory, config):
# Consume/close the unused Event.wait coroutine, then simulate a timeout.
args[0].close()
await real_sleep(0.01)
- raise asyncio.TimeoutError
+ raise TimeoutError
with (
patch.object(inventory, "_watch", new=AsyncMock()),
diff --git a/tests/stress/telnet_race_condition_test.py b/tests/stress/telnet_race_condition_test.py
index be996cc4b..0677c31c6 100644
--- a/tests/stress/telnet_race_condition_test.py
+++ b/tests/stress/telnet_race_condition_test.py
@@ -61,7 +61,7 @@ class TelnetClient:
log.warning(f"Client {self.client_id}: Send failed: {e}")
return False
- async def receive_response(self, timeout: float = 1.0) -> Optional[str]:
+ async def receive_response(self, timeout: float = 1.0) -> Optional[str]: # noqa: ASYNC109 - mock signature must match patched API
"""Receive response from server (optional)."""
if not self.reader or not self.connected:
return None
@@ -72,7 +72,7 @@ class TelnetClient:
response = data.decode("utf-8", errors="ignore")
log.debug(f"Client {self.client_id}: Received: {response[:50]}...")
return response
- except asyncio.TimeoutError:
+ except TimeoutError:
log.debug(f"Client {self.client_id}: No response (timeout)")
except Exception as e:
log.debug(f"Client {self.client_id}: Receive error: {e}")
diff --git a/tests/utils/asyncio/test_ssh_server.py b/tests/utils/asyncio/test_ssh_server.py
index 694bfd20c..3701e08bf 100644
--- a/tests/utils/asyncio/test_ssh_server.py
+++ b/tests/utils/asyncio/test_ssh_server.py
@@ -40,7 +40,7 @@ async def _connect(port):
)
-async def _wait_for_session_count(server, expected_count, timeout=2):
+async def _wait_for_session_count(server, expected_count, timeout=2): # noqa: ASYNC109 - mock signature must match patched API
deadline = asyncio.get_running_loop().time() + timeout
last_count = 0
while asyncio.get_running_loop().time() < deadline:
diff --git a/tests/utils/asyncio/test_telnet_server.py b/tests/utils/asyncio/test_telnet_server.py
index 08c2d8acd..8cb1a3a27 100644
--- a/tests/utils/asyncio/test_telnet_server.py
+++ b/tests/utils/asyncio/test_telnet_server.py
@@ -24,7 +24,7 @@ def _get_listen_port(server):
return server.sockets[0].getsockname()[1]
-async def _wait_for_connection_count(server, expected_count, timeout=2):
+async def _wait_for_connection_count(server, expected_count, timeout=2): # noqa: ASYNC109 - mock signature must match patched API
deadline = asyncio.get_running_loop().time() + timeout
last_count = 0
while asyncio.get_running_loop().time() < deadline:
diff --git a/tests/utils/test_packet_filter_validation.py b/tests/utils/test_packet_filter_validation.py
index 0cbf970b3..912ad8ce5 100644
--- a/tests/utils/test_packet_filter_validation.py
+++ b/tests/utils/test_packet_filter_validation.py
@@ -4,6 +4,7 @@ Unit tests for packet filter validation.
import pytest
+from gns3server.utils import packet_filter_validation
from gns3server.utils.packet_filter_validation import (
FilterValidationError,
filter_inactive_filters,
@@ -105,15 +106,26 @@ class TestPacketFilterValidation:
validate_filter_parameters("bpf", ["tcp port 80\nnot arp"])
validate_filter_parameters("bpf", ["tcp and not port 22\nhost 192.168.1.1\nicmp"])
- def test_bpf_multi_line_invalid(self):
+ def test_bpf_multi_line_invalid(self, monkeypatch):
"""Test multi-line BPF with invalid line."""
+ monkeypatch.setattr(
+ packet_filter_validation,
+ "validate_bpf_syntax",
+ lambda expression: {"valid": expression != "invalid!!!", "error": "syntax error"},
+ )
with pytest.raises(FilterValidationError) as excinfo:
validate_filter_parameters("bpf", ["tcp port 80\ninvalid!!!"])
err = str(excinfo.value).lower()
assert "syntax error" in err
- def test_bpf_invalid(self):
+ def test_bpf_invalid(self, monkeypatch):
"""Test invalid BPF parameters."""
+ monkeypatch.setattr(
+ packet_filter_validation,
+ "validate_bpf_syntax",
+ lambda expression: {"valid": expression != "tcp port", "error": "syntax error"},
+ )
+
# Wrong type
with pytest.raises(FilterValidationError, match="must be a string"):
validate_filter_parameters("bpf", [123])