arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-11-28 03:08:14 +00:00
Code Issues Releases Wiki Activity

Merge remote-tracking branch 'origin/asyncio' into asyncio

Browse Source
This commit is contained in:
Jeremy 2015-01-20 15:28:45 -07:00
commit 061d3223a7
4 changed files with 44 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/api/examples/post_vpcs.txt
View File

@ -20,5 +20,5 @@ X-ROUTE: /vpcs
"name": "PC TEST 1",
"project_uuid": "a1e920ca-338a-4e9f-b363-aa607b09dd80",
"script_file": null,
"uuid": "12171ed6-4234-4ba7-9ef5-db3631a3a2e4"
"uuid": "a4809caf-bfba-4f34-ad78-1d054c8b7e5a"
}

17
gns3server/modules/vpcs/vpcs_vm.py
View File

@ -80,6 +80,7 @@ class VPCSVM(BaseVM):
def __del__(self):
self._kill_process()
@asyncio.coroutine
def _check_requirements(self):
"""
Check if VPCS is available with the correct version
@ -97,7 +98,7 @@ class VPCSVM(BaseVM):
if not os.access(self._path, os.X_OK):
raise VPCSError("VPCS program '{}' is not executable".format(self._path))
self._check_vpcs_version()
yield from self._check_vpcs_version()
def __json__(self):
@ -144,14 +145,14 @@ class VPCSVM(BaseVM):
new_name=new_name))
BaseVM.name = new_name
@asyncio.coroutine
def _check_vpcs_version(self):
"""
Checks if the VPCS executable version is >= 0.5b1.
"""
# TODO: should be async
try:
output = subprocess.check_output([self._path, "-v"], cwd=self.working_dir)
match = re.search("Welcome to Virtual PC Simulator, version ([0-9a-z\.]+)", output.decode("utf-8"))
output = yield from self._get_vpcs_welcome()
match = re.search("Welcome to Virtual PC Simulator, version ([0-9a-z\.]+)", output)
if match:
version = match.group(1)
if parse_version(version) < parse_version("0.5b1"):
@ -161,13 +162,19 @@ class VPCSVM(BaseVM):
except (OSError, subprocess.SubprocessError) as e:
raise VPCSError("Error while looking for the VPCS version: {}".format(e))
@asyncio.coroutine
def _get_vpcs_welcome(self):
proc = yield from asyncio.create_subprocess_exec(' '.join([self._path, "-v"]), stdout=asyncio.subprocess.PIPE, cwd=self.working_dir)
out = yield from proc.stdout.readline()
return out.decode("utf-8")
@asyncio.coroutine
def start(self):
"""
Starts the VPCS process.
"""
self._check_requirements()
yield from self._check_requirements()
if not self.is_running():
if not self._ethernet_adapter.get_nio(0):

10
tests/api/test_vpcs.py
View File

@ -84,3 +84,13 @@ def test_vpcs_delete_nio(server, vm):
response = server.delete("/vpcs/{}/ports/0/nio".format(vm["uuid"]), example=True)
assert response.status == 200
assert response.route == "/vpcs/{uuid}/ports/{port_id}/nio"
def test_vpcs_start():
# assert True == False
pass
def test_vpcs_stop():
# assert True == False
pass

43
tests/modules/vpcs/test_vpcs_vm.py
View File

@ -36,20 +36,19 @@ def manager():
return m
@patch("subprocess.check_output", return_value="Welcome to Virtual PC Simulator, version 0.6".encode("utf-8"))
def test_vm(project, manager):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
assert vm.name == "test"
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0f"
@patch("subprocess.check_output", return_value="Welcome to Virtual PC Simulator, version 0.1".encode("utf-8"))
def test_vm_invalid_vpcs_version(project, manager, loop):
with pytest.raises(VPCSError):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.name == "test"
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0f"
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._get_vpcs_welcome", return_value="Welcome to Virtual PC Simulator, version 0.1"):
with pytest.raises(VPCSError):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.name == "test"
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0f"
@patch("gns3server.config.Config.get_section_config", return_value={"path": "/bin/test_fake"})
@ -61,28 +60,28 @@ def test_vm_invalid_vpcs_path(project, manager, loop):
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0f"
@patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True)
def test_start(project, loop, manager):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running()
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running()
@patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True)
def test_stop(project, loop, manager):
process = MagicMock()
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running()
loop.run_until_complete(asyncio.async(vm.stop()))
assert vm.is_running() is False
process.terminate.assert_called_with()
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running()
loop.run_until_complete(asyncio.async(vm.stop()))
assert vm.is_running() is False
process.terminate.assert_called_with()
def test_add_nio_binding_udp(manager, project):