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

Fix tests for VPCS

Browse Source
This commit is contained in:
Julien Duponchelle 2016-06-27 14:34:34 +02:00
parent c0b09ae76a
commit e830356e62
No known key found for this signature in database
GPG Key ID: CE8B29639E07F5E8
1 changed files with 13 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
tests/compute/vpcs/test_vpcs_vm.py
View File

@ -89,12 +89,12 @@ def test_start(loop, vm, async_run):
process.returncode = None process.returncode = None
with NotificationManager.instance().queue() as queue: with NotificationManager.instance().queue() as queue:
async_run(queue.get(0)) #  Ping async_run(queue.get(0)) # Ping
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True): with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec: with asyncio_patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec:
nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"}) nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio) async_run(vm.port_add_nio_binding(0, nio))
loop.run_until_complete(asyncio.async(vm.start())) loop.run_until_complete(asyncio.async(vm.start()))
assert mock_exec.call_args[0] == (vm.vpcs_path, assert mock_exec.call_args[0] == (vm.vpcs_path,
'-p', '-p',
@ -117,7 +117,7 @@ def test_start(loop, vm, async_run):
assert event == vm assert event == vm
def test_start_0_6_1(loop, vm): def test_start_0_6_1(loop, vm, async_run):
""" """
Version 0.6.1 doesn't have the -R options. It's not require Version 0.6.1 doesn't have the -R options. It's not require
because GNS3 provide a patch for this. because GNS3 provide a patch for this.
@ -129,8 +129,8 @@ def test_start_0_6_1(loop, vm):
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True): with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec: with asyncio_patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec:
nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"}) nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio) async_run(vm.port_add_nio_binding(0, nio))
loop.run_until_complete(asyncio.async(vm.start())) async_run(vm.start())
assert mock_exec.call_args[0] == (vm.vpcs_path, assert mock_exec.call_args[0] == (vm.vpcs_path,
'-p', '-p',
str(vm.console), str(vm.console),
@ -160,9 +160,9 @@ def test_stop(loop, vm, async_run):
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True): with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process): with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"}) nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio) async_run(vm.port_add_nio_binding(0, nio))
loop.run_until_complete(asyncio.async(vm.start())) async_run(vm.start())
assert vm.is_running() assert vm.is_running()
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"): with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
@ -182,7 +182,7 @@ def test_stop(loop, vm, async_run):
assert event == vm assert event == vm
def test_reload(loop, vm): def test_reload(loop, vm, async_run):
process = MagicMock() process = MagicMock()
# Wait process kill success # Wait process kill success
@ -194,12 +194,12 @@ def test_reload(loop, vm):
with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True): with asyncio_patch("gns3server.compute.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process): with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"}) nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio) async_run(vm.port_add_nio_binding(0, nio))
loop.run_until_complete(asyncio.async(vm.start())) async_run(vm.start())
assert vm.is_running() assert vm.is_running()
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"): with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
loop.run_until_complete(asyncio.async(vm.reload())) async_run(vm.reload())
assert vm.is_running() is True assert vm.is_running() is True
if sys.platform.startswith("win"): if sys.platform.startswith("win"):
@ -208,9 +208,9 @@ def test_reload(loop, vm):
process.terminate.assert_called_with() process.terminate.assert_called_with()
def test_add_nio_binding_udp(vm): def test_add_nio_binding_udp(vm, async_run):
nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"}) nio = VPCS.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio) async_run(vm.port_add_nio_binding(0, nio))
assert nio.lport == 4242 assert nio.lport == 4242