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

Create /hypervisor and /controller namespace

Browse Source
This commit is contained in:
Julien Duponchelle committed 2016-03-07 15:01:35 +01:00
1 parent e4b12f49d1
commit 664cec97ba
231 files changed
+1576 -1480
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/conftest.py
+37 -1
View File
@@ -79,7 +79,7 @@ def _get_unused_port():
@pytest.fixture
def server(request, loop, port_manager, monkeypatch):
def http_server(request, loop, port_manager, monkeypatch):
"""A GNS3 server"""
port = _get_unused_port()
@@ -101,9 +101,45 @@ def server(request, loop, port_manager, monkeypatch):
srv.close()
srv.wait_closed()
request.addfinalizer(tear_down)
return (host, port)
@pytest.fixture
def http_root(loop, http_server):
"""
Return an helper allowing you to call the server without any prefix
"""
host, port = http_server
return Query(loop, host=host, port=port)
@pytest.fixture
def http_controller(loop, http_server):
"""
Return an helper allowing you to call the server API without any prefix
"""
host, port = http_server
return Query(loop, host=host, port=port, prefix="/controller", api_version=1)
@pytest.fixture
def http_hypervisor(loop, http_server):
"""
Return an helper allowing you to call the hypervisor API via HTTP
"""
host, port = http_server
return Query(loop, host=host, port=port, prefix="/hypervisor", api_version=1)
@pytest.fixture
def http_api(loop, http_server):
"""
Return an helper allowing you to call the root API via HTTP
"""
host, port = http_server
return Query(loop, host=host, port=port, api_version=1)
@pytest.fixture(scope="function")
def project(tmpdir):
"""A GNS3 lab"""
tests/handlers/api/.!13639!base.py
View File
Whitespace-only changes.
tests/handlers/api/base.py
+28 -15
View File
@@ -26,11 +26,20 @@ import os
class Query:
"""
Helper to make query againt the test server
"""
def __init__(self, loop, host='localhost', port=8001):
def __init__(self, loop, host='localhost', port=8001, prefix='', api_version=None):
"""
:param prefix: Prefix added before path (ex: /hypervisor)
:param api_version: Version of the api
"""
self._loop = loop
self._port = port
self._host = host
self._prefix = prefix
self._api_version = api_version
def post(self, path, body={}, **kwargs):
return self._fetch("POST", path, body, **kwargs)
@@ -44,25 +53,24 @@ class Query:
def delete(self, path, **kwargs):
return self._fetch("DELETE", path, **kwargs)
def get_url(self, path, version):
if version is None:
return "http://{}:{}{}".format(self._host, self._port, path)
return "http://{}:{}/v{}{}".format(self._host, self._port, version, path)
def get_url(self, path):
if self._api_version is None:
return "http://{}:{}{}{}".format(self._host, self._port, self._prefix, path)
return "http://{}:{}/v{}{}{}".format(self._host, self._port, self._api_version, self._prefix, path)
def _fetch(self, method, path, body=None, api_version=1, **kwargs):
def _fetch(self, method, path, body=None, **kwargs):
"""Fetch an url, parse the JSON and return response
Options:
- example if True the session is included inside documentation
- raw do not JSON encode the query
- api_version Version of API, None if no version
"""
if body is not None and not kwargs.get("raw", False):
body = json.dumps(body)
@asyncio.coroutine
def go_request(future):
response = yield from aiohttp.request(method, self.get_url(path, api_version), data=body)
response = yield from aiohttp.request(method, self.get_url(path), data=body)
future.set_result(response)
future = asyncio.Future()
asyncio.async(go_request(future))
@@ -79,7 +87,8 @@ class Query:
response.body = future.result()
x_route = response.headers.get('X-Route', None)
if x_route is not None:
response.route = x_route.replace("/v1", "")
response.route = x_route.replace("/v{}".format(self._api_version), "")
response.route = response.route .replace(self._prefix, "")
if response.body is not None:
if response.headers.get("CONTENT-TYPE", "") == "application/json":
@@ -96,27 +105,27 @@ class Query:
response.json = {}
response.html = ""
if kwargs.get('example') and os.environ.get("PYTEST_BUILD_DOCUMENTATION") == "1":
self._dump_example(method, response.route, api_version, path, body, response)
self._dump_example(method, response.route, path, body, response)
return response
def _dump_example(self, method, route, api_version, path, body, response):
def _dump_example(self, method, route, path, body, response):
"""Dump the request for the documentation"""
if path is None:
return
with open(self._example_file_path(method, route), 'w+') as f:
f.write("curl -i -X {} 'http://localhost:8000/v{}{}'".format(method, api_version, path))
f.write("curl -i -X {} 'http://localhost:8000/v{}{}'".format(method, self._api_version, path))
if body:
f.write(" -d '{}'".format(re.sub(r"\n", "", json.dumps(json.loads(body), sort_keys=True))))
f.write("\n\n")
f.write("{} /v{}{} HTTP/1.1\n".format(method, api_version, path))
f.write("{} /v{}{} HTTP/1.1\n".format(method, self._api_version, path))
if body:
f.write(json.dumps(json.loads(body), sort_keys=True, indent=4))
f.write("\n\n\n")
f.write("HTTP/1.1 {}\n".format(response.status))
for header, value in sorted(response.headers.items()):
if header == 'DATE':
# We fix the date otherwise the example is always different
# We fix the date otherwise the example is always different and create change in git
value = "Thu, 08 Jan 2015 16:09:15 GMT"
f.write("{}: {}\n".format(header, value))
f.write("\n")
@@ -126,4 +135,8 @@ class Query:
def _example_file_path(self, method, path):
path = re.sub('[^a-z0-9]', '', path)
return "docs/api/examples/{}_{}.txt".format(method.lower(), path)
if len(self._prefix):
prefix = self._prefix.replace('/', '')
else:
prefix = 'common'
return "docs/api/examples/{}_{}_{}.txt".format(prefix, method.lower(), path)
tests/handlers/api/controller/__init__.py
View File
Whitespace-only changes.
tests/handlers/api/test_server.py → tests/handlers/api/controller/test_server.py
+2 -2
View File
@@ -16,7 +16,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
def test_server_create(server, controller):
def test_server_create(http_controller, controller):
params = {
"server_id": "my_server_id",
@@ -26,7 +26,7 @@ def test_server_create(server, controller):
"user": "julien",
"password": "secure"
}
response = server.post("/servers", params, example=True)
response = http_controller.post("/servers", params, example=True)
assert response.status == 201
assert response.route == "/servers"
assert response.json["user"] == "julien"
tests/handlers/api/hypervisor/__init__.py
View File
Whitespace-only changes.
tests/handlers/api/test_config.py → tests/handlers/api/hypervisor/test_config.py
+4 -4
View File
@@ -19,7 +19,7 @@ from unittest.mock import MagicMock, patch
from gns3server.config import Config
def test_reload_accepted(server):
def test_reload_accepted(http_hypervisor):
gns_config = MagicMock()
config = Config.instance()
@@ -27,17 +27,17 @@ def test_reload_accepted(server):
gns_config.get_section_config.return_value = config.get_section_config("Server")
with patch("gns3server.config.Config.instance", return_value=gns_config):
response = server.post('/config/reload', example=True)
response = http_hypervisor.post('/config/reload', example=True)
assert response.status == 201
assert gns_config.reload.called
def test_reload_forbidden(server):
def test_reload_forbidden(http_hypervisor):
config = Config.instance()
config.set("Server", "local", "false")
response = server.post('/config/reload')
response = http_hypervisor.post('/config/reload')
assert response.status == 403
tests/handlers/api/test_docker.py → tests/handlers/api/hypervisor/test_docker.py
+36 -36
View File
@@ -43,20 +43,20 @@ def mock_connection():
@pytest.fixture
def vm(server, project, base_params):
def vm(http_hypervisor, project, base_params):
with asyncio_patch("gns3server.modules.docker.Docker.list_images", return_value=[{"image": "nginx"}]) as mock_list:
with asyncio_patch("gns3server.modules.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}) as mock:
response = server.post("/projects/{project_id}/docker/vms".format(project_id=project.id), base_params)
response = http_hypervisor.post("/projects/{project_id}/docker/vms".format(project_id=project.id), base_params)
if response.status != 201:
print(response.body)
assert response.status == 201
return response.json
def test_docker_create(server, project, base_params):
def test_docker_create(http_hypervisor, project, base_params):
with asyncio_patch("gns3server.modules.docker.Docker.list_images", return_value=[{"image": "nginx"}]) as mock_list:
with asyncio_patch("gns3server.modules.docker.Docker.query", return_value={"Id": "8bd8153ea8f5"}) as mock:
response = server.post("/projects/{project_id}/docker/vms".format(project_id=project.id), base_params)
response = http_hypervisor.post("/projects/{project_id}/docker/vms".format(project_id=project.id), base_params)
assert response.status == 201
assert response.route == "/projects/{project_id}/docker/vms"
assert response.json["name"] == "PC TEST 1"
@@ -67,66 +67,66 @@ def test_docker_create(server, project, base_params):
assert response.json["environment"] == "YES=1\nNO=0"
def test_docker_start(server, vm):
def test_docker_start(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.start", return_value=True) as mock:
response = server.post("/projects/{project_id}/docker/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
response = http_hypervisor.post("/projects/{project_id}/docker/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
assert mock.called
assert response.status == 204
def test_docker_stop(server, vm):
def test_docker_stop(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.stop", return_value=True) as mock:
response = server.post("/projects/{project_id}/docker/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
response = http_hypervisor.post("/projects/{project_id}/docker/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
assert mock.called
assert response.status == 204
def test_docker_reload(server, vm):
def test_docker_reload(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.restart", return_value=True) as mock:
response = server.post("/projects/{project_id}/docker/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
response = http_hypervisor.post("/projects/{project_id}/docker/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
assert mock.called
assert response.status == 204
def test_docker_delete(server, vm):
def test_docker_delete(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.delete", return_value=True) as mock:
response = server.delete("/projects/{project_id}/docker/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
response = http_hypervisor.delete("/projects/{project_id}/docker/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
assert mock.called
assert response.status == 204
def test_docker_reload(server, vm):
def test_docker_reload(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.pause", return_value=True) as mock:
response = server.post("/projects/{project_id}/docker/vms/{vm_id}/suspend".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
response = http_hypervisor.post("/projects/{project_id}/docker/vms/{vm_id}/suspend".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
assert mock.called
assert response.status == 204
def test_docker_nio_create_udp(server, vm):
response = server.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
def test_docker_nio_create_udp(http_hypervisor, vm):
response = http_hypervisor.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
assert response.status == 201
assert response.route == "/projects/{project_id}/docker/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
assert response.json["type"] == "nio_udp"
def test_docker_delete_nio(server, vm):
def test_docker_delete_nio(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.adapter_remove_nio_binding") as mock:
response = server.delete("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.delete("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 204
assert response.route == "/projects/{project_id}/docker/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
def test_docker_update(server, vm, tmpdir, free_console_port):
def test_docker_update(http_hypervisor, vm, tmpdir, free_console_port):
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.update") as mock:
response = server.put("/projects/{project_id}/docker/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"name": "test",
"console": free_console_port,
"start_command": "yes",
"environment": "GNS3=1\nGNS4=0"},
example=True)
response = http_hypervisor.put("/projects/{project_id}/docker/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"name": "test",
"console": free_console_port,
"start_command": "yes",
"environment": "GNS3=1\nGNS4=0"},
example=True)
assert mock.called
assert response.status == 200
assert response.json["name"] == "test"
@@ -135,13 +135,13 @@ def test_docker_update(server, vm, tmpdir, free_console_port):
assert response.json["environment"] == "GNS3=1\nGNS4=0"
def test_docker_start_capture(server, vm, tmpdir, project):
def test_docker_start_capture(http_hypervisor, vm, tmpdir, project):
with patch("gns3server.modules.docker.docker_vm.DockerVM.is_running", return_value=True) as mock:
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.start_capture") as start_capture:
params = {"capture_file_name": "test.pcap", "data_link_type": "DLT_EN10MB"}
response = server.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/start_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=params, example=True)
response = http_hypervisor.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/start_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=params, example=True)
assert response.status == 200
@@ -149,36 +149,36 @@ def test_docker_start_capture(server, vm, tmpdir, project):
assert "test.pcap" in response.json["pcap_file_path"]
def test_docker_start_capture_not_started(server, vm, tmpdir):
def test_docker_start_capture_not_started(http_hypervisor, vm, tmpdir):
with patch("gns3server.modules.docker.docker_vm.DockerVM.is_running", return_value=False) as mock:
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.start_capture") as start_capture:
params = {"capture_file_name": "test.pcap", "data_link_type": "DLT_EN10MB"}
response = server.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/start_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=params)
response = http_hypervisor.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/start_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=params)
assert not start_capture.called
assert response.status == 409
def test_docker_stop_capture(server, vm, tmpdir, project):
def test_docker_stop_capture(http_hypervisor, vm, tmpdir, project):
with patch("gns3server.modules.docker.docker_vm.DockerVM.is_running", return_value=True) as mock:
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.stop_capture") as stop_capture:
response = server.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/stop_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/stop_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 204
assert stop_capture.called
def test_docker_stop_capture_not_started(server, vm, tmpdir):
def test_docker_stop_capture_not_started(http_hypervisor, vm, tmpdir):
with patch("gns3server.modules.docker.docker_vm.DockerVM.is_running", return_value=False) as mock:
with asyncio_patch("gns3server.modules.docker.docker_vm.DockerVM.stop_capture") as stop_capture:
response = server.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/stop_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
response = http_hypervisor.post("/projects/{project_id}/docker/vms/{vm_id}/adapters/0/ports/0/stop_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
assert not stop_capture.called
assert response.status == 409
tests/handlers/api/test_dynamips.py → tests/handlers/api/hypervisor/test_dynamips.py
+26 -26
View File
@@ -24,11 +24,11 @@ from tests.utils import asyncio_patch
# @pytest.yield_fixture(scope="module")
# def vm(server, project):
# def vm(http_hypervisor, project):
#
# dynamips_path = "/fake/dynamips"
# with asyncio_patch("gns3server.modules.dynamips.nodes.router.Router.create", return_value=True) as mock:
# response = server.post("/projects/{project_id}/dynamips/vms".format(project_id=project.id), {"name": "My router",
# response = http_hypervisor.post("/projects/{project_id}/dynamips/vms".format(project_id=project.id), {"name": "My router",
# "platform": "c3745",
# "image": "somewhere",
# "ram": 128})
@@ -39,10 +39,10 @@ from tests.utils import asyncio_patch
# yield response.json
#
#
# def test_dynamips_vm_create(server, project):
# def test_dynamips_vm_create(http_hypervisor, project):
#
# with asyncio_patch("gns3server.modules.dynamips.nodes.router.Router.create", return_value=True):
# response = server.post("/projects/{project_id}/dynamips/vms".format(project_id=project.id), {"name": "My router",
# response = http_hypervisor.post("/projects/{project_id}/dynamips/vms".format(project_id=project.id), {"name": "My router",
# "platform": "c3745",
# "image": "somewhere",
# "ram": 128},
@@ -53,46 +53,46 @@ from tests.utils import asyncio_patch
# assert response.json["dynamips_id"]
#
#
# def test_dynamips_vm_get(server, project, vm):
# response = server.get("/projects/{project_id}/dynamips/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
# def test_dynamips_vm_get(http_hypervisor, project, vm):
# response = http_hypervisor.get("/projects/{project_id}/dynamips/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
# assert response.status == 200
# assert response.route == "/projects/{project_id}/dynamips/vms/{vm_id}"
# assert response.json["name"] == "My router"
# assert response.json["project_id"] == project.id
#
#
# def test_dynamips_vm_start(server, vm):
# def test_dynamips_vm_start(http_hypervisor, vm):
# with asyncio_patch("gns3server.modules.dynamips.nodes.router.Router.start", return_value=True) as mock:
# response = server.post("/projects/{project_id}/dynamips/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
# response = http_hypervisor.post("/projects/{project_id}/dynamips/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
# assert mock.called
# assert response.status == 204
#
#
# def test_dynamips_vm_stop(server, vm):
# def test_dynamips_vm_stop(http_hypervisor, vm):
# with asyncio_patch("gns3server.modules.dynamips.nodes.router.Router.stop", return_value=True) as mock:
# response = server.post("/projects/{project_id}/dynamips/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
# response = http_hypervisor.post("/projects/{project_id}/dynamips/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
# assert mock.called
# assert response.status == 204
#
#
# def test_dynamips_vm_suspend(server, vm):
# def test_dynamips_vm_suspend(http_hypervisor, vm):
# with asyncio_patch("gns3server.modules.dynamips.nodes.router.Router.suspend", return_value=True) as mock:
# response = server.post("/projects/{project_id}/dynamips/vms/{vm_id}/suspend".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
# response = http_hypervisor.post("/projects/{project_id}/dynamips/vms/{vm_id}/suspend".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
# assert mock.called
# assert response.status == 204
#
#
# def test_dynamips_vm_resume(server, vm):
# def test_dynamips_vm_resume(http_hypervisor, vm):
# with asyncio_patch("gns3server.modules.dynamips.nodes.router.Router.resume", return_value=True) as mock:
# response = server.post("/projects/{project_id}/dynamips/vms/{vm_id}/resume".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
# response = http_hypervisor.post("/projects/{project_id}/dynamips/vms/{vm_id}/resume".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
# assert mock.called
# assert response.status == 204
# def test_vbox_nio_create_udp(server, vm):
# def test_vbox_nio_create_udp(http_hypervisor, vm):
#
# with asyncio_patch('gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_add_nio_binding') as mock:
# response = server.post("/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/0/nio".format(project_id=vm["project_id"],
# response = http_hypervisor.post("/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/0/nio".format(project_id=vm["project_id"],
# vm_id=vm["vm_id"]), {"type": "nio_udp",
# "lport": 4242,
# "rport": 4343,
@@ -108,10 +108,10 @@ from tests.utils import asyncio_patch
# assert response.json["type"] == "nio_udp"
#
#
# def test_vbox_delete_nio(server, vm):
# def test_vbox_delete_nio(http_hypervisor, vm):
#
# with asyncio_patch('gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_remove_nio_binding') as mock:
# response = server.delete("/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
# response = http_hypervisor.delete("/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
#
# assert mock.called
# args, kwgars = mock.call_args
@@ -121,8 +121,8 @@ from tests.utils import asyncio_patch
# assert response.route == "/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/{adapter_id:\d+}/nio"
#
#
# def test_vbox_update(server, vm, free_console_port):
# response = server.put("/projects/{project_id}/virtualbox/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"name": "test",
# def test_vbox_update(http_hypervisor, vm, free_console_port):
# response = http_hypervisor.put("/projects/{project_id}/virtualbox/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"name": "test",
# "console": free_console_port})
# assert response.status == 200
# assert response.json["name"] == "test"
@@ -151,17 +151,17 @@ def fake_file(tmpdir):
return path
def test_vms(server, tmpdir, fake_dynamips, fake_file):
def test_vms(http_hypervisor, tmpdir, fake_dynamips, fake_file):
with patch("gns3server.modules.Dynamips.get_images_directory", return_value=str(tmpdir), example=True):
response = server.get("/dynamips/vms")
response = http_hypervisor.get("/dynamips/vms")
assert response.status == 200
assert response.json == [{"filename": "7200.bin", "path": "7200.bin"}]
def test_upload_vm(server, tmpdir):
def test_upload_vm(http_hypervisor, tmpdir):
with patch("gns3server.modules.Dynamips.get_images_directory", return_value=str(tmpdir),):
response = server.post("/dynamips/vms/test2", body="TEST", raw=True)
response = http_hypervisor.post("/dynamips/vms/test2", body="TEST", raw=True)
assert response.status == 204
with open(str(tmpdir / "test2")) as f:
@@ -172,11 +172,11 @@ def test_upload_vm(server, tmpdir):
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
def test_upload_vm_permission_denied(server, tmpdir):
def test_upload_vm_permission_denied(http_hypervisor, tmpdir):
with open(str(tmpdir / "test2"), "w+") as f:
f.write("")
os.chmod(str(tmpdir / "test2"), 0)
with patch("gns3server.modules.Dynamips.get_images_directory", return_value=str(tmpdir),):
response = server.post("/dynamips/vms/test2", body="TEST", raw=True)
response = http_hypervisor.post("/dynamips/vms/test2", body="TEST", raw=True)
assert response.status == 409
tests/handlers/api/test_file.py → tests/handlers/api/hypervisor/test_file.py
+4 -4
View File
@@ -26,14 +26,14 @@ import aiohttp
from gns3server.version import __version__
def test_stream(server, tmpdir, loop):
def test_stream(http_hypervisor, tmpdir, loop):
with open(str(tmpdir / "test"), 'w+') as f:
f.write("hello")
def go(future):
query = json.dumps({"location": str(tmpdir / "test")})
headers = {'content-type': 'application/json'}
response = yield from aiohttp.request("GET", server.get_url("/files/stream", 1), data=query, headers=headers)
response = yield from aiohttp.request("GET", http_hypervisor.get_url("/files/stream"), data=query, headers=headers)
response.body = yield from response.content.read(5)
with open(str(tmpdir / "test"), 'a') as f:
f.write("world")
@@ -48,11 +48,11 @@ def test_stream(server, tmpdir, loop):
assert response.body == b'helloworld'
def test_stream_file_not_found(server, tmpdir, loop):
def test_stream_file_not_found(http_hypervisor, tmpdir, loop):
def go(future):
query = json.dumps({"location": str(tmpdir / "test")})
headers = {'content-type': 'application/json'}
response = yield from aiohttp.request("GET", server.get_url("/files/stream", 1), data=query, headers=headers)
response = yield from aiohttp.request("GET", http_hypervisor.get_url("/files/stream"), data=query, headers=headers)
response.close()
future.set_result(response)
tests/handlers/api/test_iou.py → tests/handlers/api/hypervisor/test_iou.py
+66 -66
View File
@@ -46,8 +46,8 @@ def base_params(tmpdir, fake_iou_bin):
@pytest.fixture
def vm(server, project, base_params):
response = server.post("/projects/{project_id}/iou/vms".format(project_id=project.id), base_params)
def vm(http_hypervisor, project, base_params):
response = http_hypervisor.post("/projects/{project_id}/iou/vms".format(project_id=project.id), base_params)
assert response.status == 201
return response.json
@@ -58,8 +58,8 @@ def startup_config_file(project, vm):
return os.path.join(directory, "startup-config.cfg")
def test_iou_create(server, project, base_params):
response = server.post("/projects/{project_id}/iou/vms".format(project_id=project.id), base_params)
def test_iou_create(http_hypervisor, project, base_params):
response = http_hypervisor.post("/projects/{project_id}/iou/vms".format(project_id=project.id), base_params)
assert response.status == 201
assert response.route == "/projects/{project_id}/iou/vms"
assert response.json["name"] == "PC TEST 1"
@@ -71,7 +71,7 @@ def test_iou_create(server, project, base_params):
assert response.json["l1_keepalives"] is False
def test_iou_create_with_params(server, project, base_params):
def test_iou_create_with_params(http_hypervisor, project, base_params):
params = base_params
params["ram"] = 1024
params["nvram"] = 512
@@ -82,7 +82,7 @@ def test_iou_create_with_params(server, project, base_params):
params["use_default_iou_values"] = True
params["iourc_content"] = "test"
response = server.post("/projects/{project_id}/iou/vms".format(project_id=project.id), params, example=True)
response = http_hypervisor.post("/projects/{project_id}/iou/vms".format(project_id=project.id), params, example=True)
assert response.status == 201
assert response.route == "/projects/{project_id}/iou/vms"
assert response.json["name"] == "PC TEST 1"
@@ -101,7 +101,7 @@ def test_iou_create_with_params(server, project, base_params):
assert "iourc" in response.json["iourc_path"]
def test_iou_create_startup_config_already_exist(server, project, base_params):
def test_iou_create_startup_config_already_exist(http_hypervisor, project, base_params):
"""We don't erase a startup-config if already exist at project creation"""
vm_id = str(uuid.uuid4())
@@ -113,7 +113,7 @@ def test_iou_create_startup_config_already_exist(server, project, base_params):
params["vm_id"] = vm_id
params["startup_config_content"] = "hostname test"
response = server.post("/projects/{project_id}/iou/vms".format(project_id=project.id), params, example=True)
response = http_hypervisor.post("/projects/{project_id}/iou/vms".format(project_id=project.id), params, example=True)
assert response.status == 201
assert response.route == "/projects/{project_id}/iou/vms"
@@ -122,8 +122,8 @@ def test_iou_create_startup_config_already_exist(server, project, base_params):
assert f.read() == "echo hello"
def test_iou_get(server, project, vm):
response = server.get("/projects/{project_id}/iou/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
def test_iou_get(http_hypervisor, project, vm):
response = http_hypervisor.get("/projects/{project_id}/iou/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 200
assert response.route == "/projects/{project_id}/iou/vms/{vm_id}"
assert response.json["name"] == "PC TEST 1"
@@ -135,50 +135,50 @@ def test_iou_get(server, project, vm):
assert response.json["l1_keepalives"] is False
def test_iou_start(server, vm):
def test_iou_start(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.iou.iou_vm.IOUVM.start", return_value=True) as mock:
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
assert mock.called
assert response.status == 200
assert response.json["name"] == "PC TEST 1"
def test_iou_start_with_iourc(server, vm, tmpdir):
def test_iou_start_with_iourc(http_hypervisor, vm, tmpdir):
body = {"iourc_content": "test"}
with asyncio_patch("gns3server.modules.iou.iou_vm.IOUVM.start", return_value=True) as mock:
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=body, example=True)
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=body, example=True)
assert mock.called
assert response.status == 200
response = server.get("/projects/{project_id}/iou/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
response = http_hypervisor.get("/projects/{project_id}/iou/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
assert response.status == 200
with open(response.json["iourc_path"]) as f:
assert f.read() == "test"
def test_iou_stop(server, vm):
def test_iou_stop(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.iou.iou_vm.IOUVM.stop", return_value=True) as mock:
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_iou_reload(server, vm):
def test_iou_reload(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.iou.iou_vm.IOUVM.reload", return_value=True) as mock:
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_iou_delete(server, vm):
def test_iou_delete(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.iou.IOU.delete_vm", return_value=True) as mock:
response = server.delete("/projects/{project_id}/iou/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.delete("/projects/{project_id}/iou/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_iou_update(server, vm, tmpdir, free_console_port, project):
def test_iou_update(http_hypervisor, vm, tmpdir, free_console_port, project):
params = {
"name": "test",
"console": free_console_port,
@@ -191,7 +191,7 @@ def test_iou_update(server, vm, tmpdir, free_console_port, project):
"use_default_iou_values": True,
"iourc_content": "test"
}
response = server.put("/projects/{project_id}/iou/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), params, example=True)
response = http_hypervisor.put("/projects/{project_id}/iou/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), params, example=True)
assert response.status == 200
assert response.json["name"] == "test"
assert response.json["console"] == free_console_port
@@ -208,65 +208,65 @@ def test_iou_update(server, vm, tmpdir, free_console_port, project):
assert "iourc" in response.json["iourc_path"]
def test_iou_nio_create_udp(server, vm):
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
def test_iou_nio_create_udp(http_hypervisor, vm):
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
assert response.status == 201
assert response.route == "/projects/{project_id}/iou/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
assert response.json["type"] == "nio_udp"
def test_iou_nio_create_ethernet(server, vm, ethernet_device):
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_generic_ethernet",
"ethernet_device": ethernet_device,
},
example=True)
def test_iou_nio_create_ethernet(http_hypervisor, vm, ethernet_device):
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_generic_ethernet",
"ethernet_device": ethernet_device,
},
example=True)
assert response.status == 201
assert response.route == "/projects/{project_id}/iou/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
assert response.json["type"] == "nio_generic_ethernet"
assert response.json["ethernet_device"] == ethernet_device
def test_iou_nio_create_ethernet_different_port(server, vm, ethernet_device):
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/3/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_generic_ethernet",
"ethernet_device": ethernet_device,
},
example=False)
def test_iou_nio_create_ethernet_different_port(http_hypervisor, vm, ethernet_device):
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/3/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_generic_ethernet",
"ethernet_device": ethernet_device,
},
example=False)
assert response.status == 201
assert response.route == "/projects/{project_id}/iou/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
assert response.json["type"] == "nio_generic_ethernet"
assert response.json["ethernet_device"] == ethernet_device
def test_iou_nio_create_tap(server, vm, ethernet_device):
def test_iou_nio_create_tap(http_hypervisor, vm, ethernet_device):
with patch("gns3server.modules.base_manager.BaseManager.has_privileged_access", return_value=True):
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_tap",
"tap_device": ethernet_device})
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_tap",
"tap_device": ethernet_device})
assert response.status == 201
assert response.route == "/projects/{project_id}/iou/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
assert response.json["type"] == "nio_tap"
def test_iou_delete_nio(server, vm):
server.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"})
response = server.delete("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
def test_iou_delete_nio(http_hypervisor, vm):
http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"})
response = http_hypervisor.delete("/projects/{project_id}/iou/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 204
assert response.route == "/projects/{project_id}/iou/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
def test_iou_start_capture(server, vm, tmpdir, project):
def test_iou_start_capture(http_hypervisor, vm, tmpdir, project):
with patch("gns3server.modules.iou.iou_vm.IOUVM.is_running", return_value=True) as mock:
with asyncio_patch("gns3server.modules.iou.iou_vm.IOUVM.start_capture") as start_capture:
params = {"capture_file_name": "test.pcap", "data_link_type": "DLT_EN10MB"}
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/0/start_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=params, example=True)
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/0/start_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=params, example=True)
assert response.status == 200
@@ -274,71 +274,71 @@ def test_iou_start_capture(server, vm, tmpdir, project):
assert "test.pcap" in response.json["pcap_file_path"]
def test_iou_start_capture_not_started(server, vm, tmpdir):
def test_iou_start_capture_not_started(http_hypervisor, vm, tmpdir):
with patch("gns3server.modules.iou.iou_vm.IOUVM.is_running", return_value=False) as mock:
with asyncio_patch("gns3server.modules.iou.iou_vm.IOUVM.start_capture") as start_capture:
params = {"capture_file_name": "test.pcap", "data_link_type": "DLT_EN10MB"}
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/0/start_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=params)
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/0/start_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), body=params)
assert not start_capture.called
assert response.status == 409
def test_iou_stop_capture(server, vm, tmpdir, project):
def test_iou_stop_capture(http_hypervisor, vm, tmpdir, project):
with patch("gns3server.modules.iou.iou_vm.IOUVM.is_running", return_value=True) as mock:
with asyncio_patch("gns3server.modules.iou.iou_vm.IOUVM.stop_capture") as stop_capture:
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/0/stop_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/0/stop_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 204
assert stop_capture.called
def test_iou_stop_capture_not_started(server, vm, tmpdir):
def test_iou_stop_capture_not_started(http_hypervisor, vm, tmpdir):
with patch("gns3server.modules.iou.iou_vm.IOUVM.is_running", return_value=False) as mock:
with asyncio_patch("gns3server.modules.iou.iou_vm.IOUVM.stop_capture") as stop_capture:
response = server.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/0/stop_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
response = http_hypervisor.post("/projects/{project_id}/iou/vms/{vm_id}/adapters/0/ports/0/stop_capture".format(project_id=vm["project_id"], vm_id=vm["vm_id"]))
assert not stop_capture.called
assert response.status == 409
def test_get_configs_without_configs_file(server, vm):
def test_get_configs_without_configs_file(http_hypervisor, vm):
response = server.get("/projects/{project_id}/iou/vms/{vm_id}/configs".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.get("/projects/{project_id}/iou/vms/{vm_id}/configs".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 200
assert "startup_config" not in response.json
assert "private_config" not in response.json
def test_get_configs_with_startup_config_file(server, project, vm):
def test_get_configs_with_startup_config_file(http_hypervisor, project, vm):
path = startup_config_file(project, vm)
with open(path, "w+") as f:
f.write("TEST")
response = server.get("/projects/{project_id}/iou/vms/{vm_id}/configs".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.get("/projects/{project_id}/iou/vms/{vm_id}/configs".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 200
assert response.json["startup_config_content"] == "TEST"
def test_vms(server, tmpdir, fake_iou_bin):
def test_vms(http_hypervisor, tmpdir, fake_iou_bin):
with patch("gns3server.modules.IOU.get_images_directory", return_value=str(tmpdir)):
response = server.get("/iou/vms", example=True)
response = http_hypervisor.get("/iou/vms", example=True)
assert response.status == 200
assert response.json == [{"filename": "iou.bin", "path": "iou.bin"}]
def test_upload_vm(server, tmpdir):
def test_upload_vm(http_hypervisor, tmpdir):
with patch("gns3server.modules.IOU.get_images_directory", return_value=str(tmpdir),):
response = server.post("/iou/vms/test2", body="TEST", raw=True)
response = http_hypervisor.post("/iou/vms/test2", body="TEST", raw=True)
assert response.status == 204
with open(str(tmpdir / "test2")) as f:
@@ -349,11 +349,11 @@ def test_upload_vm(server, tmpdir):
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
def test_upload_vm_permission_denied(server, tmpdir):
def test_upload_vm_permission_denied(http_hypervisor, tmpdir):
with open(str(tmpdir / "test2"), "w+") as f:
f.write("")
os.chmod(str(tmpdir / "test2"), 0)
with patch("gns3server.modules.IOU.get_images_directory", return_value=str(tmpdir),):
response = server.post("/iou/vms/test2", body="TEST", raw=True)
response = http_hypervisor.post("/iou/vms/test2", body="TEST", raw=True)
assert response.status == 409
tests/handlers/api/test_network.py → tests/handlers/api/hypervisor/test_network.py
+4 -4
View File
@@ -19,15 +19,15 @@ import os
import pytest
def test_udp_allocation(server, project):
response = server.post('/projects/{}/ports/udp'.format(project.id), {}, example=True)
def test_udp_allocation(http_hypervisor, project):
response = http_hypervisor.post('/projects/{}/ports/udp'.format(project.id), {}, example=True)
assert response.status == 201
assert response.json == {'udp_port': 10000}
# Netfifaces is not available on Travis
@pytest.mark.skipif(os.environ.get("TRAVIS", False) is not False, reason="Not supported on Travis")
def test_interfaces(server):
response = server.get('/interfaces', example=True)
def test_interfaces(http_hypervisor):
response = http_hypervisor.get('/interfaces', example=True)
assert response.status == 200
assert isinstance(response.json, list)
tests/handlers/api/test_project.py → tests/handlers/api/hypervisor/test_project.py
+54 -54
View File
@@ -27,49 +27,49 @@ import aiohttp
from unittest.mock import patch
from tests.utils import asyncio_patch
from gns3server.handlers.api.project_handler import ProjectHandler
from gns3server.handlers.api.hypervisor.project_handler import ProjectHandler
from gns3server.modules.project_manager import ProjectManager
def test_create_project_with_path(server, tmpdir):
def test_create_project_with_path(http_hypervisor, tmpdir):
with patch("gns3server.modules.project.Project.is_local", return_value=True):
response = server.post("/projects", {"name": "test", "path": str(tmpdir)})
response = http_hypervisor.post("/projects", {"name": "test", "path": str(tmpdir)})
assert response.status == 201
assert response.json["path"] == str(tmpdir)
assert response.json["name"] == "test"
def test_create_project_without_dir(server):
def test_create_project_without_dir(http_hypervisor):
query = {"name": "test"}
response = server.post("/projects", query, example=True)
response = http_hypervisor.post("/projects", query, example=True)
assert response.status == 201
assert response.json["project_id"] is not None
assert response.json["temporary"] is False
assert response.json["name"] == "test"
def test_create_temporary_project(server):
def test_create_temporary_project(http_hypervisor):
query = {"name": "test", "temporary": True}
response = server.post("/projects", query)
response = http_hypervisor.post("/projects", query)
assert response.status == 201
assert response.json["project_id"] is not None
assert response.json["temporary"] is True
assert response.json["name"] == "test"
def test_create_project_with_uuid(server):
def test_create_project_with_uuid(http_hypervisor):
query = {"name": "test", "project_id": "00010203-0405-0607-0809-0a0b0c0d0e0f"}
response = server.post("/projects", query)
response = http_hypervisor.post("/projects", query)
assert response.status == 201
assert response.json["project_id"] == "00010203-0405-0607-0809-0a0b0c0d0e0f"
assert response.json["name"] == "test"
def test_show_project(server):
def test_show_project(http_hypervisor):
query = {"name": "test", "project_id": "00010203-0405-0607-0809-0a0b0c0d0e02", "temporary": False}
response = server.post("/projects", query)
response = http_hypervisor.post("/projects", query)
assert response.status == 201
response = server.get("/projects/00010203-0405-0607-0809-0a0b0c0d0e02", example=True)
response = http_hypervisor.get("/projects/00010203-0405-0607-0809-0a0b0c0d0e02", example=True)
assert len(response.json.keys()) == 5
assert len(response.json["location"]) > 0
assert response.json["project_id"] == "00010203-0405-0607-0809-0a0b0c0d0e02"
@@ -77,49 +77,49 @@ def test_show_project(server):
assert response.json["name"] == "test"
def test_show_project_invalid_uuid(server):
response = server.get("/projects/00010203-0405-0607-0809-0a0b0c0d0e42")
def test_show_project_invalid_uuid(http_hypervisor):
response = http_hypervisor.get("/projects/00010203-0405-0607-0809-0a0b0c0d0e42")
assert response.status == 404
def test_list_projects(server):
def test_list_projects(http_hypervisor):
ProjectManager.instance()._projects = {}
query = {"name": "test", "project_id": "00010203-0405-0607-0809-0a0b0c0d0e0f"}
response = server.post("/projects", query)
response = http_hypervisor.post("/projects", query)
assert response.status == 201
query = {"name": "test", "project_id": "00010203-0405-0607-0809-0a0b0c0d0e0b"}
response = server.post("/projects", query)
response = http_hypervisor.post("/projects", query)
assert response.status == 201
response = server.get("/projects", example=True)
response = http_hypervisor.get("/projects", example=True)
assert response.status == 200
print(response.json)
assert len(response.json) == 2
assert response.json[0]["project_id"] == "00010203-0405-0607-0809-0a0b0c0d0e0b" or response.json[1]["project_id"] == "00010203-0405-0607-0809-0a0b0c0d0e0b"
def test_update_temporary_project(server):
def test_update_temporary_project(http_hypervisor):
query = {"name": "test", "temporary": True}
response = server.post("/projects", query)
response = http_hypervisor.post("/projects", query)
assert response.status == 201
query = {"name": "test", "temporary": False}
response = server.put("/projects/{project_id}".format(project_id=response.json["project_id"]), query, example=True)
response = http_hypervisor.put("/projects/{project_id}".format(project_id=response.json["project_id"]), query, example=True)
assert response.status == 200
assert response.json["temporary"] is False
def test_update_path_project_temporary(server, tmpdir):
def test_update_path_project_temporary(http_hypervisor, tmpdir):
os.makedirs(str(tmpdir / "a"))
os.makedirs(str(tmpdir / "b"))
with patch("gns3server.modules.project.Project.is_local", return_value=True):
response = server.post("/projects", {"name": "first_name", "path": str(tmpdir / "a"), "temporary": True})
response = http_hypervisor.post("/projects", {"name": "first_name", "path": str(tmpdir / "a"), "temporary": True})
assert response.status == 201
assert response.json["name"] == "first_name"
query = {"name": "second_name", "path": str(tmpdir / "b")}
response = server.put("/projects/{project_id}".format(project_id=response.json["project_id"]), query, example=True)
response = http_hypervisor.put("/projects/{project_id}".format(project_id=response.json["project_id"]), query, example=True)
assert response.status == 200
assert response.json["path"] == str(tmpdir / "b")
assert response.json["name"] == "second_name"
@@ -128,17 +128,17 @@ def test_update_path_project_temporary(server, tmpdir):
assert os.path.exists(str(tmpdir / "b"))
def test_update_path_project_non_temporary(server, tmpdir):
def test_update_path_project_non_temporary(http_hypervisor, tmpdir):
os.makedirs(str(tmpdir / "a"))
os.makedirs(str(tmpdir / "b"))
with patch("gns3server.modules.project.Project.is_local", return_value=True):
response = server.post("/projects", {"name": "first_name", "path": str(tmpdir / "a")})
response = http_hypervisor.post("/projects", {"name": "first_name", "path": str(tmpdir / "a")})
assert response.status == 201
assert response.json["name"] == "first_name"
query = {"name": "second_name", "path": str(tmpdir / "b")}
response = server.put("/projects/{project_id}".format(project_id=response.json["project_id"]), query, example=True)
response = http_hypervisor.put("/projects/{project_id}".format(project_id=response.json["project_id"]), query, example=True)
assert response.status == 200
assert response.json["path"] == str(tmpdir / "b")
assert response.json["name"] == "second_name"
@@ -147,66 +147,66 @@ def test_update_path_project_non_temporary(server, tmpdir):
assert os.path.exists(str(tmpdir / "b"))
def test_update_path_project_non_local(server, tmpdir):
def test_update_path_project_non_local(http_hypervisor, tmpdir):
with patch("gns3server.modules.project.Project.is_local", return_value=False):
response = server.post("/projects", {"name": "first_name"})
response = http_hypervisor.post("/projects", {"name": "first_name"})
assert response.status == 201
query = {"name": "second_name", "path": str(tmpdir)}
response = server.put("/projects/{project_id}".format(project_id=response.json["project_id"]), query, example=True)
response = http_hypervisor.put("/projects/{project_id}".format(project_id=response.json["project_id"]), query, example=True)
assert response.status == 403
def test_commit_project(server, project):
def test_commit_project(http_hypervisor, project):
with asyncio_patch("gns3server.modules.project.Project.commit", return_value=True) as mock:
response = server.post("/projects/{project_id}/commit".format(project_id=project.id), example=True)
response = http_hypervisor.post("/projects/{project_id}/commit".format(project_id=project.id), example=True)
assert response.status == 204
assert mock.called
def test_commit_project_invalid_uuid(server):
response = server.post("/projects/{project_id}/commit".format(project_id=uuid.uuid4()))
def test_commit_project_invalid_uuid(http_hypervisor):
response = http_hypervisor.post("/projects/{project_id}/commit".format(project_id=uuid.uuid4()))
assert response.status == 404
def test_delete_project(server, project):
def test_delete_project(http_hypervisor, project):
with asyncio_patch("gns3server.modules.project.Project.delete", return_value=True) as mock:
response = server.delete("/projects/{project_id}".format(project_id=project.id), example=True)
response = http_hypervisor.delete("/projects/{project_id}".format(project_id=project.id), example=True)
assert response.status == 204
assert mock.called
def test_delete_project_invalid_uuid(server):
response = server.delete("/projects/{project_id}".format(project_id=uuid.uuid4()))
def test_delete_project_invalid_uuid(http_hypervisor):
response = http_hypervisor.delete("/projects/{project_id}".format(project_id=uuid.uuid4()))
assert response.status == 404
def test_close_project(server, project):
def test_close_project(http_hypervisor, project):
with asyncio_patch("gns3server.modules.project.Project.close", return_value=True) as mock:
response = server.post("/projects/{project_id}/close".format(project_id=project.id), example=True)
response = http_hypervisor.post("/projects/{project_id}/close".format(project_id=project.id), example=True)
assert response.status == 204
assert mock.called
def test_close_project_two_client_connected(server, project):
def test_close_project_two_client_connected(http_hypervisor, project):
ProjectHandler._notifications_listening = {project.id: 2}
with asyncio_patch("gns3server.modules.project.Project.close", return_value=True) as mock:
response = server.post("/projects/{project_id}/close".format(project_id=project.id), example=True)
response = http_hypervisor.post("/projects/{project_id}/close".format(project_id=project.id), example=True)
assert response.status == 204
assert not mock.called
def test_close_project_invalid_uuid(server):
response = server.post("/projects/{project_id}/close".format(project_id=uuid.uuid4()))
def test_close_project_invalid_uuid(http_hypervisor):
response = http_hypervisor.post("/projects/{project_id}/close".format(project_id=uuid.uuid4()))
assert response.status == 404
def test_notification(server, project, loop):
def test_notification(http_hypervisor, project, loop):
@asyncio.coroutine
def go(future):
response = yield from aiohttp.request("GET", server.get_url("/projects/{project_id}/notifications".format(project_id=project.id), 1))
response = yield from aiohttp.request("GET", http_hypervisor.get_url("/projects/{project_id}/notifications".format(project_id=project.id)))
response.body = yield from response.content.read(200)
project.emit("vm.created", {"a": "b"})
response.body += yield from response.content.read(50)
@@ -222,12 +222,12 @@ def test_notification(server, project, loop):
assert b'{"action": "vm.created", "event": {"a": "b"}}\n' in response.body
def test_notification_invalid_id(server):
response = server.get("/projects/{project_id}/notifications".format(project_id=uuid.uuid4()))
def test_notification_invalid_id(http_hypervisor):
response = http_hypervisor.get("/projects/{project_id}/notifications".format(project_id=uuid.uuid4()))
assert response.status == 404
def test_list_files(server, project):
def test_list_files(http_hypervisor, project):
files = [
{
"path": "test.txt",
@@ -239,12 +239,12 @@ def test_list_files(server, project):
}
]
with asyncio_patch("gns3server.modules.project.Project.list_files", return_value=files) as mock:
response = server.get("/projects/{project_id}/files".format(project_id=project.id), example=True)
response = http_hypervisor.get("/projects/{project_id}/files".format(project_id=project.id), example=True)
assert response.status == 200
assert response.json == files
def test_get_file(server, tmpdir):
def test_get_file(http_hypervisor, tmpdir):
with patch("gns3server.config.Config.get_section_config", return_value={"project_directory": str(tmpdir)}):
project = ProjectManager.instance().create_project()
@@ -252,12 +252,12 @@ def test_get_file(server, tmpdir):
with open(os.path.join(project.path, "hello"), "w+") as f:
f.write("world")
response = server.get("/projects/{project_id}/files/hello".format(project_id=project.id), raw=True)
response = http_hypervisor.get("/projects/{project_id}/files/hello".format(project_id=project.id), raw=True)
assert response.status == 200
assert response.body == b"world"
response = server.get("/projects/{project_id}/files/false".format(project_id=project.id), raw=True)
response = http_hypervisor.get("/projects/{project_id}/files/false".format(project_id=project.id), raw=True)
assert response.status == 404
response = server.get("/projects/{project_id}/files/../hello".format(project_id=project.id), raw=True)
response = http_hypervisor.get("/projects/{project_id}/files/../hello".format(project_id=project.id), raw=True)
assert response.status == 403
tests/handlers/api/test_qemu.py → tests/handlers/api/hypervisor/test_qemu.py
+67 -66
View File
@@ -23,6 +23,7 @@ from tests.utils import asyncio_patch
from unittest.mock import patch
from gns3server.config import Config
@pytest.fixture
def fake_qemu_bin():
@@ -56,14 +57,14 @@ def base_params(tmpdir, fake_qemu_bin):
@pytest.fixture
def vm(server, project, base_params):
response = server.post("/projects/{project_id}/qemu/vms".format(project_id=project.id), base_params)
def vm(http_hypervisor, project, base_params):
response = http_hypervisor.post("/projects/{project_id}/qemu/vms".format(project_id=project.id), base_params)
assert response.status == 201
return response.json
def test_qemu_create(server, project, base_params, fake_qemu_bin):
response = server.post("/projects/{project_id}/qemu/vms".format(project_id=project.id), base_params)
def test_qemu_create(http_hypervisor, project, base_params, fake_qemu_bin):
response = http_hypervisor.post("/projects/{project_id}/qemu/vms".format(project_id=project.id), base_params)
assert response.status == 201
assert response.route == "/projects/{project_id}/qemu/vms"
assert response.json["name"] == "PC TEST 1"
@@ -72,11 +73,11 @@ def test_qemu_create(server, project, base_params, fake_qemu_bin):
assert response.json["platform"] == "x86_64"
def test_qemu_create_platform(server, project, base_params, fake_qemu_bin):
def test_qemu_create_platform(http_hypervisor, project, base_params, fake_qemu_bin):
base_params["qemu_path"] = None
base_params["platform"] = "x86_64"
response = server.post("/projects/{project_id}/qemu/vms".format(project_id=project.id), base_params)
response = http_hypervisor.post("/projects/{project_id}/qemu/vms".format(project_id=project.id), base_params)
assert response.status == 201
assert response.route == "/projects/{project_id}/qemu/vms"
assert response.json["name"] == "PC TEST 1"
@@ -85,12 +86,12 @@ def test_qemu_create_platform(server, project, base_params, fake_qemu_bin):
assert response.json["platform"] == "x86_64"
def test_qemu_create_with_params(server, project, base_params, fake_qemu_vm):
def test_qemu_create_with_params(http_hypervisor, project, base_params, fake_qemu_vm):
params = base_params
params["ram"] = 1024
params["hda_disk_image"] = "linux载.img"
response = server.post("/projects/{project_id}/qemu/vms".format(project_id=project.id), params, example=True)
response = http_hypervisor.post("/projects/{project_id}/qemu/vms".format(project_id=project.id), params, example=True)
assert response.status == 201
assert response.route == "/projects/{project_id}/qemu/vms"
@@ -101,8 +102,8 @@ def test_qemu_create_with_params(server, project, base_params, fake_qemu_vm):
assert response.json["hda_disk_image_md5sum"] == "c4ca4238a0b923820dcc509a6f75849b"
def test_qemu_get(server, project, vm):
response = server.get("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
def test_qemu_get(http_hypervisor, project, vm):
response = http_hypervisor.get("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 200
assert response.route == "/projects/{project_id}/qemu/vms/{vm_id}"
assert response.json["name"] == "PC TEST 1"
@@ -110,57 +111,57 @@ def test_qemu_get(server, project, vm):
assert response.json["vm_directory"] == os.path.join(project.path, "project-files", "qemu", vm["vm_id"])
def test_qemu_start(server, vm):
def test_qemu_start(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.qemu.qemu_vm.QemuVM.start", return_value=True) as mock:
response = server.post("/projects/{project_id}/qemu/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/qemu/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 200
assert response.json["name"] == "PC TEST 1"
def test_qemu_stop(server, vm):
def test_qemu_stop(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.qemu.qemu_vm.QemuVM.stop", return_value=True) as mock:
response = server.post("/projects/{project_id}/qemu/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/qemu/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_qemu_reload(server, vm):
def test_qemu_reload(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.qemu.qemu_vm.QemuVM.reload", return_value=True) as mock:
response = server.post("/projects/{project_id}/qemu/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/qemu/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_qemu_suspend(server, vm):
def test_qemu_suspend(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.qemu.qemu_vm.QemuVM.suspend", return_value=True) as mock:
response = server.post("/projects/{project_id}/qemu/vms/{vm_id}/suspend".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/qemu/vms/{vm_id}/suspend".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_qemu_resume(server, vm):
def test_qemu_resume(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.qemu.qemu_vm.QemuVM.resume", return_value=True) as mock:
response = server.post("/projects/{project_id}/qemu/vms/{vm_id}/resume".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/qemu/vms/{vm_id}/resume".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_qemu_delete(server, vm):
def test_qemu_delete(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.qemu.Qemu.delete_vm", return_value=True) as mock:
response = server.delete("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.delete("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_qemu_update(server, vm, tmpdir, free_console_port, project, fake_qemu_vm):
def test_qemu_update(http_hypervisor, vm, tmpdir, free_console_port, project, fake_qemu_vm):
params = {
"name": "test",
"console": free_console_port,
"ram": 1024,
"hdb_disk_image": "linux.img"
}
response = server.put("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), params, example=True)
response = http_hypervisor.put("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), params, example=True)
assert response.status == 200
assert response.json["name"] == "test"
assert response.json["console"] == free_console_port
@@ -168,71 +169,71 @@ def test_qemu_update(server, vm, tmpdir, free_console_port, project, fake_qemu_v
assert response.json["ram"] == 1024
def test_qemu_nio_create_udp(server, vm):
server.put("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"adapters": 2})
response = server.post("/projects/{project_id}/qemu/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
def test_qemu_nio_create_udp(http_hypervisor, vm):
http_hypervisor.put("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"adapters": 2})
response = http_hypervisor.post("/projects/{project_id}/qemu/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
assert response.status == 201
assert response.route == "/projects/{project_id}/qemu/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
assert response.json["type"] == "nio_udp"
def test_qemu_nio_create_ethernet(server, vm):
server.put("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"adapters": 2})
response = server.post("/projects/{project_id}/qemu/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_generic_ethernet",
"ethernet_device": "eth0",
},
example=True)
def test_qemu_nio_create_ethernet(http_hypervisor, vm):
http_hypervisor.put("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"adapters": 2})
response = http_hypervisor.post("/projects/{project_id}/qemu/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_generic_ethernet",
"ethernet_device": "eth0",
},
example=True)
assert response.status == 409
def test_qemu_delete_nio(server, vm):
server.put("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"adapters": 2})
server.post("/projects/{project_id}/qemu/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"})
response = server.delete("/projects/{project_id}/qemu/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
def test_qemu_delete_nio(http_hypervisor, vm):
http_hypervisor.put("/projects/{project_id}/qemu/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"adapters": 2})
http_hypervisor.post("/projects/{project_id}/qemu/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"})
response = http_hypervisor.delete("/projects/{project_id}/qemu/vms/{vm_id}/adapters/1/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 204
assert response.route == "/projects/{project_id}/qemu/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
def test_qemu_list_binaries(server, vm):
def test_qemu_list_binaries(http_hypervisor, vm):
ret = [{"path": "/tmp/1", "version": "2.2.0"},
{"path": "/tmp/2", "version": "2.1.0"}]
with asyncio_patch("gns3server.modules.qemu.Qemu.binary_list", return_value=ret) as mock:
response = server.get("/qemu/binaries".format(project_id=vm["project_id"]), example=True)
response = http_hypervisor.get("/qemu/binaries".format(project_id=vm["project_id"]), example=True)
assert mock.called_with(None)
assert response.status == 200
assert response.json == ret
def test_qemu_list_binaries_filter(server, vm):
def test_qemu_list_binaries_filter(http_hypervisor, vm):
ret = [
{"path": "/tmp/x86_64", "version": "2.2.0"},
{"path": "/tmp/alpha", "version": "2.1.0"},
{"path": "/tmp/i386", "version": "2.1.0"}
]
with asyncio_patch("gns3server.modules.qemu.Qemu.binary_list", return_value=ret) as mock:
response = server.get("/qemu/binaries".format(project_id=vm["project_id"]), body={"archs": ["i386"]}, example=True)
response = http_hypervisor.get("/qemu/binaries".format(project_id=vm["project_id"]), body={"archs": ["i386"]}, example=True)
assert response.status == 200
assert mock.called_with(["i386"])
assert response.json == ret
def test_vms(server, tmpdir, fake_qemu_vm):
def test_vms(http_hypervisor, tmpdir, fake_qemu_vm):
response = server.get("/qemu/vms")
response = http_hypervisor.get("/qemu/vms")
assert response.status == 200
assert response.json == [{"filename": "linux载.img", "path": "linux载.img"}]
def test_upload_vm(server, tmpdir):
def test_upload_vm(http_hypervisor, tmpdir):
with patch("gns3server.modules.Qemu.get_images_directory", return_value=str(tmpdir),):
response = server.post("/qemu/vms/test2", body="TEST", raw=True)
response = http_hypervisor.post("/qemu/vms/test2", body="TEST", raw=True)
assert response.status == 204
with open(str(tmpdir / "test2")) as f:
@@ -243,9 +244,9 @@ def test_upload_vm(server, tmpdir):
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
def test_upload_vm_ova(server, tmpdir):
def test_upload_vm_ova(http_hypervisor, tmpdir):
with patch("gns3server.modules.Qemu.get_images_directory", return_value=str(tmpdir),):
response = server.post("/qemu/vms/test2.ova/test2.vmdk", body="TEST", raw=True)
response = http_hypervisor.post("/qemu/vms/test2.ova/test2.vmdk", body="TEST", raw=True)
assert response.status == 204
with open(str(tmpdir / "test2.ova" / "test2.vmdk")) as f:
@@ -256,23 +257,23 @@ def test_upload_vm_ova(server, tmpdir):
assert checksum == "033bd94b1168d7e4f0d644c3c95e35bf"
def test_upload_vm_forbiden_location(server, tmpdir):
def test_upload_vm_forbiden_location(http_hypervisor, tmpdir):
with patch("gns3server.modules.Qemu.get_images_directory", return_value=str(tmpdir),):
response = server.post("/qemu/vms/../../test2", body="TEST", raw=True)
response = http_hypervisor.post("/qemu/vms/../../test2", body="TEST", raw=True)
assert response.status == 403
def test_upload_vm_permission_denied(server, tmpdir):
def test_upload_vm_permission_denied(http_hypervisor, tmpdir):
with open(str(tmpdir / "test2"), "w+") as f:
f.write("")
os.chmod(str(tmpdir / "test2"), 0)
with patch("gns3server.modules.Qemu.get_images_directory", return_value=str(tmpdir),):
response = server.post("/qemu/vms/test2", body="TEST", raw=True)
response = http_hypervisor.post("/qemu/vms/test2", body="TEST", raw=True)
assert response.status == 409
def test_create_img_relative(server):
def test_create_img_relative(http_hypervisor):
body = {
"qemu_img": "/tmp/qemu-img",
"path": "hda.qcow2",
@@ -284,12 +285,12 @@ def test_create_img_relative(server):
"size": 100
}
with asyncio_patch("gns3server.modules.Qemu.create_disk"):
response = server.post("/qemu/img", body=body, example=True)
response = http_hypervisor.post("/qemu/img", body=body, example=True)
assert response.status == 201
def test_create_img_absolute_non_local(server):
def test_create_img_absolute_non_local(http_hypervisor):
config = Config.instance()
config.set("Server", "local", "false")
@@ -305,12 +306,12 @@ def test_create_img_absolute_non_local(server):
"size": 100
}
with asyncio_patch("gns3server.modules.Qemu.create_disk"):
response = server.post("/qemu/img", body=body, example=True)
response = http_hypervisor.post("/qemu/img", body=body, example=True)
assert response.status == 403
def test_create_img_absolute_local(server):
def test_create_img_absolute_local(http_hypervisor):
config = Config.instance()
config.set("Server", "local", "true")
@@ -326,12 +327,12 @@ def test_create_img_absolute_local(server):
"size": 100
}
with asyncio_patch("gns3server.modules.Qemu.create_disk"):
response = server.post("/qemu/img", body=body, example=True)
response = http_hypervisor.post("/qemu/img", body=body, example=True)
assert response.status == 201
def test_capabilities(server):
def test_capabilities(http_hypervisor):
with asyncio_patch("gns3server.modules.Qemu.get_kvm_archs", return_value=["x86_64"]):
response = server.get("/qemu/capabilities", example=True)
response = http_hypervisor.get("/qemu/capabilities", example=True)
assert response.json["kvm"] == ["x86_64"]
tests/handlers/api/test_virtualbox.py → tests/handlers/api/hypervisor/test_virtualbox.py
+34 -34
View File
@@ -21,14 +21,14 @@ from unittest.mock import patch
@pytest.yield_fixture(scope="function")
def vm(server, project, monkeypatch):
def vm(http_hypervisor, project, monkeypatch):
vboxmanage_path = "/fake/VboxManage"
with asyncio_patch("gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True) as mock:
response = server.post("/projects/{project_id}/virtualbox/vms".format(project_id=project.id), {"name": "VMTEST",
"vmname": "VMTEST",
"linked_clone": False})
response = http_hypervisor.post("/projects/{project_id}/virtualbox/vms".format(project_id=project.id), {"name": "VMTEST",
"vmname": "VMTEST",
"linked_clone": False})
assert mock.called
assert response.status == 201
@@ -36,71 +36,71 @@ def vm(server, project, monkeypatch):
yield response.json
def test_vbox_create(server, project):
def test_vbox_create(http_hypervisor, project):
with asyncio_patch("gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.create", return_value=True):
response = server.post("/projects/{project_id}/virtualbox/vms".format(project_id=project.id), {"name": "VM1",
"vmname": "VM1",
"linked_clone": False},
example=True)
response = http_hypervisor.post("/projects/{project_id}/virtualbox/vms".format(project_id=project.id), {"name": "VM1",
"vmname": "VM1",
"linked_clone": False},
example=True)
assert response.status == 201
assert response.json["name"] == "VM1"
assert response.json["project_id"] == project.id
def test_vbox_get(server, project, vm):
response = server.get("/projects/{project_id}/virtualbox/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
def test_vbox_get(http_hypervisor, project, vm):
response = http_hypervisor.get("/projects/{project_id}/virtualbox/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 200
assert response.route == "/projects/{project_id}/virtualbox/vms/{vm_id}"
assert response.json["name"] == "VMTEST"
assert response.json["project_id"] == project.id
def test_vbox_start(server, vm):
def test_vbox_start(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.check_hw_virtualization", return_value=True) as mock:
with asyncio_patch("gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.start", return_value=True) as mock:
response = server.post("/projects/{project_id}/virtualbox/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/virtualbox/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_vbox_stop(server, vm):
def test_vbox_stop(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.stop", return_value=True) as mock:
response = server.post("/projects/{project_id}/virtualbox/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/virtualbox/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_vbox_suspend(server, vm):
def test_vbox_suspend(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.suspend", return_value=True) as mock:
response = server.post("/projects/{project_id}/virtualbox/vms/{vm_id}/suspend".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/virtualbox/vms/{vm_id}/suspend".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_vbox_resume(server, vm):
def test_vbox_resume(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.resume", return_value=True) as mock:
response = server.post("/projects/{project_id}/virtualbox/vms/{vm_id}/resume".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/virtualbox/vms/{vm_id}/resume".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_vbox_reload(server, vm):
def test_vbox_reload(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.reload", return_value=True) as mock:
response = server.post("/projects/{project_id}/virtualbox/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/virtualbox/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_vbox_nio_create_udp(server, vm):
def test_vbox_nio_create_udp(http_hypervisor, vm):
with asyncio_patch('gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_add_nio_binding') as mock:
response = server.post("/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"],
vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
response = http_hypervisor.post("/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"],
vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
assert mock.called
args, kwgars = mock.call_args
@@ -111,10 +111,10 @@ def test_vbox_nio_create_udp(server, vm):
assert response.json["type"] == "nio_udp"
def test_vbox_delete_nio(server, vm):
def test_vbox_delete_nio(http_hypervisor, vm):
with asyncio_patch('gns3server.modules.virtualbox.virtualbox_vm.VirtualBoxVM.adapter_remove_nio_binding') as mock:
response = server.delete("/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.delete("/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
args, kwgars = mock.call_args
@@ -124,10 +124,10 @@ def test_vbox_delete_nio(server, vm):
assert response.route == "/projects/{project_id}/virtualbox/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
def test_vbox_update(server, vm, free_console_port):
response = server.put("/projects/{project_id}/virtualbox/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"name": "test",
"console": free_console_port},
example=True)
def test_vbox_update(http_hypervisor, vm, free_console_port):
response = http_hypervisor.put("/projects/{project_id}/virtualbox/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"name": "test",
"console": free_console_port},
example=True)
assert response.status == 200
assert response.json["name"] == "test"
assert response.json["console"] == free_console_port
tests/handlers/api/test_vpcs.py → tests/handlers/api/hypervisor/test_vpcs.py
+38 -38
View File
@@ -23,22 +23,22 @@ from unittest.mock import patch
@pytest.fixture(scope="function")
def vm(server, project):
response = server.post("/projects/{project_id}/vpcs/vms".format(project_id=project.id), {"name": "PC TEST 1"})
def vm(http_hypervisor, project):
response = http_hypervisor.post("/projects/{project_id}/vpcs/vms".format(project_id=project.id), {"name": "PC TEST 1"})
assert response.status == 201
return response.json
def test_vpcs_create(server, project):
response = server.post("/projects/{project_id}/vpcs/vms".format(project_id=project.id), {"name": "PC TEST 1"}, example=True)
def test_vpcs_create(http_hypervisor, project):
response = http_hypervisor.post("/projects/{project_id}/vpcs/vms".format(project_id=project.id), {"name": "PC TEST 1"}, example=True)
assert response.status == 201
assert response.route == "/projects/{project_id}/vpcs/vms"
assert response.json["name"] == "PC TEST 1"
assert response.json["project_id"] == project.id
def test_vpcs_get(server, project, vm):
response = server.get("/projects/{project_id}/vpcs/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
def test_vpcs_get(http_hypervisor, project, vm):
response = http_hypervisor.get("/projects/{project_id}/vpcs/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 200
assert response.route == "/projects/{project_id}/vpcs/vms/{vm_id}"
assert response.json["name"] == "PC TEST 1"
@@ -47,8 +47,8 @@ def test_vpcs_get(server, project, vm):
assert response.json["status"] == "stopped"
def test_vpcs_create_startup_script(server, project):
response = server.post("/projects/{project_id}/vpcs/vms".format(project_id=project.id), {"name": "PC TEST 1", "startup_script": "ip 192.168.1.2\necho TEST"})
def test_vpcs_create_startup_script(http_hypervisor, project):
response = http_hypervisor.post("/projects/{project_id}/vpcs/vms".format(project_id=project.id), {"name": "PC TEST 1", "startup_script": "ip 192.168.1.2\necho TEST"})
assert response.status == 201
assert response.route == "/projects/{project_id}/vpcs/vms"
assert response.json["name"] == "PC TEST 1"
@@ -57,8 +57,8 @@ def test_vpcs_create_startup_script(server, project):
assert response.json["startup_script_path"] == "startup.vpc"
def test_vpcs_create_port(server, project, free_console_port):
response = server.post("/projects/{project_id}/vpcs/vms".format(project_id=project.id), {"name": "PC TEST 1", "console": free_console_port})
def test_vpcs_create_port(http_hypervisor, project, free_console_port):
response = http_hypervisor.post("/projects/{project_id}/vpcs/vms".format(project_id=project.id), {"name": "PC TEST 1", "console": free_console_port})
assert response.status == 201
assert response.route == "/projects/{project_id}/vpcs/vms"
assert response.json["name"] == "PC TEST 1"
@@ -66,72 +66,72 @@ def test_vpcs_create_port(server, project, free_console_port):
assert response.json["console"] == free_console_port
def test_vpcs_nio_create_udp(server, vm):
response = server.post("/projects/{project_id}/vpcs/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
def test_vpcs_nio_create_udp(http_hypervisor, vm):
response = http_hypervisor.post("/projects/{project_id}/vpcs/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
example=True)
assert response.status == 201
assert response.route == "/projects/{project_id}/vpcs/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
assert response.json["type"] == "nio_udp"
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
def test_vpcs_nio_create_tap(server, vm, ethernet_device):
def test_vpcs_nio_create_tap(http_hypervisor, vm, ethernet_device):
with patch("gns3server.modules.base_manager.BaseManager.has_privileged_access", return_value=True):
response = server.post("/projects/{project_id}/vpcs/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_tap",
"tap_device": ethernet_device})
response = http_hypervisor.post("/projects/{project_id}/vpcs/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_tap",
"tap_device": ethernet_device})
assert response.status == 201
assert response.route == "/projects/{project_id}/vpcs/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
assert response.json["type"] == "nio_tap"
def test_vpcs_delete_nio(server, vm):
server.post("/projects/{project_id}/vpcs/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"})
response = server.delete("/projects/{project_id}/vpcs/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
def test_vpcs_delete_nio(http_hypervisor, vm):
http_hypervisor.post("/projects/{project_id}/vpcs/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"})
response = http_hypervisor.delete("/projects/{project_id}/vpcs/vms/{vm_id}/adapters/0/ports/0/nio".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert response.status == 204
assert response.route == "/projects/{project_id}/vpcs/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
def test_vpcs_start(server, vm):
def test_vpcs_start(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM.start", return_value=True) as mock:
response = server.post("/projects/{project_id}/vpcs/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/vpcs/vms/{vm_id}/start".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 200
assert response.json["name"] == "PC TEST 1"
def test_vpcs_stop(server, vm):
def test_vpcs_stop(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM.stop", return_value=True) as mock:
response = server.post("/projects/{project_id}/vpcs/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/vpcs/vms/{vm_id}/stop".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_vpcs_reload(server, vm):
def test_vpcs_reload(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM.reload", return_value=True) as mock:
response = server.post("/projects/{project_id}/vpcs/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.post("/projects/{project_id}/vpcs/vms/{vm_id}/reload".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_vpcs_delete(server, vm):
def test_vpcs_delete(http_hypervisor, vm):
with asyncio_patch("gns3server.modules.vpcs.VPCS.delete_vm", return_value=True) as mock:
response = server.delete("/projects/{project_id}/vpcs/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
response = http_hypervisor.delete("/projects/{project_id}/vpcs/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
assert mock.called
assert response.status == 204
def test_vpcs_update(server, vm, tmpdir, free_console_port):
response = server.put("/projects/{project_id}/vpcs/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"name": "test",
"console": free_console_port,
"startup_script": "ip 192.168.1.1"},
example=True)
def test_vpcs_update(http_hypervisor, vm, tmpdir, free_console_port):
response = http_hypervisor.put("/projects/{project_id}/vpcs/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), {"name": "test",
"console": free_console_port,
"startup_script": "ip 192.168.1.1"},
example=True)
assert response.status == 200
assert response.json["name"] == "test"
assert response.json["console"] == free_console_port
tests/handlers/api/test_version.py
+10 -10
View File
@@ -25,37 +25,37 @@ from gns3server.config import Config
from gns3server.version import __version__
def test_version_output(server):
def test_version_output(http_api):
config = Config.instance()
config.set("Server", "local", "true")
response = server.get('/version', example=True)
response = http_api.get('/version', example=True)
assert response.status == 200
assert response.json == {'local': True, 'version': __version__}
def test_version_input(server):
def test_version_input(http_api):
query = {'version': __version__}
response = server.post('/version', query, example=True)
response = http_api.post('/version', query, example=True)
assert response.status == 200
assert response.json == {'version': __version__}
def test_version_invalid_input(server):
def test_version_invalid_input(http_api):
query = {'version': "0.4.2"}
response = server.post('/version', query)
response = http_api.post('/version', query)
assert response.status == 409
assert response.json == {'message': 'Client version 0.4.2 differs with server version {}'.format(__version__),
'status': 409}
def test_version_invalid_input_schema(server):
def test_version_invalid_input_schema(http_api):
query = {'version': "0.4.2", "bla": "blu"}
response = server.post('/version', query)
response = http_api.post('/version', query)
assert response.status == 400
def test_version_invalid_json(server):
def test_version_invalid_json(http_api):
query = "BOUM"
response = server.post('/version', query, raw=True)
response = http_api.post('/version', query, raw=True)
assert response.status == 400
tests/handlers/test_index.py
+4 -4
View File
@@ -23,14 +23,14 @@ from unittest.mock import patch
from gns3server.version import __version__
def test_index(server):
response = server.get('/', api_version=None)
def test_index(http_root):
response = http_root.get('/')
assert response.status == 200
html = response.html
assert "Website" in html
assert __version__ in html
def test_status(server):
response = server.get('/status', api_version=None)
def test_status(http_root):
response = http_root.get('/status')
assert response.status == 200
tests/handlers/test_upload.py
+14 -14
View File
@@ -34,7 +34,7 @@ def restore_working_dir():
os.chdir(directory)
def test_index_upload(server, tmpdir):
def test_index_upload(http_root, tmpdir):
Config.instance().set("Server", "images_path", str(tmpdir))
@@ -42,7 +42,7 @@ def test_index_upload(server, tmpdir):
open(str(tmpdir / "alpha.md5sum"), "w+").close()
open(str(tmpdir / ".beta"), "w+").close()
response = server.get('/upload', api_version=None)
response = http_root.get('/upload')
assert response.status == 200
html = response.html
assert "GNS3 Server" in html
@@ -52,7 +52,7 @@ def test_index_upload(server, tmpdir):
assert "alpha.md5sum" not in html
def test_upload(server, tmpdir):
def test_upload(http_root, tmpdir):
content = ''.join(['a' for _ in range(0, 1025)])
@@ -64,7 +64,7 @@ def test_upload(server, tmpdir):
Config.instance().set("Server", "images_path", str(tmpdir))
response = server.post('/upload', api_version=None, body=body, raw=True)
response = http_root.post('/upload', body=body, raw=True)
assert "test2" in response.body.decode("utf-8")
@@ -76,7 +76,7 @@ def test_upload(server, tmpdir):
assert checksum == "ae187e1febee2a150b64849c32d566ca"
def test_upload_previous_checksum(server, tmpdir):
def test_upload_previous_checksum(http_root, tmpdir):
content = ''.join(['a' for _ in range(0, 1025)])
@@ -93,7 +93,7 @@ def test_upload_previous_checksum(server, tmpdir):
with open(str(tmpdir / "QEMU" / "test2.md5sum"), 'w+') as f:
f.write("FAKE checksum")
response = server.post('/upload', api_version=None, body=body, raw=True)
response = http_root.post('/upload', body=body, raw=True)
assert "test2" in response.body.decode("utf-8")
@@ -105,7 +105,7 @@ def test_upload_previous_checksum(server, tmpdir):
assert checksum == "ae187e1febee2a150b64849c32d566ca"
def test_upload_images_backup(server, tmpdir):
def test_upload_images_backup(http_root, tmpdir):
Config.instance().set("Server", "images_path", str(tmpdir / 'images'))
os.makedirs(str(tmpdir / 'images' / 'IOU'))
# An old IOU image that we need to replace
@@ -126,7 +126,7 @@ def test_upload_images_backup(server, tmpdir):
body = aiohttp.FormData()
body.add_field('type', 'IMAGES')
body.add_field('file', open(str(tmpdir / 'test.tar'), 'rb'), content_type='application/x-gtar', filename='test.tar')
response = server.post('/upload', api_version=None, body=body, raw=True)
response = http_root.post('/upload', body=body, raw=True)
assert response.status == 200
with open(str(tmpdir / 'images' / 'QEMU' / 'a.img')) as f:
@@ -139,7 +139,7 @@ def test_upload_images_backup(server, tmpdir):
assert not os.path.exists(str(tmpdir / 'images' / 'archive.tar'))
def test_upload_projects_backup(server, tmpdir):
def test_upload_projects_backup(http_root, tmpdir):
Config.instance().set("Server", "projects_path", str(tmpdir / 'projects'))
os.makedirs(str(tmpdir / 'projects' / 'b'))
# An old b image that we need to replace
@@ -160,7 +160,7 @@ def test_upload_projects_backup(server, tmpdir):
body = aiohttp.FormData()
body.add_field('type', 'PROJECTS')
body.add_field('file', open(str(tmpdir / 'test.tar'), 'rb'), content_type='application/x-gtar', filename='test.tar')
response = server.post('/upload', api_version=None, body=body, raw=True)
response = http_root.post('/upload', body=body, raw=True)
assert response.status == 200
with open(str(tmpdir / 'projects' / 'a' / 'a.img')) as f:
@@ -173,7 +173,7 @@ def test_upload_projects_backup(server, tmpdir):
assert not os.path.exists(str(tmpdir / 'projects' / 'archive.tar'))
def test_backup_images(server, tmpdir, loop):
def test_backup_images(http_root, tmpdir, loop):
Config.instance().set('Server', 'images_path', str(tmpdir))
os.makedirs(str(tmpdir / 'QEMU'))
@@ -182,7 +182,7 @@ def test_backup_images(server, tmpdir, loop):
with open(str(tmpdir / 'QEMU' / 'b.img'), 'w+') as f:
f.write('world')
response = server.get('/backup/images.tar', api_version=None, raw=True)
response = http_root.get('/backup/images.tar', raw=True)
assert response.status == 200
assert response.headers['CONTENT-TYPE'] == 'application/x-gtar'
@@ -204,7 +204,7 @@ def test_backup_images(server, tmpdir, loop):
assert open(os.path.join('QEMU', 'b.img')).read() == 'world'
def test_backup_projects(server, tmpdir, loop):
def test_backup_projects(http_root, tmpdir, loop):
Config.instance().set('Server', 'projects_path', str(tmpdir))
os.makedirs(str(tmpdir / 'a'))
@@ -214,7 +214,7 @@ def test_backup_projects(server, tmpdir, loop):
with open(str(tmpdir / 'b' / 'b.gns3'), 'w+') as f:
f.write('world')
response = server.get('/backup/projects.tar', api_version=None, raw=True)
response = http_root.get('/backup/projects.tar', raw=True)
assert response.status == 200
assert response.headers['CONTENT-TYPE'] == 'application/x-gtar'
tests/web/test_documentation.py
+7 -8
View File
@@ -24,20 +24,19 @@ from gns3server.web.route import Route
def test_documentation_write(tmpdir):
os.makedirs(str(tmpdir / "api/examples"))
with open(str(tmpdir / "api/examples/post_projectsprojectidvirtualboxvms.txt"), "w+") as f:
with open(str(tmpdir / "api/examples/hypervisor_post_projectsprojectidvirtualboxvms.txt"), "w+") as f:
f.write("curl test")
Documentation(Route, str(tmpdir)).write()
assert os.path.exists(str(tmpdir / "api"))
assert os.path.exists(str(tmpdir / "api" / "v1" / "server"))
assert os.path.exists(str(tmpdir / "api" / "v1" / "server" / "virtualbox.rst"))
assert os.path.exists(str(tmpdir / "api" / "v1" / "server" / "virtualbox"))
assert os.path.exists(str(tmpdir / "api" / "v1" / "server" / "virtualbox" / "virtualboxvms.rst"))
with open(str(tmpdir / "api" / "v1" / "server" / "virtualbox" / "projectsprojectidvirtualboxvms.rst")) as f:
assert os.path.exists(str(tmpdir / "api" / "v1" / "hypervisor"))
assert os.path.exists(str(tmpdir / "api" / "v1" / "hypervisor" / "virtualbox.rst"))
assert os.path.exists(str(tmpdir / "api" / "v1" / "hypervisor" / "virtualbox"))
assert os.path.exists(str(tmpdir / "api" / "v1" / "hypervisor" / "virtualbox" / "virtualboxvms.rst"))
with open(str(tmpdir / "api" / "v1" / "hypervisor" / "virtualbox" / "projectsprojectidvirtualboxvms.rst")) as f:
content = f.read()
assert "Sample session" in content
assert "literalinclude:: ../../../examples/post_projectsprojectidvirtualboxvms.txt" in content
assert "literalinclude:: ../../../examples/hypervisor_post_projectsprojectidvirtualboxvms.txt" in content
assert os.path.exists(str(tmpdir / "api" / "v1" / "controller" / "server.rst"))