mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-11-03 05:51:15 +02:00
Migration to FastAPI
This commit is contained in:
grossmj committed
1 parent
c12b675691
commit
eb3cb8a41f
224 files changed
+11254
-13517
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,26 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_appliances_list(controller_api):
|
||||
|
||||
response = await controller_api.get("/appliances/")
|
||||
assert response.status_code == 200
|
||||
assert len(response.json) > 0
|
||||
@@ -0,0 +1,281 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import pytest
|
||||
import unittest
|
||||
from tests.utils import asyncio_patch
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_create_without_id(controller_api, controller):
|
||||
|
||||
params = {
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure"}
|
||||
|
||||
response = await controller_api.post("/computes", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["user"] == "julien"
|
||||
assert response.json["compute_id"] is not None
|
||||
assert "password" not in response.json
|
||||
assert len(controller.computes) == 1
|
||||
assert controller.computes[response.json["compute_id"]].host == "localhost"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_create_with_id(controller_api, controller):
|
||||
|
||||
params = {
|
||||
"compute_id": "my_compute_id",
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure"}
|
||||
|
||||
response = await controller_api.post("/computes", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["user"] == "julien"
|
||||
assert "password" not in response.json
|
||||
assert len(controller.computes) == 1
|
||||
assert controller.computes["my_compute_id"].host == "localhost"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_get(controller_api):
|
||||
|
||||
params = {
|
||||
"compute_id": "my_compute_id",
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure"
|
||||
}
|
||||
|
||||
response = await controller_api.post("/computes", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
response = await controller_api.get("/computes/my_compute_id")
|
||||
assert response.status_code == 200
|
||||
print(response.json)
|
||||
#assert response.json["protocol"] == "http"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_update(controller_api):
|
||||
|
||||
params = {
|
||||
"compute_id": "my_compute_id",
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure"
|
||||
}
|
||||
|
||||
response = await controller_api.post("/computes", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
response = await controller_api.get("/computes/my_compute_id")
|
||||
assert response.status_code == 200
|
||||
assert response.json["protocol"] == "http"
|
||||
|
||||
params["protocol"] = "https"
|
||||
response = await controller_api.put("/computes/my_compute_id", params)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json["protocol"] == "https"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_list(controller_api):
|
||||
|
||||
params = {
|
||||
"compute_id": "my_compute_id",
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure",
|
||||
"name": "My super server"
|
||||
}
|
||||
|
||||
response = await controller_api.post("/computes", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["user"] == "julien"
|
||||
assert "password" not in response.json
|
||||
|
||||
response = await controller_api.get("/computes")
|
||||
for compute in response.json:
|
||||
if compute['compute_id'] != 'local':
|
||||
assert compute == {
|
||||
'compute_id': 'my_compute_id',
|
||||
'connected': False,
|
||||
'host': 'localhost',
|
||||
'port': 84,
|
||||
'protocol': 'http',
|
||||
'user': 'julien',
|
||||
'name': 'My super server',
|
||||
'cpu_usage_percent': 0.0,
|
||||
'memory_usage_percent': 0.0,
|
||||
'disk_usage_percent': 0.0,
|
||||
'last_error': None,
|
||||
'capabilities': {
|
||||
'version': '',
|
||||
'platform': '',
|
||||
'cpus': 0,
|
||||
'memory': 0,
|
||||
'disk_size': 0,
|
||||
'node_types': []
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_delete(controller_api):
|
||||
|
||||
params = {
|
||||
"compute_id": "my_compute_id",
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure"
|
||||
}
|
||||
response = await controller_api.post("/computes", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
response = await controller_api.get("/computes")
|
||||
assert len(response.json) == 1
|
||||
|
||||
response = await controller_api.delete("/computes/my_compute_id")
|
||||
assert response.status_code == 204
|
||||
|
||||
response = await controller_api.get("/computes")
|
||||
assert len(response.json) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_list_images(controller_api):
|
||||
|
||||
params = {
|
||||
"compute_id": "my_compute",
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure"
|
||||
}
|
||||
response = await controller_api.post("/computes", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
with asyncio_patch("gns3server.controller.compute.Compute.images", return_value=[{"filename": "linux.qcow2"}, {"filename": "asav.qcow2"}]) as mock:
|
||||
response = await controller_api.get("/computes/my_compute/qemu/images")
|
||||
assert response.json == [{"filename": "linux.qcow2"}, {"filename": "asav.qcow2"}]
|
||||
mock.assert_called_with("qemu")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_list_vms(controller_api):
|
||||
|
||||
params = {
|
||||
"compute_id": "my_compute",
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure"
|
||||
}
|
||||
response = await controller_api.post("/computes", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
with asyncio_patch("gns3server.controller.compute.Compute.forward", return_value=[]) as mock:
|
||||
response = await controller_api.get("/computes/my_compute/virtualbox/vms")
|
||||
mock.assert_called_with("GET", "virtualbox", "vms")
|
||||
assert response.json == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_create_img(controller_api):
|
||||
|
||||
params = {
|
||||
"compute_id": "my_compute",
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure"
|
||||
}
|
||||
|
||||
response = await controller_api.post("/computes", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
params = {"path": "/test"}
|
||||
with asyncio_patch("gns3server.controller.compute.Compute.forward", return_value=[]) as mock:
|
||||
response = await controller_api.post("/computes/my_compute/qemu/img", params)
|
||||
assert response.json == []
|
||||
mock.assert_called_with("POST", "qemu", "img", data=unittest.mock.ANY)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_compute_autoidlepc(controller_api):
|
||||
|
||||
params = {
|
||||
"compute_id": "my_compute_id",
|
||||
"protocol": "http",
|
||||
"host": "localhost",
|
||||
"port": 84,
|
||||
"user": "julien",
|
||||
"password": "secure"
|
||||
}
|
||||
|
||||
await controller_api.post("/computes", params)
|
||||
|
||||
params = {
|
||||
"platform": "c7200",
|
||||
"image": "test.bin",
|
||||
"ram": 512
|
||||
}
|
||||
|
||||
with asyncio_patch("gns3server.controller.Controller.autoidlepc", return_value={"idlepc": "0x606de20c"}) as mock:
|
||||
response = await controller_api.post("/computes/my_compute_id/auto_idlepc", params)
|
||||
assert mock.called
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
# FIXME
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_compute_endpoint(controller_api):
|
||||
#
|
||||
# params = {
|
||||
# "compute_id": "my_compute",
|
||||
# "protocol": "http",
|
||||
# "host": "localhost",
|
||||
# "port": 84,
|
||||
# "user": "julien",
|
||||
# "password": "secure"
|
||||
# }
|
||||
#
|
||||
# response = await controller_api.post("/computes", params)
|
||||
# assert response.status_code == 201
|
||||
#
|
||||
# response = await controller_api.get("/computes/endpoint/my_compute/qemu/images")
|
||||
# assert response.status_code == 200
|
||||
# assert response.json['endpoint'] == 'http://localhost:84/v2/compute/qemu/images'
|
||||
@@ -0,0 +1,67 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import os
|
||||
import pytest
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shutdown_local(controller_api, config):
|
||||
|
||||
os.kill = MagicMock()
|
||||
config.set("Server", "local", True)
|
||||
response = await controller_api.post('/shutdown')
|
||||
assert response.status_code == 204
|
||||
assert os.kill.called
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shutdown_non_local(controller_api, config):
|
||||
|
||||
config.set("Server", "local", False)
|
||||
response = await controller_api.post('/shutdown')
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_debug(controller_api, config, tmpdir):
|
||||
#
|
||||
# config._main_config_file = str(tmpdir / "test.conf")
|
||||
# config.set("Server", "local", True)
|
||||
# response = await controller_api.post('/debug')
|
||||
# assert response.status_code == 201
|
||||
# debug_dir = os.path.join(config.config_dir, "debug")
|
||||
# assert os.path.exists(debug_dir)
|
||||
# assert os.path.exists(os.path.join(debug_dir, "controller.txt"))
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_debug_non_local(controller_api, config, tmpdir):
|
||||
#
|
||||
# config._main_config_file = str(tmpdir / "test.conf")
|
||||
# config.set("Server", "local", False)
|
||||
# response = await controller_api.post('/debug')
|
||||
# assert response.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_statistics_output(controller_api):
|
||||
|
||||
response = await controller_api.get('/statistics')
|
||||
assert response.status_code == 200
|
||||
@@ -0,0 +1,92 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import pytest
|
||||
from gns3server.controller.drawing import Drawing
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_drawing(controller_api, project):
|
||||
|
||||
params = {
|
||||
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
|
||||
"x": 10,
|
||||
"y": 20,
|
||||
"z": 0
|
||||
}
|
||||
|
||||
response = await controller_api.post("/projects/{}/drawings".format(project.id), params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["drawing_id"] is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_drawing(controller_api, project):
|
||||
|
||||
params = {
|
||||
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
|
||||
"x": 10,
|
||||
"y": 20,
|
||||
"z": 0
|
||||
}
|
||||
|
||||
response = await controller_api.post("/projects/{}/drawings".format(project.id), params)
|
||||
response = await controller_api.get("/projects/{}/drawings/{}".format(project.id, response.json["drawing_id"]))
|
||||
assert response.status_code == 200
|
||||
assert response.json["x"] == 10
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_drawing(controller_api, project):
|
||||
|
||||
params = {
|
||||
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
|
||||
"x": 10,
|
||||
"y": 20,
|
||||
"z": 0
|
||||
}
|
||||
|
||||
response = await controller_api.post("/projects/{}/drawings".format(project.id), params)
|
||||
response = await controller_api.put("/projects/{}/drawings/{}".format(project.id, response.json["drawing_id"]), {"x": 42})
|
||||
assert response.status_code == 200
|
||||
assert response.json["x"] == 42
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_drawing(controller_api, project):
|
||||
|
||||
params = {
|
||||
"svg": '<svg height="210" width="500"><line x1="0" y1="0" x2="200" y2="200" style="stroke:rgb(255,0,0);stroke-width:2" /></svg>',
|
||||
"x": 10,
|
||||
"y": 20,
|
||||
"z": 0
|
||||
}
|
||||
|
||||
await controller_api.post("/projects/{}/drawings".format(project.id), params)
|
||||
response = await controller_api.get("/projects/{}/drawings".format(project.id))
|
||||
assert response.status_code == 200
|
||||
assert len(response.json) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_drawing(controller_api, project):
|
||||
|
||||
drawing = Drawing(project)
|
||||
project._drawings = {drawing.id: drawing}
|
||||
response = await controller_api.delete("/projects/{}/drawings/{}".format(project.id, drawing.id))
|
||||
assert response.status_code == 204
|
||||
assert drawing.id not in project.drawings
|
||||
@@ -0,0 +1,54 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import pytest
|
||||
from tests.utils import asyncio_patch
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_vms(controller_api):
|
||||
|
||||
with asyncio_patch("gns3server.controller.gns3vm.vmware_gns3_vm.VMwareGNS3VM.list", return_value=[{"vmname": "test"}]):
|
||||
response = await controller_api.get('/gns3vm/engines/vmware/vms')
|
||||
assert response.status_code == 200
|
||||
assert response.json == [
|
||||
{
|
||||
"vmname": "test"
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_engines(controller_api):
|
||||
|
||||
response = await controller_api.get('/gns3vm/engines')
|
||||
assert response.status_code == 200
|
||||
assert len(response.json) > 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_put_gns3vm(controller_api):
|
||||
|
||||
response = await controller_api.put('/gns3vm', {"vmname": "TEST VM"})
|
||||
assert response.status_code == 200
|
||||
assert response.json["vmname"] == "TEST VM"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_gns3vm(controller_api):
|
||||
response = await controller_api.get('/gns3vm')
|
||||
assert response.status_code == 200
|
||||
@@ -0,0 +1,382 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import pytest
|
||||
|
||||
from unittest.mock import patch, MagicMock
|
||||
from tests.utils import asyncio_patch, AsyncioMagicMock
|
||||
|
||||
from gns3server.controller.ports.ethernet_port import EthernetPort
|
||||
from gns3server.controller.link import Link, FILTERS
|
||||
from gns3server.controller.udp_link import UDPLink
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@pytest.mark.asyncio
|
||||
async def nodes(compute, project):
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
node1 = await project.add_node(compute, "node1", None, node_type="qemu")
|
||||
node1._ports = [EthernetPort("E0", 0, 0, 3)]
|
||||
node2 = await project.add_node(compute, "node2", None, node_type="qemu")
|
||||
node2._ports = [EthernetPort("E0", 0, 2, 4)]
|
||||
return node1, node2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_link(controller_api, project, nodes):
|
||||
|
||||
node1, node2 = nodes
|
||||
|
||||
filters = {
|
||||
"latency": [10],
|
||||
"frequency_drop": [50]
|
||||
}
|
||||
|
||||
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
|
||||
response = await controller_api.post("/projects/{}/links".format(project.id), {
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": node1.id,
|
||||
"adapter_number": 0,
|
||||
"port_number": 3,
|
||||
"label": {
|
||||
"text": "Text",
|
||||
"x": 42,
|
||||
"y": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"node_id": node2.id,
|
||||
"adapter_number": 2,
|
||||
"port_number": 4
|
||||
}
|
||||
],
|
||||
"filters": filters
|
||||
})
|
||||
|
||||
assert mock.called
|
||||
assert response.status_code == 201
|
||||
assert response.json["link_id"] is not None
|
||||
assert len(response.json["nodes"]) == 2
|
||||
assert response.json["nodes"][0]["label"]["x"] == 42
|
||||
assert len(project.links) == 1
|
||||
assert list(project.links.values())[0].filters == filters
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_link_failure(controller_api, compute, project):
|
||||
"""
|
||||
Make sure the link is deleted if we failed to create it.
|
||||
|
||||
The failure is triggered by connecting the link to itself
|
||||
"""
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
node1 = await project.add_node(compute, "node1", None, node_type="qemu")
|
||||
node1._ports = [EthernetPort("E0", 0, 0, 3), EthernetPort("E0", 0, 0, 4)]
|
||||
|
||||
response = await controller_api.post("/projects/{}/links".format(project.id), {
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": node1.id,
|
||||
"adapter_number": 0,
|
||||
"port_number": 3,
|
||||
"label": {
|
||||
"text": "Text",
|
||||
"x": 42,
|
||||
"y": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"node_id": node1.id,
|
||||
"adapter_number": 0,
|
||||
"port_number": 4
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
assert response.status_code == 409
|
||||
assert len(project.links) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_link(controller_api, project, nodes):
|
||||
|
||||
node1, node2 = nodes
|
||||
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
|
||||
response = await controller_api.post("/projects/{}/links".format(project.id), {
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": node1.id,
|
||||
"adapter_number": 0,
|
||||
"port_number": 3,
|
||||
"label": {
|
||||
"text": "Text",
|
||||
"x": 42,
|
||||
"y": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"node_id": node2.id,
|
||||
"adapter_number": 2,
|
||||
"port_number": 4
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
assert mock.called
|
||||
link_id = response.json["link_id"]
|
||||
assert response.json["nodes"][0]["label"]["x"] == 42
|
||||
response = await controller_api.get("/projects/{}/links/{}".format(project.id, link_id))
|
||||
assert response.status_code == 200
|
||||
assert response.json["nodes"][0]["label"]["x"] == 42
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_link_suspend(controller_api, project, nodes):
|
||||
|
||||
node1, node2 = nodes
|
||||
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
|
||||
response = await controller_api.post("/projects/{}/links".format(project.id), {
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": node1.id,
|
||||
"adapter_number": 0,
|
||||
"port_number": 3,
|
||||
"label": {
|
||||
"text": "Text",
|
||||
"x": 42,
|
||||
"y": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"node_id": node2.id,
|
||||
"adapter_number": 2,
|
||||
"port_number": 4
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
assert mock.called
|
||||
link_id = response.json["link_id"]
|
||||
assert response.json["nodes"][0]["label"]["x"] == 42
|
||||
|
||||
response = await controller_api.put("/projects/{}/links/{}".format(project.id, link_id), {
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": node1.id,
|
||||
"adapter_number": 0,
|
||||
"port_number": 3,
|
||||
"label": {
|
||||
"text": "Hello",
|
||||
"x": 64,
|
||||
"y": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"node_id": node2.id,
|
||||
"adapter_number": 2,
|
||||
"port_number": 4
|
||||
}
|
||||
],
|
||||
"suspend": True
|
||||
})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json["nodes"][0]["label"]["x"] == 64
|
||||
assert response.json["suspend"]
|
||||
assert response.json["filters"] == {}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_link(controller_api, project, nodes):
|
||||
|
||||
filters = {
|
||||
"latency": [10],
|
||||
"frequency_drop": [50]
|
||||
}
|
||||
|
||||
node1, node2 = nodes
|
||||
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
|
||||
response = await controller_api.post("/projects/{}/links".format(project.id), {
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": node1.id,
|
||||
"adapter_number": 0,
|
||||
"port_number": 3,
|
||||
"label": {
|
||||
"text": "Text",
|
||||
"x": 42,
|
||||
"y": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"node_id": node2.id,
|
||||
"adapter_number": 2,
|
||||
"port_number": 4
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
assert mock.called
|
||||
link_id = response.json["link_id"]
|
||||
assert response.json["nodes"][0]["label"]["x"] == 42
|
||||
|
||||
response = await controller_api.put("/projects/{}/links/{}".format(project.id, link_id), {
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": node1.id,
|
||||
"adapter_number": 0,
|
||||
"port_number": 3,
|
||||
"label": {
|
||||
"text": "Hello",
|
||||
"x": 64,
|
||||
"y": 0
|
||||
}
|
||||
},
|
||||
{
|
||||
"node_id": node2.id,
|
||||
"adapter_number": 2,
|
||||
"port_number": 4
|
||||
}
|
||||
],
|
||||
"filters": filters
|
||||
})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json["nodes"][0]["label"]["x"] == 64
|
||||
assert list(project.links.values())[0].filters == filters
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_link(controller_api, project, nodes):
|
||||
|
||||
filters = {
|
||||
"latency": [10],
|
||||
"frequency_drop": [50]
|
||||
}
|
||||
|
||||
node1, node2 = nodes
|
||||
nodes = [
|
||||
{
|
||||
"node_id": node1.id,
|
||||
"adapter_number": 0,
|
||||
"port_number": 3
|
||||
},
|
||||
{
|
||||
"node_id": node2.id,
|
||||
"adapter_number": 2,
|
||||
"port_number": 4
|
||||
}
|
||||
]
|
||||
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as mock:
|
||||
await controller_api.post("/projects/{}/links".format(project.id), {
|
||||
"nodes": nodes,
|
||||
"filters": filters
|
||||
})
|
||||
|
||||
assert mock.called
|
||||
response = await controller_api.get("/projects/{}/links".format(project.id))
|
||||
assert response.status_code == 200
|
||||
assert len(response.json) == 1
|
||||
assert response.json[0]["filters"] == filters
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reset_link(controller_api, project):
|
||||
|
||||
link = UDPLink(project)
|
||||
project._links = {link.id: link}
|
||||
with asyncio_patch("gns3server.controller.udp_link.UDPLink.delete") as delete_mock:
|
||||
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create") as create_mock:
|
||||
response = await controller_api.post("/projects/{}/links/{}/reset".format(project.id, link.id))
|
||||
assert delete_mock.called
|
||||
assert create_mock.called
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_capture(controller_api, project):
|
||||
|
||||
link = Link(project)
|
||||
project._links = {link.id: link}
|
||||
with asyncio_patch("gns3server.controller.link.Link.start_capture") as mock:
|
||||
response = await controller_api.post("/projects/{}/links/{}/start_capture".format(project.id, link.id))
|
||||
assert mock.called
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_capture(controller_api, project):
|
||||
|
||||
link = Link(project)
|
||||
project._links = {link.id: link}
|
||||
with asyncio_patch("gns3server.controller.link.Link.stop_capture") as mock:
|
||||
response = await controller_api.post("/projects/{}/links/{}/stop_capture".format(project.id, link.id))
|
||||
assert mock.called
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
# async def test_pcap(controller_api, http_client, project):
|
||||
#
|
||||
# async def pcap_capture():
|
||||
# async with http_client.get(controller_api.get_url("/projects/{}/links/{}/pcap".format(project.id, link.id))) as response:
|
||||
# response.body = await response.content.read(5)
|
||||
# print("READ", response.body)
|
||||
# return response
|
||||
#
|
||||
# with asyncio_patch("gns3server.controller.link.Link.capture_node") as mock:
|
||||
# link = Link(project)
|
||||
# link._capture_file_name = "test"
|
||||
# link._capturing = True
|
||||
# with open(link.capture_file_path, "w+") as f:
|
||||
# f.write("hello")
|
||||
# project._links = {link.id: link}
|
||||
# response = await pcap_capture()
|
||||
# assert mock.called
|
||||
# assert response.status_code == 200
|
||||
# assert b'hello' == response.body
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_link(controller_api, project):
|
||||
|
||||
link = Link(project)
|
||||
project._links = {link.id: link}
|
||||
with asyncio_patch("gns3server.controller.link.Link.delete") as mock:
|
||||
response = await controller_api.delete("/projects/{}/links/{}".format(project.id, link.id))
|
||||
assert mock.called
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_filters(controller_api, project):
|
||||
|
||||
link = Link(project)
|
||||
project._links = {link.id: link}
|
||||
with patch("gns3server.controller.link.Link.available_filters", return_value=FILTERS) as mock:
|
||||
response = await controller_api.get("/projects/{}/links/{}/available_filters".format(project.id, link.id))
|
||||
assert mock.called
|
||||
assert response.status_code == 200
|
||||
assert response.json == FILTERS
|
||||
@@ -0,0 +1,283 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
|
||||
import pytest
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
from tests.utils import AsyncioMagicMock
|
||||
|
||||
from gns3server.controller.node import Node
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def node(project, compute):
|
||||
|
||||
node = Node(project, compute, "test", node_type="vpcs")
|
||||
project._nodes[node.id] = node
|
||||
return node
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_node(controller_api, project, compute):
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
response = await controller_api.post("/projects/{}/nodes".format(project.id), {
|
||||
"name": "test",
|
||||
"node_type": "vpcs",
|
||||
"compute_id": "example.com",
|
||||
"properties": {
|
||||
"startup_script": "echo test"
|
||||
}
|
||||
})
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.json["name"] == "test"
|
||||
assert "name" not in response.json["properties"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_node(controller_api, project, compute):
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
await controller_api.post("/projects/{}/nodes".format(project.id), {
|
||||
"name": "test",
|
||||
"node_type": "vpcs",
|
||||
"compute_id": "example.com",
|
||||
"properties": {
|
||||
"startup_script": "echo test"
|
||||
}
|
||||
})
|
||||
|
||||
response = await controller_api.get("/projects/{}/nodes".format(project.id))
|
||||
assert response.status_code == 200
|
||||
assert response.json[0]["name"] == "test"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_node(controller_api, project, compute):
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
response = await controller_api.post("/projects/{}/nodes".format(project.id), {
|
||||
"name": "test",
|
||||
"node_type": "vpcs",
|
||||
"compute_id": "example.com",
|
||||
"properties": {
|
||||
"startup_script": "echo test"
|
||||
}
|
||||
})
|
||||
|
||||
response = await controller_api.get("/projects/{}/nodes/{}".format(project.id, response.json["node_id"]))
|
||||
assert response.status_code == 200
|
||||
assert response.json["name"] == "test"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_node(controller_api, project, compute, node):
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
compute.put = AsyncioMagicMock(return_value=response)
|
||||
|
||||
response = await controller_api.put("/projects/{}/nodes/{}".format(project.id, node.id), {
|
||||
"name": "test",
|
||||
"node_type": "vpcs",
|
||||
"compute_id": "example.com",
|
||||
"properties": {
|
||||
"startup_script": "echo test"
|
||||
}
|
||||
})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json["name"] == "test"
|
||||
assert "name" not in response.json["properties"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_all_nodes(controller_api, project, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{}/nodes/start".format(project.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_all_nodes(controller_api, project, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{}/nodes/stop".format(project.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suspend_all_nodes(controller_api, project, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{}/nodes/suspend".format(project.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reload_all_nodes(controller_api, project, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{}/nodes/reload".format(project.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reset_console_all_nodes(controller_api, project, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{}/nodes/console/reset".format(project.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_node(controller_api, project, node, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{}/nodes/{}/start".format(project.id, node.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_node(controller_api, project, node, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{}/nodes/{}/stop".format(project.id, node.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suspend_node(controller_api, project, node, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{}/nodes/{}/suspend".format(project.id, node.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reload_node(controller_api, project, node, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{}/nodes/{}/reload".format(project.id, node.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_duplicate_node(controller_api, project, compute, node):
|
||||
|
||||
response = MagicMock()
|
||||
response.json({"console": 2035})
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
response = await controller_api.post("/projects/{}/nodes/{}/duplicate".format(project.id, node.id),
|
||||
{"x": 10,
|
||||
"y": 5,
|
||||
"z": 0})
|
||||
assert response.status_code == 201, response.body.decode()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_node(controller_api, project, node, compute):
|
||||
|
||||
compute.post = AsyncioMagicMock()
|
||||
response = await controller_api.delete("/projects/{}/nodes/{}".format(project.id, node.id))
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dynamips_idle_pc(controller_api, project, compute, node):
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"idlepc": "0x60606f54"}
|
||||
compute.get = AsyncioMagicMock(return_value=response)
|
||||
|
||||
response = await controller_api.get("/projects/{}/nodes/{}/dynamips/auto_idlepc".format(project.id, node.id))
|
||||
assert response.status_code == 200
|
||||
assert response.json["idlepc"] == "0x60606f54"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dynamips_idlepc_proposals(controller_api, project, compute, node):
|
||||
|
||||
response = MagicMock()
|
||||
response.json = ["0x60606f54", "0x33805a22"]
|
||||
compute.get = AsyncioMagicMock(return_value=response)
|
||||
|
||||
response = await controller_api.get("/projects/{}/nodes/{}/dynamips/idlepc_proposals".format(project.id, node.id))
|
||||
assert response.status_code == 200
|
||||
assert response.json == ["0x60606f54", "0x33805a22"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file(controller_api, project, node, compute):
|
||||
|
||||
response = MagicMock()
|
||||
response.body = b"world"
|
||||
compute.http_query = AsyncioMagicMock(return_value=response)
|
||||
|
||||
response = await controller_api.get("/projects/{project_id}/nodes/{node_id}/files/hello".format(project_id=project.id, node_id=node.id))
|
||||
assert response.status_code == 200
|
||||
assert response.content == b'world'
|
||||
|
||||
compute.http_query.assert_called_with("GET", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(project_id=project.id, node_id=node.id), timeout=None, raw=True)
|
||||
|
||||
response = await controller_api.get("/projects/{project_id}/nodes/{node_id}/files/../hello".format(project_id=project.id, node_id=node.id))
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_file(controller_api, project, node, compute):
|
||||
|
||||
compute.http_query = AsyncioMagicMock()
|
||||
response = await controller_api.post("/projects/{project_id}/nodes/{node_id}/files/hello".format(project_id=project.id, node_id=node.id), body=b"hello", raw=True)
|
||||
assert response.status_code == 201
|
||||
|
||||
compute.http_query.assert_called_with("POST", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello".format(project_id=project.id, node_id=node.id), data=b'hello', timeout=None, raw=True)
|
||||
|
||||
response = await controller_api.get("/projects/{project_id}/nodes/{node_id}/files/../hello".format(project_id=project.id, node_id=node.id))
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_get_and_post_with_nested_paths_normalization(controller_api, project, node, compute):
|
||||
#
|
||||
# response = MagicMock()
|
||||
# response.body = b"world"
|
||||
# compute.http_query = AsyncioMagicMock(return_value=response)
|
||||
# response = await controller_api.get("/projects/{project_id}/nodes/{node_id}/files/hello\\nested".format(project_id=project.id, node_id=node.id))
|
||||
# assert response.status_code == 200
|
||||
# assert response.content == b'world'
|
||||
#
|
||||
# compute.http_query.assert_called_with("GET", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello/nested".format(project_id=project.id, node_id=node.id), timeout=None, raw=True)
|
||||
#
|
||||
# compute.http_query = AsyncioMagicMock()
|
||||
# response = await controller_api.post("/projects/{project_id}/nodes/{node_id}/files/hello\\nested".format(project_id=project.id, node_id=node.id), body=b"hello", raw=True)
|
||||
# assert response.status_code == 201
|
||||
#
|
||||
# compute.http_query.assert_called_with("POST", "/projects/{project_id}/files/project-files/vpcs/{node_id}/hello/nested".format(project_id=project.id, node_id=node.id), data=b'hello', timeout=None, raw=True)
|
||||
@@ -0,0 +1,388 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import uuid
|
||||
import os
|
||||
import pytest
|
||||
import zipfile
|
||||
import json
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
from unittest.mock import patch, MagicMock
|
||||
from tests.utils import asyncio_patch
|
||||
from gns3server.app import app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@pytest.mark.asyncio
|
||||
async def project(controller_api, controller):
|
||||
|
||||
u = str(uuid.uuid4())
|
||||
params = {"name": "test", "project_id": u}
|
||||
await controller_api.post("/projects", params)
|
||||
return controller.get_project(u)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_project_with_path(controller_api, tmpdir):
|
||||
|
||||
response = await controller_api.post("/projects", {"name": "test", "path": str(tmpdir), "project_id": "00010203-0405-0607-0809-0a0b0c0d0e0f"})
|
||||
assert response.status_code == 201
|
||||
assert response.json["name"] == "test"
|
||||
assert response.json["project_id"] == "00010203-0405-0607-0809-0a0b0c0d0e0f"
|
||||
assert response.json["status"] == "opened"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_project_without_dir(controller_api):
|
||||
|
||||
params = {"name": "test", "project_id": "10010203-0405-0607-0809-0a0b0c0d0e0f"}
|
||||
response = await controller_api.post("/projects", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["project_id"] == "10010203-0405-0607-0809-0a0b0c0d0e0f"
|
||||
assert response.json["name"] == "test"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_project_with_uuid(controller_api):
|
||||
|
||||
params = {"name": "test", "project_id": "30010203-0405-0607-0809-0a0b0c0d0e0f"}
|
||||
response = await controller_api.post("/projects", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["project_id"] == "30010203-0405-0607-0809-0a0b0c0d0e0f"
|
||||
assert response.json["name"] == "test"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_project_with_variables(controller_api):
|
||||
|
||||
variables = [
|
||||
{"name": "TEST1"},
|
||||
{"name": "TEST2", "value": "value1"}
|
||||
]
|
||||
params = {"name": "test", "project_id": "30010203-0405-0607-0809-0a0b0c0d0e0f", "variables": variables}
|
||||
response = await controller_api.post("/projects", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["variables"] == [
|
||||
{"name": "TEST1"},
|
||||
{"name": "TEST2", "value": "value1"}
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_project_with_supplier(controller_api):
|
||||
|
||||
supplier = {
|
||||
'logo': 'logo.png',
|
||||
'url': 'http://example.com'
|
||||
}
|
||||
params = {"name": "test", "project_id": "30010203-0405-0607-0809-0a0b0c0d0e0f", "supplier": supplier}
|
||||
response = await controller_api.post("/projects", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["supplier"] == supplier
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_project(controller_api):
|
||||
|
||||
params = {"name": "test", "project_id": "10010203-0405-0607-0809-0a0b0c0d0e0f"}
|
||||
response = await controller_api.post("/projects", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["project_id"] == "10010203-0405-0607-0809-0a0b0c0d0e0f"
|
||||
assert response.json["name"] == "test"
|
||||
|
||||
params = {"name": "test2"}
|
||||
response = await controller_api.put("/projects/10010203-0405-0607-0809-0a0b0c0d0e0f", params)
|
||||
assert response.status_code == 200
|
||||
assert response.json["name"] == "test2"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_project_with_variables(controller_api):
|
||||
|
||||
variables = [
|
||||
{"name": "TEST1"},
|
||||
{"name": "TEST2", "value": "value1"}
|
||||
]
|
||||
params = {"name": "test", "project_id": "10010203-0405-0607-0809-0a0b0c0d0e0f", "variables": variables}
|
||||
response = await controller_api.post("/projects", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
params = {"name": "test2"}
|
||||
response = await controller_api.put("/projects/10010203-0405-0607-0809-0a0b0c0d0e0f", params)
|
||||
assert response.status_code == 200
|
||||
assert response.json["variables"] == variables
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_projects(controller_api, tmpdir):
|
||||
|
||||
await controller_api.post("/projects", {"name": "test", "path": str(tmpdir), "project_id": "00010203-0405-0607-0809-0a0b0c0d0e0f"})
|
||||
response = await controller_api.get("/projects")
|
||||
assert response.status_code == 200
|
||||
projects = response.json
|
||||
assert projects[0]["name"] == "test"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_project(controller_api, project):
|
||||
|
||||
response = await controller_api.get("/projects/{project_id}".format(project_id=project.id))
|
||||
assert response.status_code == 200
|
||||
assert response.json["name"] == "test"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_project(controller_api, project, controller):
|
||||
|
||||
with asyncio_patch("gns3server.controller.project.Project.delete", return_value=True) as mock:
|
||||
response = await controller_api.delete("/projects/{project_id}".format(project_id=project.id))
|
||||
assert response.status_code == 204
|
||||
assert mock.called
|
||||
assert project not in controller.projects
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_project_invalid_uuid(controller_api):
|
||||
|
||||
response = await controller_api.delete("/projects/{project_id}".format(project_id=uuid.uuid4()))
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_close_project(controller_api, project):
|
||||
|
||||
with asyncio_patch("gns3server.controller.project.Project.close", return_value=True) as mock:
|
||||
response = await controller_api.post("/projects/{project_id}/close".format(project_id=project.id))
|
||||
assert response.status_code == 204
|
||||
assert mock.called
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_open_project(controller_api, project):
|
||||
|
||||
with asyncio_patch("gns3server.controller.project.Project.open", return_value=True) as mock:
|
||||
response = await controller_api.post("/projects/{project_id}/open".format(project_id=project.id))
|
||||
assert response.status_code == 201
|
||||
assert mock.called
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_load_project(controller_api, project, config):
|
||||
|
||||
config.set("Server", "local", "true")
|
||||
with asyncio_patch("gns3server.controller.Controller.load_project", return_value=project) as mock:
|
||||
response = await controller_api.post("/projects/load", {"path": "/tmp/test.gns3"})
|
||||
assert response.status_code == 201
|
||||
mock.assert_called_with("/tmp/test.gns3")
|
||||
assert response.json["project_id"] == project.id
|
||||
|
||||
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_notification(controller_api, http_client, project, controller):
|
||||
#
|
||||
# async with http_client.get(controller_api.get_url("/projects/{project_id}/notifications".format(project_id=project.id))) as response:
|
||||
# response.body = await response.content.read(200)
|
||||
# controller.notification.project_emit("node.created", {"a": "b"})
|
||||
# response.body += await response.content.readany()
|
||||
# assert response.status_code == 200
|
||||
# assert b'"action": "ping"' in response.body
|
||||
# assert b'"cpu_usage_percent"' in response.body
|
||||
# assert b'{"action": "node.created", "event": {"a": "b"}}\n' in response.body
|
||||
# assert project.status_code == "opened"
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_notification_invalid_id(controller_api):
|
||||
#
|
||||
# response = await controller_api.get("/projects/{project_id}/notifications".format(project_id=uuid.uuid4()))
|
||||
# assert response.status_code == 404
|
||||
|
||||
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_notification_ws(controller_api, http_client, controller, project):
|
||||
#
|
||||
# ws = await http_client.ws_connect(controller_api.get_url("/projects/{project_id}/notifications/ws".format(project_id=project.id)))
|
||||
# answer = await ws.receive()
|
||||
# answer = json.loads(answer.data)
|
||||
# assert answer["action"] == "ping"
|
||||
#
|
||||
# controller.notification.project_emit("test", {})
|
||||
# answer = await ws.receive()
|
||||
# answer = json.loads(answer.data)
|
||||
# assert answer["action"] == "test"
|
||||
#
|
||||
# if not ws.closed:
|
||||
# await ws.close()
|
||||
#
|
||||
# assert project.status_code == "opened"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_export_with_images(controller_api, tmpdir, project):
|
||||
|
||||
project.dump = MagicMock()
|
||||
os.makedirs(project.path, exist_ok=True)
|
||||
with open(os.path.join(project.path, 'a'), 'w+') as f:
|
||||
f.write('hello')
|
||||
|
||||
os.makedirs(str(tmpdir / "IOS"))
|
||||
with open(str(tmpdir / "IOS" / "test.image"), "w+") as f:
|
||||
f.write("AAA")
|
||||
|
||||
topology = {
|
||||
"topology": {
|
||||
"nodes": [
|
||||
{
|
||||
"properties": {
|
||||
"image": "test.image"
|
||||
},
|
||||
"node_type": "dynamips"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
with open(os.path.join(project.path, "test.gns3"), 'w+') as f:
|
||||
json.dump(topology, f)
|
||||
|
||||
with patch("gns3server.compute.Dynamips.get_images_directory", return_value=str(tmpdir / "IOS")):
|
||||
response = await controller_api.get("/projects/{project_id}/export?include_images=yes".format(project_id=project.id))
|
||||
assert response.status_code == 200
|
||||
assert response.headers['CONTENT-TYPE'] == 'application/gns3project'
|
||||
assert response.headers['CONTENT-DISPOSITION'] == 'attachment; filename="{}.gns3project"'.format(project.name)
|
||||
|
||||
with open(str(tmpdir / 'project.zip'), 'wb+') as f:
|
||||
f.write(response.content)
|
||||
|
||||
with zipfile.ZipFile(str(tmpdir / 'project.zip')) as myzip:
|
||||
with myzip.open("a") as myfile:
|
||||
content = myfile.read()
|
||||
assert content == b"hello"
|
||||
myzip.getinfo("images/IOS/test.image")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_export_without_images(controller_api, tmpdir, project):
|
||||
|
||||
project.dump = MagicMock()
|
||||
os.makedirs(project.path, exist_ok=True)
|
||||
with open(os.path.join(project.path, 'a'), 'w+') as f:
|
||||
f.write('hello')
|
||||
|
||||
os.makedirs(str(tmpdir / "IOS"))
|
||||
with open(str(tmpdir / "IOS" / "test.image"), "w+") as f:
|
||||
f.write("AAA")
|
||||
|
||||
topology = {
|
||||
"topology": {
|
||||
"nodes": [
|
||||
{
|
||||
"properties": {
|
||||
"image": "test.image"
|
||||
},
|
||||
"node_type": "dynamips"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
with open(os.path.join(project.path, "test.gns3"), 'w+') as f:
|
||||
json.dump(topology, f)
|
||||
|
||||
with patch("gns3server.compute.Dynamips.get_images_directory", return_value=str(tmpdir / "IOS"),):
|
||||
response = await controller_api.get("/projects/{project_id}/export?include_images=0".format(project_id=project.id))
|
||||
assert response.status_code == 200
|
||||
assert response.headers['CONTENT-TYPE'] == 'application/gns3project'
|
||||
assert response.headers['CONTENT-DISPOSITION'] == 'attachment; filename="{}.gns3project"'.format(project.name)
|
||||
|
||||
with open(str(tmpdir / 'project.zip'), 'wb+') as f:
|
||||
f.write(response.content)
|
||||
|
||||
with zipfile.ZipFile(str(tmpdir / 'project.zip')) as myzip:
|
||||
with myzip.open("a") as myfile:
|
||||
content = myfile.read()
|
||||
assert content == b"hello"
|
||||
# Image should not exported
|
||||
with pytest.raises(KeyError):
|
||||
myzip.getinfo("images/IOS/test.image")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file(controller_api, project):
|
||||
|
||||
os.makedirs(project.path, exist_ok=True)
|
||||
with open(os.path.join(project.path, 'hello'), 'w+') as f:
|
||||
f.write('world')
|
||||
|
||||
response = await controller_api.get("/projects/{project_id}/files/hello".format(project_id=project.id))
|
||||
assert response.status_code == 200
|
||||
assert response.content == b"world"
|
||||
|
||||
response = await controller_api.get("/projects/{project_id}/files/false".format(project_id=project.id))
|
||||
assert response.status_code == 404
|
||||
|
||||
response = await controller_api.get("/projects/{project_id}/files/../hello".format(project_id=project.id))
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_write_file(controller_api, project):
|
||||
|
||||
response = await controller_api.post("/projects/{project_id}/files/hello".format(project_id=project.id), body="world", raw=True)
|
||||
assert response.status_code == 204
|
||||
|
||||
with open(os.path.join(project.path, "hello")) as f:
|
||||
assert f.read() == "world"
|
||||
|
||||
response = await controller_api.post("/projects/{project_id}/files/../hello".format(project_id=project.id), raw=True)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_write_and_get_file_with_leading_slashes_in_filename(controller_api, project):
|
||||
|
||||
response = await controller_api.post("/projects/{project_id}/files//hello".format(project_id=project.id), body="world", raw=True)
|
||||
assert response.status_code == 204
|
||||
|
||||
response = await controller_api.get("/projects/{project_id}/files//hello".format(project_id=project.id), raw=True)
|
||||
assert response.status_code == 200
|
||||
assert response.content == b"world"
|
||||
|
||||
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_import(controller_api, tmpdir, controller):
|
||||
#
|
||||
# with zipfile.ZipFile(str(tmpdir / "test.zip"), 'w') as myzip:
|
||||
# myzip.writestr("project.gns3", b'{"project_id": "c6992992-ac72-47dc-833b-54aa334bcd05", "version": "2.0.0", "name": "test"}')
|
||||
# myzip.writestr("demo", b"hello")
|
||||
#
|
||||
# project_id = str(uuid.uuid4())
|
||||
# with open(str(tmpdir / "test.zip"), "rb") as f:
|
||||
# response = await controller_api.post("/projects/{project_id}/import".format(project_id=project_id), body=f.read(), raw=True)
|
||||
# assert response.status_code == 201
|
||||
#
|
||||
# project = controller.get_project(project_id)
|
||||
# with open(os.path.join(project.path, "demo")) as f:
|
||||
# content = f.read()
|
||||
# assert content == "hello"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_duplicate(controller_api, project):
|
||||
|
||||
response = await controller_api.post("/projects/{project_id}/duplicate".format(project_id=project.id), {"name": "hello"})
|
||||
assert response.status_code == 201
|
||||
assert response.json["name"] == "hello"
|
||||
@@ -0,0 +1,72 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import os
|
||||
import uuid
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@pytest.mark.asyncio
|
||||
async def project(controller_api, controller):
|
||||
|
||||
u = str(uuid.uuid4())
|
||||
params = {"name": "test", "project_id": u}
|
||||
await controller_api.post("/projects", params)
|
||||
project = controller.get_project(u)
|
||||
return project
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@pytest.mark.asyncio
|
||||
async def snapshot(project):
|
||||
|
||||
snapshot = await project.snapshot("test")
|
||||
return snapshot
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_snapshots(controller_api, project, snapshot):
|
||||
|
||||
assert snapshot.name == "test"
|
||||
response = await controller_api.get("/projects/{}/snapshots".format(project.id))
|
||||
assert response.status_code == 200
|
||||
assert len(response.json) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_snapshot(controller_api, project, snapshot):
|
||||
|
||||
response = await controller_api.delete("/projects/{}/snapshots/{}".format(project.id, snapshot.id))
|
||||
assert response.status_code == 204
|
||||
assert not os.path.exists(snapshot.path)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restore_snapshot(controller_api, project, snapshot):
|
||||
|
||||
response = await controller_api.post("/projects/{}/snapshots/{}/restore".format(project.id, snapshot.id))
|
||||
assert response.status_code == 201
|
||||
assert response.json["name"] == project.name
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_snapshot(controller_api, project):
|
||||
|
||||
response = await controller_api.post("/projects/{}/snapshots".format(project.id), {"name": "snap1"})
|
||||
assert response.status_code == 201
|
||||
assert len(os.listdir(os.path.join(project.path, "snapshots"))) == 1
|
||||
@@ -0,0 +1,63 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
|
||||
import pytest
|
||||
import os
|
||||
import urllib.parse
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_symbols(controller_api):
|
||||
|
||||
response = await controller_api.get('/symbols')
|
||||
|
||||
assert response.status_code == 200
|
||||
assert {
|
||||
'symbol_id': ':/symbols/classic/firewall.svg',
|
||||
'filename': 'firewall.svg',
|
||||
'builtin': True,
|
||||
'theme': 'Classic'
|
||||
} in response.json
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get(controller_api, controller):
|
||||
|
||||
controller.symbols.theme = "Classic"
|
||||
response = await controller_api.get('/symbols/' + urllib.parse.quote(':/symbols/classic/firewall.svg') + '/raw')
|
||||
assert response.status_code == 200
|
||||
assert response.headers['CONTENT-TYPE'] == 'image/svg+xml'
|
||||
assert response.headers['CONTENT-LENGTH'] == '9381'
|
||||
assert '</svg>' in response.text
|
||||
|
||||
# Reply with the default symbol
|
||||
response = await controller_api.get('/symbols/404.png/raw')
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload(controller_api, symbols_dir):
|
||||
|
||||
response = await controller_api.post("/symbols/test2/raw", body=b"TEST", raw=True)
|
||||
assert response.status_code == 204
|
||||
|
||||
with open(os.path.join(symbols_dir, "test2")) as f:
|
||||
assert f.read() == "TEST"
|
||||
|
||||
response = await controller_api.get('/symbols/test2/raw')
|
||||
assert response.status_code == 200
|
||||
@@ -0,0 +1,982 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import pytest
|
||||
import uuid
|
||||
|
||||
from tests.utils import asyncio_patch
|
||||
|
||||
from gns3server.controller.template import Template
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_template_list(controller_api, controller):
|
||||
|
||||
id = str(uuid.uuid4())
|
||||
controller.template_manager.load_templates()
|
||||
controller.template_manager._templates[id] = Template(id, {
|
||||
"template_type": "qemu",
|
||||
"category": 0,
|
||||
"name": "test",
|
||||
"symbol": "guest.svg",
|
||||
"default_name_format": "{name}-{0}",
|
||||
"compute_id": "local"
|
||||
})
|
||||
response = await controller_api.get("/templates")
|
||||
assert response.status_code == 200
|
||||
assert len(response.json) > 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_template_create_without_id(controller_api, controller):
|
||||
|
||||
params = {"base_script_file": "vpcs_base_config.txt",
|
||||
"category": "guest",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "PC{0}",
|
||||
"name": "VPCS_TEST",
|
||||
"compute_id": "local",
|
||||
"symbol": ":/symbols/vpcs_guest.svg",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
assert len(controller.template_manager.templates) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_template_create_with_id(controller_api, controller):
|
||||
|
||||
params = {"template_id": str(uuid.uuid4()),
|
||||
"base_script_file": "vpcs_base_config.txt",
|
||||
"category": "guest",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "PC{0}",
|
||||
"name": "VPCS_TEST",
|
||||
"compute_id": "local",
|
||||
"symbol": ":/symbols/vpcs_guest.svg",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
assert len(controller.template_manager.templates) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_template_create_wrong_type(controller_api, controller):
|
||||
|
||||
params = {"template_id": str(uuid.uuid4()),
|
||||
"base_script_file": "vpcs_base_config.txt",
|
||||
"category": "guest",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "PC{0}",
|
||||
"name": "VPCS_TEST",
|
||||
"compute_id": "local",
|
||||
"symbol": ":/symbols/vpcs_guest.svg",
|
||||
"template_type": "invalid_template_type"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 422
|
||||
assert len(controller.template_manager.templates) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_template_get(controller_api):
|
||||
|
||||
template_id = str(uuid.uuid4())
|
||||
params = {"template_id": template_id,
|
||||
"base_script_file": "vpcs_base_config.txt",
|
||||
"category": "guest",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "PC{0}",
|
||||
"name": "VPCS_TEST",
|
||||
"compute_id": "local",
|
||||
"symbol": ":/symbols/vpcs_guest.svg",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
response = await controller_api.get("/templates/{}".format(template_id))
|
||||
assert response.status_code == 200
|
||||
assert response.json["template_id"] == template_id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_template_update(controller_api):
|
||||
|
||||
template_id = str(uuid.uuid4())
|
||||
params = {"template_id": template_id,
|
||||
"base_script_file": "vpcs_base_config.txt",
|
||||
"category": "guest",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "PC{0}",
|
||||
"name": "VPCS_TEST",
|
||||
"compute_id": "local",
|
||||
"symbol": ":/symbols/vpcs_guest.svg",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
response = await controller_api.get("/templates/{}".format(template_id))
|
||||
assert response.status_code == 200
|
||||
assert response.json["template_id"] == template_id
|
||||
|
||||
params["name"] = "VPCS_TEST_RENAMED"
|
||||
response = await controller_api.put("/templates/{}".format(template_id), params)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json["name"] == "VPCS_TEST_RENAMED"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_template_delete(controller_api, controller):
|
||||
|
||||
template_id = str(uuid.uuid4())
|
||||
params = {"template_id": template_id,
|
||||
"base_script_file": "vpcs_base_config.txt",
|
||||
"category": "guest",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "PC{0}",
|
||||
"name": "VPCS_TEST",
|
||||
"compute_id": "local",
|
||||
"symbol": ":/symbols/vpcs_guest.svg",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
response = await controller_api.get("/templates")
|
||||
assert len(response.json) == 1
|
||||
assert len(controller.template_manager._templates) == 1
|
||||
|
||||
response = await controller_api.delete("/templates/{}".format(template_id))
|
||||
assert response.status_code == 204
|
||||
|
||||
response = await controller_api.get("/templates")
|
||||
assert len(response.json) == 0
|
||||
assert len(controller.template_manager.templates) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_template_duplicate(controller_api, controller):
|
||||
|
||||
template_id = str(uuid.uuid4())
|
||||
params = {"template_id": template_id,
|
||||
"base_script_file": "vpcs_base_config.txt",
|
||||
"category": "guest",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "PC{0}",
|
||||
"name": "VPCS_TEST",
|
||||
"compute_id": "local",
|
||||
"symbol": ":/symbols/vpcs_guest.svg",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
|
||||
response = await controller_api.post("/templates/{}/duplicate".format(template_id))
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] != template_id
|
||||
params.pop("template_id")
|
||||
for param, value in params.items():
|
||||
assert response.json[param] == value
|
||||
|
||||
response = await controller_api.get("/templates")
|
||||
assert len(response.json) == 2
|
||||
assert len(controller.template_manager.templates) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c7200_dynamips_template_create(controller_api):
|
||||
|
||||
params = {"name": "Cisco c7200 template",
|
||||
"platform": "c7200",
|
||||
"compute_id": "local",
|
||||
"image": "c7200-adventerprisek9-mz.124-24.T5.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "dynamips",
|
||||
"auto_delete_disks": False,
|
||||
"builtin": False,
|
||||
"category": "router",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "R{0}",
|
||||
"disk0": 0,
|
||||
"disk1": 0,
|
||||
"exec_area": 64,
|
||||
"idlemax": 500,
|
||||
"idlepc": "",
|
||||
"idlesleep": 30,
|
||||
"image": "c7200-adventerprisek9-mz.124-24.T5.image",
|
||||
"mac_addr": "",
|
||||
"midplane": "vxr",
|
||||
"mmap": True,
|
||||
"name": "Cisco c7200 template",
|
||||
"npe": "npe-400",
|
||||
"nvram": 512,
|
||||
"platform": "c7200",
|
||||
"private_config": "",
|
||||
"ram": 512,
|
||||
"sparsemem": True,
|
||||
"startup_config": "ios_base_startup-config.txt",
|
||||
"symbol": ":/symbols/router.svg",
|
||||
"system_id": "FTX0945W0MY"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c3745_dynamips_template_create(controller_api):
|
||||
|
||||
params = {"name": "Cisco c3745 template",
|
||||
"platform": "c3745",
|
||||
"compute_id": "local",
|
||||
"image": "c3745-adventerprisek9-mz.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "dynamips",
|
||||
"auto_delete_disks": False,
|
||||
"builtin": False,
|
||||
"category": "router",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "R{0}",
|
||||
"disk0": 0,
|
||||
"disk1": 0,
|
||||
"exec_area": 64,
|
||||
"idlemax": 500,
|
||||
"idlepc": "",
|
||||
"idlesleep": 30,
|
||||
"image": "c3745-adventerprisek9-mz.124-25d.image",
|
||||
"mac_addr": "",
|
||||
"mmap": True,
|
||||
"name": "Cisco c3745 template",
|
||||
"iomem": 5,
|
||||
"nvram": 256,
|
||||
"platform": "c3745",
|
||||
"private_config": "",
|
||||
"ram": 256,
|
||||
"sparsemem": True,
|
||||
"startup_config": "ios_base_startup-config.txt",
|
||||
"symbol": ":/symbols/router.svg",
|
||||
"system_id": "FTX0945W0MY"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c3725_dynamips_template_create(controller_api):
|
||||
|
||||
params = {"name": "Cisco c3725 template",
|
||||
"platform": "c3725",
|
||||
"compute_id": "local",
|
||||
"image": "c3725-adventerprisek9-mz.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "dynamips",
|
||||
"auto_delete_disks": False,
|
||||
"builtin": False,
|
||||
"category": "router",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "R{0}",
|
||||
"disk0": 0,
|
||||
"disk1": 0,
|
||||
"exec_area": 64,
|
||||
"idlemax": 500,
|
||||
"idlepc": "",
|
||||
"idlesleep": 30,
|
||||
"image": "c3725-adventerprisek9-mz.124-25d.image",
|
||||
"mac_addr": "",
|
||||
"mmap": True,
|
||||
"name": "Cisco c3725 template",
|
||||
"iomem": 5,
|
||||
"nvram": 256,
|
||||
"platform": "c3725",
|
||||
"private_config": "",
|
||||
"ram": 128,
|
||||
"sparsemem": True,
|
||||
"startup_config": "ios_base_startup-config.txt",
|
||||
"symbol": ":/symbols/router.svg",
|
||||
"system_id": "FTX0945W0MY"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c3600_dynamips_template_create(controller_api):
|
||||
|
||||
params = {"name": "Cisco c3600 template",
|
||||
"platform": "c3600",
|
||||
"chassis": "3660",
|
||||
"compute_id": "local",
|
||||
"image": "c3660-a3jk9s-mz.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "dynamips",
|
||||
"auto_delete_disks": False,
|
||||
"builtin": False,
|
||||
"category": "router",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "R{0}",
|
||||
"disk0": 0,
|
||||
"disk1": 0,
|
||||
"exec_area": 64,
|
||||
"idlemax": 500,
|
||||
"idlepc": "",
|
||||
"idlesleep": 30,
|
||||
"image": "c3660-a3jk9s-mz.124-25d.image",
|
||||
"mac_addr": "",
|
||||
"mmap": True,
|
||||
"name": "Cisco c3600 template",
|
||||
"iomem": 5,
|
||||
"nvram": 128,
|
||||
"platform": "c3600",
|
||||
"chassis": "3660",
|
||||
"private_config": "",
|
||||
"ram": 192,
|
||||
"sparsemem": True,
|
||||
"startup_config": "ios_base_startup-config.txt",
|
||||
"symbol": ":/symbols/router.svg",
|
||||
"system_id": "FTX0945W0MY"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c3600_dynamips_template_create_wrong_chassis(controller_api):
|
||||
|
||||
params = {"name": "Cisco c3600 template",
|
||||
"platform": "c3600",
|
||||
"chassis": "3650",
|
||||
"compute_id": "local",
|
||||
"image": "c3660-a3jk9s-mz.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c2691_dynamips_template_create(controller_api):
|
||||
|
||||
params = {"name": "Cisco c2691 template",
|
||||
"platform": "c2691",
|
||||
"compute_id": "local",
|
||||
"image": "c2691-adventerprisek9-mz.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "dynamips",
|
||||
"auto_delete_disks": False,
|
||||
"builtin": False,
|
||||
"category": "router",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "R{0}",
|
||||
"disk0": 0,
|
||||
"disk1": 0,
|
||||
"exec_area": 64,
|
||||
"idlemax": 500,
|
||||
"idlepc": "",
|
||||
"idlesleep": 30,
|
||||
"image": "c2691-adventerprisek9-mz.124-25d.image",
|
||||
"mac_addr": "",
|
||||
"mmap": True,
|
||||
"name": "Cisco c2691 template",
|
||||
"iomem": 5,
|
||||
"nvram": 256,
|
||||
"platform": "c2691",
|
||||
"private_config": "",
|
||||
"ram": 192,
|
||||
"sparsemem": True,
|
||||
"startup_config": "ios_base_startup-config.txt",
|
||||
"symbol": ":/symbols/router.svg",
|
||||
"system_id": "FTX0945W0MY"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c2600_dynamips_template_create(controller_api):
|
||||
|
||||
params = {"name": "Cisco c2600 template",
|
||||
"platform": "c2600",
|
||||
"chassis": "2651XM",
|
||||
"compute_id": "local",
|
||||
"image": "c2600-adventerprisek9-mz.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "dynamips",
|
||||
"auto_delete_disks": False,
|
||||
"builtin": False,
|
||||
"category": "router",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "R{0}",
|
||||
"disk0": 0,
|
||||
"disk1": 0,
|
||||
"exec_area": 64,
|
||||
"idlemax": 500,
|
||||
"idlepc": "",
|
||||
"idlesleep": 30,
|
||||
"image": "c2600-adventerprisek9-mz.124-25d.image",
|
||||
"mac_addr": "",
|
||||
"mmap": True,
|
||||
"name": "Cisco c2600 template",
|
||||
"iomem": 15,
|
||||
"nvram": 128,
|
||||
"platform": "c2600",
|
||||
"chassis": "2651XM",
|
||||
"private_config": "",
|
||||
"ram": 160,
|
||||
"sparsemem": True,
|
||||
"startup_config": "ios_base_startup-config.txt",
|
||||
"symbol": ":/symbols/router.svg",
|
||||
"system_id": "FTX0945W0MY"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c2600_dynamips_template_create_wrong_chassis(controller_api):
|
||||
|
||||
params = {"name": "Cisco c2600 template",
|
||||
"platform": "c2600",
|
||||
"chassis": "2660XM",
|
||||
"compute_id": "local",
|
||||
"image": "c2600-adventerprisek9-mz.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c1700_dynamips_template_create(controller_api):
|
||||
|
||||
params = {"name": "Cisco c1700 template",
|
||||
"platform": "c1700",
|
||||
"chassis": "1760",
|
||||
"compute_id": "local",
|
||||
"image": "c1700-adventerprisek9-mz.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "dynamips",
|
||||
"auto_delete_disks": False,
|
||||
"builtin": False,
|
||||
"category": "router",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "R{0}",
|
||||
"disk0": 0,
|
||||
"disk1": 0,
|
||||
"exec_area": 64,
|
||||
"idlemax": 500,
|
||||
"idlepc": "",
|
||||
"idlesleep": 30,
|
||||
"image": "c1700-adventerprisek9-mz.124-25d.image",
|
||||
"mac_addr": "",
|
||||
"mmap": True,
|
||||
"name": "Cisco c1700 template",
|
||||
"iomem": 15,
|
||||
"nvram": 128,
|
||||
"platform": "c1700",
|
||||
"chassis": "1760",
|
||||
"private_config": "",
|
||||
"ram": 160,
|
||||
"sparsemem": False,
|
||||
"startup_config": "ios_base_startup-config.txt",
|
||||
"symbol": ":/symbols/router.svg",
|
||||
"system_id": "FTX0945W0MY"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_c1700_dynamips_template_create_wrong_chassis(controller_api):
|
||||
|
||||
params = {"name": "Cisco c1700 template",
|
||||
"platform": "c1700",
|
||||
"chassis": "1770",
|
||||
"compute_id": "local",
|
||||
"image": "c1700-adventerprisek9-mz.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dynamips_template_create_wrong_platform(controller_api):
|
||||
|
||||
params = {"name": "Cisco c3900 template",
|
||||
"platform": "c3900",
|
||||
"compute_id": "local",
|
||||
"image": "c3900-test.124-25d.image",
|
||||
"template_type": "dynamips"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_iou_template_create(controller_api):
|
||||
|
||||
params = {"name": "IOU template",
|
||||
"compute_id": "local",
|
||||
"path": "/path/to/i86bi_linux-ipbase-ms-12.4.bin",
|
||||
"template_type": "iou"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "iou",
|
||||
"builtin": False,
|
||||
"category": "router",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "IOU{0}",
|
||||
"ethernet_adapters": 2,
|
||||
"name": "IOU template",
|
||||
"nvram": 128,
|
||||
"path": "/path/to/i86bi_linux-ipbase-ms-12.4.bin",
|
||||
"private_config": "",
|
||||
"ram": 256,
|
||||
"serial_adapters": 2,
|
||||
"startup_config": "iou_l3_base_startup-config.txt",
|
||||
"symbol": ":/symbols/multilayer_switch.svg",
|
||||
"use_default_iou_values": True,
|
||||
"l1_keepalives": False}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_docker_template_create(controller_api):
|
||||
|
||||
params = {"name": "Docker template",
|
||||
"compute_id": "local",
|
||||
"image": "gns3/endhost:latest",
|
||||
"template_type": "docker"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"adapters": 1,
|
||||
"template_type": "docker",
|
||||
"builtin": False,
|
||||
"category": "guest",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_http_path": "/",
|
||||
"console_http_port": 80,
|
||||
"console_resolution": "1024x768",
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "{name}-{0}",
|
||||
"environment": "",
|
||||
"extra_hosts": "",
|
||||
"image": "gns3/endhost:latest",
|
||||
"name": "Docker template",
|
||||
"start_command": "",
|
||||
"symbol": ":/symbols/docker_guest.svg",
|
||||
"custom_adapters": []}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_qemu_template_create(controller_api):
|
||||
|
||||
params = {"name": "Qemu template",
|
||||
"compute_id": "local",
|
||||
"platform": "i386",
|
||||
"hda_disk_image": "IOSvL2-15.2.4.0.55E.qcow2",
|
||||
"ram": 512,
|
||||
"template_type": "qemu"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"adapter_type": "e1000",
|
||||
"adapters": 1,
|
||||
"template_type": "qemu",
|
||||
"bios_image": "",
|
||||
"boot_priority": "c",
|
||||
"builtin": False,
|
||||
"category": "guest",
|
||||
"cdrom_image": "",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"cpu_throttling": 0,
|
||||
"cpus": 1,
|
||||
"default_name_format": "{name}-{0}",
|
||||
"first_port_name": "",
|
||||
"hda_disk_image": "IOSvL2-15.2.4.0.55E.qcow2",
|
||||
"hda_disk_interface": "none",
|
||||
"hdb_disk_image": "",
|
||||
"hdb_disk_interface": "none",
|
||||
"hdc_disk_image": "",
|
||||
"hdc_disk_interface": "none",
|
||||
"hdd_disk_image": "",
|
||||
"hdd_disk_interface": "none",
|
||||
"initrd": "",
|
||||
"kernel_command_line": "",
|
||||
"kernel_image": "",
|
||||
"legacy_networking": False,
|
||||
"linked_clone": True,
|
||||
"mac_address": "",
|
||||
"name": "Qemu template",
|
||||
"on_close": "power_off",
|
||||
"options": "",
|
||||
"platform": "i386",
|
||||
"port_name_format": "Ethernet{0}",
|
||||
"port_segment_size": 0,
|
||||
"process_priority": "normal",
|
||||
"qemu_path": "",
|
||||
"ram": 512,
|
||||
"symbol": ":/symbols/qemu_guest.svg",
|
||||
"usage": "",
|
||||
"custom_adapters": []}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_vmware_template_create(controller_api):
|
||||
|
||||
params = {"name": "VMware template",
|
||||
"compute_id": "local",
|
||||
"template_type": "vmware",
|
||||
"vmx_path": "/path/to/vm.vmx"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"adapter_type": "e1000",
|
||||
"adapters": 1,
|
||||
"template_type": "vmware",
|
||||
"builtin": False,
|
||||
"category": "guest",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "none",
|
||||
"default_name_format": "{name}-{0}",
|
||||
"first_port_name": "",
|
||||
"headless": False,
|
||||
"linked_clone": False,
|
||||
"name": "VMware template",
|
||||
"on_close": "power_off",
|
||||
"port_name_format": "Ethernet{0}",
|
||||
"port_segment_size": 0,
|
||||
"symbol": ":/symbols/vmware_guest.svg",
|
||||
"use_any_adapter": False,
|
||||
"vmx_path": "/path/to/vm.vmx",
|
||||
"custom_adapters": []}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_virtualbox_template_create(controller_api):
|
||||
|
||||
params = {"name": "VirtualBox template",
|
||||
"compute_id": "local",
|
||||
"template_type": "virtualbox",
|
||||
"vmname": "My VirtualBox VM"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"adapter_type": "Intel PRO/1000 MT Desktop (82540EM)",
|
||||
"adapters": 1,
|
||||
"template_type": "virtualbox",
|
||||
"builtin": False,
|
||||
"category": "guest",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "none",
|
||||
"default_name_format": "{name}-{0}",
|
||||
"first_port_name": "",
|
||||
"headless": False,
|
||||
"linked_clone": False,
|
||||
"name": "VirtualBox template",
|
||||
"on_close": "power_off",
|
||||
"port_name_format": "Ethernet{0}",
|
||||
"port_segment_size": 0,
|
||||
"ram": 256,
|
||||
"symbol": ":/symbols/vbox_guest.svg",
|
||||
"use_any_adapter": False,
|
||||
"vmname": "My VirtualBox VM",
|
||||
"custom_adapters": []}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_vpcs_template_create(controller_api):
|
||||
|
||||
params = {"name": "VPCS template",
|
||||
"compute_id": "local",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "vpcs",
|
||||
"base_script_file": "vpcs_base_config.txt",
|
||||
"builtin": False,
|
||||
"category": "guest",
|
||||
"compute_id": "local",
|
||||
"console_auto_start": False,
|
||||
"console_type": "telnet",
|
||||
"default_name_format": "PC{0}",
|
||||
"name": "VPCS template",
|
||||
"symbol": ":/symbols/vpcs_guest.svg"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ethernet_switch_template_create(controller_api):
|
||||
|
||||
params = {"name": "Ethernet switch template",
|
||||
"compute_id": "local",
|
||||
"template_type": "ethernet_switch"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "ethernet_switch",
|
||||
"builtin": False,
|
||||
"category": "switch",
|
||||
"compute_id": "local",
|
||||
"console_type": "none",
|
||||
"default_name_format": "Switch{0}",
|
||||
"name": "Ethernet switch template",
|
||||
"ports_mapping": [{"ethertype": "",
|
||||
"name": "Ethernet0",
|
||||
"port_number": 0,
|
||||
"type": "access",
|
||||
"vlan": 1
|
||||
},
|
||||
{"ethertype": "",
|
||||
"name": "Ethernet1",
|
||||
"port_number": 1,
|
||||
"type": "access",
|
||||
"vlan": 1
|
||||
},
|
||||
{"ethertype": "",
|
||||
"name": "Ethernet2",
|
||||
"port_number": 2,
|
||||
"type": "access",
|
||||
"vlan": 1
|
||||
},
|
||||
{"ethertype": "",
|
||||
"name": "Ethernet3",
|
||||
"port_number": 3,
|
||||
"type": "access",
|
||||
"vlan": 1
|
||||
},
|
||||
{"ethertype": "",
|
||||
"name": "Ethernet4",
|
||||
"port_number": 4,
|
||||
"type": "access",
|
||||
"vlan": 1
|
||||
},
|
||||
{"ethertype": "",
|
||||
"name": "Ethernet5",
|
||||
"port_number": 5,
|
||||
"type": "access",
|
||||
"vlan": 1
|
||||
},
|
||||
{"ethertype": "",
|
||||
"name": "Ethernet6",
|
||||
"port_number": 6,
|
||||
"type": "access",
|
||||
"vlan": 1
|
||||
},
|
||||
{"ethertype": "",
|
||||
"name": "Ethernet7",
|
||||
"port_number": 7,
|
||||
"type": "access",
|
||||
"vlan": 1
|
||||
}],
|
||||
"symbol": ":/symbols/ethernet_switch.svg"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cloud_template_create(controller_api):
|
||||
|
||||
params = {"name": "Cloud template",
|
||||
"compute_id": "local",
|
||||
"template_type": "cloud"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"template_type": "cloud",
|
||||
"builtin": False,
|
||||
"category": "guest",
|
||||
"compute_id": "local",
|
||||
"default_name_format": "Cloud{0}",
|
||||
"name": "Cloud template",
|
||||
"symbol": ":/symbols/cloud.svg"}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ethernet_hub_template_create(controller_api):
|
||||
|
||||
params = {"name": "Ethernet hub template",
|
||||
"compute_id": "local",
|
||||
"template_type": "ethernet_hub"}
|
||||
|
||||
response = await controller_api.post("/templates", params)
|
||||
assert response.status_code == 201
|
||||
assert response.json["template_id"] is not None
|
||||
|
||||
expected_response = {"ports_mapping": [{"port_number": 0,
|
||||
"name": "Ethernet0"
|
||||
},
|
||||
{"port_number": 1,
|
||||
"name": "Ethernet1"
|
||||
},
|
||||
{"port_number": 2,
|
||||
"name": "Ethernet2"
|
||||
},
|
||||
{"port_number": 3,
|
||||
"name": "Ethernet3"
|
||||
},
|
||||
{"port_number": 4,
|
||||
"name": "Ethernet4"
|
||||
},
|
||||
{"port_number": 5,
|
||||
"name": "Ethernet5"
|
||||
},
|
||||
{"port_number": 6,
|
||||
"name": "Ethernet6"
|
||||
},
|
||||
{"port_number": 7,
|
||||
"name": "Ethernet7"
|
||||
}],
|
||||
"compute_id": "local",
|
||||
"name": "Ethernet hub template",
|
||||
"symbol": ":/symbols/hub.svg",
|
||||
"default_name_format": "Hub{0}",
|
||||
"template_type": "ethernet_hub",
|
||||
"category": "switch",
|
||||
"builtin": False}
|
||||
|
||||
for item, value in expected_response.items():
|
||||
assert response.json.get(item) == value
|
||||
|
||||
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_create_node_from_template(controller_api, controller, project):
|
||||
#
|
||||
# id = str(uuid.uuid4())
|
||||
# controller.template_manager._templates = {id: Template(id, {
|
||||
# "template_type": "qemu",
|
||||
# "category": 0,
|
||||
# "name": "test",
|
||||
# "symbol": "guest.svg",
|
||||
# "default_name_format": "{name}-{0}",
|
||||
# "compute_id": "example.com"
|
||||
# })}
|
||||
# with asyncio_patch("gns3server.controller.project.Project.add_node_from_template", return_value={"name": "test", "node_type": "qemu", "compute_id": "example.com"}) as mock:
|
||||
# response = await controller_api.post("/projects/{}/templates/{}".format(project.id, id), {
|
||||
# "x": 42,
|
||||
# "y": 12
|
||||
# })
|
||||
# mock.assert_called_with(id, x=42, y=12, compute_id=None)
|
||||
# assert response.status_code == 201
|
||||
@@ -0,0 +1,63 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import pytest
|
||||
|
||||
from gns3server.version import __version__
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_version_output(controller_api, config):
|
||||
|
||||
config.set("Server", "local", "true")
|
||||
response = await controller_api.get('/version')
|
||||
assert response.status_code == 200
|
||||
assert response.json == {'local': True, 'version': __version__}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_version_input(controller_api):
|
||||
|
||||
params = {'version': __version__}
|
||||
response = await controller_api.post('/version', params)
|
||||
assert response.status_code == 200
|
||||
assert response.json == {'version': __version__}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_version_invalid_input(controller_api):
|
||||
|
||||
params = {'version': "0.4.2"}
|
||||
response = await controller_api.post('/version', params)
|
||||
assert response.status_code == 409
|
||||
assert response.json == {'message': 'Client version 0.4.2 is not the same as server version {}'.format(__version__)}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_version_invalid_input_schema(controller_api):
|
||||
|
||||
params = {'version': "0.4.2", "bla": "blu"}
|
||||
response = await controller_api.post('/version', params)
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_version_invalid_json(controller_api):
|
||||
|
||||
params = "BOUM"
|
||||
response = await controller_api.post('/version', params, raw=True)
|
||||
assert response.status_code == 422
|
||||
Reference in new issue
Block a user