NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-10-26 03:42:22 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2933 from markparonyan/ruff/trivia-fixups

Browse Source 
refactor: ruff fixes
This commit is contained in:
Jeremy Grossmann 2026-10-01 23:18:29 +02:00 committed by GitHub
commit 7f8d9c2f54
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
127 changed files with 181 additions and 265 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
gns3server/agent/__init__.py
View File

@ -103,6 +103,6 @@ except ImportError:
__all__ = [
"AI_COPILOT_AVAILABLE",
"MCP_AVAILABLE",
"get_project_agent_manager",
"ProjectAgentManager",
"get_project_agent_manager",
]

8
gns3server/agent/gns3_copilot/agent/context_manager.py
View File

@ -391,10 +391,10 @@ def create_pre_model_hook(
max_tokens_for_trim = max_input_tokens - tool_tokens
# Validate budget and provide actionable warnings
if system_tokens + tool_tokens > max_input_tokens: # noqa: E501
if system_tokens + tool_tokens > max_input_tokens:
logger.error(
"System prompt (%d tokens) + tools (%d tokens) EXCEED input "
"budget (%d tokens). This will likely cause LLM call failures. " # noqa: E501
"budget (%d tokens). This will likely cause LLM call failures. "
"Recommendations: 1) Reduce system prompt length, 2) Reduce "
"number of tools, 3) Use a model with larger context window, "
"or 4) Switch to 'conservative' strategy.",
@ -524,7 +524,7 @@ def _inject_topology_into_system(
if filtered_count > 0:
logger.debug("Filtered out %d existing SystemMessage(s)", filtered_count)
return [SystemMessage(content=formatted_prompt)] + non_system_messages
return [SystemMessage(content=formatted_prompt), *non_system_messages]
# ============================================================================
@ -556,7 +556,7 @@ def prepare_context_messages(
else:
formatted_prompt = system_prompt.replace("{{topology_info}}", "(No topology information available)")
return [SystemMessage(content=formatted_prompt)] + state_messages
return [SystemMessage(content=formatted_prompt), *state_messages]
# ============================================================================

2
gns3server/agent/gns3_copilot/agent/gns3_copilot.py
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

8
gns3server/agent/gns3_copilot/gns3_client/__init__.py
View File

@ -71,16 +71,16 @@ __description__ = "AI-powered network automation assistant for GNS3"
__url__ = "https://github.com/yueguobin/gns3-copilot"
__all__ = [
"GNS3ProjectInfoTool",
"GNS3TopologyTool",
"Gns3Connector",
"build_gns3_ctx",
"GNS3TopologyTool",
"GNS3ProjectInfoTool",
"get_current_jwt_token",
"get_current_llm_config",
"get_gns3_connector",
"get_gns3_connector_with_llm_config",
"get_gns3_server_host",
"get_llm_config",
"set_current_jwt_token",
"get_current_jwt_token",
"set_current_llm_config",
"get_current_llm_config",
]

4
gns3server/agent/gns3_copilot/gns3_client/connector.py
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

4
gns3server/agent/gns3_copilot/gns3_client/context_helpers.py
View File

@ -124,8 +124,8 @@ def get_current_llm_config() -> Optional[dict]:
__all__ = [
"set_current_jwt_token",
"get_current_jwt_token",
"set_current_llm_config",
"get_current_llm_config",
"set_current_jwt_token",
"set_current_llm_config",
]

2
gns3server/agent/gns3_copilot/gns3_client/gns3_project_info.py
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__":

2
gns3server/agent/gns3_copilot/gns3_client/gns3_topology_reader.py
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:
"""

12
gns3server/agent/gns3_copilot/skills/__init__.py
View File

@ -55,18 +55,18 @@ from .manager import SkillsManager
from .loader import SkillsLoader
__all__ = [
"SKILLS_REGISTRY",
"INJECTION_SKILLS_REGISTRY",
"get_skill",
"get_injection_skill",
"SKILLS_REGISTRY",
"DeviceSkillsTool",
"InjectionSkillsTool",
"PacketAnalysisSkillsTool",
"SkillsManager",
"SkillsLoader",
"set_skills_manager",
"SkillsManager",
"get_injection_skill",
"get_skill",
"get_skills_manager",
"get_skills_repository_info",
"reload_injection_skills",
"reload_skills_repository",
"get_skills_repository_info",
"set_skills_manager",
]

6
gns3server/agent/gns3_copilot/skills/loader.py
View File

@ -278,7 +278,7 @@ class SkillsLoader:
return ""
try:
with open(prompt_file, "r", encoding="utf-8") as f:
with open(prompt_file, encoding="utf-8") as f:
content = f.read()
logger.debug(f"Loaded prompt: {prompt_name} from {prompt_file}")
return content
@ -302,7 +302,7 @@ class SkillsLoader:
try:
commands = []
with open(config_file, "r", encoding="utf-8") as f:
with open(config_file, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
@ -364,7 +364,7 @@ class SkillsLoader:
Returns:
Parsed YAML content as dictionary, or empty dict if file is empty
"""
with open(file_path, "r", encoding="utf-8") as f:
with open(file_path, encoding="utf-8") as f:
data = yaml.safe_load(f)
if not isinstance(data, dict):
logger.warning(f"YAML file {file_path} is empty or has invalid format")

2
gns3server/agent/gns3_copilot/skills/manager.py
View File

@ -89,7 +89,7 @@ class SkillsManager:
self.branch = branch
self.auto_update = auto_update
self.loader = SkillsLoader(str(local_path))
self._repo: Optional["git.Repo"] = None
self._repo: Optional[git.Repo] = None
self._prompt_count = 0
if not GIT_AVAILABLE:

6
gns3server/agent/gns3_copilot/skills/registry.py
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:

6
gns3server/agent/gns3_copilot/tools_v2/__init__.py
View File

@ -80,17 +80,17 @@ __url__ = "https://github.com/yueguobin/gns3-copilot"
# Export main tool classes
__all__ = [
"ExecuteMultipleDeviceConfigCommands",
"ExecuteMultipleDeviceCommands",
"ExecuteMultipleDeviceConfigCommands",
"GNS3CreateNodeTool",
"GNS3LinkTool",
"GNS3PacketFilterTool",
"GNS3StartNodeTool",
"GNS3StartNodeQuickTool",
"GNS3StartNodeTool",
"GNS3StopNodeTool",
"GNS3SuspendNodeTool",
"GNS3UpdateNodeNameTool",
"GNS3TemplateTool",
"GNS3UpdateNodeNameTool",
"GNS3WaitTool",
"PacketAnalysisTool",
]

8
gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py
View File

@ -57,7 +57,7 @@ from gns3server.agent.gns3_copilot.utils.command_filter import (
# Import custom Netmiko device types for GNS3 emulation
# This registers gns3_huawei_telnet_ce and other custom device types
# NOTE: Must be imported BEFORE any Nornir operations to ensure device types are registered
from gns3server.agent.gns3_copilot.utils import custom_netmiko # noqa: F401
from gns3server.agent.gns3_copilot.utils import custom_netmiko
# Explicitly register custom device types to ensure they are available
# This is a safety measure in case the auto-registration on import doesn't work
@ -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,
)

8
gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py
View File

@ -57,7 +57,7 @@ from gns3server.agent.gns3_copilot.utils.command_filter import (
# Import custom Netmiko device types for GNS3 emulation
# This registers gns3_huawei_telnet_ce and other custom device types
# NOTE: Must be imported BEFORE any Nornir operations to ensure device types are registered
from gns3server.agent.gns3_copilot.utils import custom_netmiko # noqa: F401
from gns3server.agent.gns3_copilot.utils import custom_netmiko
# Explicitly register custom device types to ensure they are available
# This is a safety measure in case the auto-registration on import doesn't work
@ -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,
)

4
gns3server/agent/gns3_copilot/tools_v2/gns3_create_link.py
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__":

4
gns3server/agent/gns3_copilot/tools_v2/gns3_create_node.py
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__":

2
gns3server/agent/gns3_copilot/tools_v2/gns3_get_node_temp.py
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__":

12
gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py
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",
}

2
gns3server/agent/gns3_copilot/tools_v2/gns3_start_node.py
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

2
gns3server/agent/gns3_copilot/tools_v2/gns3_stop_node.py
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__":

2
gns3server/agent/gns3_copilot/tools_v2/gns3_suspend_node.py
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__":

2
gns3server/agent/gns3_copilot/tools_v2/gns3_update_node_name.py
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__":

2
gns3server/agent/gns3_copilot/tools_v2/gns3_wait.py
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}"}

6
gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py
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
@ -444,7 +444,7 @@ class PacketAnalysisTool(BaseTool):
# Build command: tshark -r <file> <user_args>
import shlex
cmd = ["tshark", "-r", pcap_file] + shlex.split(tshark_args)
cmd = ["tshark", "-r", pcap_file, *shlex.split(tshark_args)]
logger.info(f"Running tshark: {' '.join(cmd)}")
@ -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__":

8
gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py
View File

@ -54,7 +54,7 @@ from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology
# Import custom Netmiko device types for GNS3 emulation
# This registers gns3_vpcs_telnet and other custom device types
# NOTE: Must be imported BEFORE any Nornir operations to ensure device types are registered
from gns3server.agent.gns3_copilot.utils import custom_netmiko # noqa: F401
from gns3server.agent.gns3_copilot.utils import custom_netmiko
# Explicitly register VPCS device type to ensure it is available
try:
@ -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,
)

4
gns3server/agent/gns3_copilot/utils/__init__.py
View File

@ -58,8 +58,8 @@ __url__ = "https://github.com/yueguobin/gns3-copilot"
# Export main utility functions
__all__ = [
"format_error_message",
"get_device_ports_from_topology",
"parse_tool_content",
"format_tool_response",
"get_device_ports_from_topology",
"normalize_tool_response",
"parse_tool_content",
]

6
gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py
View File

@ -61,17 +61,17 @@ logger = logging.getLogger(__name__)
# Import all custom drivers (auto-registers them with Netmiko)
try:
from . import huawei_ce # noqa: F401
from . import huawei_ce
except Exception as e:
logger.warning(f"Failed to import Huawei CE driver: {e}", exc_info=True)
try:
from .ruijie_telnet import RuijieTelnetEnhanced # noqa: F401
from .ruijie_telnet import RuijieTelnetEnhanced
except Exception as e:
logger.warning(f"Failed to import Ruijie driver: {e}", exc_info=True)
try:
from . import vpcs_telnet # noqa: F401
from . import vpcs_telnet
except Exception as e:
logger.warning(f"Failed to import VPCS driver: {e}", exc_info=True)

2
gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py
View File

@ -338,7 +338,7 @@ def save_markdown(content: str, filename: str = "netmiko_devices.md") -> None:
with open(output_path, "w", encoding="utf-8") as f:
f.write(content)
print(f"Markdown document saved to: {output_path}")
except IOError as e:
except OSError as e:
print(f"Error: Failed to save Markdown file: {e}", file=sys.stderr)
sys.exit(1)

2
gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py
View File

@ -157,7 +157,7 @@ class TestHuaweiTelnetCEIntegration(unittest.TestCase):
def test_mock_telnet_connection(self):
"""Test telnet_login logic with mocked connection."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.huawei_ce import ( # noqa: E501
from gns3server.agent.gns3_copilot.utils.custom_netmiko.huawei_ce import (
GNS3HuaweiTelnetCE,
)

4
gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_ruijie_telnet.py
View File

@ -178,7 +178,7 @@ class TestRuijieTelnetEnhancedIntegration(unittest.TestCase):
def test_send_config_set_calls_preprocess(self):
"""Test that send_config_set calls preprocessing."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.ruijie_telnet import ( # noqa: E501
from gns3server.agent.gns3_copilot.utils.custom_netmiko.ruijie_telnet import (
RuijieTelnetEnhanced,
)
@ -205,7 +205,7 @@ class TestRuijieTelnetEnhancedIntegration(unittest.TestCase):
def test_multiple_interactive_patterns(self):
"""Test that multiple interactive commands are detected."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.ruijie_telnet import ( # noqa: E501
from gns3server.agent.gns3_copilot.utils.custom_netmiko.ruijie_telnet import (
RuijieTelnetEnhanced,
)

10
gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_vpcs_telnet.py
View File

@ -116,7 +116,7 @@ class TestVPCSTelnetInit(unittest.TestCase):
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import (
VPCSTelnet,
)
@ -165,7 +165,7 @@ class TestVPCSTelnetMethods(unittest.TestCase):
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import (
VPCSTelnet,
)
@ -208,7 +208,7 @@ class TestVPCSTelnetSendCommand(unittest.TestCase):
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import (
VPCSTelnet,
)
@ -273,7 +273,7 @@ class TestVPCSTelnetTelnetLogin(unittest.TestCase):
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import (
VPCSTelnet,
)
@ -353,7 +353,7 @@ class TestVPCSTelnetAnsiStripping(unittest.TestCase):
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import ( # noqa: E501
from gns3server.agent.gns3_copilot.utils.custom_netmiko.vpcs_telnet import (
VPCSTelnet,
)

6
gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py
View File

@ -238,7 +238,7 @@ def generate_ellipse_svg(
svg_height: int,
) -> str:
"""Generate SVG for ellipse."""
return f"""<svg width="{svg_width}" height="{svg_height}"><ellipse cx="{rx}" cy="{ry}" rx="{rx}" ry="{ry}" fill="{color_scheme["fill"]}" fill-opacity="{color_scheme["fill_opacity"]}"/></svg>""" # noqa: E501
return f"""<svg width="{svg_width}" height="{svg_height}"><ellipse cx="{rx}" cy="{ry}" rx="{rx}" ry="{ry}" fill="{color_scheme["fill"]}" fill-opacity="{color_scheme["fill_opacity"]}"/></svg>"""
def generate_rectangle_svg(
@ -247,7 +247,7 @@ def generate_rectangle_svg(
color_scheme: dict[str, Any],
) -> str:
"""Generate SVG for rectangle."""
return f"""<svg width="{width}" height="{height}"><rect x="0" y="0" width="{width}" height="{height}" fill="{color_scheme["fill"]}" fill-opacity="{color_scheme["fill_opacity"]}"/></svg>""" # noqa: E501
return f"""<svg width="{width}" height="{height}"><rect x="0" y="0" width="{width}" height="{height}" fill="{color_scheme["fill"]}" fill-opacity="{color_scheme["fill_opacity"]}"/></svg>"""
def generate_text_svg(text: str, color_scheme: dict[str, Any]) -> str:
@ -255,7 +255,7 @@ def generate_text_svg(text: str, color_scheme: dict[str, Any]) -> str:
text_width = len(text) * 8 + 20
text_height = DEFAULT_FONT_SIZE + 16
return f"""<svg width="{text_width}" height="{text_height}"><text font-family="TypeWriter" font-size="{DEFAULT_FONT_SIZE}.0" font-weight="bold" fill="{color_scheme["stroke"]}" text-anchor="middle" x="{text_width / 2}" y="{text_height / 2 + 4}">{text}</text></svg>""" # noqa: E501
return f"""<svg width="{text_width}" height="{text_height}"><text font-family="TypeWriter" font-size="{DEFAULT_FONT_SIZE}.0" font-weight="bold" fill="{color_scheme["stroke"]}" text-anchor="middle" x="{text_width / 2}" y="{text_height / 2 + 4}">{text}</text></svg>"""
def _hsv_to_hex(h: int, s: int, v: int) -> str:

4
gns3server/agent/gns3_copilot/utils/parse_tool_content.py
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__":

2
gns3server/agent/mcp/templates.py
View File

@ -173,7 +173,7 @@ def delete_template_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) ->
template_id = matches[0]["template_id"]
conn.http_call("delete", f"{conn.base_url}/templates/{template_id}")
return {"message": f"Template deleted"}
return {"message": "Template deleted"}
# ── Tool definitions ───────────────────────────────────────────────────────

2
gns3server/api/routes/compute/notifications.py
View File

@ -67,7 +67,7 @@ if __name__ == "__main__":
<ul id='messages'>
</ul>
<script>
var ws = new WebSocket("ws://localhost:8000/notifications/ws");
var ws = new WebSocket("ws://localhost:8000/notifications/ws");
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')

2
gns3server/api/routes/controller/controller.py
View File

@ -272,7 +272,7 @@ async def controller_http_notifications(request: Request) -> StreamingResponse:
with Controller.instance().notification.controller_queue() as queue:
while not app.state.exiting:
msg = await queue.get_json(5)
yield f"{msg}\n".encode("utf-8")
yield f"{msg}\n".encode()
finally:
log.info(f"Client {client} has disconnected from controller HTTP notification stream")

2
gns3server/api/routes/controller/netmiko.py
View File

@ -64,7 +64,7 @@ def _load_netmiko_device_types() -> schemas.NetmikoDeviceTypeList:
# imported them yet); failures are logged by the package itself, do not
# fail the whole endpoint.
try:
from gns3server.agent.gns3_copilot.utils import custom_netmiko # noqa: F401
from gns3server.agent.gns3_copilot.utils import custom_netmiko
except Exception as e:
log.warning(f"Could not register GNS3-copilot custom Netmiko drivers: {e}")

2
gns3server/api/routes/controller/nodes.py
View File

@ -577,7 +577,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 (IOError, OSError, asyncio.TimeoutError) as e:
except (OSError, asyncio.TimeoutError) as e:
log.error(f"Error streaming file '{path}' from compute: {e}")
raise
finally:

2
gns3server/api/routes/controller/projects.py
View File

@ -543,7 +543,7 @@ async def project_http_notifications(project_id: UUID) -> StreamingResponse:
with controller.notification.project_queue(project.id) as queue:
while not app.state.exiting:
msg = await queue.get_json(5)
yield f"{msg}\n".encode("utf-8")
yield f"{msg}\n".encode()
finally:
log.info(f"Client has disconnected from notification for project ID '{project.id}' (HTTP stream method)")
if project.auto_close:

2
gns3server/api/routes/index.py
View File

@ -62,7 +62,7 @@ async def web_ui(file_path: str):
# which results with warnings in Firefox 66 on Windows
# Ref. gns3-server#1559
_, ext = os.path.splitext(static)
mimetype = ext == ".js" and "application/javascript" or None
mimetype = (ext == ".js" and "application/javascript") or None
return FileResponse(static, media_type=mimetype)

2
gns3server/compute/builtin/nodes/cloud.py
View File

@ -323,7 +323,7 @@ class Cloud(BaseNode):
if port_info["type"] == "ethernet":
network_interfaces = [interface["name"] for interface in self._interfaces()]
if not port_info["interface"] in network_interfaces:
if port_info["interface"] not in network_interfaces:
raise NodeError(
f"Interface '{port_info['interface']}' could not be found on this system, please update '{self.name}'"
)

8
gns3server/compute/docker/docker_vm.py
View File

@ -277,7 +277,7 @@ class DockerVM(BaseNode):
if not mac_address:
# use the node UUID to generate a random MAC address
self._mac_address = "02:42:%s:%s:%s:00" % (self.id[2:4], self.id[4:6], self.id[6:8])
self._mac_address = "02:42:{}:{}:{}:00".format(self.id[2:4], self.id[4:6], self.id[6:8])
else:
self._mac_address = mac_address
@ -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(

8
gns3server/compute/dynamips/nodes/router.py
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]

2
gns3server/compute/qemu/__init__.py
View File

@ -178,7 +178,7 @@ class Qemu(BaseManager):
qemu_img_path = shutil.which("qemu-img")
if not qemu_img_path:
raise QemuError(f"Could not find qemu-img binary")
raise QemuError("Could not find qemu-img binary")
try:
if os.path.exists(disk_image_path):

10
gns3server/compute/qemu/qemu_vm.py
View File

@ -2102,7 +2102,7 @@ class QemuVM(BaseNode):
raise OSError(f"mcopy failure: return code {retcode}")
async def _export_config(self):
disk_name = getattr(self, "config_disk_name")
disk_name = self.config_disk_name
if not disk_name:
return
disk = os.path.join(self.working_dir, disk_name)
@ -2122,7 +2122,7 @@ class QemuVM(BaseNode):
shutil.rmtree(config_dir, ignore_errors=True)
async def _import_config(self):
disk_name = getattr(self, "config_disk_name")
disk_name = self.config_disk_name
zip_file = os.path.join(self.working_dir, "config.zip")
if not disk_name or not os.path.exists(zip_file):
return
@ -2131,7 +2131,7 @@ class QemuVM(BaseNode):
disk_tmp = disk + ".tmp"
try:
os.mkdir(config_dir)
shutil.copyfile(getattr(self, "config_disk_image"), disk_tmp)
shutil.copyfile(self.config_disk_image, disk_tmp)
unpack_zip(zip_file, config_dir)
config_files = [os.path.join(config_dir, fname) for fname in os.listdir(config_dir)]
if config_files:
@ -2279,9 +2279,9 @@ class QemuVM(BaseNode):
options.extend(await self._disk_interface_options(disk, disk_index, interface))
# config disk
disk_image = getattr(self, "config_disk_image")
disk_image = self.config_disk_image
if disk_image and self._create_config_disk:
disk_name = getattr(self, "config_disk_name")
disk_name = self.config_disk_name
disk = os.path.join(self.working_dir, disk_name)
if self.hdd_disk_interface == "none":
# use the HDA interface type if none has been configured for HDD

4
gns3server/compute/qemu/utils/ziputils.py
View File

@ -40,7 +40,7 @@ def unpack_zip(filename, extract_dir=None):
with zipfile.ZipFile(filename, "r") as zfile:
for zinfo in zfile.infolist():
fname = os.path.join(extract_dir, zinfo.filename)
date_time = time.mktime(zinfo.date_time + (0, 0, -1))
date_time = time.mktime((*zinfo.date_time, 0, 0, -1))
zfile.extract(zinfo, extract_dir)
# update timestamp
@ -52,4 +52,4 @@ def unpack_zip(filename, extract_dir=None):
for fname, date_time in reversed(dirs):
os.utime(fname, (date_time, date_time))
except zipfile.BadZipFile:
raise shutil.ReadError("%s is not a zip file" % filename)
raise shutil.ReadError("{} is not a zip file".format(filename))

8
gns3server/compute/virtualbox/virtualbox_vm.py
View File

@ -146,7 +146,7 @@ class VirtualBoxVM(BaseNode):
"""
args = shlex.split(params)
result = await self.manager.execute("controlvm", [self._uuid] + args)
result = await self.manager.execute("controlvm", [self._uuid, *args])
return result
async def _modify_vm(self, params):
@ -157,7 +157,7 @@ class VirtualBoxVM(BaseNode):
"""
args = shlex.split(params)
await self.manager.execute("modifyvm", [self._uuid] + args)
await self.manager.execute("modifyvm", [self._uuid, *args])
async def _check_duplicate_linked_clone(self):
"""
@ -241,7 +241,7 @@ class VirtualBoxVM(BaseNode):
try:
tree = ET.parse(linked_vbox_file)
except ET.ParseError:
raise VirtualBoxError(f"Cannot modify VirtualBox linked node file. File {{linked_vbox_file}} is corrupted.")
raise VirtualBoxError("Cannot modify VirtualBox linked node file. File {linked_vbox_file} is corrupted.")
except OSError as e:
raise VirtualBoxError(f"Cannot modify VirtualBox linked nodes file '{self._linked_vbox_file()}': {e}")
@ -866,7 +866,7 @@ class VirtualBoxVM(BaseNode):
"""
args = shlex.split(params)
await self.manager.execute("storageattach", [self._uuid] + args)
await self.manager.execute("storageattach", [self._uuid, *args])
async def _get_nic_attachements(self, maximum_adapters):
"""

2
gns3server/compute/vmware/__init__.py
View File

@ -237,7 +237,7 @@ class VMware(BaseManager):
vmnet = f"vmnet{match.group(1)}"
if vmnet not in ("vmnet0", "vmnet1", "vmnet8"):
vmnet_interfaces[vmnet] = {}
with open(vmware_networking_file, "r", encoding="utf-8") as f:
with open(vmware_networking_file, encoding="utf-8") as f:
for line in f.read().splitlines():
match = re.search(r"VNET_([0-9]+)_HOSTONLY_SUBNET\s+(.*)", line)
if match:

2
gns3server/controller/__init__.py
View File

@ -176,7 +176,7 @@ class Controller:
protocol = server_config.protocol.value
if self._ssl_context and protocol != "https":
log.warning(f"Protocol changed to 'https' for local compute because SSL is enabled")
log.warning("Protocol changed to 'https' for local compute because SSL is enabled")
protocol = "https"
try:
self._local_server = await self.add_compute(

1
gns3server/controller/appliance_to_template.py
View File

@ -1,5 +1,4 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2021 GNS3 Technologies Inc.
#

2
gns3server/controller/gns3vm/__init__.py
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

2
gns3server/controller/gns3vm/virtualbox_gns3_vm.py
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):
"""

8
gns3server/controller/gns3vm/vmware_gns3_vm.py
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

6
gns3server/controller/link.py
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:

2
gns3server/controller/node.py
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

2
gns3server/controller/ports/port_factory.py
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")

6
gns3server/controller/project.py
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()

8
gns3server/controller/topology.py
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

4
gns3server/db/models/base.py
View File

@ -59,10 +59,10 @@ class GUID(TypeDecorator):
return str(value)
else:
if not isinstance(value, uuid.UUID):
return "%.32x" % uuid.UUID(value).int
return "{:032x}".format(uuid.UUID(value).int)
else:
# hexstring
return "%.32x" % value.int
return "{:032x}".format(value.int)
def process_result_value(self, value, dialect):
if value is None:

4
gns3server/db/repositories/llm_model_configs.py
View File

@ -163,7 +163,7 @@ class LLMModelConfigsRepository(BaseRepository):
# Handle extra config fields
for key, value in updates.items():
if key not in ["name", "model_type", "is_default", "expected_version"] + config_fields:
if key not in ["name", "model_type", "is_default", "expected_version", *config_fields]:
if value is not None:
current_config[key] = value
@ -322,7 +322,7 @@ class LLMModelConfigsRepository(BaseRepository):
# Handle extra config fields
for key, value in updates.items():
if key not in ["name", "model_type", "is_default", "expected_version"] + config_fields:
if key not in ["name", "model_type", "is_default", "expected_version", *config_fields]:
if value is not None:
current_config[key] = value

4
gns3server/db/tasks.py
View File

@ -172,9 +172,9 @@ async def connect_to_db(app: FastAPI) -> None:
async def disconnect_from_db(app: FastAPI) -> None:
# dispose of the connection pool used by the database engine
if getattr(app.state, "_db_engine"):
if app.state._db_engine:
await app.state._db_engine.dispose()
log.info(f"Disconnected from database")
log.info("Disconnected from database")
async def get_computes(app: FastAPI) -> List[schemas.Compute]:

4
gns3server/logger.py
View File

@ -125,8 +125,8 @@ class CompressedRotatingFileHandler(RotatingFileHandler):
self.stream.close()
if self.backupCount > 0:
for i in range(self.backupCount - 1, 0, -1):
sfn = "%s.%d.gz" % (self.baseFilename, i)
dfn = "%s.%d.gz" % (self.baseFilename, i + 1)
sfn = f"{self.baseFilename}.{i}.gz"
dfn = f"{self.baseFilename}.{i + 1}.gz"
if os.path.exists(sfn):
if os.path.exists(dfn):
os.remove(dfn)

4
gns3server/main.py
View File

@ -45,7 +45,7 @@ def daemonize():
# Exit first parent
sys.exit(0)
except OSError as e:
print("First fork failed: %d (%s)\n" % (e.errno, e.strerror), file=sys.stderr)
print(f"First fork failed: {e.errno} ({e.strerror})\n", file=sys.stderr)
sys.exit(1)
# Decouple from parent environment
@ -59,7 +59,7 @@ def daemonize():
# Exit from second parent
sys.exit(0)
except OSError as e:
print("Second fork failed: %d (%s)\n" % (e.errno, e.strerror), file=sys.stderr)
print(f"Second fork failed: {e.errno} ({e.strerror})\n", file=sys.stderr)
sys.exit(1)

4
gns3server/utils/__init__.py
View File

@ -50,7 +50,7 @@ def int_to_macaddress(integer):
"""
Convert an integer to a mac address
"""
return ":".join(textwrap.wrap("%012x" % (integer), width=2))
return ":".join(textwrap.wrap("{:012x}".format(integer), width=2))
def parse_version(version):
@ -72,7 +72,7 @@ def parse_version(version):
info = int(info)
# We pad with zero to compare only on string
# This avoid issue when comparing version with different length
version.append("%06d" % (info,))
version.append(f"{info:06d}")
except ValueError:
# Force to a version with three number
if len(version) == 1:

4
gns3server/utils/asyncio/aiozipstream.py
View File

@ -183,11 +183,11 @@ class ZipFile(zipfile.ZipFile):
"""
if not isinstance(comment, bytes):
raise TypeError("comment: expected bytes, got %s" % type(comment))
raise TypeError("comment: expected bytes, got {}".format(type(comment)))
# check for valid comment length
if len(comment) >= zipfile.ZIP_MAX_COMMENT:
if self.debug:
print("Archive comment is too long; truncating to %d bytes" % zipfile.ZIP_MAX_COMMENT)
print(f"Archive comment is too long; truncating to {zipfile.ZIP_MAX_COMMENT} bytes")
comment = comment[: zipfile.ZIP_MAX_COMMENT]
self._comment = comment
self._didModify = True

6
gns3server/utils/images.py
View File

@ -117,7 +117,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
@ -186,7 +186,7 @@ async def discover_images(image_type: str, skip_image_paths: list | None = None)
if filename.endswith(".tmp") or filename.endswith(".md5sum") or filename.startswith("."):
continue
path = os.path.join(root, filename)
if not os.path.isfile(path) or skip_image_paths and path in skip_image_paths or path in files:
if not os.path.isfile(path) or (skip_image_paths and path in skip_image_paths) or path in files:
continue
if "/lib/" in path or "/lib64/" in path:
# ignore custom IOU libraries
@ -230,7 +230,7 @@ def default_images_directory(image_type):
elif image_type == "dynamips" or image_type == "ios":
return os.path.join(img_dir, "IOS")
else:
raise NotImplementedError(f"%s node type is not supported", image_type)
raise NotImplementedError("%s node type is not supported", image_type)
def images_directories(image_type, include_parent_directory=True):

2
gns3server/utils/packet_filter_validation.py
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:

2
gns3server/utils/uninstall_ai_features.py
View File

@ -25,7 +25,7 @@ def _read_requirements(filename):
base_dir = _find_base_dir()
filepath = os.path.join(base_dir, filename)
packages = []
with open(filepath, "r") as f:
with open(filepath) as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):

