arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-11-15 12:59:06 +00:00
Code Issues Releases Wiki Activity

Raise error if qemu image already exist when creating disk

Browse Source
This commit is contained in:
Julien Duponchelle 2015-07-28 16:15:01 +02:00
parent 54448ab936
commit e92d9ca5ff
2 changed files with 18 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
gns3server/modules/qemu/__init__.py
View File

@ -205,6 +205,8 @@ class Qemu(BaseManager):
directory = self.get_images_directory() directory = self.get_images_directory()
os.makedirs(directory, exist_ok=True) os.makedirs(directory, exist_ok=True)
path = os.path.join(directory, os.path.basename(path)) path = os.path.join(directory, os.path.basename(path))
if os.path.exists(path):
raise QemuError("Could not create disk image {} already exist".format(path))
command = [qemu_img, "create", "-f", img_format] command = [qemu_img, "create", "-f", img_format]
for option in sorted(options.keys()): for option in sorted(options.keys()):
@ -215,4 +217,4 @@ class Qemu(BaseManager):
process = yield from asyncio.create_subprocess_exec(*command) process = yield from asyncio.create_subprocess_exec(*command)
yield from process.wait() yield from process.wait()
except (OSError, subprocess.SubprocessError) as e: except (OSError, subprocess.SubprocessError) as e:
raise QemuError("Could create disk image {}:{}".format(path, e)) raise QemuError("Could not create disk image {}:{}".format(path, e))

15
tests/modules/qemu/test_qemu_manager.py
View File

@ -22,6 +22,7 @@ import sys
import pytest import pytest
from gns3server.modules.qemu import Qemu from gns3server.modules.qemu import Qemu
from gns3server.modules.qemu.qemu_error import QemuError
from tests.utils import asyncio_patch from tests.utils import asyncio_patch
from unittest.mock import patch, MagicMock from unittest.mock import patch, MagicMock
@ -145,3 +146,17 @@ def test_create_image_relative_path(loop, tmpdir, fake_qemu_img_binary):
str(tmpdir / "hda.qcow2"), str(tmpdir / "hda.qcow2"),
"100M" "100M"
) )
def test_create_image_exist(loop, 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.modules.qemu.Qemu.get_images_directory", return_value=str(tmpdir)):
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