NgTech-LTD/gns3-server
2
0 Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-11-05 14:51:13 +02:00
Code Issues Packages Projects Releases Wiki Activity

Support for source and destination for traceNG.

Browse Source
This commit is contained in:
grossmj committed 2018-03-27 16:58:49 +07:00
1 parent 8e695c8af1
commit d08c08617c
10 files changed
+117 -93
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/compute/traceng/test_traceng_vm.py
+39 -43
View File
@@ -71,17 +71,15 @@ def test_start(loop, vm, async_run):
with asyncio_patch("gns3server.compute.traceng.traceng_vm.TraceNGVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec:
with asyncio_patch("gns3server.compute.traceng.traceng_vm.TraceNGVM.start_wrap_console"):
loop.run_until_complete(asyncio.async(vm.start()))
assert mock_exec.call_args[0] == (vm._traceng_path(),
'-p',
str(vm._internal_console_port),
'-s',
ANY,
'-c',
ANY,
'-t',
'127.0.0.1')
loop.run_until_complete(asyncio.async(vm.start()))
assert mock_exec.call_args[0] == (vm._traceng_path(),
'-c',
ANY,
'-v',
ANY,
'-b',
'127.0.0.1',
'10.0.0.1')
assert vm.is_running()
assert vm.command_line == ' '.join(mock_exec.call_args[0])
(action, event, kwargs) = async_run(queue.get(0))
@@ -100,29 +98,28 @@ def test_stop(loop, vm, async_run):
with NotificationManager.instance().queue() as queue:
with asyncio_patch("gns3server.compute.traceng.traceng_vm.TraceNGVM._check_requirements", return_value=True):
with asyncio_patch("gns3server.compute.traceng.traceng_vm.TraceNGVM.start_wrap_console"):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = TraceNG.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1", "filters": {}})
async_run(vm.port_add_nio_binding(0, nio))
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = TraceNG.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1", "filters": {}})
async_run(vm.port_add_nio_binding(0, nio))
async_run(vm.start())
assert vm.is_running()
async_run(vm.start())
assert vm.is_running()
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
loop.run_until_complete(asyncio.async(vm.stop()))
assert vm.is_running() is False
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
loop.run_until_complete(asyncio.async(vm.stop()))
assert vm.is_running() is False
if sys.platform.startswith("win"):
process.send_signal.assert_called_with(1)
else:
process.terminate.assert_called_with()
if sys.platform.startswith("win"):
process.send_signal.assert_called_with(1)
else:
process.terminate.assert_called_with()
async_run(queue.get(0)) #  Ping
async_run(queue.get(0)) #  Started
async_run(queue.get(0)) #  Ping
async_run(queue.get(0)) #  Started
(action, event, kwargs) = async_run(queue.get(0))
assert action == "node.updated"
assert event == vm
(action, event, kwargs) = async_run(queue.get(0))
assert action == "node.updated"
assert event == vm
def test_reload(loop, vm, async_run):
@@ -135,22 +132,21 @@ def test_reload(loop, vm, async_run):
process.returncode = None
with asyncio_patch("gns3server.compute.traceng.traceng_vm.TraceNGVM._check_requirements", return_value=True):
with asyncio_patch("gns3server.compute.traceng.traceng_vm.TraceNGVM.start_wrap_console"):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = TraceNG.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1", "filters": {}})
async_run(vm.port_add_nio_binding(0, nio))
async_run(vm.start())
assert vm.is_running()
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
nio = TraceNG.instance().create_nio({"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1", "filters": {}})
async_run(vm.port_add_nio_binding(0, nio))
async_run(vm.start())
assert vm.is_running()
vm._ubridge_send = AsyncioMagicMock()
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
async_run(vm.reload())
assert vm.is_running() is True
vm._ubridge_send = AsyncioMagicMock()
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
async_run(vm.reload())
assert vm.is_running() is True
#if sys.platform.startswith("win"):
# process.send_signal.assert_called_with(1)
#else:
process.terminate.assert_called_with()
#if sys.platform.startswith("win"):
# process.send_signal.assert_called_with(1)
#else:
process.terminate.assert_called_with()
def test_add_nio_binding_udp(vm, async_run):
tests/handlers/api/compute/test_capabilities.py
+2 -2
View File
@@ -31,11 +31,11 @@ from gns3server.version import __version__
def test_get(http_compute, windows_platform):
response = http_compute.get('/capabilities', example=True)
assert response.status == 200
assert response.json == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou'], 'version': __version__, 'platform': sys.platform}
assert response.json == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou', 'traceng'], 'version': __version__, 'platform': sys.platform}
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
def test_get_on_gns3vm(http_compute, on_gns3vm):
response = http_compute.get('/capabilities', example=True)
assert response.status == 200
assert response.json == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou'], 'version': __version__, 'platform': sys.platform}
assert response.json == {'node_types': ['cloud', 'ethernet_hub', 'ethernet_switch', 'nat', 'vpcs', 'virtualbox', 'dynamips', 'frame_relay_switch', 'atm_switch', 'qemu', 'vmware', 'docker', 'iou', 'traceng'], 'version': __version__, 'platform': sys.platform}
tests/handlers/api/compute/test_traceng.py
-8
View File
@@ -47,14 +47,6 @@ def test_traceng_get(http_compute, project, vm):
assert response.json["status"] == "stopped"
def test_traceng_create_startup_script(http_compute, project):
response = http_compute.post("/projects/{project_id}/traceng/nodes".format(project_id=project.id), {"name": "TraceNG TEST 1", "startup_script": "ip 192.168.1.2\necho TEST"})
assert response.status == 201
assert response.route == "/projects/{project_id}/traceng/nodes"
assert response.json["name"] == "TraceNG TEST 1"
assert response.json["project_id"] == project.id
def test_traceng_create_port(http_compute, project, free_console_port):
response = http_compute.post("/projects/{project_id}/traceng/nodes".format(project_id=project.id), {"name": "TraceNG TEST 1", "console": free_console_port})
assert response.status == 201