Improve PR #386

* Add more test
* Check to kvmi-ok is asynchronous

gns3server/handlers/api/qemu_handler.py

@@ -306,7 +306,7 @@ class QEMUHandler:
         output=QEMU_BINARY_LIST_SCHEMA)
     def list_binaries(request, response):
 
-        binaries = yield from Qemu.binary_list(request.json["archs"] if "archs" in request.json else None)
+        binaries = yield from Qemu.binary_list(request.json.get("archs", None))
         response.json(binaries)
 
     @classmethod
@@ -333,8 +333,8 @@ class QEMUHandler:
         output=QEMU_CAPABILITY_LIST_SCHEMA
     )
     def get_capabilities(request, response):
-        capabilities = {}
-        kvms = Qemu.get_kvm_archs()
+        capabilities = {"kvm": []}
+        kvms = yield from Qemu.get_kvm_archs()
         if kvms:
             capabilities["kvm"] = kvms
         response.json(capabilities)

gns3server/modules/qemu/__init__.py

@@ -40,6 +40,7 @@ class Qemu(BaseManager):
     _VM_CLASS = QemuVM
 
     @staticmethod
+    @asyncio.coroutine
     def get_kvm_archs():
         """
         Gets a list of architectures for which KVM is available on this server.
@@ -47,14 +48,19 @@ class Qemu(BaseManager):
         :returns: List of architectures for which KVM is available on this server.
         """
         kvm = []
-        x86_64_aliases = ["x86_64", "x86-64", "x64", "AMD64", "amd64", "Intel 64", "EM64T"]
-        i386_aliases = ["i386", "x86", "x32"]
-        if sys.platform.startswith("linux") and subprocess.call("kvm-ok") == 0:
+
+        try:
+            process = yield from asyncio.create_subprocess_exec("kvm-ok")
+            yield from process.wait()
+        except OSError:
+            return kvm
+
+        if process.returncode == 0:
             arch = platform.machine()
-            if arch in x86_64_aliases:
+            if arch == "x86_64":
                 kvm.append("x86_64")
                 kvm.append("i386")
-            elif arch in i386_aliases:
+            elif arch == "i386":
                 kvm.append("i386")
             else:
                 kvm.append(platform.machine())

tests/handlers/api/test_qemu.py

@@ -200,11 +200,24 @@ def test_qemu_list_binaries(server, vm):
            {"path": "/tmp/2", "version": "2.1.0"}]
     with asyncio_patch("gns3server.modules.qemu.Qemu.binary_list", return_value=ret) as mock:
         response = server.get("/qemu/binaries".format(project_id=vm["project_id"]), example=True)
-        assert mock.called
+        assert mock.called_with(None)
         assert response.status == 200
         assert response.json == ret
 
 
+def test_qemu_list_binaries_filter(server, vm):
+    ret = [
+        {"path": "/tmp/x86_64", "version": "2.2.0"},
+        {"path": "/tmp/alpha", "version": "2.1.0"},
+        {"path": "/tmp/i386", "version": "2.1.0"}
+    ]
+    with asyncio_patch("gns3server.modules.qemu.Qemu.binary_list", return_value=ret) as mock:
+        response = server.get("/qemu/binaries".format(project_id=vm["project_id"]), body={"archs": ["i386"]}, example=True)
+        assert response.status == 200
+        assert mock.called_with(["i386"])
+        assert response.json == ret
+
+
 def test_vms(server, tmpdir, fake_qemu_vm):
 
     with patch("gns3server.modules.Qemu.get_images_directory", return_value=str(tmpdir), example=True):
@@ -312,3 +325,9 @@ def test_create_img_absolute_local(server):
         response = server.post("/qemu/img", body=body, example=True)
 
     assert response.status == 201
+
+
+def test_capabilities(server):
+    with asyncio_patch("gns3server.modules.Qemu.get_kvm_archs", return_value=["x86_64"]):
+        response = server.get("/qemu/capabilities", example=True)
+        assert response.json["kvm"] == ["x86_64"]

tests/modules/qemu/test_qemu_manager.py

@@ -20,6 +20,7 @@ import stat
 import asyncio
 import sys
 import pytest
+import platform
 
 from gns3server.modules.qemu import Qemu
 from gns3server.modules.qemu.qemu_error import QemuError
@@ -174,3 +175,21 @@ def test_create_image_exist(loop, tmpdir, fake_qemu_img_binary):
             with pytest.raises(QemuError):
                 loop.run_until_complete(asyncio.async(Qemu.instance().create_disk(fake_qemu_img_binary, "hda.qcow2", options)))
                 assert not process.called
+
+
+def test_get_kvm_archs_no_kvm(loop):
+    with asyncio_patch("asyncio.create_subprocess_exec", side_effect=FileNotFoundError('kvm-ok')):
+        archs = loop.run_until_complete(asyncio.async(Qemu.get_kvm_archs()))
+        assert archs == []
+
+
+def test_get_kvm_archs_kvm_ok(loop):
+
+    process = MagicMock()
+    with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
+        process.returncode = 0
+        archs = loop.run_until_complete(asyncio.async(Qemu.get_kvm_archs()))
+        if platform.machine() == 'x86_64':
+            assert archs == ['x86_64', 'i386']
+        else:
+            assert archs == platform.machine()

tests/utils.py

@@ -50,6 +50,11 @@ class _asyncio_patch:
         future = asyncio.Future()
         if "return_value" in self.kwargs:
             future.set_result(self.kwargs["return_value"])
+        elif "side_effect" in self.kwargs:
+            if isinstance(self.kwargs["side_effect"], Exception):
+                future.set_exception(self.kwargs["side_effect"])
+            else:
+                raise NotImplementedError
         else:
             future.set_result(True)
         return future