NgTech-LTD/gns3-server
2
0 Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-11-03 05:51:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch '3.1' into fix-api

Browse Source
This commit is contained in:
Mark Paronyan authored and GitHub committed 2026-10-05 21:23:03 +03:00
2 parents b8e552ee9d 65c858d455
commit cbd4f457be
73 files changed
+591 -1741
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/api/routes/controller/test_nodes.py
+3 -9
View File
@@ -465,9 +465,7 @@ class TestNodeRoutes:
compute.http_query.assert_called_with(
"GET",
"/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(
project_id=project.id, node_id=node.id
),
f"/projects/{project.id}/files/project-files/vpcs/{node.id}/hello",
timeout=None,
stream=True,
)
@@ -491,16 +489,12 @@ class TestNodeRoutes:
compute.http_query.assert_called_once()
call_args = compute.http_query.call_args
assert call_args[0][0] == "POST"
assert call_args[0][1] == "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(
project_id=project.id, node_id=node.id
)
assert call_args[0][1] == f"/projects/{project.id}/files/project-files/vpcs/{node.id}/hello"
assert call_args[1]["timeout"] is None
# data should be an async generator from request.stream()
assert hasattr(call_args[1]["data"], "__aiter__")
response = await client.get(
"/projects/{project_id}/nodes/{node_id}/files/../hello".format(project_id=project.id, node_id=node.id)
)
response = await client.get(f"/projects/{project.id}/nodes/{node.id}/files/../hello")
assert response.status_code == status.HTTP_404_NOT_FOUND
# @pytest.mark.asyncio
tests/api/routes/controller/test_projects.py
+9 -12
View File
@@ -244,10 +244,9 @@ class TestControllerProjectRoutes:
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["CONTENT-TYPE"] == "application/gns3project"
assert response.headers[
"CONTENT-DISPOSITION"
] == "attachment; filename=\"{name}.gns3project\"; filename*=UTF-8''{name}.gns3project".format(
name=project.name
assert (
response.headers["CONTENT-DISPOSITION"]
== f"attachment; filename=\"{project.name}.gns3project\"; filename*=UTF-8''{project.name}.gns3project"
)
with open(str(tmpdir / "project.zip"), "wb+") as f:
@@ -283,10 +282,9 @@ class TestControllerProjectRoutes:
)
assert response.status_code == status.HTTP_200_OK
assert response.headers["CONTENT-TYPE"] == "application/gns3project"
assert response.headers[
"CONTENT-DISPOSITION"
] == "attachment; filename=\"{name}.gns3project\"; filename*=UTF-8''{name}.gns3project".format(
name=project.name
assert (
response.headers["CONTENT-DISPOSITION"]
== f"attachment; filename=\"{project.name}.gns3project\"; filename*=UTF-8''{project.name}.gns3project"
)
with open(str(tmpdir / "project.zip"), "wb+") as f:
f.write(response.content)
@@ -343,10 +341,9 @@ class TestControllerProjectRoutes:
if response.status_code == status.HTTP_200_OK:
assert response.headers["CONTENT-TYPE"] == "application/gns3project"
assert response.headers[
"CONTENT-DISPOSITION"
] == "attachment; filename=\"{name}.gns3project\"; filename*=UTF-8''{name}.gns3project".format(
name=project.name
assert (
response.headers["CONTENT-DISPOSITION"]
== f"attachment; filename=\"{project.name}.gns3project\"; filename*=UTF-8''{project.name}.gns3project"
)
with open(str(tmpdir / "project.zip"), "wb+") as f:
f.write(response.content)
tests/api/routes/controller/test_version.py
+1 -3
View File
@@ -42,9 +42,7 @@ class TestVersionRoutes:
params = {"version": "0.4.2"}
response = await client.post(app.url_path_for("check_version"), json=params)
assert response.status_code == status.HTTP_409_CONFLICT
assert response.json() == {
"message": "Client version 0.4.2 is not the same as server version {}".format(__version__)
}
assert response.json() == {"message": f"Client version 0.4.2 is not the same as server version {__version__}"}
async def test_version_invalid_input_schema(self, app: FastAPI, client: AsyncClient) -> None:
tests/compute/builtin/nodes/test_cloud.py
+11 -11
View File
@@ -189,11 +189,11 @@ async def test_linux_ethernet_raw_add_nio(linux_platform, compute_project, nio):
ubridge_mock.assert_has_calls(
[
call("bridge create {}-0".format(cloud._id)),
call("bridge add_nio_udp {}-0 4242 127.0.0.1 4343".format(cloud._id)),
call("bridge reset_packet_filters {}-0".format(cloud._id)),
call('bridge add_nio_linux_raw {}-0 "eth0"'.format(cloud._id)),
call("bridge start {}-0".format(cloud._id)),
call(f"bridge create {cloud._id}-0"),
call(f"bridge add_nio_udp {cloud._id}-0 4242 127.0.0.1 4343"),
call(f"bridge reset_packet_filters {cloud._id}-0"),
call(f'bridge add_nio_linux_raw {cloud._id}-0 "eth0"'),
call(f"bridge start {cloud._id}-0"),
]
)
@@ -218,11 +218,11 @@ async def test_linux_ethernet_raw_add_nio_bridge(linux_platform, compute_project
tap = "gns3tap0-0"
ubridge_mock.assert_has_calls(
[
call("bridge create {}-0".format(cloud._id)),
call("bridge add_nio_udp {}-0 4242 127.0.0.1 4343".format(cloud._id)),
call("bridge reset_packet_filters {}-0".format(cloud._id)),
call('bridge add_nio_tap "{}-0" "{}"'.format(cloud._id, tap)),
call('brctl addif "bridge0" "{}"'.format(tap)),
call("bridge start {}-0".format(cloud._id)),
call(f"bridge create {cloud._id}-0"),
call(f"bridge add_nio_udp {cloud._id}-0 4242 127.0.0.1 4343"),
call(f"bridge reset_packet_filters {cloud._id}-0"),
call(f'bridge add_nio_tap "{cloud._id}-0" "{tap}"'),
call(f'brctl addif "bridge0" "{tap}"'),
call(f"bridge start {cloud._id}-0"),
]
)
tests/compute/docker/test_docker.py
+3 -3
View File
@@ -51,7 +51,7 @@ async def test_query_success(vm):
data = await vm.query("POST", "test", data={"a": True}, params={"b": 1})
vm._session.request.assert_called_with(
"POST",
"http://docker/v{}/test".format(DOCKER_MINIMUM_API_VERSION),
f"http://docker/v{DOCKER_MINIMUM_API_VERSION}/test",
data='{"a": true}',
headers={"content-type": "application/json"},
params={"b": 1},
@@ -76,7 +76,7 @@ async def test_query_error(vm):
await vm.query("POST", "test", data={"a": True}, params={"b": 1})
vm._session.request.assert_called_with(
"POST",
"http://docker/v{}/test".format(DOCKER_MINIMUM_API_VERSION),
f"http://docker/v{DOCKER_MINIMUM_API_VERSION}/test",
data='{"a": true}',
headers={"content-type": "application/json"},
params={"b": 1},
@@ -99,7 +99,7 @@ async def test_query_error_json(vm):
await vm.query("POST", "test", data={"a": True}, params={"b": 1})
vm._session.request.assert_called_with(
"POST",
"http://docker/v{}/test".format(DOCKER_MINIMUM_API_VERSION),
f"http://docker/v{DOCKER_MINIMUM_API_VERSION}/test",
data='{"a": true}',
headers={"content-type": "application/json"},
params={"b": 1},
tests/compute/docker/test_docker_vm.py
+22 -26
View File
@@ -103,7 +103,7 @@ async def test_create(compute_project, manager):
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -145,7 +145,7 @@ async def test_create_with_tag(compute_project, manager):
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -192,7 +192,7 @@ async def test_create_vnc(compute_project, manager):
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -423,7 +423,7 @@ async def test_create_start_cmd(compute_project, manager):
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -520,7 +520,7 @@ async def test_create_image_digest_match(compute_project, manager):
)
await vm.create()
# the last query is the container creation: the digest check let it through
assert mock.call_args[0] == ("POST", "containers/create?name={}".format(vm.docker_name))
assert mock.call_args[0] == ("POST", f"containers/create?name={vm.docker_name}")
assert vm._cid == "sha256:" + "a" * 64
@@ -554,7 +554,7 @@ async def test_create_with_user(compute_project, manager):
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -647,7 +647,7 @@ async def test_create_with_extra_volumes_duplicate_1_image(compute_project, mana
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -699,7 +699,7 @@ async def test_create_with_extra_volumes_duplicate_2_user(compute_project, manag
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -751,7 +751,7 @@ async def test_create_with_extra_volumes_duplicate_3_subdir(compute_project, man
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -803,7 +803,7 @@ async def test_create_with_extra_volumes_duplicate_4_backslash(compute_project,
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -853,7 +853,7 @@ async def test_create_with_extra_volumes_duplicate_5_subdir_issue_1595(compute_p
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -905,7 +905,7 @@ async def test_create_with_extra_volumes_duplicate_6_subdir_issue_1595(compute_p
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -956,7 +956,7 @@ async def test_create_with_extra_volumes(compute_project, manager):
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -1376,7 +1376,7 @@ async def test_update(vm):
mock_query.assert_any_call("DELETE", "containers/e90e34656842", params={"force": 1, "v": 1})
mock_query.assert_any_call(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -1448,7 +1448,7 @@ async def test_update_running(vm):
mock_query.assert_any_call("DELETE", "containers/e90e34656842", params={"force": 1, "v": 1})
mock_query.assert_any_call(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -1633,7 +1633,7 @@ async def test_add_ubridge_connection_no_free_interface(vm):
nio = vm.manager.create_nio(nio)
with pytest.raises(DockerError):
# We create fake ethernet interfaces for docker
interfaces = ["tap-gns3-e{}".format(index) for index in range(4096)]
interfaces = [f"tap-gns3-e{index}" for index in range(4096)]
with patch("psutil.net_if_addrs", return_value=interfaces):
await vm._add_ubridge_connection(nio, 0)
@@ -1821,9 +1821,9 @@ async def test_start_vnc(vm):
"None",
"-desktop",
"test",
":{}".format(vm._display),
f":{vm._display}",
)
mock_wait.assert_called_with("/tmp/.X11-unix/X{}".format(vm._display))
mock_wait.assert_called_with(f"/tmp/.X11-unix/X{vm._display}")
@pytest.mark.asyncio
@@ -1882,9 +1882,7 @@ async def test_fix_permission(vm):
"/gns3/bin/busybox",
"sh",
"-c",
'(/gns3/bin/busybox find "/etc" -depth -print0 | /gns3/bin/busybox xargs -0 /gns3/bin/busybox stat -c \'%a:%u:%g:%n\' > "/etc/.gns3_perms") && /gns3/bin/busybox chmod -R u+rX "/etc" && /gns3/bin/busybox chown {}:{} -R "/etc"'.format(
os.getuid(), os.getgid()
),
f'(/gns3/bin/busybox find "/etc" -depth -print0 | /gns3/bin/busybox xargs -0 /gns3/bin/busybox stat -c \'%a:%u:%g:%n\' > "/etc/.gns3_perms") && /gns3/bin/busybox chmod -R u+rX "/etc" && /gns3/bin/busybox chown {os.getuid()}:{os.getgid()} -R "/etc"',
stderr=asyncio.subprocess.PIPE,
)
assert process.wait.called
@@ -1907,9 +1905,7 @@ async def test_fix_permission_not_running(vm):
"/gns3/bin/busybox",
"sh",
"-c",
'(/gns3/bin/busybox find "/etc" -depth -print0 | /gns3/bin/busybox xargs -0 /gns3/bin/busybox stat -c \'%a:%u:%g:%n\' > "/etc/.gns3_perms") && /gns3/bin/busybox chmod -R u+rX "/etc" && /gns3/bin/busybox chown {}:{} -R "/etc"'.format(
os.getuid(), os.getgid()
),
f'(/gns3/bin/busybox find "/etc" -depth -print0 | /gns3/bin/busybox xargs -0 /gns3/bin/busybox stat -c \'%a:%u:%g:%n\' > "/etc/.gns3_perms") && /gns3/bin/busybox chmod -R u+rX "/etc" && /gns3/bin/busybox chown {os.getuid()}:{os.getgid()} -R "/etc"',
stderr=asyncio.subprocess.PIPE,
)
assert mock_start.called
@@ -2080,7 +2076,7 @@ async def test_cpus(compute_project, manager):
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
@@ -2122,7 +2118,7 @@ async def test_memory(compute_project, manager):
await vm.create()
mock.assert_called_with(
"POST",
"containers/create?name={}".format(vm.docker_name),
f"containers/create?name={vm.docker_name}",
data={
"Tty": True,
"OpenStdin": True,
tests/compute/iou/test_iou_vm.py
+7 -7
View File
@@ -56,7 +56,7 @@ def iourc_file(tmpdir):
path = str(tmpdir / "iourc")
with open(path, "w+") as f:
hostname = socket.gethostname()
f.write("[license]\n{} = aaaaaaaaaaaaaaaa;".format(hostname))
f.write(f"[license]\n{hostname} = aaaaaaaaaaaaaaaa;")
return path
@@ -182,15 +182,15 @@ async def test_rename_nvram_file(vm):
It should rename the nvram file to the correct name before launching the VM
"""
with open(os.path.join(vm.working_dir, "nvram_0000{}".format(vm.application_id + 1)), "w+") as f:
with open(os.path.join(vm.working_dir, f"nvram_0000{vm.application_id + 1}"), "w+") as f:
f.write("1")
with open(os.path.join(vm.working_dir, "vlan.dat-0000{}".format(vm.application_id + 1)), "w+") as f:
with open(os.path.join(vm.working_dir, f"vlan.dat-0000{vm.application_id + 1}"), "w+") as f:
f.write("1")
vm._rename_nvram_file()
assert os.path.exists(os.path.join(vm.working_dir, "nvram_0000{}".format(vm.application_id)))
assert os.path.exists(os.path.join(vm.working_dir, "vlan.dat-0000{}".format(vm.application_id)))
assert os.path.exists(os.path.join(vm.working_dir, f"nvram_0000{vm.application_id}"))
assert os.path.exists(os.path.join(vm.working_dir, f"vlan.dat-0000{vm.application_id}"))
@pytest.mark.asyncio
@@ -589,13 +589,13 @@ async def test_invalid_iou_file(vm, iourc_file):
# Missing ;
with pytest.raises(IOUError):
with open(iourc_file, "w+") as f:
f.write("[license]\n{} = aaaaaaaaaaaaaaaa".format(hostname))
f.write(f"[license]\n{hostname} = aaaaaaaaaaaaaaaa")
await vm._check_iou_license()
# Key too short
with pytest.raises(IOUError):
with open(iourc_file, "w+") as f:
f.write("[license]\n{} = aaaaaaaaaaaaaa;".format(hostname))
f.write(f"[license]\n{hostname} = aaaaaaaaaaaaaa;")
await vm._check_iou_license()
# Invalid hostname
tests/compute/qemu/test_qemu_vm.py
+24 -26
View File
@@ -384,7 +384,7 @@ async def test_disk_options(vm, tmpdir, fake_qemu_img_binary):
fake_qemu_img_binary,
"create",
"-o",
"backing_file={}".format(vm._hda_disk_image),
f"backing_file={vm._hda_disk_image}",
"-F",
"qcow2",
"-f",
@@ -394,7 +394,7 @@ async def test_disk_options(vm, tmpdir, fake_qemu_img_binary):
fake_qemu_img_binary,
"create",
"-o",
"backing_file={}".format(vm._hdb_disk_image),
f"backing_file={vm._hdb_disk_image}",
"-F",
"qcow2",
"-f",
@@ -449,7 +449,7 @@ async def test_uefi_boot_mode_option(vm, tmpdir, images_dir, fake_qemu_img_binar
f.write("1")
options = await vm._build_command()
assert " ".join(["-drive", "if=pflash,format=raw,readonly,file={}".format(ovmf_code_path)]) in " ".join(options)
assert " ".join(["-drive", f"if=pflash,format=raw,readonly,file={ovmf_code_path}"]) in " ".join(options)
assert " ".join(
["-drive", "if=pflash,format=raw,file={}".format(os.path.join(vm.working_dir, "OVMF_VARS_4M.fd"))]
) in " ".join(options)
@@ -490,7 +490,7 @@ async def test_tpm_option(vm, tmpdir, fake_qemu_img_binary):
tpm_sock = os.path.join(vm.temporary_directory, "swtpm.sock")
with patch("os.path.exists", return_value=True) as os_path:
options = await vm._build_command()
assert "-chardev socket,id=chrtpm,path={}".format(tpm_sock) in " ".join(options)
assert f"-chardev socket,id=chrtpm,path={tpm_sock}" in " ".join(options)
assert "-tpmdev emulator,id=tpm0,chardev=chrtpm" in " ".join(options)
assert "-device tpm-tis,tpmdev=tpm0" in " ".join(options)
@@ -608,7 +608,7 @@ async def test_build_command(vm, fake_qemu_binary):
"-uuid",
vm.id,
"-serial",
"telnet:127.0.0.1:{},server,nowait".format(vm._internal_console_port),
f"telnet:127.0.0.1:{vm._internal_console_port},server,nowait",
"-net",
"none",
"-device",
@@ -616,9 +616,9 @@ async def test_build_command(vm, fake_qemu_binary):
"-device",
"pci-bridge,id=pci-bridge1,bus=dmi_pci_bridge1,chassis_nr=0x1,addr=0x1,shpc=off",
"-device",
"e1000,mac={},bus=pci-bridge1,addr=0x00,netdev=gns3-0".format(vm._mac_address),
f"e1000,mac={vm._mac_address},bus=pci-bridge1,addr=0x00,netdev=gns3-0",
"-netdev",
"socket,id=gns3-0,udp=127.0.0.1:{},localaddr=127.0.0.1:{}".format(nio.rport, nio.lport),
f"socket,id=gns3-0,udp=127.0.0.1:{nio.rport},localaddr=127.0.0.1:{nio.lport}",
"-display",
"none",
]
@@ -679,7 +679,7 @@ async def test_build_command_kvm_2_4(linux_platform, vm, fake_qemu_binary):
"-uuid",
vm.id,
"-serial",
"telnet:127.0.0.1:{},server,nowait".format(vm._internal_console_port),
f"telnet:127.0.0.1:{vm._internal_console_port},server,nowait",
"-net",
"none",
"-device",
@@ -687,9 +687,9 @@ async def test_build_command_kvm_2_4(linux_platform, vm, fake_qemu_binary):
"-device",
"pci-bridge,id=pci-bridge1,bus=dmi_pci_bridge1,chassis_nr=0x1,addr=0x1,shpc=off",
"-device",
"e1000,mac={},bus=pci-bridge1,addr=0x00,netdev=gns3-0".format(vm._mac_address),
f"e1000,mac={vm._mac_address},bus=pci-bridge1,addr=0x00,netdev=gns3-0",
"-netdev",
"socket,id=gns3-0,udp=127.0.0.1:{},localaddr=127.0.0.1:{}".format(nio.rport, nio.lport),
f"socket,id=gns3-0,udp=127.0.0.1:{nio.rport},localaddr=127.0.0.1:{nio.lport}",
"-nographic",
]
@@ -727,7 +727,7 @@ async def test_build_command_two_adapters(vm, fake_qemu_binary):
"-uuid",
vm.id,
"-serial",
"telnet:127.0.0.1:{},server,nowait".format(vm._internal_console_port),
f"telnet:127.0.0.1:{vm._internal_console_port},server,nowait",
"-net",
"none",
"-device",
@@ -735,15 +735,13 @@ async def test_build_command_two_adapters(vm, fake_qemu_binary):
"-device",
"pci-bridge,id=pci-bridge1,bus=dmi_pci_bridge1,chassis_nr=0x1,addr=0x1,shpc=off",
"-device",
"e1000,mac={},bus=pci-bridge1,addr=0x00,netdev=gns3-0".format(vm._mac_address),
f"e1000,mac={vm._mac_address},bus=pci-bridge1,addr=0x00,netdev=gns3-0",
"-netdev",
"socket,id=gns3-0,udp=127.0.0.1:{},localaddr=127.0.0.1:{}".format(nio1.rport, nio1.lport),
f"socket,id=gns3-0,udp=127.0.0.1:{nio1.rport},localaddr=127.0.0.1:{nio1.lport}",
"-device",
"e1000,mac={},bus=pci-bridge1,addr=0x01,netdev=gns3-1".format(
int_to_macaddress(macaddress_to_int(vm._mac_address) + 1)
),
f"e1000,mac={int_to_macaddress(macaddress_to_int(vm._mac_address) + 1)},bus=pci-bridge1,addr=0x01,netdev=gns3-1",
"-netdev",
"socket,id=gns3-1,udp=127.0.0.1:{},localaddr=127.0.0.1:{}".format(nio2.rport, nio2.lport),
f"socket,id=gns3-1,udp=127.0.0.1:{nio2.rport},localaddr=127.0.0.1:{nio2.lport}",
"-nographic",
]
@@ -761,8 +759,8 @@ async def test_build_command_two_adapters_mac_address(vm):
assert mac_0[:8] == "00:00:ab"
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()):
cmd = await vm._build_command()
assert "e1000,mac={},bus=pci-bridge1,addr=0x00,netdev=gns3-0".format(mac_0) in cmd
assert "e1000,mac={},bus=pci-bridge1,addr=0x01,netdev=gns3-1".format(mac_1) in cmd
assert f"e1000,mac={mac_0},bus=pci-bridge1,addr=0x00,netdev=gns3-0" in cmd
assert f"e1000,mac={mac_1},bus=pci-bridge1,addr=0x01,netdev=gns3-1" in cmd
vm.mac_address = "00:42:ab:0e:0f:0a"
mac_0 = vm._mac_address
@@ -770,8 +768,8 @@ async def test_build_command_two_adapters_mac_address(vm):
assert mac_0[:8] == "00:42:ab"
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()):
cmd = await vm._build_command()
assert "e1000,mac={},bus=pci-bridge1,addr=0x00,netdev=gns3-0".format(mac_0) in cmd
assert "e1000,mac={},bus=pci-bridge1,addr=0x01,netdev=gns3-1".format(mac_1) in cmd
assert f"e1000,mac={mac_0},bus=pci-bridge1,addr=0x00,netdev=gns3-0" in cmd
assert f"e1000,mac={mac_1},bus=pci-bridge1,addr=0x01,netdev=gns3-1" in cmd
@pytest.mark.asyncio
@@ -793,19 +791,19 @@ async def test_build_command_large_number_of_adapters(vm):
assert len([l for l in cmd if "e1000" in l]) == 100
assert len(vm._ethernet_adapters) == 100
assert "e1000,mac={},bus=pci-bridge1,addr=0x00,netdev=gns3-0".format(mac_0) in cmd
assert "e1000,mac={},bus=pci-bridge1,addr=0x01,netdev=gns3-1".format(mac_1) in cmd
assert f"e1000,mac={mac_0},bus=pci-bridge1,addr=0x00,netdev=gns3-0" in cmd
assert f"e1000,mac={mac_1},bus=pci-bridge1,addr=0x01,netdev=gns3-1" in cmd
assert "pci-bridge,id=pci-bridge0,bus=dmi_pci_bridge0,chassis_nr=0x1,addr=0x0,shpc=off" not in cmd
assert "pci-bridge,id=pci-bridge1,bus=dmi_pci_bridge1,chassis_nr=0x1,addr=0x1,shpc=off" in cmd
assert "pci-bridge,id=pci-bridge2,bus=dmi_pci_bridge2,chassis_nr=0x1,addr=0x2,shpc=off" in cmd
assert "i82801b11-bridge,id=dmi_pci_bridge1" in cmd
mac_29 = int_to_macaddress(macaddress_to_int(vm._mac_address) + 29)
assert "e1000,mac={},bus=pci-bridge1,addr=0x1d,netdev=gns3-29".format(mac_29) in cmd
assert f"e1000,mac={mac_29},bus=pci-bridge1,addr=0x1d,netdev=gns3-29" in cmd
mac_30 = int_to_macaddress(macaddress_to_int(vm._mac_address) + 30)
assert "e1000,mac={},bus=pci-bridge1,addr=0x1e,netdev=gns3-30".format(mac_30) in cmd
assert f"e1000,mac={mac_30},bus=pci-bridge1,addr=0x1e,netdev=gns3-30" in cmd
mac_74 = int_to_macaddress(macaddress_to_int(vm._mac_address) + 74)
assert "e1000,mac={},bus=pci-bridge3,addr=0x0a,netdev=gns3-74".format(mac_74) in cmd
assert f"e1000,mac={mac_74},bus=pci-bridge3,addr=0x0a,netdev=gns3-74" in cmd
@pytest.mark.asyncio
tests/compute/virtualbox/test_virtualbox_manager.py
+1 -1
View File
@@ -91,7 +91,7 @@ async def test_list_vms(manager):
return ["memory=512"]
elif args[0] == "ccd8c50b-c172-457d-99fa-dd69371ede0e":
return ["memory=256"]
assert False, "Unknow {} {}".format(cmd, args)
assert False, f"Unknow {cmd} {args}"
with asyncio_patch("gns3server.compute.virtualbox.VirtualBox.execute") as mock:
mock.side_effect = execute_mock
tests/compute/vpcs/test_vpcs_manager.py
+1 -1
View File
@@ -74,5 +74,5 @@ async def test_get_mac_id_no_id_available(compute_project, port_manager):
with pytest.raises(VPCSError):
for i in range(0, 256):
node_id = str(uuid.uuid4())
await vpcs.create_node("PC {}".format(i), compute_project.id, node_id)
await vpcs.create_node(f"PC {i}", compute_project.id, node_id)
assert vpcs.get_mac_id(node_id) == i
tests/compute/vpcs/test_vpcs_vm.py
+1 -1
View File
@@ -298,7 +298,7 @@ def test_update_startup_script_with_escaping_characters_in_name(vm):
vm.startup_script = "set pcname initial-name\n"
vm.name = "test\\"
assert vm.startup_script == "set pcname test{}".format(os.linesep)
assert vm.startup_script == f"set pcname test{os.linesep}"
def test_get_startup_script(vm):
tests/controller/test_compute.py
+2 -2
View File
@@ -400,7 +400,7 @@ async def test_downloadFile(project, compute):
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
await compute.download_file(project, "test/titi")
mock.assert_called_with(
"GET", "https://example.com:84/v3/compute/projects/{}/files/test/titi".format(project.id), auth=None
"GET", f"https://example.com:84/v3/compute/projects/{project.id}/files/test/titi", auth=None
)
await compute.close()
@@ -492,7 +492,7 @@ async def test_list_files(project, compute):
assert await compute.list_files(project) == res
mock.assert_any_call(
"GET",
"https://example.com:84/v3/compute/projects/{}/files".format(project.id),
f"https://example.com:84/v3/compute/projects/{project.id}/files",
headers={"content-type": "application/json"},
data=None,
auth=None,
tests/controller/test_node.py
+19 -21
View File
@@ -312,7 +312,7 @@ async def test_create(node, compute):
assert await node.create() is True
data = {"console": 2048, "console_type": "vnc", "node_id": node.id, "startup_script": "echo test", "name": "demo"}
compute.post.assert_called_with("/projects/{}/vpcs/nodes".format(node.project.id), data=data, timeout=1200)
compute.post.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes", data=data, timeout=1200)
assert node._console == 2048
assert node._properties == {"startup_script": "echo test"}
@@ -359,7 +359,7 @@ async def test_create_image_missing_kept_in_degraded_state(project, compute, tmp
async def resp(*args, **kwargs):
raise ComputeConflictError(
"/projects/{}/qemu/nodes".format(project.id),
f"/projects/{project.id}/qemu/nodes",
{"message": "The image is missing", "image": "missing.qcow2", "exception": "ImageMissingError"},
)
@@ -390,7 +390,7 @@ async def test_create_image_missing_raises_by_default(project, compute, tmpdir,
async def resp(*args, **kwargs):
raise ComputeConflictError(
"/projects/{}/qemu/nodes".format(project.id),
f"/projects/{project.id}/qemu/nodes",
{"message": "The image is missing", "image": "missing.qcow2", "exception": "ImageMissingError"},
)
@@ -513,7 +513,7 @@ async def test_create_docker_image_missing_after_failed_pull(project, compute):
async def resp(*args, **kwargs):
raise ComputeConflictError(
"/projects/{}/docker/nodes".format(project.id),
f"/projects/{project.id}/docker/nodes",
{"message": "The image is missing", "image": "ghost:latest", "exception": "ImageMissingError"},
)
@@ -544,7 +544,7 @@ async def test_create_docker_image_missing_pull_error_raises_by_default(project,
async def resp(*args, **kwargs):
raise ComputeConflictError(
"/projects/{}/docker/nodes".format(project.id),
f"/projects/{project.id}/docker/nodes",
{"message": "The image is missing", "image": "ghost:latest", "exception": "ImageMissingError"},
)
@@ -714,7 +714,7 @@ async def test_create_base_script(node, config, compute, tmpdir):
"name": "demo",
}
compute.post.assert_called_with("/projects/{}/vpcs/nodes".format(node.project.id), data=data, timeout=1200)
compute.post.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes", data=data, timeout=1200)
def test_symbol(node, symbols_dir):
@@ -779,7 +779,7 @@ async def test_update(node, compute, project, controller):
await node.update(x=42, console=2048, console_type="vnc", properties={"startup_script": "echo test"}, name="demo")
data = {"console": 2048, "console_type": "vnc", "startup_script": "echo test", "name": "demo"}
compute.put.assert_called_with("/projects/{}/vpcs/nodes/{}".format(node.project.id, node.id), data=data)
compute.put.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes/{node.id}", data=data)
assert node._console == 2048
assert node.x == 42
assert node._properties == {"startup_script": "echo test"}
@@ -799,7 +799,7 @@ async def test_update_properties(node, compute, controller):
await node.update(x=42, console=2048, console_type="vnc", properties={"startup_script": "hello world"}, name="demo")
data = {"console": 2048, "console_type": "vnc", "startup_script": "hello world", "name": "demo"}
compute.put.assert_called_with("/projects/{}/vpcs/nodes/{}".format(node.project.id, node.id), data=data)
compute.put.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes/{node.id}", data=data)
assert node._console == 2048
assert node.x == 42
assert node._properties == {"startup_script": "echo test"}
@@ -933,7 +933,7 @@ async def test_start(node, compute):
compute.post = AsyncioMagicMock()
await node.start()
compute.post.assert_called_with("/projects/{}/vpcs/nodes/{}/start".format(node.project.id, node.id), timeout=240)
compute.post.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes/{node.id}/start", timeout=240)
@pytest.mark.asyncio
@@ -949,7 +949,7 @@ async def test_start_iou(compute, project, controller):
controller._iou_license_settings = {"license_check": True, "iourc_content": "aa"}
await node.start()
compute.post.assert_called_with(
"/projects/{}/iou/nodes/{}/start".format(node.project.id, node.id),
f"/projects/{node.project.id}/iou/nodes/{node.id}/start",
timeout=240,
data={"license_check": True, "iourc_content": "aa"},
)
@@ -962,7 +962,7 @@ async def test_stop(node, compute):
await node.stop()
compute.post.assert_called_with(
"/projects/{}/vpcs/nodes/{}/stop".format(node.project.id, node.id), timeout=240, dont_connect=True
f"/projects/{node.project.id}/vpcs/nodes/{node.id}/stop", timeout=240, dont_connect=True
)
@@ -971,7 +971,7 @@ async def test_suspend(node, compute):
compute.post = AsyncioMagicMock()
await node.suspend()
compute.post.assert_called_with("/projects/{}/vpcs/nodes/{}/suspend".format(node.project.id, node.id), timeout=240)
compute.post.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes/{node.id}/suspend", timeout=240)
@pytest.mark.asyncio
@@ -979,7 +979,7 @@ async def test_reload(node, compute):
compute.post = AsyncioMagicMock()
await node.reload()
compute.post.assert_called_with("/projects/{}/vpcs/nodes/{}/reload".format(node.project.id, node.id), timeout=240)
compute.post.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes/{node.id}/reload", timeout=240)
@pytest.mark.asyncio
@@ -994,7 +994,7 @@ async def test_create_without_console(node, compute):
await node.create()
data = {"console_type": "vnc", "node_id": node.id, "startup_script": "echo test", "name": "demo"}
compute.post.assert_called_with("/projects/{}/vpcs/nodes".format(node.project.id), data=data, timeout=1200)
compute.post.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes", data=data, timeout=1200)
assert node._console == 2048
assert node._properties == {"test_value": "success", "startup_script": "echo test"}
@@ -1003,21 +1003,21 @@ async def test_create_without_console(node, compute):
async def test_delete(node, compute):
await node.destroy()
compute.delete.assert_called_with("/projects/{}/vpcs/nodes/{}".format(node.project.id, node.id))
compute.delete.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes/{node.id}")
@pytest.mark.asyncio
async def test_post(node, compute):
await node.post("/test", {"a": "b"})
compute.post.assert_called_with("/projects/{}/vpcs/nodes/{}/test".format(node.project.id, node.id), data={"a": "b"})
compute.post.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes/{node.id}/test", data={"a": "b"})
@pytest.mark.asyncio
async def test_delete(node, compute):
await node.delete("/test")
compute.delete.assert_called_with("/projects/{}/vpcs/nodes/{}/test".format(node.project.id, node.id))
compute.delete.assert_called_with(f"/projects/{node.project.id}/vpcs/nodes/{node.id}/test")
@pytest.mark.asyncio
@@ -1028,9 +1028,7 @@ async def test_dynamips_idle_pc(node, compute):
response.json = {"idlepc": "0x60606f54"}
compute.get = AsyncioMagicMock(return_value=response)
await node.dynamips_auto_idlepc()
compute.get.assert_called_with(
"/projects/{}/dynamips/nodes/{}/auto_idlepc".format(node.project.id, node.id), timeout=240
)
compute.get.assert_called_with(f"/projects/{node.project.id}/dynamips/nodes/{node.id}/auto_idlepc", timeout=240)
@pytest.mark.asyncio
@@ -1042,7 +1040,7 @@ async def test_dynamips_idlepc_proposals(node, compute):
compute.get = AsyncioMagicMock(return_value=response)
await node.dynamips_idlepc_proposals()
compute.get.assert_called_with(
"/projects/{}/dynamips/nodes/{}/idlepc_proposals".format(node.project.id, node.id), timeout=240
f"/projects/{node.project.id}/dynamips/nodes/{node.id}/idlepc_proposals", timeout=240
)
tests/controller/test_project.py
+6 -6
View File
@@ -122,7 +122,7 @@ async def test_update_on_compute(controller):
project._project_created_on_compute = [compute]
project.emit_notification = MagicMock()
await project.update(variables=variables)
compute.put.assert_any_call("/projects/{}".format(project.id), {"variables": variables})
compute.put.assert_any_call(f"/projects/{project.id}", {"variables": variables})
@pytest.mark.asyncio
@@ -203,7 +203,7 @@ async def test_add_node_local(controller):
},
)
compute.post.assert_any_call(
"/projects/{}/vpcs/nodes".format(project.id),
f"/projects/{project.id}/vpcs/nodes",
data={"node_id": node.id, "startup_script": "test.cfg", "name": "test"},
timeout=1200,
)
@@ -272,7 +272,7 @@ async def test_add_node_non_local(controller):
compute.post.assert_any_call("/projects", data={"name": project._name, "project_id": project._id})
compute.post.assert_any_call(
"/projects/{}/vpcs/nodes".format(project.id),
f"/projects/{project.id}/vpcs/nodes",
data={"node_id": node.id, "startup_script": "test.cfg", "name": "test"},
timeout=1200,
)
@@ -448,7 +448,7 @@ async def test_add_node_iou_no_id_available(controller):
with pytest.raises(ControllerError):
for i in range(1, 513):
prop = {"properties": {"application_id": i}}
project._nodes[i] = Node(project, compute, "Node{}".format(i), node_id=i, node_type="iou", **prop)
project._nodes[i] = Node(project, compute, f"Node{i}", node_id=i, node_type="iou", **prop)
await project.add_node(compute, "test1", None, node_type="iou")
@@ -538,7 +538,7 @@ async def test_delete_node(controller):
await project.delete_node(node.id)
assert node.id not in project._nodes
compute.delete.assert_any_call("/projects/{}/vpcs/nodes/{}".format(project.id, node.id))
compute.delete.assert_any_call(f"/projects/{project.id}/vpcs/nodes/{node.id}")
project.emit_notification.assert_any_call("node.deleted", node.asdict())
@@ -585,7 +585,7 @@ async def test_delete_node_delete_link(controller):
assert node.id not in project._nodes
assert link.id not in project._links
compute.delete.assert_any_call("/projects/{}/vpcs/nodes/{}".format(project.id, node.id))
compute.delete.assert_any_call(f"/projects/{project.id}/vpcs/nodes/{node.id}")
project.emit_notification.assert_any_call("node.deleted", node.asdict())
project.emit_notification.assert_any_call("link.deleted", link.asdict())
tests/controller/test_udp_link.py
+16 -24
View File
@@ -74,7 +74,7 @@ async def test_create(project):
await link.add_node(node2, 3, 1)
compute1.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/nio".format(project.id, node1.id),
f"/projects/{project.id}/vpcs/nodes/{node1.id}/adapters/0/ports/4/nio",
data={
"lport": 1024,
"rhost": "192.168.1.2",
@@ -88,7 +88,7 @@ async def test_create(project):
)
compute2.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/3/ports/1/nio".format(project.id, node2.id),
f"/projects/{project.id}/vpcs/nodes/{node2.id}/adapters/3/ports/1/nio",
data={
"lport": 2048,
"rhost": "192.168.1.1",
@@ -152,7 +152,7 @@ async def test_create_one_side_failure(project):
await link.add_node(node2, 3, 1)
compute1.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/nio".format(project.id, node1.id),
f"/projects/{project.id}/vpcs/nodes/{node1.id}/adapters/0/ports/4/nio",
data={
"lport": 1024,
"rhost": "192.168.1.2",
@@ -166,7 +166,7 @@ async def test_create_one_side_failure(project):
)
compute2.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/3/ports/1/nio".format(project.id, node2.id),
f"/projects/{project.id}/vpcs/nodes/{node2.id}/adapters/3/ports/1/nio",
data={
"lport": 2048,
"rhost": "192.168.1.1",
@@ -179,9 +179,7 @@ async def test_create_one_side_failure(project):
timeout=120,
)
# The link creation has failed we rollback the nio
compute1.delete.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/nio".format(project.id, node1.id), timeout=120
)
compute1.delete.assert_any_call(f"/projects/{project.id}/vpcs/nodes/{node1.id}/adapters/0/ports/4/nio", timeout=120)
@pytest.mark.asyncio
@@ -202,12 +200,8 @@ async def test_delete(project):
await link.delete()
compute1.delete.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/nio".format(project.id, node1.id), timeout=120
)
compute2.delete.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/3/ports/1/nio".format(project.id, node2.id), timeout=120
)
compute1.delete.assert_any_call(f"/projects/{project.id}/vpcs/nodes/{node1.id}/adapters/0/ports/4/nio", timeout=120)
compute2.delete.assert_any_call(f"/projects/{project.id}/vpcs/nodes/{node2.id}/adapters/3/ports/1/nio", timeout=120)
@pytest.mark.asyncio
@@ -310,7 +304,7 @@ async def test_reset(project):
assert link.debug_link_data[1]["lport"] != link.debug_link_data[1]["rport"]
# the committed NIO carries the fresh pair, not the released one
compute1.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/nio".format(project.id, node1.id),
f"/projects/{project.id}/vpcs/nodes/{node1.id}/adapters/0/ports/4/nio",
data={
"lport": 4096,
"rhost": "192.168.1.2",
@@ -401,16 +395,14 @@ async def test_capture(project):
assert link.capturing
compute1.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/capture/start".format(project.id, node_vpcs.id),
f"/projects/{project.id}/vpcs/nodes/{node_vpcs.id}/adapters/0/ports/4/capture/start",
data={"capture_file_name": link.default_capture_file_name(), "data_link_type": "DLT_EN10MB"},
)
await link.stop_capture()
assert link.capturing is False
compute1.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/capture/stop".format(project.id, node_vpcs.id)
)
compute1.post.assert_any_call(f"/projects/{project.id}/vpcs/nodes/{node_vpcs.id}/adapters/0/ports/4/capture/stop")
@pytest.mark.asyncio
@@ -483,7 +475,7 @@ async def test_update(project):
await link.add_node(node2, 3, 1)
compute1.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/nio".format(project.id, node1.id),
f"/projects/{project.id}/vpcs/nodes/{node1.id}/adapters/0/ports/4/nio",
data={
"lport": 1024,
"rhost": "192.168.1.2",
@@ -497,7 +489,7 @@ async def test_update(project):
)
compute2.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/3/ports/1/nio".format(project.id, node2.id),
f"/projects/{project.id}/vpcs/nodes/{node2.id}/adapters/3/ports/1/nio",
data={
"lport": 2048,
"rhost": "192.168.1.1",
@@ -513,7 +505,7 @@ async def test_update(project):
assert link.created
await link.update_filters({"frequency_drop": [5], "bpf": ["icmp[icmptype] == 8"]})
compute1.put.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/nio".format(project.id, node1.id),
f"/projects/{project.id}/vpcs/nodes/{node1.id}/adapters/0/ports/4/nio",
data={
"lport": 1024,
"rhost": "192.168.1.2",
@@ -563,7 +555,7 @@ async def test_update_ethernet_switch_nio(project):
await link.update_filters({"delay": [10, 0]})
compute1.put.assert_any_call(
"/projects/{}/ethernet_switch/nodes/{}/adapters/3/ports/1/nio".format(project.id, node_switch.id),
f"/projects/{project.id}/ethernet_switch/nodes/{node_switch.id}/adapters/3/ports/1/nio",
data={
"lport": 1024,
"rhost": "192.168.1.1",
@@ -625,7 +617,7 @@ async def test_update_suspend(project):
await link.add_node(node2, 3, 1)
compute1.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/0/ports/4/nio".format(project.id, node1.id),
f"/projects/{project.id}/vpcs/nodes/{node1.id}/adapters/0/ports/4/nio",
data={
"lport": 1024,
"rhost": "192.168.1.2",
@@ -639,7 +631,7 @@ async def test_update_suspend(project):
)
compute2.post.assert_any_call(
"/projects/{}/vpcs/nodes/{}/adapters/3/ports/1/nio".format(project.id, node2.id),
f"/projects/{project.id}/vpcs/nodes/{node2.id}/adapters/3/ports/1/nio",
data={
"lport": 2048,
"rhost": "192.168.1.1",
tests/test_topologies.py
+15 -15
View File
@@ -46,15 +46,15 @@ def test_convert(directory, tmpdir):
before_directory = os.path.join(topologies_directory, directory, "before")
after_directory = os.path.join(topologies_directory, directory, "after")
assert os.path.exists(before_directory), "No before directory found file for {}".format(directory)
assert os.path.exists(after_directory), "No after directory found file for {}".format(directory)
assert os.path.exists(before_directory), f"No before directory found file for {directory}"
assert os.path.exists(after_directory), f"No after directory found file for {directory}"
gns3_file = None
for file in os.listdir(before_directory):
if file.endswith(".gns3"):
gns3_file = file
assert gns3_file, "No .gns3 found file for {}".format(before_directory)
assert gns3_file, f"No .gns3 found file for {before_directory}"
with open(os.path.join(before_directory, gns3_file)) as f:
before_topology = json.load(f)
@@ -76,19 +76,19 @@ def test_convert(directory, tmpdir):
for file in files:
directory = os.path.relpath(root, after_directory)
file_path = os.path.join(work_directory, directory, file)
assert os.path.exists(file_path), "{} is missing".format(os.path.join(directory, file))
assert os.path.exists(file_path), f"{os.path.join(directory, file)} is missing"
# For gns3project we check if size are not too much differents
if file_path.endswith(".gns3project"):
size = os.stat(file_path).st_size
other_size = os.stat(os.path.join(os.path.join(root, file))).st_size
assert size in range(other_size - 100, other_size + 100), "File {} is different".format(
os.path.join(directory, file)
assert size in range(other_size - 100, other_size + 100), (
f"File {os.path.join(directory, file)} is different"
)
# For non .gns3 file we check if the file are the same
elif not file_path.endswith(".gns3"):
assert os.stat(file_path).st_size == os.stat(os.path.join(os.path.join(root, file))).st_size, (
"File {} is different".format(os.path.join(directory, file))
f"File {os.path.join(directory, file)} is different"
)
# Check if we don't have unexpected file in work directory
@@ -98,7 +98,7 @@ def test_convert(directory, tmpdir):
file_path = os.path.join(after_directory, directory, file)
# .backup are created by the conversion process
if ".backup" not in file_path:
assert os.path.exists(file_path), "{} should not be here".format(os.path.join(directory, file))
assert os.path.exists(file_path), f"{os.path.join(directory, file)} should not be here"
with open(os.path.join(after_directory, gns3_file)) as f:
after_topology = json.load(f)
@@ -109,12 +109,12 @@ def compare_dict(path, source, reference):
"""
Compare two dictionary of a topology
"""
assert isinstance(source, dict), "Source is not a dict in {}".format(path)
assert isinstance(source, dict), f"Source is not a dict in {path}"
for key in source:
assert key in reference, "Unexpected {} in {} it should be {}".format(key, source, reference)
assert key in reference, f"Unexpected {key} in {source} it should be {reference}"
for key in sorted(reference.keys()):
val = reference[key]
assert key in source, "{} is missing in {}".format(key, source)
assert key in source, f"{key} is missing in {source}"
if (
isinstance(val, str)
or isinstance(val, float)
@@ -133,12 +133,12 @@ def compare_dict(path, source, reference):
elif key == "version":
assert source[key] == __version__
else:
assert val == source[key], "Wrong value for {}: \n{}\nit should be\n{}".format(key, source[key], val)
assert val == source[key], f"Wrong value for {key}: \n{source[key]}\nit should be\n{val}"
elif isinstance(val, dict):
compare_dict(path + key + "/", source[key], val)
elif isinstance(val, list):
assert len(val) == len(source[key]), "Not enough value in {} ({}/{}) it shoud be {} not {}".format(
key, len(val), len(source[key]), val, source[key]
assert len(val) == len(source[key]), (
f"Not enough value in {key} ({len(val)}/{len(source[key])}) it shoud be {val} not {source[key]}"
)
for idx, element in enumerate(source[key]):
if isinstance(element, dict):
@@ -146,4 +146,4 @@ def compare_dict(path, source, reference):
else:
assert element == val[idx]
else:
assert False, "Value type for {} is not supported".format(key)
assert False, f"Value type for {key} is not supported"