arno
/
gns3-server
mirror of https://github.com/GNS3/gns3-server
1
0
Fork 0
Code Issues Releases Wiki Activity

Fix VPCS tests for recent version

Browse Source
pull/979/head
Julien Duponchelle 7 years ago
parent 6d150bb62c
commit 85136bc2e4
No known key found for this signature in database
GPG Key ID: CE8B29639E07F5E8
3 changed files with 19 additions and 28 deletions
Show Stats Download Patch File Download Diff File

2
gns3server/handlers/api/vpcs_handler.py
Unescape View File

@ -202,7 +202,7 @@ class VPCSHandler:
nio_type = request.json["type"]
if nio_type not in ("nio_udp", "nio_tap"):
raise HTTPConflict(text="NIO of type {} is not supported".format(nio_type))
nio = vpcs_manager.create_nio(vm.vpcs_path, request.json)
nio = vpcs_manager.create_nio(vm.vpcs_path(), request.json)
vm.port_add_nio_binding(int(request.match_info["port_number"]), nio)
response.set_status(201)
response.json(nio)

13
gns3server/modules/vpcs/vpcs_vm.py
Unescape View File

@ -94,7 +94,7 @@ class VPCSVM(BaseVM):
Check if VPCS is available with the correct version.
"""
path = self.vpcs_path
path = self.vpcs_path()
if not path:
raise VPCSError("No path to a VPCS executable has been set")
@ -132,7 +132,6 @@ class VPCSVM(BaseVM):
else:
return None
@property
def vpcs_path(self):
"""
Returns the VPCS executable path.
@ -203,7 +202,7 @@ class VPCSVM(BaseVM):
Checks if the VPCS executable version is >= 0.8b or == 0.6.1.
"""
try:
output = yield from subprocess_check_output(self.vpcs_path, "-v", cwd=self.working_dir)
output = yield from subprocess_check_output(self.vpcs_path(), "-v", cwd=self.working_dir)
match = re.search("Welcome to Virtual PC Simulator, version ([0-9a-z\.]+)", output)
if match:
version = match.group(1)
@ -211,7 +210,7 @@ class VPCSVM(BaseVM):
if self._vpcs_version < parse_version("0.8b") and self._vpcs_version != parse_version("0.6.1"):
raise VPCSError("VPCS executable version must be >= 0.8b or 0.6.1")
else:
raise VPCSError("Could not determine the VPCS version for {}".format(self.vpcs_path))
raise VPCSError("Could not determine the VPCS version for {}".format(self.vpcs_path()))
except (OSError, subprocess.SubprocessError) as e:
raise VPCSError("Error while looking for the VPCS version: {}".format(e))
@ -247,8 +246,8 @@ class VPCSVM(BaseVM):
self.status = "started"
except (OSError, subprocess.SubprocessError) as e:
vpcs_stdout = self.read_vpcs_stdout()
log.error("Could not start VPCS {}: {}\n{}".format(self.vpcs_path, e, vpcs_stdout))
raise VPCSError("Could not start VPCS {}: {}\n{}".format(self.vpcs_path, e, vpcs_stdout))
log.error("Could not start VPCS {}: {}\n{}".format(self.vpcs_path(), e, vpcs_stdout))
raise VPCSError("Could not start VPCS {}: {}\n{}".format(self.vpcs_path(), e, vpcs_stdout))
def _termination_callback(self, returncode):
"""
@ -417,7 +416,7 @@ class VPCSVM(BaseVM):
"""
command = [self.vpcs_path]
command = [self.vpcs_path()]
command.extend(["-p", str(self._console)]) # listen to console port
command.extend(["-m", str(self._manager.get_mac_id(self.id))]) # the unique ID is used to set the MAC address offset
command.extend(["-i", "1"]) # option to start only one VPC instance

32
tests/modules/vpcs/test_vpcs_vm.py
Unescape View File

@ -65,7 +65,7 @@ def test_vm_check_vpcs_version_0_6_1(loop, vm, manager):
def test_vm_invalid_vpcs_version(loop, manager, vm):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.subprocess_check_output", return_value="Welcome to Virtual PC Simulator, version 0.1"):
with pytest.raises(VPCSError):
nio = manager.create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
nio = manager.create_nio(vm.vpcs_path(), {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm._check_vpcs_version()))
assert vm.name == "test"
@ -73,9 +73,9 @@ def test_vm_invalid_vpcs_version(loop, manager, vm):
def test_vm_invalid_vpcs_path(vm, manager, loop):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM.vpcs_path", return_value="/tmp/fake/path/vpcs"):
with patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM.vpcs_path", return_value="/tmp/fake/path/vpcs"):
with pytest.raises(VPCSError):
nio = manager.create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
nio = manager.create_nio(vm.vpcs_path(), {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.name == "test"
@ -89,10 +89,10 @@ def test_start(loop, vm):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec:
nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
nio = VPCS.instance().create_nio(vm.vpcs_path(), {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio)
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',
str(vm.console),
'-m', '1',
@ -125,10 +125,10 @@ def test_start_0_6_1(loop, vm):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec:
nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
nio = VPCS.instance().create_nio(vm.vpcs_path(), {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio)
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',
str(vm.console),
'-m', '1',
@ -158,7 +158,7 @@ def test_stop(loop, vm):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
nio = VPCS.instance().create_nio(vm.vpcs_path(), {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
@ -191,7 +191,7 @@ def test_reload(loop, vm):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
nio = VPCS.instance().create_nio(vm.vpcs_path(), {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running()
@ -207,7 +207,7 @@ def test_reload(loop, vm):
def test_add_nio_binding_udp(vm):
nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
nio = VPCS.instance().create_nio(vm.vpcs_path(), {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio)
assert nio.lport == 4242
@ -215,21 +215,13 @@ def test_add_nio_binding_udp(vm):
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
def test_add_nio_binding_tap(vm, ethernet_device):
with patch("gns3server.modules.base_manager.BaseManager.has_privileged_access", return_value=True):
nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_tap", "tap_device": ethernet_device})
nio = VPCS.instance().create_nio(vm.vpcs_path(), {"type": "nio_tap", "tap_device": ethernet_device})
vm.port_add_nio_binding(0, nio)
assert nio.tap_device == ethernet_device
# def test_add_nio_binding_tap_no_privileged_access(vm):
# with patch("gns3server.modules.base_manager.BaseManager.has_privileged_access", return_value=False):
# with pytest.raises(aiohttp.web.HTTPForbidden):
# nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_tap", "tap_device": "test"})
# vm.port_add_nio_binding(0, nio)
# assert vm._ethernet_adapter.ports[0] is None
#
def test_port_remove_nio_binding(vm):
nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
nio = VPCS.instance().create_nio(vm.vpcs_path(), {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_add_nio_binding(0, nio)
vm.port_remove_nio_binding(0)
assert vm._ethernet_adapter.ports[0] is None

Write Preview
Loading…
Cancel
Save