refactor: use explicit conversion in f-strings

This commit is contained in:
Mark Paronyan 2026-09-30 20:45:43 +03:00
parent 0f66fd6c2b
commit c5b5091f5e
No known key found for this signature in database
GPG Key ID: 8DC7338A87FE22F8
32 changed files with 65 additions and 65 deletions

View File

@ -527,7 +527,7 @@ def tool_node(state: dict, config: RunnableConfig | None = None):
)
except Exception as e:
logger.error("Tool %s failed: %s", tool_name, e, exc_info=True)
observation = f"Error: {str(e)}"
observation = f"Error: {e!s}"
# Serialize observation to JSON string if it's not already a string
# This ensures ToolMessage.content is always JSON format, not Python

View File

@ -175,7 +175,7 @@ class Gns3Connector:
else:
raise HTTPError(f"v3 API authentication failed: {response.status_code} - {response.text}")
except Exception as e:
raise HTTPError(f"v3 API authentication error: {str(e)}") from e
raise HTTPError(f"v3 API authentication error: {e!s}") from e
def _is_token_expired(self) -> bool:
"""
@ -273,7 +273,7 @@ class Gns3Connector:
except Exception:
# If JSON parsing fails, return error with original text
return HTTPError(
f"Original Error: {str(e)}. GNS3 response text: {response.text}",
f"Original Error: {e!s}. GNS3 response text: {response.text}",
response=response,
)
return e

View File

@ -143,7 +143,7 @@ class GNS3ProjectInfoTool(BaseTool):
except Exception as e:
logger.error("Error retrieving GNS3 project info: %s", str(e))
return {"error": f"Failed to retrieve project info: {str(e)}"}
return {"error": f"Failed to retrieve project info: {e!s}"}
if __name__ == "__main__":

View File

@ -145,7 +145,7 @@ class GNS3TopologyTool(BaseTool):
except Exception as e:
logger.error("Error retrieving GNS3 topology: %s", str(e))
return {"error": f"Failed to retrieve topology: {str(e)}"}
return {"error": f"Failed to retrieve topology: {e!s}"}
def _clean_nodes_ports(self, data: dict) -> dict:
"""

View File

@ -239,7 +239,7 @@ def reload_injection_skills() -> dict[str, Any]:
logger.error(f"Error during skills reload: {e}")
return {
"success": False,
"message": f"Error: {str(e)}",
"message": f"Error: {e!s}",
"skill_count": len(INJECTION_SKILLS_REGISTRY),
"version": "",
}
@ -278,7 +278,7 @@ def reload_prompts() -> dict[str, Any]:
manager = get_skills_manager()
return {
"success": False,
"message": f"Error: {str(e)}",
"message": f"Error: {e!s}",
"prompt_count": manager.get_prompt_count() if manager else 0,
"version": "",
}
@ -314,7 +314,7 @@ def reload_forbidden_commands() -> dict[str, Any]:
}
except Exception as e:
logger.error(f"Error during forbidden commands reload: {e}")
return {"success": False, "message": f"Error: {str(e)}", "command_count": 0, "version": ""}
return {"success": False, "message": f"Error: {e!s}", "command_count": 0, "version": ""}
def get_prompt(prompt_name: str) -> str:

View File

@ -255,7 +255,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
except Exception as e:
# Overall execution failed
logger.error("Error executing configurations on all devices: %s", e)
return [{"status": "failed", "error": f"Execution error: {str(e)}"}]
return [{"status": "failed", "error": f"Execution error: {e!s}"}]
logger.info(
"Multiple device configuration execution completed. Results: %s",
@ -285,7 +285,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
)
return Result(
host=task.host,
result=f"Configuration failed (ReadTimeout): {str(e)}",
result=f"Configuration failed (ReadTimeout): {e!s}",
failed=True,
)
@ -305,7 +305,7 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
)
return Result(
host=task.host,
result=f"Configuration failed (Unhandled Exception): {str(e)}",
result=f"Configuration failed (Unhandled Exception): {e!s}",
failed=True,
)

View File

@ -253,7 +253,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
except Exception as e:
# Overall execution failed
logger.error("Error executing display on all devices: %s", e)
return [{"status": "failed", "error": f"Execution error: {str(e)}"}]
return [{"status": "failed", "error": f"Execution error: {e!s}"}]
logger.debug(
"Multiple device display execution completed. Results: %s",
@ -288,7 +288,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
)
return Result(
host=task.host,
result=f"diagnostic command failed (ReadTimeout): {str(e)}",
result=f"diagnostic command failed (ReadTimeout): {e!s}",
failed=True,
)
@ -313,7 +313,7 @@ class ExecuteMultipleDeviceCommands(BaseTool):
)
return Result(
host=task.host,
result=f"diagnostic command failed (Unhandled): {str(e)}",
result=f"diagnostic command failed (Unhandled): {e!s}",
failed=True,
)

