ExternalVendorCode/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2025-06-20 00:03:56 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Replace asyncio.async() by asyncio.ensure_future() in tests. Ref #2566.

Browse Source
This commit is contained in:
grossmj
2018-08-24 18:52:15 +07:00
parent 089d25c79d
commit a3d1e865a8
19 changed files with 171 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
tests/compute/qemu/test_qemu_vm.py
View File

@ -118,7 +118,7 @@ def test_is_running(vm, running_subprocess_mock):
def test_start(loop, vm, running_subprocess_mock):
with asyncio_patch("gns3server.compute.qemu.QemuVM.start_wrap_console"):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=running_subprocess_mock) as mock:
loop.run_until_complete(asyncio.async(vm.start()))
loop.run_until_complete(asyncio.ensure_future(vm.start()))
assert vm.is_running()
assert vm.command_line == ' '.join(mock.call_args[0])
@ -135,9 +135,9 @@ def test_stop(loop, vm, running_subprocess_mock):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = Qemu.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.adapter_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
loop.run_until_complete(asyncio.ensure_future(vm.start()))
assert vm.is_running()
loop.run_until_complete(asyncio.async(vm.stop()))
loop.run_until_complete(asyncio.ensure_future(vm.stop()))
assert vm.is_running() is False
process.terminate.assert_called_with()
@ -184,7 +184,7 @@ def test_termination_callback_error(vm, tmpdir, async_run):
def test_reload(loop, vm):
with asyncio_patch("gns3server.compute.qemu.QemuVM._control_vm") as mock:
loop.run_until_complete(asyncio.async(vm.reload()))
loop.run_until_complete(asyncio.ensure_future(vm.reload()))
assert mock.called_with("system_reset")
@ -193,33 +193,33 @@ def test_suspend(loop, vm):
control_vm_result = MagicMock()
control_vm_result.match.group.decode.return_value = "running"
with asyncio_patch("gns3server.compute.qemu.QemuVM._control_vm", return_value=control_vm_result) as mock:
loop.run_until_complete(asyncio.async(vm.suspend()))
loop.run_until_complete(asyncio.ensure_future(vm.suspend()))
assert mock.called_with("system_reset")
def test_add_nio_binding_udp(vm, loop):
nio = Qemu.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1", "filters": {}})
assert nio.lport == 4242
loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, nio)))
loop.run_until_complete(asyncio.ensure_future(vm.adapter_add_nio_binding(0, nio)))
assert nio.lport == 4242
def test_port_remove_nio_binding(vm, loop):
nio = Qemu.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1", "filters": {}})
loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, nio)))
loop.run_until_complete(asyncio.async(vm.adapter_remove_nio_binding(0)))
loop.run_until_complete(asyncio.ensure_future(vm.adapter_add_nio_binding(0, nio)))
loop.run_until_complete(asyncio.ensure_future(vm.adapter_remove_nio_binding(0)))
assert vm._ethernet_adapters[0].ports[0] is None
def test_close(vm, port_manager, loop):
with asyncio_patch("gns3server.compute.qemu.QemuVM.start_wrap_console"):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()):
loop.run_until_complete(asyncio.async(vm.start()))
loop.run_until_complete(asyncio.ensure_future(vm.start()))
console_port = vm.console
loop.run_until_complete(asyncio.async(vm.close()))
loop.run_until_complete(asyncio.ensure_future(vm.close()))
# Raise an exception if the port is not free
port_manager.reserve_tcp_port(console_port, vm.project)
@ -328,7 +328,7 @@ 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:
options = loop.run_until_complete(asyncio.async(vm._disk_options()))
options = loop.run_until_complete(asyncio.ensure_future(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"))
@ -341,7 +341,7 @@ def test_cdrom_option(vm, tmpdir, loop, fake_qemu_img_binary):
vm._cdrom_image = str(tmpdir / "test.iso")
open(vm._cdrom_image, "w+").close()
options = loop.run_until_complete(asyncio.async(vm._build_command()))
options = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
assert ' '.join(['-cdrom', str(tmpdir / "test.iso")]) in ' '.join(options)
@ -351,7 +351,7 @@ def test_bios_option(vm, tmpdir, loop, fake_qemu_img_binary):
vm._bios_image = str(tmpdir / "test.img")
open(vm._bios_image, "w+").close()
options = loop.run_until_complete(asyncio.async(vm._build_command()))
options = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
assert ' '.join(['-bios', str(tmpdir / "test.img")]) in ' '.join(options)
@ -359,14 +359,14 @@ def test_bios_option(vm, tmpdir, loop, fake_qemu_img_binary):
def test_vnc_option(vm, tmpdir, loop, fake_qemu_img_binary):
vm._console_type = 'vnc'
vm._console = 5905
options = loop.run_until_complete(asyncio.async(vm._build_command()))
options = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
assert '-vnc 127.0.0.1:5' in ' '.join(options)
def test_spice_option(vm, tmpdir, loop, fake_qemu_img_binary):
vm._console_type = 'spice'
vm._console = 5905
options = loop.run_until_complete(asyncio.async(vm._build_command()))
options = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
assert '-spice addr=127.0.0.1,port=5905,disable-ticketing' in ' '.join(options)
assert '-vga qxl' in ' '.join(options)
@ -383,7 +383,7 @@ def test_disk_options_multiple_disk(vm, tmpdir, loop, fake_qemu_img_binary):
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()))
options = loop.run_until_complete(asyncio.ensure_future(vm._disk_options()))
assert options == [
'-drive', 'file=' + os.path.join(vm.working_dir, "hda_disk.qcow2") + ',if=ide,index=0,media=disk',
@ -400,7 +400,7 @@ def test_set_process_priority(vm, loop, fake_qemu_img_binary):
vm._process = MagicMock()
vm._process.pid = 42
vm._process_priority = "low"
loop.run_until_complete(asyncio.async(vm._set_process_priority()))
loop.run_until_complete(asyncio.ensure_future(vm._set_process_priority()))
assert process.called
args, kwargs = process.call_args
assert args == ("renice", "-n", "5", "-p", "42")
@ -412,7 +412,7 @@ def test_set_process_priority_normal(vm, loop, fake_qemu_img_binary):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
vm._process = MagicMock()
vm._process.pid = 42
loop.run_until_complete(asyncio.async(vm._set_process_priority()))
loop.run_until_complete(asyncio.ensure_future(vm._set_process_priority()))
assert not process.called
@ -429,7 +429,7 @@ def test_control_vm(vm, loop):
reader = MagicMock()
writer = MagicMock()
with asyncio_patch("asyncio.open_connection", return_value=(reader, writer)) as open_connect:
res = loop.run_until_complete(asyncio.async(vm._control_vm("test")))
res = loop.run_until_complete(asyncio.ensure_future(vm._control_vm("test")))
assert writer.write.called_with("test")
assert res is None
@ -446,7 +446,7 @@ def test_control_vm_expect_text(vm, loop, running_subprocess_mock):
reader.readline.return_value = future
vm._monitor = 4242
res = loop.run_until_complete(asyncio.async(vm._control_vm("test", [b"epic"])))
res = loop.run_until_complete(asyncio.ensure_future(vm._control_vm("test", [b"epic"])))
assert writer.write.called_with("test")
assert res == "epic product"
@ -456,7 +456,7 @@ def test_build_command(vm, loop, fake_qemu_binary, port_manager):
os.environ["DISPLAY"] = "0:0"
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
nio = vm._local_udp_tunnels[0][0]
assert cmd == [
fake_qemu_binary,
@ -489,7 +489,7 @@ def test_build_command_manual_uuid(vm, loop, fake_qemu_binary, port_manager):
vm.options = "-uuid e1c307a4-896f-11e6-81a5-3c07547807cc"
os.environ["DISPLAY"] = "0:0"
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
assert "e1c307a4-896f-11e6-81a5-3c07547807cc" in cmd
assert vm.id not in cmd
@ -502,7 +502,7 @@ def test_build_command_kvm(linux_platform, vm, loop, fake_qemu_binary, port_mana
vm.manager.get_qemu_version = AsyncioMagicMock(return_value="2.3.2")
os.environ["DISPLAY"] = "0:0"
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
nio = vm._local_udp_tunnels[0][0]
assert cmd == [
fake_qemu_binary,
@ -536,7 +536,7 @@ def test_build_command_kvm_2_4(linux_platform, vm, loop, fake_qemu_binary, port_
vm.manager.get_qemu_version = AsyncioMagicMock(return_value="2.4.2")
os.environ["DISPLAY"] = "0:0"
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
nio = vm._local_udp_tunnels[0][0]
assert cmd == [
fake_qemu_binary,
@ -569,7 +569,7 @@ def test_build_command_without_display(vm, loop, fake_qemu_binary):
os.environ["DISPLAY"] = ""
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
assert "-nographic" in cmd
@ -578,7 +578,7 @@ def test_build_command_two_adapters(vm, loop, fake_qemu_binary, port_manager):
os.environ["DISPLAY"] = "0:0"
vm.adapters = 2
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
nio1 = vm._local_udp_tunnels[0][0]
nio2 = vm._local_udp_tunnels[1][0]
assert cmd == [
@ -619,7 +619,7 @@ def test_build_command_two_adapters_mac_address(vm, loop, fake_qemu_binary, port
mac_1 = int_to_macaddress(macaddress_to_int(vm._mac_address) + 1)
assert mac_0[:8] == "00:00:ab"
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
assert "e1000,mac={},netdev=gns3-0".format(mac_0) in cmd
assert "e1000,mac={},netdev=gns3-1".format(mac_1) in cmd
@ -628,7 +628,7 @@ def test_build_command_two_adapters_mac_address(vm, loop, fake_qemu_binary, port
mac_1 = int_to_macaddress(macaddress_to_int(vm._mac_address) + 1)
assert mac_0[:8] == "00:42:ab"
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
assert "e1000,mac={},netdev=gns3-0".format(mac_0) in cmd
assert "e1000,mac={},netdev=gns3-1".format(mac_1) in cmd
@ -648,7 +648,7 @@ def test_build_command_large_number_of_adapters(vm, loop, fake_qemu_binary, port
mac_1 = int_to_macaddress(macaddress_to_int(vm._mac_address) + 1)
assert mac_0[:8] == "00:00:ab"
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
# Count if we have 100 e1000 adapters in the command
assert len([l for l in cmd if "e1000" in l ]) == 100
@ -672,10 +672,10 @@ def test_build_command_large_number_of_adapters(vm, loop, fake_qemu_binary, port
vm.manager.get_qemu_version = AsyncioMagicMock(return_value="2.0.0")
with pytest.raises(QemuError):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
vm.adapters = 5
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
# Windows accept this kind of mistake
@ -685,7 +685,7 @@ def test_build_command_with_invalid_options(vm, loop, fake_qemu_binary):
vm.options = "'test"
with pytest.raises(QemuError):
cmd = loop.run_until_complete(asyncio.async(vm._build_command()))
cmd = loop.run_until_complete(asyncio.ensure_future(vm._build_command()))
def test_hda_disk_image(vm, images_dir):