diff --git a/gns3server/handlers/api/vpcs_handler.py b/gns3server/handlers/api/vpcs_handler.py index 59dd1dcd..6af82ad7 100644 --- a/gns3server/handlers/api/vpcs_handler.py +++ b/gns3server/handlers/api/vpcs_handler.py @@ -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) diff --git a/gns3server/modules/vpcs/vpcs_vm.py b/gns3server/modules/vpcs/vpcs_vm.py index 08efcf10..70474576 100644 --- a/gns3server/modules/vpcs/vpcs_vm.py +++ b/gns3server/modules/vpcs/vpcs_vm.py @@ -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 diff --git a/tests/modules/vpcs/test_vpcs_vm.py b/tests/modules/vpcs/test_vpcs_vm.py index ce7cf577..949bdf74 100644 --- a/tests/modules/vpcs/test_vpcs_vm.py +++ b/tests/modules/vpcs/test_vpcs_vm.py @@ -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