View File

@ -238,7 +238,7 @@ class GNS3LinkTool(BaseTool):
created_links.append(link_info)
except Exception as e:
error_msg = f"Failed to create link {i}: {str(e)}"
error_msg = f"Failed to create link {i}: {e!s}"
logger.error(error_msg)
created_links.append({"error": error_msg})
@ -257,7 +257,7 @@ class GNS3LinkTool(BaseTool):
return [{"error": f"Invalid JSON input: {e}"}]
except Exception as e:
logger.error("Failed to process link creation: %s", e)
return [{"error": f"Failed to process link creation: {str(e)}"}]
return [{"error": f"Failed to process link creation: {e!s}"}]
if __name__ == "__main__":

View File

@ -244,7 +244,7 @@ class GNS3CreateNodeTool(BaseTool):
except Exception as e:
error_info = {
"error": f"Node {i + 1} creation failed: {str(e)}",
"error": f"Node {i + 1} creation failed: {e!s}",
"status": "failed",
}
results.append(error_info)
@ -280,7 +280,7 @@ class GNS3CreateNodeTool(BaseTool):
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to process node creation request: %s", e)
return {"error": f"Failed to process node creation request: {str(e)}"}
return {"error": f"Failed to process node creation request: {e!s}"}
if __name__ == "__main__":

View File

@ -168,7 +168,7 @@ class GNS3TemplateTool(BaseTool):
except Exception as e:
logger.error("Failed to connect to GNS3 server or retrieve templates: %s", e)
return {"error": f"Failed to retrieve templates: {str(e)}"}
return {"error": f"Failed to retrieve templates: {e!s}"}
if __name__ == "__main__":

View File

@ -239,7 +239,7 @@ class GNS3PacketFilterTool(BaseTool):
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to process packet filter request: %s", e)
return {"error": f"Failed to process packet filter request: {str(e)}"}
return {"error": f"Failed to process packet filter request: {e!s}"}
def _get_available_filters(self, gns3_ctx: dict, project_id: str, link_id: str) -> dict[str, Any]:
"""Get available filter types for the link."""
@ -260,7 +260,7 @@ class GNS3PacketFilterTool(BaseTool):
return {
"action": "get_available",
"link_id": link_id,
"error": f"Failed to get available filters: {str(e)}",
"error": f"Failed to get available filters: {e!s}",
"status": "failed",
}
@ -318,7 +318,7 @@ class GNS3PacketFilterTool(BaseTool):
except Exception as e:
logger.error("Unexpected error during BPF validation: %s", e)
return {"valid": False, "error": f"BPF validation error: {str(e)}"}
return {"valid": False, "error": f"BPF validation error: {e!s}"}
def _set_filters(
self,
@ -384,7 +384,7 @@ class GNS3PacketFilterTool(BaseTool):
return {
"action": "set",
"link_id": link_id,
"error": f"Failed to set filters: {str(e)}",
"error": f"Failed to set filters: {e!s}",
"status": "failed",
}
@ -407,7 +407,7 @@ class GNS3PacketFilterTool(BaseTool):
return {
"action": "get",
"link_id": link_id,
"error": f"Failed to get filters: {str(e)}",
"error": f"Failed to get filters: {e!s}",
"status": "failed",
}
@ -449,7 +449,7 @@ class GNS3PacketFilterTool(BaseTool):
return {
"action": "clear",
"link_id": link_id,
"error": f"Failed to clear filters: {str(e)}",
"error": f"Failed to clear filters: {e!s}",
"status": "failed",
}

View File

@ -220,7 +220,7 @@ class GNS3StartNodeTool(BaseTool):
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to start nodes: %s", e)
return {"error": f"Failed to start nodes: {str(e)}"}
return {"error": f"Failed to start nodes: {e!s}"}
# Backward-compat alias: the waiting variant was removed; both names now

View File

@ -198,7 +198,7 @@ class GNS3StopNodeTool(BaseTool):
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to stop nodes: %s", e)
return {"error": f"Failed to stop nodes: {str(e)}"}
return {"error": f"Failed to stop nodes: {e!s}"}
if __name__ == "__main__":

View File

@ -209,7 +209,7 @@ class GNS3SuspendNodeTool(BaseTool):
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to suspend nodes: %s", e)
return {"error": f"Failed to suspend nodes: {str(e)}"}
return {"error": f"Failed to suspend nodes: {e!s}"}
if __name__ == "__main__":

View File

@ -241,7 +241,7 @@ class GNS3UpdateNodeNameTool(BaseTool):
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Failed to update node names: %s", e)
return {"error": f"Failed to update node names: {str(e)}"}
return {"error": f"Failed to update node names: {e!s}"}
if __name__ == "__main__":

View File

@ -104,4 +104,4 @@ class GNS3WaitTool(BaseTool):
return {"error": f"Invalid JSON input: {e}"}
except Exception as e:
logger.error("Wait tool failed: %s", e)
return {"error": f"Wait tool failed: {str(e)}"}
return {"error": f"Wait tool failed: {e!s}"}

View File

@ -325,7 +325,7 @@ class PacketAnalysisTool(BaseTool):
except Exception as e:
logger.error(f"PacketAnalysisTool error: {e}", exc_info=True)
return f'{{"error": "Analysis failed: {str(e)}"}}'
return f'{{"error": "Analysis failed: {e!s}"}}'
finally:
# Clean up temp file
@ -497,7 +497,7 @@ class PacketAnalysisTool(BaseTool):
return '{"error": "tshark not installed. Please install tshark: apt install tshark"}'
except Exception as e:
logger.error(f"tshark execution error: {e}", exc_info=True)
return f'{{"error": "tshark failed: {str(e)}"}}'
return f'{{"error": "tshark failed: {e!s}"}}'
if __name__ == "__main__":

View File

@ -227,7 +227,7 @@ class VPCSCommands(BaseTool):
except Exception as e:
# Overall execution failed
logger.error("Error executing commands on all VPCS devices: %s", e)
return [{"status": "failed", "error": f"Execution error: {str(e)}"}]
return [{"status": "failed", "error": f"Execution error: {e!s}"}]
logger.debug(
"VPCS command execution completed. Results: %s",
@ -262,7 +262,7 @@ class VPCSCommands(BaseTool):
)
return Result(
host=task.host,
result=f"Command failed (ReadTimeout): {str(e)}",
result=f"Command failed (ReadTimeout): {e!s}",
failed=True,
)
@ -289,7 +289,7 @@ class VPCSCommands(BaseTool):
)
return Result(
host=task.host,
result=f"Command failed: {str(retry_e)}",
result=f"Command failed: {retry_e!s}",
failed=True,
)