67
pyproject.toml
View File

@ -97,14 +97,11 @@ ignore = [
"E402", # 129 module-import-not-at-top-of-file
"E501", # 655 line-too-long
"F401", # 300 unused-import
"F541", # 16 f-string-missing-placeholders
"F841", # 32 unused-variable
"I001", # 441 unsorted-imports
"PLW0603", # 21 global-statement
"PLW2901", # 30 redefined-loop-name
"RUF010", # 65 explicit-f-string-type-conversion
"RUF012", # 23 mutable-class-default
"RUF013", # 40 implicit-optional
"RUF036", # 22 none-not-at-end-of-union
"RUF059", # 29 unused-unpacked-variable
"S101", # 19 assert
@ -112,11 +109,9 @@ ignore = [
"S110", # 23 try-except-pass
"UP006", # 270 non-pep585-annotation
"UP007", # 81 non-pep604-annotation-union
"UP009", # 51 utf8-encoding-declaration
"UP032", # 560 f-string
"UP035", # 106 deprecated-import
"UP045", # 1100 non-pep604-annotation-optional
"W293", # 661 blank-line-with-whitespace
]
[tool.ruff.lint.mccabe]
@ -141,33 +136,17 @@ extend-immutable-calls = [
"tests/**" = ["S101", "S105", "S106", "S108"]
# Baseline: existing violations of rules that are enforced everywhere else.
# Remove entries one small PR at a time. Never add new ones.
"gns3server/agent/__init__.py" = ["LOG015", "RUF022"]
"gns3server/agent/gns3_copilot/agent/context_manager.py" = ["RUF005", "RUF100"]
"gns3server/agent/__init__.py" = ["LOG015"]
"gns3server/agent/gns3_copilot/agent/gns3_copilot.py" = ["DTZ003"]
"gns3server/agent/gns3_copilot/agent_service.py" = ["C901", "DTZ003"]
"gns3server/agent/gns3_copilot/chat_sessions_repository.py" = ["DTZ003", "S608"]
"gns3server/agent/gns3_copilot/gns3_client/__init__.py" = ["RUF022"]
"gns3server/agent/gns3_copilot/gns3_client/context_helpers.py" = ["RUF022"]
"gns3server/agent/gns3_copilot/skills/__init__.py" = ["RUF022"]
"gns3server/agent/gns3_copilot/skills/loader.py" = ["UP015"]
"gns3server/agent/gns3_copilot/skills/manager.py" = ["UP037"]
"gns3server/agent/gns3_copilot/skills/registry.py" = ["C901"]
"gns3server/agent/gns3_copilot/tools_v2/__init__.py" = ["RUF022"]
"gns3server/agent/gns3_copilot/tools_v2/config_tools_nornir.py" = ["RUF100"]
"gns3server/agent/gns3_copilot/tools_v2/display_tools_nornir.py" = ["RUF100"]
"gns3server/agent/gns3_copilot/tools_v2/gns3_packet_filter.py" = ["PLW1510", "S603", "S607"]
"gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py" = ["PLW1510", "RUF005", "S603", "S607"]
"gns3server/agent/gns3_copilot/tools_v2/vpcs_tools_netmiko.py" = ["RUF100"]
"gns3server/agent/gns3_copilot/utils/__init__.py" = ["RUF022"]
"gns3server/agent/gns3_copilot/utils/custom_netmiko/__init__.py" = ["RUF100"]
"gns3server/agent/gns3_copilot/tools_v2/packet_analysis_tool.py" = ["PLW1510", "S603", "S607"]
"gns3server/agent/gns3_copilot/utils/custom_netmiko/huawei_ce.py" = ["S107"]
"gns3server/agent/gns3_copilot/utils/custom_netmiko/ruijie_telnet.py" = ["LOG015"]
"gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py" = ["DTZ005", "UP024"]
"gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_huawei_ce.py" = ["RUF100"]
"gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_ruijie_telnet.py" = ["RUF100"]
"gns3server/agent/gns3_copilot/utils/custom_netmiko/tests/test_vpcs_telnet.py" = ["RUF100"]
"gns3server/agent/gns3_copilot/utils/custom_netmiko/scripts/list_netmiko_telnet_devices.py" = ["DTZ005"]
"gns3server/agent/gns3_copilot/utils/custom_netmiko/vpcs_telnet.py" = ["S107"]
"gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py" = ["RUF100"]
"gns3server/agent/gns3_copilot/utils/message_converters.py" = ["RUF019"]
"gns3server/agent/gns3_copilot/utils/parse_tool_content.py" = ["C901", "DTZ003"]
"gns3server/agent/mcp/__init__.py" = ["RUF003"]
@ -175,21 +154,15 @@ extend-immutable-calls = [
"gns3server/agent/web_wireshark/manager.py" = ["C901"]
"gns3server/agent/web_wireshark/setup_wireshark_image.py" = ["PLW1510", "S603", "S607"]
"gns3server/agent/web_wireshark/stats.py" = ["ASYNC221", "PLW1510", "S603", "S607"]
"gns3server/api/routes/compute/notifications.py" = ["W291"]
"gns3server/api/routes/compute/qemu_nodes.py" = ["F811"]
"gns3server/api/routes/controller/controller.py" = ["UP012"]
"gns3server/api/routes/controller/dependencies/authentication.py" = ["S105", "S106"]
"gns3server/api/routes/controller/links.py" = ["E722"]
"gns3server/api/routes/controller/netmiko.py" = ["RUF100"]
"gns3server/api/routes/controller/nodes.py" = ["C901", "RUF019", "UP024"]
"gns3server/api/routes/controller/projects.py" = ["UP012"]
"gns3server/api/routes/controller/nodes.py" = ["C901", "RUF019"]
"gns3server/api/routes/controller/templates.py" = ["S324"]
"gns3server/api/routes/controller/users.py" = ["S105", "S106"]
"gns3server/api/routes/index.py" = ["RUF021"]
"gns3server/compute/base_node.py" = ["C901", "RUF019"]
"gns3server/compute/builtin/nodes/cloud.py" = ["E713"]
"gns3server/compute/docker/__init__.py" = ["B006", "C901"]
"gns3server/compute/docker/docker_vm.py" = ["B006", "C901", "E731", "RUF003", "RUF006", "S108", "UP031"]
"gns3server/compute/docker/docker_vm.py" = ["B006", "C901", "E731", "RUF003", "RUF006", "S108"]
"gns3server/compute/docker/iol_docker_vm.py" = ["RUF002", "S108"]
"gns3server/compute/docker/vendor_docker_vm.py" = ["RUF002", "S108"]
"gns3server/compute/dynamips/__init__.py" = ["C901"]
@ -198,11 +171,10 @@ extend-immutable-calls = [
"gns3server/compute/iou/iou_vm.py" = ["C901", "S108", "S324"]
"gns3server/compute/iou/utils/iou_export.py" = ["C901", "RUF003"]
"gns3server/compute/project.py" = ["DTZ006", "S324"]
"gns3server/compute/qemu/qemu_vm.py" = ["ASYNC220", "B009", "C901", "RUF003", "S603"]
"gns3server/compute/qemu/utils/ziputils.py" = ["RUF005", "UP031"]
"gns3server/compute/qemu/qemu_vm.py" = ["ASYNC220", "C901", "RUF003", "S603"]
"gns3server/compute/ubridge/ubridge_hypervisor.py" = ["C901"]
"gns3server/compute/virtualbox/virtualbox_vm.py" = ["C901", "RUF005", "S314"]
"gns3server/compute/vmware/__init__.py" = ["C901", "UP015"]
"gns3server/compute/virtualbox/virtualbox_vm.py" = ["C901", "S314"]
"gns3server/compute/vmware/__init__.py" = ["C901"]
"gns3server/compute/vmware/vmware_vm.py" = ["C901"]
"gns3server/compute/vpcs/vpcs_vm.py" = ["B018"]
"gns3server/config.py" = ["B905", "C901"]
@ -220,52 +192,47 @@ extend-immutable-calls = [
"gns3server/controller/project.py" = ["ASYNC110", "C901", "RUF006", "RUF019"]
"gns3server/controller/topology.py" = ["C901"]
"gns3server/controller/udp_link.py" = ["C901"]
"gns3server/db/models/base.py" = ["UP031"]
"gns3server/db/repositories/llm_model_configs.py" = ["DTZ003", "RUF005", "RUF019"]
"gns3server/db/repositories/llm_model_configs.py" = ["DTZ003", "RUF019"]
"gns3server/db/repositories/rbac.py" = ["C901"]
"gns3server/db/tasks.py" = ["B009", "RUF006", "RUF019", "UP004"]
"gns3server/db_migrations/versions/7ceeddd9c9a8_init.py" = ["W291"]
"gns3server/db/tasks.py" = ["RUF006", "RUF019"]
"gns3server/db_migrations/versions/ec4b7b198555_convert_pickletype_fields_to_json.py" = ["S301", "S608"]
"gns3server/logger.py" = ["PLW0127", "UP031"]
"gns3server/main.py" = ["UP031"]
"gns3server/logger.py" = ["PLW0127"]
"gns3server/schemas/controller/settings.py" = ["S105"]
"gns3server/schemas/controller/templates/docker_templates.py" = ["W292"]
"gns3server/schemas/controller/tokens.py" = ["S105"]
"gns3server/server.py" = ["DTZ011", "RUF006", "UP036"]
"gns3server/services/authentication.py" = ["S105"]
"gns3server/services/templates.py" = ["W292"]
"gns3server/utils/__init__.py" = ["S324", "UP031"]
"gns3server/utils/__init__.py" = ["S324"]
"gns3server/utils/asyncio/__init__.py" = ["RUF006"]
"gns3server/utils/asyncio/aiozipstream.py" = ["C901", "UP031"]
"gns3server/utils/asyncio/aiozipstream.py" = ["C901"]
"gns3server/utils/asyncio/raw_command_server.py" = ["B006"]
"gns3server/utils/asyncio/telnet_server.py" = ["RUF006"]
"gns3server/utils/images.py" = ["RUF021", "S324"]
"gns3server/utils/images.py" = ["S324"]
"gns3server/utils/packet_filter_validation.py" = ["PLW1510", "S603", "S607", "W292"]
"gns3server/utils/picture.py" = ["C901"]
"gns3server/utils/port_allocator.py" = ["S324"]
"gns3server/utils/uninstall_ai_features.py" = ["PLW1510", "S603", "UP015"]
"gns3server/utils/uninstall_ai_features.py" = ["PLW1510", "S603"]
"gns3server/utils/vmnet.py" = ["S605", "S607"]
"gns3server/utils/websocket_to_websocket.py" = ["C901"]
"gns3server/utils/zipfile_zstd/__init__.py" = ["F403"]
"gns3server/version.py" = ["S607"]
"scripts/extract_mermaid.py" = ["C901", "PLW1510", "S603", "S607"]
"scripts/tag.py" = ["S603", "S607"]
"scripts/welcome.py" = ["C901", "E722", "PLW1510", "S108", "S112", "S603", "S605", "S606", "S607", "W291"]
"scripts/welcome.py" = ["C901", "E722", "PLW1510", "S108", "S112", "S603", "S605", "S606", "S607"]
"tests/agent/mcp/test_handlers.py" = ["B905"]
"tests/agent/mcp/test_tool_params.py" = ["C901"]
"tests/api/routes/controller/test_images.py" = ["S324", "W292"]
"tests/api/routes/controller/test_links.py" = ["RUF015", "W292"]
"tests/api/routes/controller/test_users.py" = ["F811"]
"tests/compute/docker/test_docker.py" = ["B018"]
"tests/compute/docker/test_docker_vm.py" = ["F811", "UP004"]
"tests/compute/docker/test_docker_vm.py" = ["F811"]
"tests/compute/docker/test_iol_docker_vm.py" = ["RUF003", "RUF015"]
"tests/compute/docker/test_vendor_docker_vm.py" = ["RUF015"]
"tests/compute/qemu/test_qemu_vm.py" = ["E741"]
"tests/compute/test_project.py" = ["RUF003", "RUF064", "S103"]
"tests/compute/virtualbox/test_virtualbox_manager.py" = ["B011"]
"tests/compute/vpcs/test_vpcs_vm.py" = ["RUF003"]
"tests/controller/gns3vm/test_vmware_gns3_vm.py" = ["UP015"]
"tests/controller/test_export_project.py" = ["E713"]
"tests/controller/test_import_project.py" = ["F811"]
"tests/controller/test_marker.py" = ["B006"]
"tests/controller/test_node.py" = ["E711", "F811"]

2
scripts/extract_mermaid.py
View File

@ -337,7 +337,7 @@ def main():
versions = os.listdir(item_path)
print(f" {item}: {', '.join(versions)}")
else:
print(f" Puppeteer cache: NOT FOUND")
print(" Puppeteer cache: NOT FOUND")
print(f"\n Result: {'✓ ALL CHECKS PASSED' if ok else '✗ SOME CHECKS FAILED'}")
sys.exit(0 if ok else 1)

1
scripts/welcome.py
View File

@ -1,5 +1,4 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Copyright (C) 2015 GNS3 Technologies Inc.
#

2
tests/agent/mcp/test_handlers.py
View File

@ -553,7 +553,7 @@ class TestLink:
# a short path-bound ticket replaces the long Bearer JWT in the URL
assert "?token=gns3t_" in result["download_url"]
assert "Bearer" not in result["curl_command"]
path = f"/v3/projects/p1/links/l1/capture/file"
path = "/v3/projects/p1/links/l1/capture/file"
ticket = result["download_url"].split("token=")[1]
assert result["curl_command"] == f"curl -L -o capture.pcap '{result['download_url']}'"
redeemed = access_ticket_service.redeem_for_path(ticket, path)

1
tests/api/routes/compute/test_capabilities.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_cloud_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_compute.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_docker_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_dynamips_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_ethernet_switch_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2022 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_iou_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_nat_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_projects.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_qemu_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_virtualbox_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_vmware_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/compute/test_vpcs_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

12
tests/api/routes/controller/test_acl.py
View File

@ -59,7 +59,7 @@ class TestACLRoutes:
await rbac_repo.create_ace(ace)
# add an ACE on /projects to allow user to create a project
path = f"/projects"
path = "/projects"
new_ace = {"path": path, "ace_type": "user", "user_id": str(test_user.user_id), "role_id": role_id}
response = await authorized_client.post(app.url_path_for("create_ace"), json=new_ace)
@ -99,7 +99,7 @@ class TestACLRoutes:
async def test_get_ace(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
rbac_repo = RbacRepository(db_session)
ace_in_db = await rbac_repo.get_ace_by_path(f"/projects")
ace_in_db = await rbac_repo.get_ace_by_path("/projects")
response = await client.get(app.url_path_for("get_ace", ace_id=ace_in_db.ace_id))
assert response.status_code == status.HTTP_200_OK
assert response.json()["ace_id"] == str(ace_in_db.ace_id)
@ -115,13 +115,13 @@ class TestACLRoutes:
) -> None:
rbac_repo = RbacRepository(db_session)
ace_in_db = await rbac_repo.get_ace_by_path(f"/projects")
ace_in_db = await rbac_repo.get_ace_by_path("/projects")
update_ace = {"path": f"/appliances", "ace_type": "user", "user_id": str(test_user.user_id), "role_id": role_id}
update_ace = {"path": "/appliances", "ace_type": "user", "user_id": str(test_user.user_id), "role_id": role_id}
response = await client.put(app.url_path_for("update_ace", ace_id=ace_in_db.ace_id), json=update_ace)
assert response.status_code == status.HTTP_200_OK
updated_ace_in_db = await rbac_repo.get_ace(ace_in_db.ace_id)
assert updated_ace_in_db.path == f"/appliances"
assert updated_ace_in_db.path == "/appliances"
async def test_delete_ace(
self,
@ -131,7 +131,7 @@ class TestACLRoutes:
) -> None:
rbac_repo = RbacRepository(db_session)
ace_in_db = await rbac_repo.get_ace_by_path(f"/appliances")
ace_in_db = await rbac_repo.get_ace_by_path("/appliances")
response = await client.delete(app.url_path_for("delete_ace", ace_id=ace_in_db.ace_id))
assert response.status_code == status.HTTP_204_NO_CONTENT

1
tests/api/routes/controller/test_drawings.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/controller/test_gns3vm.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/controller/test_links.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/controller/test_markers.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2025 GNS3 Technologies Inc.
#

1
tests/api/routes/controller/test_nodes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/controller/test_projects.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/controller/test_settings.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2026 GNS3 Technologies Inc.
#

1
tests/api/routes/controller/test_symbols.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/controller/test_version.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/test_index.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

1
tests/api/routes/test_routes.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

3
tests/compute/docker/test_docker_vm.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#
@ -2052,7 +2051,7 @@ async def test_delete_reports_root_files_when_reclaim_fails(vm):
@pytest.mark.asyncio
async def test_read_console_output_with_binary_mode(vm):
class InputStreamMock(object):
class InputStreamMock:
def __init__(self):
self.sent = False

1
tests/compute/dynamips/test_dynamips_manager.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#

Some files were not shown because too many files have changed in this diff Show More