NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2025-01-18 07:23:47 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix tests

Browse Source
This commit is contained in:
Julien Duponchelle 2015-01-20 12:46:15 +01:00
parent 927e6b540d
commit 0695e75e77
14 changed files with 139 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
docs/api/examples/delete_vpcsuuidportsportidnio.txt Normal file
View File

@ -0,0 +1,15 @@
curl -i -X DELETE 'http://localhost:8000/vpcs/{uuid}/ports/{port_id}/nio'
DELETE /vpcs/{uuid}/ports/{port_id}/nio HTTP/1.1
HTTP/1.1 200
CONNECTION: close
CONTENT-LENGTH: 2
CONTENT-TYPE: application/json
DATE: Thu, 08 Jan 2015 16:09:15 GMT
SERVER: Python/3.4 aiohttp/0.13.1
X-ROUTE: /vpcs/{uuid}/ports/{port_id}/nio
{}

8
docs/api/examples/post_project.txt
View File

@ -1,8 +1,8 @@
curl -i -X POST 'http://localhost:8000/project' -d '{"location": "/private/var/folders/3s/r2wbv07n7wg4vrsn874lmxxh0000gn/T/pytest-262/test_create_project_with_dir0"}'
curl -i -X POST 'http://localhost:8000/project' -d '{"location": "/private/var/folders/3s/r2wbv07n7wg4vrsn874lmxxh0000gn/T/pytest-301/test_create_project_with_dir0"}'
POST /project HTTP/1.1
{
"location": "/private/var/folders/3s/r2wbv07n7wg4vrsn874lmxxh0000gn/T/pytest-262/test_create_project_with_dir0"
"location": "/private/var/folders/3s/r2wbv07n7wg4vrsn874lmxxh0000gn/T/pytest-301/test_create_project_with_dir0"
}
@ -15,6 +15,6 @@ SERVER: Python/3.4 aiohttp/0.13.1
X-ROUTE: /project
{
"location": "/private/var/folders/3s/r2wbv07n7wg4vrsn874lmxxh0000gn/T/pytest-262/test_create_project_with_dir0",
"uuid": "66821e79-aa05-4490-8c26-ffb7aeddc0d2"
"location": "/private/var/folders/3s/r2wbv07n7wg4vrsn874lmxxh0000gn/T/pytest-301/test_create_project_with_dir0",
"uuid": "d4d66c3f-439b-4ae9-972b-59040189c995"
}

25
docs/api/examples/post_vpcsuuidportsportidnio.txt Normal file
View File

@ -0,0 +1,25 @@
curl -i -X POST 'http://localhost:8000/vpcs/{uuid}/ports/{port_id}/nio' -d '{"lport": 4242, "rhost": "127.0.0.1", "rport": 4343, "type": "nio_udp"}'
POST /vpcs/{uuid}/ports/{port_id}/nio HTTP/1.1
{
"lport": 4242,
"rhost": "127.0.0.1",
"rport": 4343,
"type": "nio_udp"
}
HTTP/1.1 200
CONNECTION: close
CONTENT-LENGTH: 89
CONTENT-TYPE: application/json
DATE: Thu, 08 Jan 2015 16:09:15 GMT
SERVER: Python/3.4 aiohttp/0.13.1
X-ROUTE: /vpcs/{uuid}/ports/{port_id}/nio
{
"lport": 4242,
"rhost": "127.0.0.1",
"rport": 4343,
"type": "nio_udp"
}

5
gns3server/handlers/vpcs_handler.py
View File