View File

@ -496,7 +496,7 @@ def _test_parse_tool_content() -> None:
for i, (input_data, expected) in enumerate(test_cases):
result = parse_tool_content(input_data)
status = "✓" if result == expected else "✗"
print(f"Test {i + 1}: {status} Input: {repr(input_data)} -> {result}")
print(f"Test {i + 1}: {status} Input: {input_data!r} -> {result}")
print("\nTesting format_tool_response function:")
format_tests = [
@ -518,7 +518,7 @@ def _test_parse_tool_content() -> None:
valid = "✓"
except Exception:
valid = "✗"
print(f"Format Test {i + 1}: {valid} Input: {repr(input_data)} -> {result}")
print(f"Format Test {i + 1}: {valid} Input: {input_data!r} -> {result}")
if __name__ == "__main__":

View File

@ -1468,7 +1468,7 @@ class DockerVM(BaseNode):
pass
# Ignore runtime error because when closing the server
except RuntimeError as e:
log.debug(f"Docker runtime error when closing: {str(e)}")
log.debug(f"Docker runtime error when closing: {e!s}")
return
self.status = "stopped"
@ -1571,7 +1571,7 @@ class DockerVM(BaseNode):
self.manager.port_manager.release_udp_port(nio.lport, self._project)
# Ignore runtime error because when closing the server
except (DockerHttp404Error, RuntimeError) as e:
log.debug(f"Docker error when closing: {str(e)}")
log.debug(f"Docker error when closing: {e!s}")
return
def _get_container_ifname(self, adapter_number):
@ -1775,7 +1775,7 @@ class DockerVM(BaseNode):
for index in range(4096):
if f"tap-gns3-e{index}" not in psutil.net_if_addrs():
adapter.host_ifc = f"tap-gns3-e{str(index)}"
adapter.host_ifc = f"tap-gns3-e{index!s}"
break
if adapter.host_ifc is None:
raise DockerError(

View File

@ -97,7 +97,7 @@ class Router(BaseNode):
try:
os.makedirs(os.path.join(self._working_directory, "configs"), exist_ok=True)
except OSError as e:
raise DynamipsError(f"Can't create the dynamips config directory: {str(e)}")
raise DynamipsError(f"Can't create the dynamips config directory: {e!s}")
if dynamips_id:
self._convert_before_2_0_0_b3(dynamips_id)
@ -153,7 +153,7 @@ class Router(BaseNode):
try:
shutil.move(path, dst)
except OSError as e:
log.error(f"Can't move {path}: {str(e)}")
log.error(f"Can't move {path}: {e!s}")
continue
for path in glob.glob(os.path.join(glob.escape(dynamips_dir), f"*_i{dynamips_id}_*")):
dst = os.path.join(self._working_directory, os.path.basename(path))
@ -161,7 +161,7 @@ class Router(BaseNode):
try:
shutil.move(path, dst)
except OSError as e:
log.error(f"Can't move {path}: {str(e)}")
log.error(f"Can't move {path}: {e!s}")
continue
def asdict(self):
@ -1608,7 +1608,7 @@ class Router(BaseNode):
try:
open(output_file, "w+").close()
except OSError as e:
raise DynamipsError(f'Can not write capture to "{output_file}": {str(e)}')
raise DynamipsError(f'Can not write capture to "{output_file}": {e!s}')
try:
adapter = self._slots[slot_number]

View File

@ -334,7 +334,7 @@ class GNS3VM:
await engine.start()
except Exception as e:
await self._controller.delete_compute("vm")
log.error(f"Cannot start the GNS3 VM: {str(e)}")
log.error(f"Cannot start the GNS3 VM: {e!s}")
await compute.update(name=f"GNS3 VM ({engine.vmname})")
compute.set_last_error(str(e))
raise e

View File

@ -260,7 +260,7 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
await self._check_requirements()
return await self._virtualbox_manager.list_vms()
except VirtualBoxError as e:
raise GNS3VMError(f"Could not list VirtualBox VMs: {str(e)}")
raise GNS3VMError(f"Could not list VirtualBox VMs: {e!s}")
async def start(self):
"""

View File

@ -116,7 +116,7 @@ class VMwareGNS3VM(BaseGNS3VM):
try:
return await self._vmware_manager.list_vms()
except VMwareError as e:
raise GNS3VMError(f"Could not list VMware VMs: {str(e)}")
raise GNS3VMError(f"Could not list VMware VMs: {e!s}")
async def start(self):
"""
@ -143,7 +143,7 @@ class VMwareGNS3VM(BaseGNS3VM):
try:
running = await self._is_running()
except VMwareError as e:
raise GNS3VMError(f"Could not list VMware VMs: {str(e)}")
raise GNS3VMError(f"Could not list VMware VMs: {e!s}")
if not running:
# set the number of vCPUs and amount of RAM
if self.allocate_vcpus_ram:
@ -197,7 +197,7 @@ class VMwareGNS3VM(BaseGNS3VM):
try:
await self._execute("suspend", [self._vmx_path])
except GNS3VMError as e:
log.warning(f"Error when suspending the VM: {str(e)}")
log.warning(f"Error when suspending the VM: {e!s}")
log.info("GNS3 VM has been suspended")
self.running = False
@ -211,6 +211,6 @@ class VMwareGNS3VM(BaseGNS3VM):
try:
await self._execute("stop", [self._vmx_path, "soft"])
except GNS3VMError as e:
log.warning(f"Error when stopping the VM: {str(e)}")
log.warning(f"Error when stopping the VM: {e!s}")
log.info("GNS3 VM has been stopped")
self.running = False

View File

@ -222,7 +222,7 @@ class Link:
try:
validate_all_filters(new_filters)
except FilterValidationError as e:
raise ControllerError(f"Invalid packet filter parameters: {str(e)}")
raise ControllerError(f"Invalid packet filter parameters: {e!s}")
if new_filters != self.filters:
self._filters = new_filters
@ -483,7 +483,7 @@ class Link:
# Re-raise ControllerError to return error to client
raise
except Exception as e:
error_msg = f"Error starting Web Wireshark: {str(e)}"
error_msg = f"Error starting Web Wireshark: {e!s}"
log.error(error_msg)
raise ControllerError(error_msg)
finally:
@ -532,7 +532,7 @@ class Link:
except ControllerError:
raise
except Exception as e:
error_msg = f"Error restarting Web Wireshark: {str(e)}"
error_msg = f"Error restarting Web Wireshark: {e!s}"
log.error(error_msg)
raise ControllerError(error_msg)
finally:

View File

@ -1050,7 +1050,7 @@ class Node:
with open(image, "rb") as f:
await self._compute.post(f"/{self._node_type}/images/{os.path.basename(img)}", data=f, timeout=None)
except OSError as e:
raise ControllerError(f"Can't upload {image}: {str(e)}")
raise ControllerError(f"Can't upload {image}: {e!s}")
self.project.emit_notification("log.info", {"message": f"Upload finished for {img}"})
return True

View File

@ -88,7 +88,7 @@ class StandardPortFactory:
**cls._generate_replacement(interface_number, segment_number),
)
except (IndexError, ValueError, KeyError) as e:
raise ControllerError(f"Invalid port name format {port_name_format}: {str(e)}")
raise ControllerError(f"Invalid port name format {port_name_format}: {e!s}")
port_name = custom_adapter_settings.get("port_name", port_name)
port = PortFactory(port_name, segment_number, adapter_number, port_number, "ethernet")

View File

@ -1731,7 +1731,7 @@ class Project:
)
shutil.rmtree(self.path)
except OSError as e:
raise ControllerError(f"Cannot delete project directory {self.path}: {str(e)}")
raise ControllerError(f"Cannot delete project directory {self.path}: {e!s}")
self.emit_controller_notification("project.deleted", self.asdict())
def _get_disconnected_computes(self):
@ -2081,7 +2081,7 @@ class Project:
else:
log.info("Fast duplication failed, fallback to normal duplication")
except Exception as e:
raise ControllerError(f"Cannot duplicate project: {str(e)}")
raise ControllerError(f"Cannot duplicate project: {e!s}")
try:
begin = time.time()
@ -2119,7 +2119,7 @@ class Project:
log.info(f"Project '{project.name}' duplicated in {time.time() - begin:.4f} seconds")
except (ValueError, OSError, UnicodeEncodeError) as e:
raise ControllerError(f"Cannot duplicate project: {str(e)}")
raise ControllerError(f"Cannot duplicate project: {e!s}")
if previous_status == "closed":
await self.close()

