Handle creating Qemu disk images and resizing

tests/api/routes/compute/test_qemu_nodes.py

@@ -17,13 +17,13 @@
 
 import pytest
 import os
-import sys
 import stat
+import shutil
 
 from fastapi import FastAPI, status
 from httpx import AsyncClient
 from tests.utils import asyncio_patch
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 
 from gns3server.compute.project import Project
 
@@ -52,6 +52,16 @@ def fake_qemu_vm(images_dir) -> str:
     return bin_path
 
 
+@pytest.fixture
+def fake_qemu_img_binary(tmpdir):
+
+    bin_path = str(tmpdir / "qemu-img")
+    with open(bin_path, "w+") as f:
+        f.write("1")
+    os.chmod(bin_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+    return bin_path
+
+
 @pytest.fixture
 def base_params(tmpdir, fake_qemu_bin) -> dict:
     """Return standard parameters"""
@@ -60,9 +70,12 @@ def base_params(tmpdir, fake_qemu_bin) -> dict:
 
 
 @pytest.fixture
-async def vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None:
+async def qemu_vm(app: FastAPI, compute_client: AsyncClient, compute_project: Project, base_params: dict) -> None:
 
-    response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=base_params)
+    response = await compute_client.post(
+        app.url_path_for("compute:create_qemu_node", project_id=compute_project.id),
+        json=base_params
+    )
     assert response.status_code == status.HTTP_201_CREATED
     return response.json()
 
@@ -116,99 +129,107 @@ async def test_qemu_create_with_params(app: FastAPI,
     assert response.json()["hda_disk_image_md5sum"] == "c4ca4238a0b923820dcc509a6f75849b"
 
 
-async def test_qemu_create_with_project_file(app: FastAPI,
-                                             compute_client: AsyncClient,
-                                             compute_project: Project,
-                                             base_params: dict,
-                                             fake_qemu_vm: str) -> None:
-
-    response = await compute_client.post(app.url_path_for("compute:write_compute_project_file",
-                                                  project_id=compute_project.id,
-                                                  file_path="hello.img"), content=b"world")
-    assert response.status_code == status.HTTP_204_NO_CONTENT
-    params = base_params
-    params["hda_disk_image"] = "hello.img"
-    response = await compute_client.post(app.url_path_for("compute:create_qemu_node", project_id=compute_project.id), json=params)
-    assert response.status_code == status.HTTP_201_CREATED
-    assert response.json()["hda_disk_image"] == "hello.img"
-    assert response.json()["hda_disk_image_md5sum"] == "7d793037a0760186574b0282f2f435e7"
+# async def test_qemu_create_with_project_file(app: FastAPI,
+#                                              compute_client: AsyncClient,
+#                                              compute_project: Project,
+#                                              base_params: dict,
+#                                              fake_qemu_vm: str) -> None:
+#
+#     response = await compute_client.post(
+#         app.url_path_for("compute:write_compute_project_file", project_id=compute_project.id, file_path="hello.img"),
+#         content=b"world"
+#     )
+#     assert response.status_code == status.HTTP_204_NO_CONTENT
+#     params = base_params
+#     params["hda_disk_image"] = "hello.img"
+#     response = await compute_client.post(
+#         app.url_path_for("compute:create_qemu_node", project_id=compute_project.id),
+#         json=params
+#     )
+#     assert response.status_code == status.HTTP_201_CREATED
+#     assert response.json()["hda_disk_image"] == "hello.img"
+#     assert response.json()["hda_disk_image_md5sum"] == "7d793037a0760186574b0282f2f435e7"
 
 
-async def test_qemu_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, vm: dict):
+async def test_qemu_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, qemu_vm: dict):
 
