mirror of
https://github.com/GNS3/gns3-server.git
synced 2024-12-19 12:57:56 +00:00
Fix some tests on Windows
This commit is contained in:
parent
bf3444933e
commit
77f54848e3
@ -102,6 +102,7 @@ class Project:
|
||||
|
||||
server_config = Config.instance().get_section_config("Server")
|
||||
path = os.path.expanduser(server_config.get("projects_path", "~/GNS3/projects"))
|
||||
path = os.path.normpath(path)
|
||||
try:
|
||||
os.makedirs(path, exist_ok=True)
|
||||
except OSError as e:
|
||||
|
@ -182,12 +182,12 @@ class VPCSVM(BaseVM):
|
||||
|
||||
try:
|
||||
script_file = os.path.join(self.working_dir, 'startup.vpc')
|
||||
with open(script_file, "w+", encoding="utf-8") as f:
|
||||
with open(script_file, "wb+") as f:
|
||||
if startup_script is None:
|
||||
f.write('')
|
||||
f.write(b'')
|
||||
else:
|
||||
startup_script = startup_script.replace("%h", self._name)
|
||||
f.write(startup_script)
|
||||
f.write(startup_script.encode("utf-8"))
|
||||
except OSError as e:
|
||||
raise VPCSError('Cannot write the startup script file "{}": {}'.format(self.script_file, e))
|
||||
|
||||
|
@ -18,6 +18,7 @@
|
||||
|
||||
import pytest
|
||||
import tempfile
|
||||
import sys
|
||||
|
||||
from gns3server.modules.dynamips import Dynamips
|
||||
from gns3server.modules.dynamips.dynamips_error import DynamipsError
|
||||
@ -36,7 +37,7 @@ def test_vm_invalid_dynamips_path(manager):
|
||||
with pytest.raises(DynamipsError):
|
||||
manager.find_dynamips()
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported by Windows")
|
||||
def test_vm_non_executable_dynamips_path(manager):
|
||||
tmpfile = tempfile.NamedTemporaryFile()
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"dynamips_path": tmpfile.name}):
|
||||
|
@ -26,8 +26,8 @@ pytestmark = pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supp
|
||||
|
||||
if not sys.platform.startswith("win"):
|
||||
from gns3server.modules.iou import IOU
|
||||
from gns3server.modules.iou.iou_error import IOUError
|
||||
|
||||
from gns3server.modules.iou.iou_error import IOUError
|
||||
from gns3server.modules.project_manager import ProjectManager
|
||||
|
||||
|
||||
|
@ -18,6 +18,7 @@
|
||||
import os
|
||||
import stat
|
||||
import asyncio
|
||||
import sys
|
||||
|
||||
from gns3server.modules.qemu import Qemu
|
||||
from tests.utils import asyncio_patch
|
||||
@ -27,7 +28,10 @@ def test_get_qemu_version(loop):
|
||||
|
||||
with asyncio_patch("gns3server.modules.qemu.subprocess_check_output", return_value="QEMU emulator version 2.2.0, Copyright (c) 2003-2008 Fabrice Bellard") as mock:
|
||||
version = loop.run_until_complete(asyncio.async(Qemu._get_qemu_version("/tmp/qemu-test")))
|
||||
assert version == "2.2.0"
|
||||
if sys.platform.startswith("win"):
|
||||
assert version == ""
|
||||
else:
|
||||
assert version == "2.2.0"
|
||||
|
||||
|
||||
def test_binary_list(loop):
|
||||
@ -43,12 +47,17 @@ def test_binary_list(loop):
|
||||
with asyncio_patch("gns3server.modules.qemu.subprocess_check_output", return_value="QEMU emulator version 2.2.0, Copyright (c) 2003-2008 Fabrice Bellard") as mock:
|
||||
qemus = loop.run_until_complete(asyncio.async(Qemu.binary_list()))
|
||||
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-system-x86"), "version": "2.2.0"} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-kvm"), "version": "2.2.0"} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-system-x42"), "version": "2.2.0"} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "hello"), "version": "2.2.0"} not in qemus
|
||||
if sys.platform.startswith("win"):
|
||||
version = ""
|
||||
else:
|
||||
version = "2.2.0"
|
||||
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-system-x86"), "version": version} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-kvm"), "version": version} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-system-x42"), "version": version} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "hello"), "version": version} not in qemus
|
||||
|
||||
|
||||
def test_get_legacy_vm_workdir():
|
||||
|
||||
assert Qemu.get_legacy_vm_workdir(42, "bla") == "qemu/vm-42"
|
||||
assert Qemu.get_legacy_vm_workdir(42, "bla") == os.path.join("qemu", "vm-42")
|
||||
|
@ -19,6 +19,7 @@ import pytest
|
||||
import aiohttp
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
import stat
|
||||
import re
|
||||
from tests.utils import asyncio_patch
|
||||
@ -51,7 +52,10 @@ def fake_qemu_img_binary():
|
||||
@pytest.fixture
|
||||
def fake_qemu_binary():
|
||||
|
||||
bin_path = os.path.join(os.environ["PATH"], "qemu_x42")
|
||||
if sys.platform.startswith("win"):
|
||||
bin_path = os.path.join(os.environ["PATH"], "qemu_x42.EXE")
|
||||
else:
|
||||
bin_path = os.path.join(os.environ["PATH"], "qemu_x42")
|
||||
with open(bin_path, "w+") as f:
|
||||
f.write("1")
|
||||
os.chmod(bin_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
|
||||
@ -178,8 +182,9 @@ def test_set_qemu_path(vm, tmpdir, fake_qemu_binary):
|
||||
f.write("1")
|
||||
|
||||
# Raise because file is not executable
|
||||
with pytest.raises(QemuError):
|
||||
vm.qemu_path = path
|
||||
if not sys.platform.startswith("win"):
|
||||
with pytest.raises(QemuError):
|
||||
vm.qemu_path = path
|
||||
|
||||
os.chmod(path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
|
||||
|
||||
@ -204,6 +209,7 @@ def test_disk_options(vm, loop, fake_qemu_img_binary):
|
||||
assert args == (fake_qemu_img_binary, "create", "-f", "qcow2", os.path.join(vm.working_dir, "flash.qcow2"), "256M")
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
def test_set_process_priority(vm, loop, fake_qemu_img_binary):
|
||||
|
||||
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
|
||||
@ -273,7 +279,7 @@ def test_build_command(vm, loop, fake_qemu_binary, port_manager):
|
||||
"user,id=gns3-0"
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
def test_build_command_without_display(vm, loop, fake_qemu_binary):
|
||||
|
||||
os.environ["DISPLAY"] = ""
|
||||
@ -282,7 +288,9 @@ def test_build_command_without_display(vm, loop, fake_qemu_binary):
|
||||
assert "-nographic" in cmd
|
||||
|
||||
|
||||
def test_build_command_witht_invalid_options(vm, loop, fake_qemu_binary):
|
||||
# Windows accept this kind of mistake
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
def test_build_command_with_invalid_options(vm, loop, fake_qemu_binary):
|
||||
|
||||
vm.options = "'test"
|
||||
with pytest.raises(QemuError):
|
||||
|
@ -22,15 +22,15 @@ from unittest.mock import patch
|
||||
|
||||
|
||||
from gns3server.modules.vpcs import VPCS
|
||||
from gns3server.modules.iou import IOU
|
||||
from gns3server.modules.qemu import Qemu
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def iou(port_manager):
|
||||
IOU._instance = None
|
||||
iou = IOU.instance()
|
||||
iou.port_manager = port_manager
|
||||
return iou
|
||||
def qemu(port_manager):
|
||||
Qemu._instance = None
|
||||
qemu = Qemu.instance()
|
||||
qemu.port_manager = port_manager
|
||||
return qemu
|
||||
|
||||
|
||||
def test_create_vm_new_topology(loop, project, port_manager):
|
||||
@ -93,31 +93,31 @@ def test_create_vm_old_topology(loop, project, tmpdir, port_manager):
|
||||
assert f.read() == "1"
|
||||
|
||||
|
||||
def test_get_abs_image_path(iou, tmpdir):
|
||||
os.makedirs(str(tmpdir / "IOU"))
|
||||
def test_get_abs_image_path(qemu, tmpdir):
|
||||
os.makedirs(str(tmpdir / "QEMU"))
|
||||
path1 = str(tmpdir / "test1.bin")
|
||||
open(path1, 'w+').close()
|
||||
|
||||
path2 = str(tmpdir / "IOU" / "test2.bin")
|
||||
path2 = str(tmpdir / "QEMU" / "test2.bin")
|
||||
open(path2, 'w+').close()
|
||||
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"images_path": str(tmpdir)}):
|
||||
assert iou.get_abs_image_path(path1) == path1
|
||||
assert iou.get_abs_image_path(path2) == path2
|
||||
assert iou.get_abs_image_path("test2.bin") == path2
|
||||
assert iou.get_abs_image_path("../test1.bin") == path1
|
||||
assert qemu.get_abs_image_path(path1) == path1
|
||||
assert qemu.get_abs_image_path(path2) == path2
|
||||
assert qemu.get_abs_image_path("test2.bin") == path2
|
||||
assert qemu.get_abs_image_path("../test1.bin") == path1
|
||||
|
||||
|
||||
def test_get_relative_image_path(iou, tmpdir):
|
||||
os.makedirs(str(tmpdir / "IOU"))
|
||||
def test_get_relative_image_path(qemu, tmpdir):
|
||||
os.makedirs(str(tmpdir / "Qemu"))
|
||||
path1 = str(tmpdir / "test1.bin")
|
||||
open(path1, 'w+').close()
|
||||
|
||||
path2 = str(tmpdir / "IOU" / "test2.bin")
|
||||
path2 = str(tmpdir / "QEMU" / "test2.bin")
|
||||
open(path2, 'w+').close()
|
||||
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"images_path": str(tmpdir)}):
|
||||
assert iou.get_relative_image_path(path1) == path1
|
||||
assert iou.get_relative_image_path(path2) == "test2.bin"
|
||||
assert iou.get_relative_image_path("test2.bin") == "test2.bin"
|
||||
assert iou.get_relative_image_path("../test1.bin") == path1
|
||||
assert qemu.get_relative_image_path(path1) == path1
|
||||
assert qemu.get_relative_image_path(path2) == "test2.bin"
|
||||
assert qemu.get_relative_image_path("test2.bin") == "test2.bin"
|
||||
assert qemu.get_relative_image_path("../test1.bin") == path1
|
||||
|
@ -19,6 +19,7 @@ import pytest
|
||||
import aiohttp
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
from tests.utils import asyncio_patch
|
||||
|
||||
|
||||
@ -100,7 +101,11 @@ def test_stop(loop, vm):
|
||||
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
|
||||
loop.run_until_complete(asyncio.async(vm.stop()))
|
||||
assert vm.is_running() is False
|
||||
process.terminate.assert_called_with()
|
||||
|
||||
if sys.platform.startswith("win"):
|
||||
process.send_signal.assert_called_with(1)
|
||||
else:
|
||||
process.terminate.assert_called_with()
|
||||
|
||||
|
||||
def test_reload(loop, vm):
|
||||
@ -122,7 +127,11 @@ def test_reload(loop, vm):
|
||||
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
|
||||
loop.run_until_complete(asyncio.async(vm.reload()))
|
||||
assert vm.is_running() is True
|
||||
process.terminate.assert_called_with()
|
||||
|
||||
if sys.platform.startswith("win"):
|
||||
process.send_signal.assert_called_with(1)
|
||||
else:
|
||||
process.terminate.assert_called_with()
|
||||
|
||||
|
||||
def test_add_nio_binding_udp(vm):
|
||||
|
Loading…
Reference in New Issue
Block a user