Merge pull request #478 from GNS3/qcow2_rebase

Qcow2 rebase

tests/modules/qemu/test_qcow2.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2016 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import pytest
+import shutil
+import asyncio
+
+from gns3server.modules.qemu.qcow2 import Qcow2, Qcow2Error
+
+
+def qemu_img():
+    """
+    Return the path of qemu-img on system.
+    We can't use shutil.which because for safety reason we break
+    the PATH to avoid test interacting with real binaries
+    """
+    paths = [
+        "/usr/bin/qemu-img",
+        "/usr/local/bin/qemu-img"
+    ]
+    for path in paths:
+        if os.path.exists(path):
+            return path
+    return None
+
+
+def test_valid_base_file():
+    qcow2 = Qcow2("tests/resources/empty8G.qcow2")
+    assert qcow2.version == 3
+    assert qcow2.backing_file is None
+
+
+def test_valid_linked_file():
+    qcow2 = Qcow2("tests/resources/linked.qcow2")
+    assert qcow2.version == 3
+    assert qcow2.backing_file == "empty8G.qcow2"
+
+
+def test_invalid_file():
+    with pytest.raises(Qcow2Error):
+        Qcow2("tests/resources/nvram_iou")
+
+
+@pytest.mark.skipif(qemu_img() is None, reason="qemu-img is not available")
+def test_rebase(tmpdir, loop):
+    shutil.copy("tests/resources/empty8G.qcow2", str(tmpdir / "empty16G.qcow2"))
+    shutil.copy("tests/resources/linked.qcow2", str(tmpdir / "linked.qcow2"))
+    qcow2 = Qcow2(str(tmpdir / "linked.qcow2"))
+    assert qcow2.version == 3
+    assert qcow2.backing_file == "empty8G.qcow2"
+    loop.run_until_complete(asyncio.async(qcow2.rebase(qemu_img(), str(tmpdir / "empty16G.qcow2"))))
+    assert qcow2.backing_file == str(tmpdir / "empty16G.qcow2")

tests/modules/qemu/test_qemu_vm.py

@@ -321,11 +321,35 @@ def test_disk_options(vm, tmpdir, loop, fake_qemu_img_binary):
     open(vm._hda_disk_image, "w+").close()
 
     with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
-        loop.run_until_complete(asyncio.async(vm._disk_options()))
+        options = loop.run_until_complete(asyncio.async(vm._disk_options()))
         assert process.called
         args, kwargs = process.call_args
         assert args == (fake_qemu_img_binary, "create", "-o", "backing_file={}".format(vm._hda_disk_image), "-f", "qcow2", os.path.join(vm.working_dir, "hda_disk.qcow2"))
 
+    assert options == ['-drive', 'file=' + os.path.join(vm.working_dir, "hda_disk.qcow2") + ',if=ide,index=0,media=disk']
+
+
+def test_disk_options_multiple_disk(vm, tmpdir, loop, fake_qemu_img_binary):
+
+    vm._hda_disk_image = str(tmpdir / "test0.qcow2")
+    vm._hdb_disk_image = str(tmpdir / "test1.qcow2")
+    vm._hdc_disk_image = str(tmpdir / "test2.qcow2")
+    vm._hdd_disk_image = str(tmpdir / "test3.qcow2")
+    open(vm._hda_disk_image, "w+").close()
+    open(vm._hdb_disk_image, "w+").close()
+    open(vm._hdc_disk_image, "w+").close()
+    open(vm._hdd_disk_image, "w+").close()
+
+    with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
+        options = loop.run_until_complete(asyncio.async(vm._disk_options()))
+
+    assert options == [
+        '-drive', 'file=' + os.path.join(vm.working_dir, "hda_disk.qcow2") + ',if=ide,index=0,media=disk',
+        '-drive', 'file=' + os.path.join(vm.working_dir, "hdb_disk.qcow2") + ',if=ide,index=1,media=disk',
+        '-drive', 'file=' + os.path.join(vm.working_dir, "hdc_disk.qcow2") + ',if=ide,index=2,media=disk',
+        '-drive', 'file=' + os.path.join(vm.working_dir, "hdd_disk.qcow2") + ',if=ide,index=3,media=disk'
+    ]
+
 
 @pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
 def test_set_process_priority(vm, loop, fake_qemu_img_binary):