View File

@ -139,7 +139,7 @@ def load_topology(path):
with open(path, encoding="utf-8") as f:
topo = json.load(f)
except (OSError, UnicodeDecodeError, ValueError) as e:
raise ControllerError(f"Could not load topology {path}: {str(e)}")
raise ControllerError(f"Could not load topology {path}: {e!s}")
if topo.get("revision", 0) > GNS3_FILE_FORMAT_REVISION:
raise ControllerError(
@ -154,7 +154,7 @@ def load_topology(path):
try:
shutil.copy(path, path + ".backup{}".format(topo.get("revision", 0)))
except OSError as e:
raise ControllerError(f"Can't write backup of the topology {path}: {str(e)}")
raise ControllerError(f"Can't write backup of the topology {path}: {e!s}")
changed = True
# update the version because we converted the topology
topo["version"] = __version__
@ -210,7 +210,7 @@ def load_topology(path):
with open(path, "w+", encoding="utf-8") as f:
json.dump(topo, f, indent=4, sort_keys=True)
except OSError as e:
raise ControllerError(f"Can't write the topology {path}: {str(e)}")
raise ControllerError(f"Can't write the topology {path}: {e!s}")
return topo
@ -326,7 +326,7 @@ def _convert_2_0_0_beta_2(topo, topo_path):
for path in glob.glob(os.path.join(glob.escape(dynamips_dir), "configs", f"i{dynamips_id}_*")):
shutil.move(path, os.path.join(node_dir, "configs", os.path.basename(path)))
except OSError as e:
raise ControllerError(f"Can't convert project {topo_path}: {str(e)}")
raise ControllerError(f"Can't convert project {topo_path}: {e!s}")
return topo

View File

@ -114,7 +114,7 @@ async def list_images(image_type):
}
)
except OSError as e:
log.warning(f"Can't add image {path}: {str(e)}")
log.warning(f"Can't add image {path}: {e!s}")
return images

View File

@ -62,7 +62,7 @@ def validate_bpf_syntax(bpf_expression: str) -> Dict[str, Any]:
except Exception as e:
log.error("Unexpected error during BPF validation: %s", e)
return {"valid": False, "error": f"BPF validation error: {str(e)}"}
return {"valid": False, "error": f"BPF validation error: {e!s}"}
def validate_filter_parameters(filter_type: str, values: List[Any]) -> None:

View File

@ -134,7 +134,7 @@ async def benchmark_get_projects(base_url, headers, project_count, iterations=10
else:
print(f" Iteration {i + 1}: ERROR {response.status_code}")
except Exception as e:
print(f" Iteration {i + 1}: Exception - {type(e).__name__}: {str(e)}")
print(f" Iteration {i + 1}: Exception - {type(e).__name__}: {e!s}")
# Cleanup test projects
if project_count > 0: