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

style: apply ruff format to the files this branch touches

Browse Source
This commit is contained in:
YueGuobin 2026-10-14 22:13:58 +08:00
parent 97b5514134
commit 3e6a53c4b9
No known key found for this signature in database
12 changed files with 17 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -795,7 +795,7 @@ class EthernetSwitch(KernelDatapathMixin, BaseNode):
anchor = self._kernel_ports.get(port_number)
if anchor is not None:
# Per-interface stop: only this port's capture is stopped.
await self._ubridge_send(f'capture stop_kernel {anchor}')
await self._ubridge_send(f"capture stop_kernel {anchor}")
else:
ubridge_bridge = self._ubridge_bridge_name(port_number)
await self._ubridge_send(f"bridge stop_capture {ubridge_bridge}")

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

@ -2044,7 +2044,7 @@ class DockerVM(DockerKernelDatapathMixin, BaseNode):
# capturing flag above gates it, so uBridge's 208-on-idle-
# interface can only surface on a genuine desync).
host_ifc = self._kernel_host_ifc(adapter_number, port_number)
await self._ubridge_send(f'capture stop_kernel {host_ifc}')
await self._ubridge_send(f"capture stop_kernel {host_ifc}")
return
bridge_name = self._bridge_name(adapter_number, port_number)
await self._ubridge_send(f"bridge stop_capture {bridge_name}")

2
gns3server/compute/dynamips/nodes/router.py
View File

@ -1536,7 +1536,7 @@ class Router(KernelDatapathMixin, BaseNode):
anchor = self._kernel_host_ifc(slot_number, port_number)
if self.ubridge and anchor is not None:
# Per-interface stop: only this port's capture is stopped.
await self._ubridge_send(f'capture stop_kernel {anchor}')
await self._ubridge_send(f"capture stop_kernel {anchor}")
return
await nio.stop_packet_capture()

2
gns3server/compute/iou/iou_vm.py
View File

@ -1869,7 +1869,7 @@ class IOUVM(KernelDatapathMixin, BaseNode):
# Per-interface stop: only this port's capture is stopped.
anchor = self._kernel_host_ifc(adapter_number, port_number)
if anchor is not None:
await self._ubridge_send(f'capture stop_kernel {anchor}')
await self._ubridge_send(f"capture stop_kernel {anchor}")
else:
await self._ubridge_send(
f"iol_bridge stop_capture {self._iol_bridge_name()} {adapter_number} {port_number}"

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

@ -1790,7 +1790,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
anchor = self._kernel_host_ifc(adapter_number)
if isinstance(nio, NIOBridge) and anchor is not None:
# Per-interface stop: only this port's capture is stopped.
await self._ubridge_send(f'capture stop_kernel {anchor}')
await self._ubridge_send(f"capture stop_kernel {anchor}")
else:
await self._ubridge_send("bridge stop_capture {name}".format(name=f"QEMU-{self._id}-{adapter_number}"))

2
gns3server/compute/vpcs/vpcs_vm.py
View File

@ -691,7 +691,7 @@ class VPCSVM(KernelDatapathMixin, BaseNode):
anchor = self._kernel_host_ifc(0, port_number)
if isinstance(nio, NIOBridge) and anchor is not None:
# Per-interface stop: only this port's capture is stopped.
await self._ubridge_send(f'capture stop_kernel {anchor}')
await self._ubridge_send(f"capture stop_kernel {anchor}")
else:
await self._ubridge_send("bridge stop_capture {name}".format(name=f"VPCS-{self._id}"))

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

@ -792,7 +792,7 @@ class TestEthernetSwitchNodesRoutes:
)
)
assert response.status_code == status.HTTP_204_NO_CONTENT
node._ubridge_send.assert_any_call(f'capture stop_kernel {anchor}')
node._ubridge_send.assert_any_call(f"capture stop_kernel {anchor}")
async def test_ethernet_switch_batch_create_anchor_nio(
self, app: FastAPI, compute_client: AsyncClient, compute_project: Project, ethernet_switch: dict

8
tests/compute/docker/test_docker_kernel_datapath.py
View File

@ -997,7 +997,7 @@ async def test_stop_capture_kernel(vm):
await vm.stop_capture(0)
assert nio.capturing is False
vm._ubridge_send.assert_any_call(f'capture stop_kernel {host_ifc}')
vm._ubridge_send.assert_any_call(f"capture stop_kernel {host_ifc}")
@pytest.mark.asyncio
@ -1061,11 +1061,11 @@ async def test_capture_two_ports_at_once_and_stop_them_independently(vm):
assert nio1.capturing is True
assert nio2.capturing is False
vm._ubridge_send.assert_any_call(f'capture stop_kernel {host_ifc2}')
assert not any(f'stop_kernel {host_ifc1}' in str(c) for c in vm._ubridge_send.call_args_list)
vm._ubridge_send.assert_any_call(f"capture stop_kernel {host_ifc2}")
assert not any(f"stop_kernel {host_ifc1}" in str(c) for c in vm._ubridge_send.call_args_list)
await vm.stop_capture(0)
vm._ubridge_send.assert_any_call(f'capture stop_kernel {host_ifc1}')
vm._ubridge_send.assert_any_call(f"capture stop_kernel {host_ifc1}")
@pytest.mark.asyncio

2
tests/compute/dynamips/test_dynamips_kernel_datapath.py
View File

@ -581,7 +581,7 @@ async def test_capture_on_a_kernel_link_uses_the_anchor(router):
assert not any("bind_filter" in c for c in _hypervisor_commands(router))
await router.stop_capture(0, 0)
router._ubridge_send.assert_any_call(f'capture stop_kernel {TAP00}')
router._ubridge_send.assert_any_call(f"capture stop_kernel {TAP00}")
@pytest.mark.asyncio

2
tests/compute/iou/test_iou_kernel_datapath.py
View File

@ -371,7 +371,7 @@ async def test_capture_on_a_kernel_link_uses_the_anchor(vm):
vm._ubridge_send.assert_any_call(f'capture start_kernel {TAP00} "/tmp/test.pcap"')
await vm.stop_capture(0, 0)
vm._ubridge_send.assert_any_call(f'capture stop_kernel {TAP00}')
vm._ubridge_send.assert_any_call(f"capture stop_kernel {TAP00}")
@pytest.mark.asyncio

2
tests/compute/qemu/test_qemu_kernel_datapath.py
View File

@ -328,7 +328,7 @@ async def test_capture_on_a_kernel_link_uses_the_anchor(vm):
vm._ubridge_send.assert_any_call(f'capture start_kernel {TAP0} "/tmp/test.pcap"')
await vm.stop_capture(0)
vm._ubridge_send.assert_any_call(f'capture stop_kernel {TAP0}')
vm._ubridge_send.assert_any_call(f"capture stop_kernel {TAP0}")
@pytest.mark.asyncio

4
tests/e2e/test_docker_kernel_datapath.py
View File

@ -352,7 +352,9 @@ def test_docker_kernel_ten_concurrent_captures():
# every later frame (up to and including the ARP each first ping
# triggers) lands in a pcap
captures = [
compute.call("POST", f"/projects/{pid}/links/{link['link_id']}/capture/start", {"data_link_type": "DLT_EN10MB"})
compute.call(
"POST", f"/projects/{pid}/links/{link['link_id']}/capture/start", {"data_link_type": "DLT_EN10MB"}
)
for link in links
]