-    response = await compute_client.get(app.url_path_for("compute:get_qemu_node", project_id=vm["project_id"], node_id=vm["node_id"]))
+    response = await compute_client.get(
+        app.url_path_for("compute:get_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+    )
     assert response.status_code == status.HTTP_200_OK
     assert response.json()["name"] == "PC TEST 1"
     assert response.json()["project_id"] == compute_project.id
-    assert response.json()["node_directory"] == os.path.join(compute_project.path,
-                                                             "project-files",
-                                                             "qemu",
-                                                             vm["node_id"])
+    assert response.json()["node_directory"] == os.path.join(
+        compute_project.path,
+        "project-files",
+        "qemu",
+        qemu_vm["node_id"]
+    )
 
 
-async def test_qemu_start(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+async def test_qemu_start(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
     with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.start", return_value=True) as mock:
-        response = await compute_client.post(app.url_path_for("compute:start_qemu_node",
-                                                      project_id=vm["project_id"],
-                                                      node_id=vm["node_id"]))
+        response = await compute_client.post(
+            app.url_path_for("compute:start_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+        )
         assert mock.called
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
 
-async def test_qemu_stop(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+async def test_qemu_stop(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
     with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.stop", return_value=True) as mock:
-        response = await compute_client.post(app.url_path_for("compute:stop_qemu_node",
-                                                      project_id=vm["project_id"],
-                                                      node_id=vm["node_id"]))
+        response = await compute_client.post(
+            app.url_path_for("compute:stop_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+        )
         assert mock.called
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
 
-async def test_qemu_reload(app: FastAPI, compute_client: AsyncClient, vm) -> None:
+async def test_qemu_reload(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
     with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.reload", return_value=True) as mock:
-        response = await compute_client.post(app.url_path_for("compute:reload_qemu_node",
-                                                      project_id=vm["project_id"],
-                                                      node_id=vm["node_id"]))
+        response = await compute_client.post(
+            app.url_path_for("compute:reload_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+        )
         assert mock.called
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
 
-async def test_qemu_suspend(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+async def test_qemu_suspend(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
     with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.suspend", return_value=True) as mock:
-        response = await compute_client.post(app.url_path_for("compute:suspend_qemu_node",
-                                                      project_id=vm["project_id"],
-                                                      node_id=vm["node_id"]))
+        response = await compute_client.post(
+            app.url_path_for("compute:suspend_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+        )
         assert mock.called
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
 
-async def test_qemu_resume(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+async def test_qemu_resume(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
     with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.resume", return_value=True) as mock:
-        response = await compute_client.post(app.url_path_for("compute:resume_qemu_node",
-                                                      project_id=vm["project_id"],
-                                                      node_id=vm["node_id"]))
+        response = await compute_client.post(
+            app.url_path_for("compute:resume_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+        )
         assert mock.called
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
 
-async def test_qemu_delete(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+async def test_qemu_delete(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
     with asyncio_patch("gns3server.compute.qemu.Qemu.delete_node", return_value=True) as mock:
-        response = await compute_client.delete(app.url_path_for("compute:delete_qemu_node",
-                                                        project_id=vm["project_id"],
-                                                        node_id=vm["node_id"]))
+        response = await compute_client.delete(
+            app.url_path_for("compute:delete_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
+        )
         assert mock.called
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
 
 async def test_qemu_update(app: FastAPI,
                            compute_client: AsyncClient,
-                           vm: dict,
+                           qemu_vm: dict,
                            free_console_port: int,
                            fake_qemu_vm: str) -> None:
 
@@ -219,9 +240,10 @@ async def test_qemu_update(app: FastAPI,
         "hdb_disk_image": "linux载.img"
     }
 
-    response = await compute_client.put(app.url_path_for("compute:update_qemu_node",
-                                                 project_id=vm["project_id"],
-                                                 node_id=vm["node_id"]), json=params)
+    response = await compute_client.put(
+        app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"]),
+        json=params
+    )
     assert response.status_code == status.HTTP_200_OK
     assert response.json()["name"] == "test"
     assert response.json()["console"] == free_console_port
@@ -229,7 +251,7 @@ async def test_qemu_update(app: FastAPI,
     assert response.json()["ram"] == 1024
 
 
-async def test_qemu_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+async def test_qemu_nio_create_udp(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
     params = {
         "type": "nio_udp",
@@ -239,21 +261,25 @@ async def test_qemu_nio_create_udp(app: FastAPI, compute_client: AsyncClient, vm
     }
 
     with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.add_ubridge_udp_connection"):
-        await compute_client.put(app.url_path_for("compute:update_qemu_node",
-                                                 project_id=vm["project_id"],
-                                                 node_id=vm["node_id"]), json={"adapters": 2})
+        await compute_client.put(
+            app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"]),
+            json={"adapters": 2}
+        )
 
-        url = app.url_path_for("compute:create_qemu_node_nio",
-                               project_id=vm["project_id"],
-                               node_id=vm["node_id"],
-                               adapter_number="1",
-                               port_number="0")
+        url = app.url_path_for(
+            "compute:create_qemu_node_nio",
+            project_id=qemu_vm["project_id"],
+            node_id=qemu_vm["node_id"],
+            adapter_number="1",
+            port_number="0"
+        )
         response = await compute_client.post(url, json=params)
+
     assert response.status_code == status.HTTP_201_CREATED
     assert response.json()["type"] == "nio_udp"
 
 
-async def test_qemu_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+async def test_qemu_nio_update_udp(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
     params = {
         "type": "nio_udp",
@@ -262,31 +288,35 @@ async def test_qemu_nio_update_udp(app: FastAPI, compute_client: AsyncClient, vm
         "rhost": "127.0.0.1"
     }
 
-    await compute_client.put(app.url_path_for("compute:update_qemu_node",
-                                      project_id=vm["project_id"],
-                                      node_id=vm["node_id"]), json={"adapters": 2})
+    await compute_client.put(
+        app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"]),
+        json={"adapters": 2}
+    )
 
-    url = app.url_path_for("compute:create_qemu_node_nio",
-                           project_id=vm["project_id"],
-                           node_id=vm["node_id"],
-                           adapter_number="1",
-                           port_number="0")
+    url = app.url_path_for(
+        "compute:create_qemu_node_nio",
+        project_id=qemu_vm["project_id"],
+        node_id=qemu_vm["node_id"],
+        adapter_number="1",
+        port_number="0"
+    )
 
     await compute_client.post(url, json=params)
-
     params["filters"] = {}
 
-    url = app.url_path_for("compute:update_qemu_node_nio",
-                           project_id=vm["project_id"],
-                           node_id=vm["node_id"],
-                           adapter_number="1",
-                           port_number="0")
+    url = app.url_path_for(
+        "compute:update_qemu_node_nio",
+        project_id=qemu_vm["project_id"],
+        node_id=qemu_vm["node_id"],
+        adapter_number="1",
+        port_number="0"
+    )
     response = await compute_client.put(url, json=params)
     assert response.status_code == status.HTTP_201_CREATED
     assert response.json()["type"] == "nio_udp"
 
 
-async def test_qemu_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+async def test_qemu_delete_nio(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
 
     params = {
         "type": "nio_udp",
@@ -296,27 +326,32 @@ async def test_qemu_delete_nio(app: FastAPI, compute_client: AsyncClient, vm: di
     }
 
     with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM._ubridge_send"):
-        await compute_client.put(app.url_path_for("compute:update_qemu_node",
-                                                 project_id=vm["project_id"],
-                                                 node_id=vm["node_id"]), json={"adapters": 2})
+        await compute_client.put(
+            app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"]),
+            json={"adapters": 2}
+        )
 
-        url = app.url_path_for("compute:create_qemu_node_nio",
-                               project_id=vm["project_id"],
-                               node_id=vm["node_id"],
-                               adapter_number="1",
-                               port_number="0")
+        url = app.url_path_for(
+            "compute:create_qemu_node_nio",
+            project_id=qemu_vm["project_id"],
+            node_id=qemu_vm["node_id"],
+            adapter_number="1",
+            port_number="0"
+        )
         await compute_client.post(url, json=params)
 
-        url = app.url_path_for("compute:delete_qemu_node_nio",
-                               project_id=vm["project_id"],
-                               node_id=vm["node_id"],
-                               adapter_number="1",
-                               port_number="0")
+        url = app.url_path_for(
+            "compute:delete_qemu_node_nio",
+            project_id=qemu_vm["project_id"],
+            node_id=qemu_vm["node_id"],
+            adapter_number="1",
+            port_number="0"
+        )
         response = await compute_client.delete(url)
     assert response.status_code == status.HTTP_204_NO_CONTENT
 
 
-async def test_qemu_list_binaries(app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+async def test_qemu_list_binaries(app: FastAPI, compute_client: AsyncClient) -> None:
 
     ret = [{"path": "/tmp/1", "version": "2.2.0"},
            {"path": "/tmp/2", "version": "2.1.0"}]
@@ -480,33 +515,177 @@ async def test_capabilities(app: FastAPI, compute_client: AsyncClient) -> None:
 async def test_qemu_duplicate(app: FastAPI,
                               compute_client: AsyncClient,
                               compute_project: Project,
-                              vm: dict,
+                              qemu_vm: dict,
                               base_params: dict) -> None:
 
     # create destination node first
-    response = await compute_client.post(app.url_path_for("compute:create_qemu_node",
-                                                  project_id=vm["project_id"]), json=base_params)
+    response = await compute_client.post(
+        app.url_path_for("compute:create_qemu_node", project_id=qemu_vm["project_id"]),
+        json=base_params
+    )
 
     assert response.status_code == status.HTTP_201_CREATED
     params = {"destination_node_id": response.json()["node_id"]}
-    response = await compute_client.post(app.url_path_for("compute:duplicate_qemu_node",
-                                                  project_id=vm["project_id"], node_id=vm["node_id"]), json=params)
+    response = await compute_client.post(
+        app.url_path_for("compute:duplicate_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"]),
+        json=params
+    )
     assert response.status_code == status.HTTP_201_CREATED
 
 
+async def test_qemu_create_disk_image(
+        app: FastAPI,
+        compute_client: AsyncClient,
+        compute_project: Project,
+        fake_qemu_img_binary: str,
+        qemu_vm: dict,
+):
+
+    options = {
+        "format": "qcow2",
+        "preallocation": "metadata",
+        "cluster_size": 64,
+        "refcount_bits": 12,
+        "lazy_refcounts": "off",
+        "size": 30
+    }
+
+    with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as qemu_img:
+        response = await compute_client.post(
+            app.url_path_for(
+                "compute:create_qemu_disk_image",
+                project_id=qemu_vm["project_id"],
+                node_id=qemu_vm["node_id"],
+                disk_name="disk.qcow2"
+            ),
+            json=options
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        args, kwargs = qemu_img.call_args
+        assert args == (
+            fake_qemu_img_binary,
+            "create",
+            "-f",
+            "qcow2",
+            "-o",
+            "cluster_size=64",
+            "-o",
+            "lazy_refcounts=off",
+            "-o",
+            "preallocation=metadata",
+            "-o",
+            "refcount_bits=12",
+            os.path.join(qemu_vm["node_directory"], "disk.qcow2"),
+            "30M"
+        )
+
+
+async def test_qemu_create_disk_image_already_exists(
+        app: FastAPI,
+        compute_client: AsyncClient,
+        compute_project: Project,
+        fake_qemu_img_binary: str,
+        qemu_vm: dict,
+):
+
+    node = compute_project.get_node(qemu_vm["node_id"])
+    shutil.copy("tests/resources/empty8G.qcow2", os.path.join(node.working_dir, "disk.qcow2"))
+
+    options = {
+        "format": "qcow2",
+        "size": 100
+    }
+
+    response = await compute_client.post(
+        app.url_path_for(
+            "compute:create_qemu_disk_image",
+            project_id=qemu_vm["project_id"],
+            node_id=qemu_vm["node_id"],
+            disk_name="disk.qcow2"
+        ),
+        json=options
+    )
+    assert response.status_code == status.HTTP_409_CONFLICT
+
+
+# async def test_qemu_create_disk_image_with_not_supported_characters_by_filesystem(
+#         app: FastAPI,
+#         compute_client: AsyncClient,
+#         compute_project: Project,
+#         fake_qemu_img_binary: str,
+#         qemu_vm: dict,
+# ):
+#
+#     node = compute_project.get_node(qemu_vm["node_id"])
+#     shutil.copy("tests/resources/empty8G.qcow2", os.path.join(node.working_dir, "disk.qcow2"))
+#
+#     options = {
+#         "format": "qcow2",
+#         "size": 100
+#     }
+#
+#     with patch("os.path.exists", side_effect=UnicodeEncodeError('error', u"", 1, 2, 'Emulated Unicode Err')):
+#         response = await compute_client.post(
+#             app.url_path_for(
+#                 "compute:create_qemu_disk_image",
+#                 project_id=qemu_vm["project_id"],
+#                 node_id=qemu_vm["node_id"],
+#                 disk_name=u"\u2019"
+#             ),
+#             json=options
+#         )
+#         assert response.status_code == status.HTTP_409_CONFLICT
+
+
+async def test_qemu_update_disk_image(
+        app: FastAPI,
+        compute_client: AsyncClient,
+        compute_project: Project,
+        fake_qemu_img_binary: str,
+        qemu_vm: dict,
+) -> None:
+
+    node = compute_project.get_node(qemu_vm["node_id"])
+    shutil.copy("tests/resources/empty8G.qcow2", os.path.join(node.working_dir, "disk.qcow2"))
+
+    with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as qemu_img:
+        response = await compute_client.put(
+            app.url_path_for(
+                "compute:update_qemu_disk_image",
+                project_id=qemu_vm["project_id"],
+                node_id=qemu_vm["node_id"],
+                disk_name="disk.qcow2"
+            ),
+            json={"extend": 10}
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        assert qemu_img.called
+        args, kwargs = qemu_img.call_args
+        assert args == (
+            fake_qemu_img_binary,
+            "resize",
+            os.path.join(qemu_vm["node_directory"], "disk.qcow2"),
+            "+10M"
+        )
+
+
 @pytest.mark.asyncio
-async def test_qemu_start_capture(app: FastAPI, compute_client: AsyncClient, vm):
+async def test_qemu_start_capture(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict):
 
     params = {
         "capture_file_name": "test.pcap",
         "data_link_type": "DLT_EN10MB"
     }
 
-    url = app.url_path_for("compute:start_qemu_node_capture",
-                           project_id=vm["project_id"],
-                           node_id=vm["node_id"],
-                           adapter_number="0",
-                           port_number="0")
+    url = app.url_path_for(
+        "compute:start_qemu_node_capture",
+        project_id=qemu_vm["project_id"],
+        node_id=qemu_vm["node_id"],
+        adapter_number="0",
+        port_number="0"
+    )
 
     with patch("gns3server.compute.qemu.qemu_vm.QemuVM.is_running", return_value=True):
         with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.start_capture") as mock:
@@ -517,13 +696,15 @@ async def test_qemu_start_capture(app: FastAPI, compute_client: AsyncClient, vm)
 
 
 @pytest.mark.asyncio
-async def test_qemu_stop_capture(app: FastAPI, compute_client: AsyncClient, vm):
+async def test_qemu_stop_capture(app: FastAPI, compute_client: AsyncClient, qemu_vm: dict):
 
-    url = app.url_path_for("compute:stop_qemu_node_capture",
-                           project_id=vm["project_id"],
-                           node_id=vm["node_id"],
-                           adapter_number="0",
-                           port_number="0")
+    url = app.url_path_for(
+        "compute:stop_qemu_node_capture",
+        project_id=qemu_vm["project_id"],
+        node_id=qemu_vm["node_id"],
+        adapter_number="0",
+        port_number="0"
+    )
 
     with patch("gns3server.compute.qemu.qemu_vm.QemuVM.is_running", return_value=True):
         with asyncio_patch("gns3server.compute.qemu.qemu_vm.QemuVM.stop_capture") as mock:

tests/api/routes/controller/test_nodes.py

@@ -223,33 +223,129 @@ async def test_dynamips_idle_pc(
         client: AsyncClient,
         project: Project,
         compute: Compute,
-        node: Node) -> None:
+        node: Node
+) -> None:
 
     response = MagicMock()
     response.json = {"idlepc": "0x60606f54"}
     compute.get = AsyncioMagicMock(return_value=response)
 
+    node._node_type = "dynamips"  # force Dynamips node type
     response = await client.get(app.url_path_for("auto_idlepc", project_id=project.id, node_id=node.id))
     assert response.status_code == status.HTTP_200_OK
     assert response.json()["idlepc"] == "0x60606f54"
 
 
+async def test_dynamips_idle_pc_wrong_node_type(
+        app: FastAPI,
+        client: AsyncClient,
+        project: Project,
+        compute: Compute,
+        node: Node
+) -> None:
+
+    response = await client.get(app.url_path_for("auto_idlepc", project_id=project.id, node_id=node.id))
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
 async def test_dynamips_idlepc_proposals(
         app: FastAPI,
         client: AsyncClient,
         project: Project,
         compute: Compute,
-        node: Node) -> None:
+        node: Node
+) -> None:
 
     response = MagicMock()
     response.json = ["0x60606f54", "0x33805a22"]
     compute.get = AsyncioMagicMock(return_value=response)
 
+    node._node_type = "dynamips"  # force Dynamips node type
     response = await client.get(app.url_path_for("idlepc_proposals", project_id=project.id, node_id=node.id))
     assert response.status_code == status.HTTP_200_OK
     assert response.json() == ["0x60606f54", "0x33805a22"]
 
 
+async def test_dynamips_idlepc_proposals_wrong_node_type(
+        app: FastAPI,
+        client: AsyncClient,
+        project: Project,
+        compute: Compute,
+        node: Node
+) -> None:
+
+    response = await client.get(app.url_path_for("idlepc_proposals", project_id=project.id, node_id=node.id))
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
+async def test_qemu_disk_image_create(
+        app: FastAPI,
+        client: AsyncClient,
+        project: Project,
+        compute: Compute,
+        node: Node
+) -> None:
+
+    response = MagicMock()
+    compute.post = AsyncioMagicMock(return_value=response)
+
+    node._node_type = "qemu"  # force Qemu node type
+    response = await client.post(
+        app.url_path_for("create_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
+        json={"format": "qcow2", "size": 30}
+    )
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+
+async def test_qemu_disk_image_create_wrong_node_type(
+        app: FastAPI,
+        client: AsyncClient,
+        project: Project,
+        compute: Compute,
+        node: Node
+) -> None:
+
+    response = await client.post(
+        app.url_path_for("create_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
+        json={"format": "qcow2", "size": 30}
+    )
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
+async def test_qemu_disk_image_update(
+        app: FastAPI,
+        client: AsyncClient,
+        project: Project,
+        compute: Compute,
+        node: Node
+) -> None:
+
+    response = MagicMock()
+    compute.put = AsyncioMagicMock(return_value=response)
+
+    node._node_type = "qemu"  # force Qemu node type
+    response = await client.put(
+        app.url_path_for("update_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
+        json={"extend": 10}
+    )
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+
+async def test_qemu_disk_image_update_wrong_node_type(
+        app: FastAPI,
+        client: AsyncClient,
+        project: Project,
+        compute: Compute,
+        node: Node
+) -> None:
+
+    response = await client.put(
+        app.url_path_for("update_disk_image", project_id=project.id, node_id=node.id, disk_name="hda_disk.qcow2"),
+        json={"extend": 10}
+    )
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
 async def test_get_file(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node) -> None:
 
     response = MagicMock()

tests/compute/qemu/test_qemu_manager.py

@@ -17,7 +17,6 @@
 
 import os
 import stat
-import sys
 import pytest
 import platform
 
@@ -113,95 +112,6 @@ def test_get_legacy_vm_workdir():
     assert Qemu.get_legacy_vm_workdir(42, "bla") == os.path.join("qemu", "vm-42")
 
 
-@pytest.mark.asyncio
-async def test_create_image_abs_path(tmpdir, fake_qemu_img_binary):
-
-    options = {
-        "format": "qcow2",
-        "preallocation": "metadata",
-        "cluster_size": 64,
-        "refcount_bits": 12,
-        "lazy_refcounts": "off",
-        "size": 100
-    }
-    with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
-        await Qemu.instance().create_disk(fake_qemu_img_binary, str(tmpdir / "hda.qcow2"), options)
-        args, kwargs = process.call_args
-        assert args == (
-            fake_qemu_img_binary,
-            "create",
-            "-f",
-            "qcow2",
-            "-o",
-            "cluster_size=64",
-            "-o",
-            "lazy_refcounts=off",
-            "-o",
-            "preallocation=metadata",
-            "-o",
-            "refcount_bits=12",
-            str(tmpdir / "hda.qcow2"),
-            "100M"
-        )
-
-
-@pytest.mark.asyncio
-async def test_create_image_relative_path(tmpdir, fake_qemu_img_binary):
-
-    options = {
-        "format": "raw",
-        "size": 100
-    }
-    with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
-        with patch("gns3server.compute.qemu.Qemu.get_images_directory", return_value=str(tmpdir)):
-            await Qemu.instance().create_disk(fake_qemu_img_binary, "hda.qcow2", options)
-            args, kwargs = process.call_args
-            assert args == (
-                fake_qemu_img_binary,
-                "create",
-                "-f",
-                "raw",
-                str(tmpdir / "hda.qcow2"),
-                "100M"
-            )
-
-
-@pytest.mark.asyncio
-async def test_create_image_exist(tmpdir, fake_qemu_img_binary):
-
-    open(str(tmpdir / "hda.qcow2"), "w+").close()
-    options = {
-        "format": "raw",
-        "size": 100
-    }
-    with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
-        with patch("gns3server.compute.qemu.Qemu.get_images_directory", return_value=str(tmpdir)):
-            with pytest.raises(QemuError):
-                await Qemu.instance().create_disk(fake_qemu_img_binary, "hda.qcow2", options)
-                assert not process.called
-
-
-@pytest.mark.asyncio
-async def test_create_image_with_not_supported_characters_by_filesystem(tmpdir, fake_qemu_img_binary):
-
-    open(str(tmpdir / "hda.qcow2"), "w+").close()
-
-    options = {
-        "format": "raw",
-        "size": 100
-    }
-
-    # patching os.makedirs is necessary as it depends on already mocked os.path.exists
-    with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process, \
-            patch("gns3server.compute.qemu.Qemu.get_images_directory", return_value=str(tmpdir)), \
-            patch("os.path.exists", side_effect=UnicodeEncodeError('error', u"", 1, 2, 'Emulated Unicode Err')),\
-            patch("os.makedirs"):
-
-        with pytest.raises(QemuError):
-            await Qemu.instance().create_disk(fake_qemu_img_binary, "hda.qcow2", options)
-            assert not process.called
-
-
 @pytest.mark.asyncio
 async def test_get_kvm_archs_kvm_ok():
 

tests/compute/qemu/test_qemu_vm.py

@@ -18,7 +18,6 @@
 import pytest
 import asyncio
 import os
-import sys
 import stat
 from tests.utils import asyncio_patch, AsyncioMagicMock
 

tests/controller/test_compute.py

@@ -20,8 +20,8 @@ import pytest
 from unittest.mock import patch, MagicMock
 
 from gns3server.controller.project import Project
-from gns3server.controller.compute import Compute, ComputeConflict
-from gns3server.controller.controller_error import ControllerError, ControllerNotFoundError
+from gns3server.controller.compute import Compute
+from gns3server.controller.controller_error import ControllerError, ControllerNotFoundError, ComputeConflictError
 from pydantic import SecretStr
 from tests.utils import asyncio_patch, AsyncioMagicMock
 
@@ -212,7 +212,7 @@ async def test_compute_httpQueryConflictError(compute):
     with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
         response.status = 409
         response.read = AsyncioMagicMock(return_value=b'{"message": "Test"}')
-        with pytest.raises(ComputeConflict):
+        with pytest.raises(ComputeConflictError):
             await compute.post("/projects", {"a": "b"})
         assert mock.called
         await compute.close()