mirror of
https://github.com/GNS3/gns3-server
synced 2024-12-25 00:08:11 +00:00
PEP8
This commit is contained in:
parent
0695e75e77
commit
68d0e5f42d
@ -19,7 +19,7 @@
|
||||
Simple file upload & listing handler.
|
||||
"""
|
||||
|
||||
#TODO: file upload with aiohttp
|
||||
# TODO: file upload with aiohttp
|
||||
|
||||
|
||||
import os
|
||||
|
@ -31,7 +31,7 @@ class ProjectHandler:
|
||||
def create_project(request, response):
|
||||
pm = ProjectManager.instance()
|
||||
p = pm.create_project(
|
||||
location = request.json.get("location"),
|
||||
uuid = request.json.get("uuid")
|
||||
location=request.json.get("location"),
|
||||
uuid=request.json.get("uuid")
|
||||
)
|
||||
response.json(p)
|
||||
|
@ -40,11 +40,11 @@ class VPCSHandler:
|
||||
def create(request, response):
|
||||
|
||||
vpcs = VPCS.instance()
|
||||
vm = yield from vpcs.create_vm(request.json["name"], request.json["project_uuid"], uuid = 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,
|
||||
"project_uuid": vm.project.uuid})
|
||||
"project_uuid": vm.project.uuid})
|
||||
|
||||
@classmethod
|
||||
@Route.post(
|
||||
|
@ -20,8 +20,3 @@ from .vpcs import VPCS
|
||||
from .virtualbox import VirtualBox
|
||||
|
||||
MODULES = [VPCS, VirtualBox]
|
||||
|
||||
#if sys.platform.startswith("linux"):
|
||||
# # IOU runs only on Linux
|
||||
# from .iou import IOU
|
||||
# MODULES.append(IOU)
|
||||
|
@ -96,9 +96,9 @@ class BaseManager:
|
||||
"""
|
||||
project = ProjectManager.instance().get_project(project_identifier)
|
||||
|
||||
#TODO: support for old projects VM with normal IDs.
|
||||
# TODO: support for old projects VM with normal IDs.
|
||||
|
||||
#TODO: supports specific args: pass kwargs to VM_CLASS?
|
||||
# TODO: supports specific args: pass kwargs to VM_CLASS?
|
||||
|
||||
if not uuid:
|
||||
uuid = str(uuid4())
|
||||
|
@ -31,7 +31,7 @@ class BaseVM:
|
||||
self._manager = manager
|
||||
self._config = Config.instance()
|
||||
|
||||
#TODO: When delete release console ports
|
||||
# TODO: When delete release console ports
|
||||
|
||||
@property
|
||||
def project(self):
|
||||
@ -98,4 +98,3 @@ class BaseVM:
|
||||
"""
|
||||
|
||||
raise NotImplementedError
|
||||
|
||||
|
@ -61,7 +61,7 @@ class VPCSDevice(BaseVM):
|
||||
|
||||
self._console = console
|
||||
|
||||
#TODO: remove working_dir
|
||||
# TODO: remove working_dir
|
||||
self._working_dir = "/tmp"
|
||||
|
||||
self._command = []
|
||||
@ -120,7 +120,7 @@ class VPCSDevice(BaseVM):
|
||||
|
||||
return self._console
|
||||
|
||||
#FIXME: correct way to subclass a property?
|
||||
# FIXME: correct way to subclass a property?
|
||||
@BaseVM.name.setter
|
||||
def name(self, new_name):
|
||||
"""
|
||||
@ -151,7 +151,7 @@ class VPCSDevice(BaseVM):
|
||||
"""
|
||||
Checks if the VPCS executable version is >= 0.5b1.
|
||||
"""
|
||||
#TODO: should be async
|
||||
# TODO: should be async
|
||||
try:
|
||||
output = subprocess.check_output([self._path, "-v"], cwd=self._working_dir)
|
||||
match = re.search("Welcome to Virtual PC Simulator, version ([0-9a-z\.]+)", output.decode("utf-8"))
|
||||
@ -219,7 +219,7 @@ class VPCSDevice(BaseVM):
|
||||
Reads the standard output of the VPCS process.
|
||||
Only use when the process has been stopped or has crashed.
|
||||
"""
|
||||
#TODO: should be async
|
||||
# TODO: should be async
|
||||
output = ""
|
||||
if self._vpcs_stdout_file:
|
||||
try:
|
||||
@ -258,7 +258,7 @@ class VPCSDevice(BaseVM):
|
||||
rhost = nio_settings["rhost"]
|
||||
rport = nio_settings["rport"]
|
||||
try:
|
||||
#TODO: handle IPv6
|
||||
# TODO: handle IPv6
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
||||
sock.connect((rhost, rport))
|
||||
except OSError as e:
|
||||
@ -272,7 +272,6 @@ class VPCSDevice(BaseVM):
|
||||
if not nio:
|
||||
raise VPCSError("Requested NIO does not exist or is not supported: {}".format(nio_settings["type"]))
|
||||
|
||||
|
||||
self._ethernet_adapter.add_nio(port_id, nio)
|
||||
log.info("VPCS {name} {uuid}]: {nio} added to port {port_id}".format(name=self._name,
|
||||
uuid=self.uuid,
|
||||
@ -355,8 +354,8 @@ class VPCSDevice(BaseVM):
|
||||
command.extend(["-e"])
|
||||
command.extend(["-d", nio.tap_device])
|
||||
|
||||
#FIXME: find workaround
|
||||
#command.extend(["-m", str(self._id)]) # the unique ID is used to set the MAC address offset
|
||||
# FIXME: find workaround
|
||||
# command.extend(["-m", str(self._id)]) # the unique ID is used to set the MAC address offset
|
||||
command.extend(["-i", "1"]) # option to start only one VPC instance
|
||||
command.extend(["-F"]) # option to avoid the daemonization of VPCS
|
||||
if self._script_file:
|
||||
|
@ -36,4 +36,3 @@ PROJECT_OBJECT_SCHEMA = {
|
||||
},
|
||||
"additionalProperties": False
|
||||
}
|
||||
|
||||
|
@ -78,4 +78,3 @@ VBOX_OBJECT_SCHEMA = {
|
||||
"additionalProperties": False,
|
||||
"required": ["name", "uuid"]
|
||||
}
|
||||
|
||||
|
@ -148,4 +148,3 @@ VPCS_OBJECT_SCHEMA = {
|
||||
"additionalProperties": False,
|
||||
"required": ["name", "uuid", "console", "project_uuid"]
|
||||
}
|
||||
|
||||
|
@ -22,6 +22,7 @@ import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Response(aiohttp.web.Response):
|
||||
|
||||
def __init__(self, route=None, output_schema=None, headers={}, **kwargs):
|
||||
|
@ -15,6 +15,7 @@
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
|
||||
"""Base code use for all API tests"""
|
||||
|
||||
import json
|
||||
@ -27,7 +28,7 @@ import aiohttp
|
||||
|
||||
|
||||
from gns3server.web.route import Route
|
||||
#TODO: get rid of *
|
||||
# TODO: get rid of *
|
||||
from gns3server.handlers import *
|
||||
from gns3server.modules import MODULES
|
||||
from gns3server.modules.port_manager import PortManager
|
||||
@ -143,6 +144,7 @@ def loop(request):
|
||||
request.addfinalizer(tear_down)
|
||||
return loop
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def server(request, loop):
|
||||
port = _get_unused_port()
|
||||
@ -167,5 +169,4 @@ def server(request, loop):
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def project():
|
||||
return ProjectManager.instance().create_project(uuid = "a1e920ca-338a-4e9f-b363-aa607b09dd80")
|
||||
|
||||
return ProjectManager.instance().create_project(uuid="a1e920ca-338a-4e9f-b363-aa607b09dd80")
|
||||
|
@ -30,12 +30,14 @@ def test_create_project_with_dir(server, tmpdir):
|
||||
assert response.status == 200
|
||||
assert response.json['location'] == str(tmpdir)
|
||||
|
||||
|
||||
def test_create_project_without_dir(server):
|
||||
query = {}
|
||||
response = server.post('/project', query)
|
||||
assert response.status == 200
|
||||
assert response.json['uuid'] is not None
|
||||
|
||||
|
||||
def test_create_project_with_uuid(server):
|
||||
query = {'uuid': '00010203-0405-0607-0809-0a0b0c0d0e0f'}
|
||||
response = server.post('/project', query)
|
||||
|
@ -38,12 +38,11 @@ def test_vpcs_create(server, project):
|
||||
assert response.json["project_uuid"] == "61d61bdd-aa7d-4912-817f-65a9eb54d3ab"
|
||||
|
||||
|
||||
#FIXME
|
||||
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"},
|
||||
"lport": 4242,
|
||||
"rport": 4343,
|
||||
"rhost": "127.0.0.1"},
|
||||
example=True)
|
||||
assert response.status == 200
|
||||
assert response.route == "/vpcs/{uuid}/ports/{port_id}/nio"
|
||||
@ -53,18 +52,17 @@ def test_vpcs_nio_create_udp(server, vm):
|
||||
@patch("gns3server.modules.vpcs.vpcs_device.has_privileged_access", return_value=True)
|
||||
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"})
|
||||
"tap_device": "test"})
|
||||
assert response.status == 200
|
||||
assert response.route == "/vpcs/{uuid}/ports/{port_id}/nio"
|
||||
assert response.json["type"] == "nio_tap"
|
||||
|
||||
|
||||
#FIXME
|
||||
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"})
|
||||
"lport": 4242,
|
||||
"rport": 4343,
|
||||
"rhost": "127.0.0.1"})
|
||||
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"
|
||||
|
@ -1,19 +0,0 @@
|
||||
import sys
|
||||
import os
|
||||
import pytest
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def server(request):
|
||||
"""
|
||||
Starts GNS3 server for all the tests.
|
||||
"""
|
||||
|
||||
cwd = os.path.dirname(os.path.abspath(__file__))
|
||||
server_script = os.path.join(cwd, "../gns3server/main.py")
|
||||
process = subprocess.Popen([sys.executable, server_script, "--port=8000"])
|
||||
#time.sleep(1) # give some time for the process to start
|
||||
request.addfinalizer(process.terminate)
|
||||
return process
|
@ -16,8 +16,9 @@
|
||||
# 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 gns3server.modules.project import Project
|
||||
import os
|
||||
from gns3server.modules.project import Project
|
||||
|
||||
|
||||
def test_affect_uuid():
|
||||
p = Project()
|
||||
@ -26,16 +27,19 @@ def test_affect_uuid():
|
||||
p = Project(uuid='00010203-0405-0607-0809-0a0b0c0d0e0f')
|
||||
assert p.uuid == '00010203-0405-0607-0809-0a0b0c0d0e0f'
|
||||
|
||||
|
||||
def test_path(tmpdir):
|
||||
p = Project(location = str(tmpdir))
|
||||
p = Project(location=str(tmpdir))
|
||||
assert p.path == os.path.join(str(tmpdir), p.uuid)
|
||||
assert os.path.exists(os.path.join(str(tmpdir), p.uuid))
|
||||
assert os.path.exists(os.path.join(str(tmpdir), p.uuid, 'files'))
|
||||
|
||||
|
||||
def test_temporary_path():
|
||||
p = Project()
|
||||
assert os.path.exists(p.path)
|
||||
|
||||
|
||||
def test_json(tmpdir):
|
||||
p = Project()
|
||||
assert p.__json__() == {"location": p.location, "uuid": p.uuid}
|
||||
|
@ -25,8 +25,8 @@ def test_create_project():
|
||||
project = pm.create_project(uuid='00010203-0405-0607-0809-0a0b0c0d0e0f')
|
||||
assert project == pm.get_project('00010203-0405-0607-0809-0a0b0c0d0e0f')
|
||||
|
||||
|
||||
def test_project_not_found():
|
||||
pm = ProjectManager.instance()
|
||||
with pytest.raises(aiohttp.web.HTTPNotFound):
|
||||
pm.get_project('00010203-0405-0607-0809-000000000000')
|
||||
|
||||
|
@ -19,7 +19,7 @@ import pytest
|
||||
import asyncio
|
||||
from tests.utils import asyncio_patch
|
||||
|
||||
#TODO: Move loop to util
|
||||
# TODO: Move loop to util
|
||||
from tests.api.base import loop, project
|
||||
from asyncio.subprocess import Process
|
||||
from unittest.mock import patch, MagicMock
|
||||
@ -42,6 +42,7 @@ def test_vm(manager):
|
||||
assert vm.name == "test"
|
||||
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0f"
|
||||
|
||||
|
||||
@patch("subprocess.check_output", return_value="Welcome to Virtual PC Simulator, version 0.1".encode("utf-8"))
|
||||
def test_vm_invalid_vpcs_version(project, manager):
|
||||
with pytest.raises(VPCSError):
|
||||
@ -49,20 +50,23 @@ def test_vm_invalid_vpcs_version(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"})
|
||||
|
||||
@patch("gns3server.config.Config.get_section_config", return_value={"path": "/bin/test_fake"})
|
||||
def test_vm_invalid_vpcs_path(project, manager):
|
||||
with pytest.raises(VPCSError):
|
||||
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(project, loop, manager):
|
||||
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()):
|
||||
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
|
||||
assert vm.is_running()
|
||||
|
||||
|
||||
def test_stop(project, loop, manager):
|
||||
process = MagicMock()
|
||||
@ -71,22 +75,25 @@ def test_stop(project, loop, 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
|
||||
assert vm.is_running()
|
||||
loop.run_until_complete(asyncio.async(vm.stop()))
|
||||
assert vm.is_running() == False
|
||||
assert vm.is_running() is False
|
||||
process.terminate.assert_called_with()
|
||||
|
||||
|
||||
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(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(manager):
|
||||
vm = VPCSDevice("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
|
||||
with patch("gns3server.modules.vpcs.vpcs_device.has_privileged_access", return_value=False):
|
||||
@ -94,6 +101,7 @@ def test_add_nio_binding_tap_no_privileged_access(manager):
|
||||
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(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"})
|
||||
|
Loading…
Reference in New Issue
Block a user