@ -40,10 +40,11 @@ class VPCSHandler:
def create(request, response):
vpcs = VPCS.instance()
vm = yield from vpcs.create_vm(request.json["name"], request.json.get("uuid"))
vm = yield from vpcs.create_vm(request.json["name"], request.json["project_uuid"], uuid = request.json.get("uuid"))
response.json({"name": vm.name,
"uuid": vm.uuid,
"console": vm.console})
"console": vm.console,
"project_uuid": vm.project.uuid})
@classmethod
@Route.post(

15
gns3server/modules/base_manager.py
View File

@ -20,6 +20,7 @@ import asyncio
import aiohttp
from uuid import UUID, uuid4
from .project_manager import ProjectManager
class BaseManager:
@ -85,16 +86,24 @@ class BaseManager:
return self._vms[uuid]
@asyncio.coroutine
def create_vm(self, name, uuid=None):
def create_vm(self, name, project_identifier, uuid=None):
"""
Create a new VM
#TODO: support for old projects with normal IDs.
:param name VM name
:param project_identifier UUID of Project
:param uuid Force UUID force VM
"""
project = ProjectManager.instance().get_project(project_identifier)
#TODO: support for old projects VM with normal IDs.
#TODO: supports specific args: pass kwargs to VM_CLASS?
if not uuid:
uuid = str(uuid4())
vm = self._VM_CLASS(name, uuid, self)
vm = self._VM_CLASS(name, uuid, project, self)
future = vm.create()
if isinstance(future, asyncio.Future):
yield from future

8
gns3server/modules/base_vm.py
View File

@ -23,15 +23,21 @@ log = logging.getLogger(__name__)
class BaseVM:
def __init__(self, name, uuid, manager):
def __init__(self, name, uuid, project, manager):
self._name = name
self._uuid = uuid
self._project = project
self._manager = manager
self._config = Config.instance()
#TODO: When delete release console ports
@property
def project(self):
"""Return VM current project"""
return self._project
@property
def name(self):
"""

1
gns3server/modules/project.py
View File

@ -34,6 +34,7 @@ class Project:
if uuid is None:
self._uuid = str(uuid4())
else:
assert len(uuid) == 36
self._uuid = uuid
self._location = location

2
gns3server/modules/project_manager.py
View File

@ -48,6 +48,8 @@ class ProjectManager:
:returns: Project instance
"""
assert len(project_id) == 36
if project_id not in self._projects:
raise aiohttp.web.HTTPNotFound(text="Project UUID {} doesn't exist".format(project_id))
return self._projects[project_id]

12
gns3server/modules/vpcs/vpcs_device.py
View File

@ -48,22 +48,22 @@ class VPCSDevice(BaseVM):
:param name: name of this VPCS device
:param uuid: VPCS instance UUID
:param project: Project instance
:param manager: parent VM Manager
:param working_dir: path to a working directory
:param console: TCP console port
"""
def __init__(self, name, uuid, manager, working_dir=None, console=None):
def __init__(self, name, uuid, project, manager, working_dir=None, console=None):
super().__init__(name, uuid, manager)
# TODO: Hardcodded for testing
#self._working_dir = working_dir
self._working_dir = "/tmp"
super().__init__(name, uuid, project, manager)
self._path = self._config.get_section_config("VPCS").get("path", "vpcs")
self._console = console
#TODO: remove working_dir
self._working_dir = "/tmp"
self._command = []
self._process = None
self._vpcs_stdout_file = ""

18
gns3server/schemas/vpcs.py
View File

@ -37,6 +37,13 @@ VPCS_CREATE_SCHEMA = {
"maxLength": 36,
"pattern": "^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}$"
},
"project_uuid": {
"description": "Project UUID",
"type": "string",
"minLength": 36,
"maxLength": 36,
"pattern": "^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}$"
},
"console": {
"description": "console TCP port",
"minimum": 1,
@ -45,7 +52,7 @@ VPCS_CREATE_SCHEMA = {
},
},
"additionalProperties": False,
"required": ["name"]
"required": ["name", "project_uuid"]
}
@ -130,8 +137,15 @@ VPCS_OBJECT_SCHEMA = {
"maximum": 65535,
"type": "integer"
},
"project_uuid": {
"description": "Project UUID",
"type": "string",
"minLength": 36,
"maxLength": 36,
"pattern": "^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}$"
}
},
"additionalProperties": False,
"required": ["name", "uuid", "console"]
"required": ["name", "uuid", "console", "project_uuid"]
}

33
old_tests/vpcs/test_vpcs_device.py
View File

@ -1,33 +0,0 @@
from gns3server.modules.vpcs import VPCSDevice
import os
import pytest
@pytest.fixture(scope="session")
def vpcs(request):
if os.path.isfile("/usr/bin/vpcs"):
vpcs_path = "/usr/bin/vpcs"
else:
cwd = os.path.dirname(os.path.abspath(__file__))
vpcs_path = os.path.join(cwd, "vpcs")
vpcs_device = VPCSDevice("VPCS1", vpcs_path, "/tmp")
vpcs_device.port_add_nio_binding(0, 'nio_tap:tap0')
vpcs_device.start()
request.addfinalizer(vpcs_device.delete)
return vpcs_device
def test_vpcs_is_started(vpcs):
print(vpcs.command())
assert vpcs.id == 1 # we should have only one VPCS running!
assert vpcs.is_running()
def test_vpcs_restart(vpcs):
vpcs.stop()
assert not vpcs.is_running()
vpcs.start()
assert vpcs.is_running()

7
tests/api/base.py
View File

@ -31,6 +31,7 @@ from gns3server.web.route import Route
from gns3server.handlers import *
from gns3server.modules import MODULES
from gns3server.modules.port_manager import PortManager
from gns3server.modules.project_manager import ProjectManager
class Query:
@ -162,3 +163,9 @@ def server(request, loop):
srv.wait_closed()
request.addfinalizer(tear_down)
return Query(loop, host=host, port=port)
@pytest.fixture(scope="module")
def project():
return ProjectManager.instance().create_project(uuid = "a1e920ca-338a-4e9f-b363-aa607b09dd80")

32
tests/api/test_vpcs.py
View File

@ -15,24 +15,32 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from tests.api.base import server, loop
import pytest
from tests.api.base import server, loop, project
from tests.utils import asyncio_patch
from unittest.mock import patch
@pytest.fixture(scope="module")
def vm(server, project):
response = server.post("/vpcs", {"name": "PC TEST 1", "project_uuid": project.uuid})
assert response.status == 200
return response.json
@asyncio_patch("gns3server.modules.VPCS.create_vm", return_value="61d61bdd-aa7d-4912-817f-65a9eb54d3ab")
def test_vpcs_create(server):
response = server.post("/vpcs", {"name": "PC TEST 1"}, example=False)
def test_vpcs_create(server, project):
response = server.post("/vpcs", {"name": "PC TEST 1", "project_uuid": project.uuid}, example=True)
assert response.status == 200
assert response.route == "/vpcs"
assert response.json["name"] == "PC TEST 1"
assert response.json["uuid"] == "61d61bdd-aa7d-4912-817f-65a9eb54d3ab"
assert response.json["project_uuid"] == "61d61bdd-aa7d-4912-817f-65a9eb54d3ab"
#FIXME
def test_vpcs_nio_create_udp(server):
vm = server.post("/vpcs", {"name": "PC TEST 1"})
response = server.post("/vpcs/{}/ports/0/nio".format(vm.json["uuid"]), {"type": "nio_udp",
def test_vpcs_nio_create_udp(server, vm):
response = server.post("/vpcs/{}/ports/0/nio".format(vm["uuid"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"},
@ -43,9 +51,8 @@ def test_vpcs_nio_create_udp(server):
@patch("gns3server.modules.vpcs.vpcs_device.has_privileged_access", return_value=True)
def test_vpcs_nio_create_tap(mock, server):
vm = server.post("/vpcs", {"name": "PC TEST 1"})
response = server.post("/vpcs/{}/ports/0/nio".format(vm.json["uuid"]), {"type": "nio_tap",
def test_vpcs_nio_create_tap(mock, server, vm):
response = server.post("/vpcs/{}/ports/0/nio".format(vm["uuid"]), {"type": "nio_tap",
"tap_device": "test"})
assert response.status == 200
assert response.route == "/vpcs/{uuid}/ports/{port_id}/nio"
@ -53,12 +60,11 @@ def test_vpcs_nio_create_tap(mock, server):
#FIXME
def test_vpcs_delete_nio(server):
vm = server.post("/vpcs", {"name": "PC TEST 1"})
response = server.post("/vpcs/{}/ports/0/nio".format(vm.json["uuid"]), {"type": "nio_udp",
def test_vpcs_delete_nio(server, vm):
response = server.post("/vpcs/{}/ports/0/nio".format(vm["uuid"]), {"type": "nio_udp",
"lport": 4242,
"rport": 4343,
"rhost": "127.0.0.1"})
response = server.delete("/vpcs/{}/ports/0/nio".format(vm.json["uuid"]), example=True)
response = server.delete("/vpcs/{}/ports/0/nio".format(vm["uuid"]), example=True)
assert response.status == 200
assert response.route == "/vpcs/{uuid}/ports/{port_id}/nio"

42
tests/modules/vpcs/test_vpcs_device.py
View File

@ -19,8 +19,8 @@ import pytest
import asyncio
from tests.utils import asyncio_patch
#Move loop to util
from tests.api.base import loop
#TODO: Move loop to util
from tests.api.base import loop, project
from asyncio.subprocess import Process
from unittest.mock import patch, MagicMock
from gns3server.modules.vpcs.vpcs_device import VPCSDevice
@ -28,44 +28,46 @@ from gns3server.modules.vpcs.vpcs_error import VPCSError
from gns3server.modules.vpcs import VPCS
from gns3server.modules.port_manager import PortManager
@pytest.fixture(scope="module")
def manager():
m = VPCS.instance()
m.port_manager = PortManager("127.0.0.1", False)
return m
@patch("subprocess.check_output", return_value="Welcome to Virtual PC Simulator, version 0.6".encode("utf-8"))
def test_vm(tmpdir, manager):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", manager, working_dir=str(tmpdir))
def test_vm(manager):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
assert vm.name == "test"
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0f"
@patch("subprocess.check_output", return_value="Welcome to Virtual PC Simulator, version 0.1".encode("utf-8"))
def test_vm_invalid_vpcs_version(tmpdir, manager):
def test_vm_invalid_vpcs_version(project, manager):
with pytest.raises(VPCSError):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", manager, working_dir=str(tmpdir))
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
assert vm.name == "test"
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0f"
@patch("gns3server.config.Config.get_section_config", return_value = {"path": "/bin/test_fake"})
def test_vm_invalid_vpcs_path(tmpdir, manager):
def test_vm_invalid_vpcs_path(project, manager):
with pytest.raises(VPCSError):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", manager, working_dir=str(tmpdir))
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
assert vm.name == "test"
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0f"
def test_start(tmpdir, loop, manager):
def test_start(project, loop, manager):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", manager, working_dir=str(tmpdir))
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running() == True
def test_stop(tmpdir, loop, manager):
def test_stop(project, loop, manager):
process = MagicMock()
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", manager, working_dir=str(tmpdir))
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
loop.run_until_complete(asyncio.async(vm.start()))
@ -74,26 +76,26 @@ def test_stop(tmpdir, loop, manager):
assert vm.is_running() == False
process.terminate.assert_called_with()
def test_add_nio_binding_udp(tmpdir, manager):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", manager, working_dir=str(tmpdir))
def test_add_nio_binding_udp(manager):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
assert nio.lport == 4242
def test_add_nio_binding_tap(tmpdir, manager):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", manager, working_dir=str(tmpdir))
def test_add_nio_binding_tap(project, manager):
vm = VPCSDevice("test", 42, project, manager)
with patch("gns3server.modules.vpcs.vpcs_device.has_privileged_access", return_value=True):
nio = vm.port_add_nio_binding(0, {"type": "nio_tap", "tap_device": "test"})
assert nio.tap_device == "test"
def test_add_nio_binding_tap_no_privileged_access(tmpdir, manager):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", manager, working_dir=str(tmpdir))
def test_add_nio_binding_tap_no_privileged_access(manager):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
with patch("gns3server.modules.vpcs.vpcs_device.has_privileged_access", return_value=False):
with pytest.raises(VPCSError):
vm.port_add_nio_binding(0, {"type": "nio_tap", "tap_device": "test"})
assert vm._ethernet_adapter.ports[0] is None
def test_port_remove_nio_binding(tmpdir, manager):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", manager, working_dir=str(tmpdir))
def test_port_remove_nio_binding(manager):
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
vm.port_remove_nio_binding(0)
assert vm._ethernet_adapter.ports[0] is None