Get a working Qemu handler. Next step add all parameters

This commit is contained in:
Julien Duponchelle committed 2015-02-19 19:43:45 +01:00
1 parent b03b9226ff
commit d0244824bf
8 files changed
+799 -30

No files matched your search

+18 -8
View File
@@ -20,6 +20,7 @@ import aiohttp
import asyncio
import os
import stat
import re
from tests.utils import asyncio_patch
@@ -83,7 +84,7 @@ def test_stop(loop, vm):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = Qemu.instance().create_nio(vm.qemu_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio)
vm.adapter_add_nio_binding(0, 0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running()
loop.run_until_complete(asyncio.async(vm.stop()))
@@ -98,23 +99,32 @@ def test_reload(loop, vm):
assert mock.called_with("system_reset")
def test_suspend(loop, vm):
control_vm_result = MagicMock()
control_vm_result.match.group.decode.return_value = "running"
with asyncio_patch("gns3server.modules.qemu.QemuVM._control_vm", return_value=control_vm_result) as mock:
loop.run_until_complete(asyncio.async(vm.suspend()))
assert mock.called_with("system_reset")
def test_add_nio_binding_udp(vm, loop):
nio = Qemu.instance().create_nio(vm.qemu_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
loop.run_until_complete(asyncio.async(vm.port_add_nio_binding(0, nio)))
loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, 0, nio)))
assert nio.lport == 4242
def test_add_nio_binding_tap(vm, loop):
def test_add_nio_binding_ethernet(vm, loop):
with patch("gns3server.modules.base_manager.BaseManager._has_privileged_access", return_value=True):
nio = Qemu.instance().create_nio(vm.qemu_path, {"type": "nio_tap", "tap_device": "test"})
loop.run_until_complete(asyncio.async(vm.port_add_nio_binding(0, nio)))
assert nio.tap_device == "test"
nio = Qemu.instance().create_nio(vm.qemu_path, {"type": "nio_generic_ethernet", "ethernet_device": "eth0"})
loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, 0, nio)))
assert nio.ethernet_device == "eth0"
def test_port_remove_nio_binding(vm, loop):
nio = Qemu.instance().create_nio(vm.qemu_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
loop.run_until_complete(asyncio.async(vm.port_add_nio_binding(0, nio)))
loop.run_until_complete(asyncio.async(vm.port_remove_nio_binding(0)))
loop.run_until_complete(asyncio.async(vm.adapter_add_nio_binding(0, 0, nio)))
loop.run_until_complete(asyncio.async(vm.adapter_remove_nio_binding(0, 0)))
assert vm._ethernet_adapters[0].ports[0] is None