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

refactor: use f-strings in controller tests

Browse Source
This commit is contained in:
Mark Paronyan 2026-09-30 20:58:12 +03:00
parent bee6b3f8b1
commit be943fcdf3
No known key found for this signature in database
GPG Key ID: 8DC7338A87FE22F8
4 changed files with 43 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tests/controller/test_compute.py
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,

40
tests/controller/test_node.py
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
)

12
tests/controller/test_project.py
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())

40
tests/controller/test_udp_link.py
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",