refactor: use f-strings in controller modules

This commit is contained in:
Mark Paronyan 2026-09-30 20:58:11 +03:00
parent 552c0ca3e0
commit 7a9c600f54
No known key found for this signature in database
GPG Key ID: 8DC7338A87FE22F8
9 changed files with 47 additions and 86 deletions

View File

@ -362,7 +362,7 @@ class Controller:
# install the built-in appliances if needed
if Config.instance().settings.Server.install_builtin_appliances:
previous_version = controller_vars.get("version")
log.info("Comparing controller version {} with config version {}".format(__version__, previous_version))
log.info(f"Comparing controller version {__version__} with config version {previous_version}")
builtin_appliances_path = self._appliance_manager.builtin_appliances_path()
if not previous_version or parse_version(__version__.split("+")[0]) > parse_version(
previous_version.split("+")[0]

View File

@ -72,9 +72,7 @@ class Drawing:
return data.decode()
except UnicodeError:
width, height, filetype = get_size(data)
return '<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" height="{height}" width="{width}">\n<image height="{height}" width="{width}" xlink:href="data:image/{filetype};base64,{b64}" />\n</svg>'.format(
b64=base64.b64encode(data).decode(), filetype=filetype, width=width, height=height
)
return f'<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" height="{height}" width="{width}">\n<image height="{height}" width="{width}" xlink:href="data:image/{filetype};base64,{base64.b64encode(data).decode()}" />\n</svg>'
except OSError:
log.warning("Image file %s missing", filename)
return "<svg></svg>"

View File

@ -175,7 +175,7 @@ def _is_exportable(path, include_snapshots=False):
return False
# do not export directories of snapshots
if include_snapshots is False and "{sep}snapshots{sep}".format(sep=os.path.sep) in path:
if include_snapshots is False and f"{os.path.sep}snapshots{os.path.sep}" in path:
return False
try:

View File

@ -60,9 +60,7 @@ class GNS3VM:
:returns: Return list of engines supported by GNS3 for the GNS3VM
"""
download_url = "https://github.com/GNS3/gns3-gui/releases/download/v{version}/GNS3.VM.VMware.Workstation.{version}.zip".format(
version=__version__
)
download_url = f"https://github.com/GNS3/gns3-gui/releases/download/v{__version__}/GNS3.VM.VMware.Workstation.{__version__}.zip"
vmware_info = {
"engine_id": "vmware",
"description": f'VMware is the recommended choice for best performances.<br>The GNS3 VM can be <a href="{download_url}">downloaded here</a>.',
@ -76,9 +74,7 @@ class GNS3VM:
vmware_info["name"] = "VMware Workstation / Player (recommended)"
download_url = (
"https://github.com/GNS3/gns3-gui/releases/download/v{version}/GNS3.VM.Hyper-V.{version}.zip".format(
version=__version__
)
f"https://github.com/GNS3/gns3-gui/releases/download/v{__version__}/GNS3.VM.Hyper-V.{__version__}.zip"
)
hyperv_info = {
"engine_id": "hyper-v",
@ -90,9 +86,7 @@ class GNS3VM:
}
download_url = (
"https://github.com/GNS3/gns3-gui/releases/download/v{version}/GNS3.VM.VirtualBox.{version}.zip".format(
version=__version__
)
f"https://github.com/GNS3/gns3-gui/releases/download/v{__version__}/GNS3.VM.VirtualBox.{__version__}.zip"
)
virtualbox_info = {
"engine_id": "virtualbox",
@ -377,9 +371,7 @@ class GNS3VM:
if netmask:
compute_network = ipaddress.ip_interface(f"{compute.host_ip}/{netmask}").network
if vm_network.compare_networks(compute_network) != 0:
msg = "The GNS3 VM (IP={}, NETWORK={}) is not on the same network as the {} server (IP={}, NETWORK={}), please make sure the local server binding is in the same network as the GNS3 VM".format(
self.ip_address, vm_network, compute_id, compute.host_ip, compute_network
)
msg = f"The GNS3 VM (IP={self.ip_address}, NETWORK={vm_network}) is not on the same network as the {compute_id} server (IP={compute.host_ip}, NETWORK={compute_network}), please make sure the local server binding is in the same network as the GNS3 VM"
self._controller.notification.controller_emit("log.warning", {"message": msg})
except ComputeError as e:
log.warning(f"Could not check the VM is in the same subnet as the local server: {e}")

View File

@ -138,7 +138,7 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
for info in result.splitlines():
if "=" in info:
name, value = info.split("=", 1)
if name == "{}{}".format(backend_type, interface_number):
if name == f"{backend_type}{interface_number}":
return value.strip('"')
return None
@ -179,7 +179,7 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
ip_address = match.group(1)
netmask = match.group(2)
else:
raise GNS3VMError("Could not find IP address and netmask for vboxnet {}".format(vboxnet))
raise GNS3VMError(f"Could not find IP address and netmask for vboxnet {vboxnet}")
try:
interface = ipaddress.IPv4Interface(f"{ip_address}/{netmask}")
@ -189,17 +189,15 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
lower_ip = str(interface.ip + 2)
upper_ip = str(subnet.network_address + subnet.num_addresses - 2)
except ValueError:
raise GNS3VMError(
"Invalid IP address and netmask for vboxnet {}: {}/{}".format(vboxnet, ip_address, netmask)
)
raise GNS3VMError(f"Invalid IP address and netmask for vboxnet {vboxnet}: {ip_address}/{netmask}")
dhcp_server_args = [
"add",
"--network=HostInterfaceNetworking-{}".format(vboxnet),
"--server-ip={}".format(dhcp_server_ip),
"--netmask={}".format(netmask),
"--lower-ip={}".format(lower_ip),
"--upper-ip={}".format(upper_ip),
f"--network=HostInterfaceNetworking-{vboxnet}",
f"--server-ip={dhcp_server_ip}",
f"--netmask={netmask}",
f"--lower-ip={lower_ip}",
f"--upper-ip={upper_ip}",
"--enable",
]
await self._execute("dhcpserver", dhcp_server_args)
@ -212,7 +210,7 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
:returns: boolean
"""
properties = await self._execute("list", ["{}".format(vboxnet_type)])
properties = await self._execute("list", [f"{vboxnet_type}"])
for prop in properties.splitlines():
try:
name, value = prop.split(":", 1)
@ -287,9 +285,7 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
interface_number = await self._look_for_interface("hostonlynetwork")
if interface_number < 0:
raise GNS3VMError(
'VM "{}" must have a network adapter attached to a host-only network in order to start'.format(
self.vmname
)
f'VM "{self.vmname}" must have a network adapter attached to a host-only network in order to start'
)
else:
backend_type = "hostonlyadapter"
@ -299,17 +295,13 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
if interface_number < 0:
raise GNS3VMError(
'VM "{}" must have a network adapter attached to a {} in order to start'.format(
self.vmname, backend_description
)
f'VM "{self.vmname}" must have a network adapter attached to a {backend_description} in order to start'
)
vboxnet = await self._look_for_vboxnet(backend_type, interface_number)
if vboxnet is None:
raise GNS3VMError(
'A VirtualBox host-only network could not be found on network adapter {} for "{}"'.format(
interface_number, self._vmname
)
f'A VirtualBox host-only network could not be found on network adapter {interface_number} for "{self._vmname}"'
)
if not (await self._check_vboxnet_exists(vboxnet, vboxnet_type)):
@ -319,9 +311,7 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
first_available_vboxnet = await self._find_first_available_vboxnet()
if first_available_vboxnet is None:
raise GNS3VMError(
'Please add a VirtualBox host-only network with DHCP enabled and attached it to network adapter {} for "{}"'.format(
interface_number, self._vmname
)
f'Please add a VirtualBox host-only network with DHCP enabled and attached it to network adapter {interface_number} for "{self._vmname}"'
)
await self.set_hostonly_network(interface_number, first_available_vboxnet)
vboxnet = first_available_vboxnet
@ -330,18 +320,14 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
await self._execute("hostonlyif", ["create"])
except GNS3VMError:
raise GNS3VMError(
'VirtualBox host-only network "{}" does not exist and could not be automatically created, please make the sure the network adapter {} configuration is valid for "{}"'.format(
vboxnet, interface_number, self._vmname
)
f'VirtualBox host-only network "{vboxnet}" does not exist and could not be automatically created, please make the sure the network adapter {interface_number} configuration is valid for "{self._vmname}"'
)
if backend_type == "hostonlyadapter" and not (await self._check_dhcp_server(vboxnet)):
try:
await self._add_dhcp_server(vboxnet)
except GNS3VMError as e:
raise GNS3VMError(
"Could not add DHCP server for vboxnet {}: {}, please configure manually".format(vboxnet, e)
)
raise GNS3VMError(f"Could not add DHCP server for vboxnet {vboxnet}: {e}, please configure manually")
vm_state = await self._get_state()
log.info(f'"{self._vmname}" state is {vm_state}')
@ -372,7 +358,7 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
else:
# if we can't get the IP address from the guest property, we try to get it from the GNS3 server (a NAT interface is required)
if nat_interface_number < 0:
raise GNS3VMError("Could not find guest IP address for {}".format(self.vmname))
raise GNS3VMError(f"Could not find guest IP address for {self.vmname}")
log.warning("Could not find IP address from guest property, trying to get it from GNS3 server")
ip_address = "127.0.0.1"
try:
@ -382,33 +368,27 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
s.bind((ip_address, 0))
api_port = s.getsockname()[1]
except OSError as e:
raise GNS3VMError("Error while getting random port: {}".format(e))
raise GNS3VMError(f"Error while getting random port: {e}")
if await self._check_vbox_port_forwarding():
# delete the GNS3VM NAT port forwarding rule if it exists
log.info("Removing GNS3VM NAT port forwarding rule from interface {}".format(nat_interface_number))
await self._execute(
"controlvm", [self._vmname, "natpf{}".format(nat_interface_number), "delete", "GNS3VM"]
)
log.info(f"Removing GNS3VM NAT port forwarding rule from interface {nat_interface_number}")
await self._execute("controlvm", [self._vmname, f"natpf{nat_interface_number}", "delete", "GNS3VM"])
# add a GNS3VM NAT port forwarding rule to redirect 127.0.0.1 with random port to the port in the VM
log.info(
"Adding GNS3VM NAT port forwarding rule with port {} to interface {}".format(
api_port, nat_interface_number
)
)
log.info(f"Adding GNS3VM NAT port forwarding rule with port {api_port} to interface {nat_interface_number}")
await self._execute(
"controlvm",
[
self._vmname,
"natpf{}".format(nat_interface_number),
"GNS3VM,tcp,{},{},,{}".format(ip_address, api_port, self.port),
f"natpf{nat_interface_number}",
f"GNS3VM,tcp,{ip_address},{api_port},,{self.port}",
],
)
self.ip_address = await self._get_ip_from_server(interface_number, api_port)
log.info("GNS3 VM has been started with IP '{}'".format(self.ip_address))
log.info(f"GNS3 VM has been started with IP '{self.ip_address}'")
self.running = True
async def _get_ip_from_guest_property(self):
@ -446,8 +426,9 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
json_data = await resp.json()
if json_data:
for interface in json_data:
if "name" in interface and interface["name"] == "eth{}".format(
hostonly_interface_number - 1
if (
"name" in interface
and interface["name"] == f"eth{hostonly_interface_number - 1}"
):
if "ip_address" in interface and len(interface["ip_address"]) > 0:
return interface["ip_address"]
@ -537,7 +518,5 @@ class VirtualBoxGNS3VM(BaseGNS3VM):
"modifyvm", [self._vmname, f"--hostonlyadapter{adapter_number}", hostonly_network_name], timeout=3
)
log.info(
'VirtualBox host-only network "{}" set on network adapter {} for "{}"'.format(
hostonly_network_name, adapter_number, self._vmname
)
f'VirtualBox host-only network "{hostonly_network_name}" set on network adapter {adapter_number} for "{self._vmname}"'
)

View File

@ -388,7 +388,7 @@ async def update_snapshots(snapshots_dir, project_path, project_name, project_id
async with aiofiles.open(snapshot_path, "wb+") as f:
async for chunk in zstream:
await f.write(chunk)
log.info("Project '{}': updated and repacked snapshot file '{}'".format(project_name, snapshot))
log.info(f"Project '{project_name}': updated and repacked snapshot file '{snapshot}'")
except OSError as e:
raise ControllerError(
f"Cannot update snapshot '{os.path.basename(snapshot)}': the snapshot cannot be recreated: {e}"

View File

@ -552,14 +552,8 @@ class Link:
adapter_number = self.capture_node["adapter_number"]
port_number = self.capture_node["port_number"]
url = (
"/projects/{project_id}/{node_type}/nodes/{node_id}/adapters/{adapter_number}/"
"ports/{port_number}/capture/stream".format(
project_id=self.project.id,
node_type=node_type,
node_id=node_id,
adapter_number=adapter_number,
port_number=port_number,
)
f"/projects/{self.project.id}/{node_type}/nodes/{node_id}/adapters/{adapter_number}/"
f"ports/{port_number}/capture/stream"
)
return compute._getUrl(url)

View File

@ -1483,7 +1483,7 @@ class Project:
try:
snapshot = Snapshot(self, filename=snap)
except ValueError:
log.error("Invalid snapshot file: {}".format(snap))
log.error(f"Invalid snapshot file: {snap}")
continue
self._snapshots[snapshot.id] = snapshot
else:
@ -1492,13 +1492,13 @@ class Project:
try:
path = os.path.join(snapshot_dir, snapshot_entry["filename"])
if not os.path.isfile(path):
log.warning("Snapshot file '{}' does not exist".format(path))
log.warning(f"Snapshot file '{path}' does not exist")
continue
snapshot_entry.pop("project_id")
snapshot = Snapshot(self, **snapshot_entry)
self._snapshots[snapshot.id] = snapshot
except KeyError:
log.error("Invalid entry in snapshot config file: {}".format(snapshot_entry))
log.error(f"Invalid entry in snapshot config file: {snapshot_entry}")
continue
self._save_snapshot_config()
@ -1515,7 +1515,7 @@ class Project:
with open(self._snapshot_conf_path, "w+") as f:
json.dump(self._snapshot_conf, f, indent=4)
except OSError as e:
log.error("Cannot write snapshot config '{}': {}".format(self._snapshot_conf_path, e))
log.error(f"Cannot write snapshot config '{self._snapshot_conf_path}': {e}")
@open_required
async def snapshot(self, name):
@ -2142,7 +2142,7 @@ class Project:
# remote replication is not supported with remote computes
for compute in self.computes:
if compute.id != "local":
log.warning("Fast duplication is not supported with remote compute: '{}'".format(compute.id))
log.warning(f"Fast duplication is not supported with remote compute: '{compute.id}'")
return None
# work dir
p_work = pathlib.Path(location or self.path).parent.absolute()
@ -2156,9 +2156,7 @@ class Project:
await wait_run_in_executor(
shutil.copytree, self.path, new_project_path.as_posix(), symlinks=True, ignore_dangling_symlinks=True
)
log.info(
"Project content copied from '{}' to '{}' in {}s".format(self.path, new_project_path, time.time() - t0)
)
log.info(f"Project content copied from '{self.path}' to '{new_project_path}' in {time.time() - t0}s")
# Read the topology file using the actual filename (self._filename), not self.name
# This handles the case where a project has been renamed but we need to read the actual file
@ -2177,7 +2175,7 @@ class Project:
regenerate_topology_ids(topology, new_project_path, reset_mac_addresses)
# dump the updated .gns3 project file
dot_gns3_path = new_project_path.joinpath("{}.gns3".format(project_name))
dot_gns3_path = new_project_path.joinpath(f"{project_name}.gns3")
topology["project_id"] = new_project_id
with open(dot_gns3_path, "w+") as f:
json.dump(topology, f, indent=4, sort_keys=True)
@ -2190,7 +2188,7 @@ class Project:
# Remove the old .gns3 file (which has the original project name)
os.remove(old_gns3_file)
project = await self.controller.load_project(dot_gns3_path, load=False)
log.info("Project '{}': fast duplicated in {:.4f} seconds".format(project.name, time.time() - t0))
log.info(f"Project '{project.name}': fast duplicated in {time.time() - t0:.4f} seconds")
return project
def is_running(self):

View File

@ -79,7 +79,7 @@ class Snapshot:
.replace(tzinfo=None)
.strftime(DESCRIPTION_DATETIME_FORMAT)
)
description = "Snapshot '{}' taken on {}".format(self._name, date)
description = f"Snapshot '{self._name}' taken on {date}"
self._description = description
self._filename = filename
@ -157,7 +157,7 @@ class Snapshot:
auto_open=self._project.auto_open,
auto_close=self._project.auto_close,
)
log.info("Snapshot '{}' restored in {:.4f} seconds".format(self.name, time.time() - begin))
log.info(f"Snapshot '{self.name}' restored in {time.time() - begin:.4f} seconds")
except (OSError, PermissionError) as e:
raise ControllerError(str(e))
await project.open()