mirror of
https://github.com/GNS3/gns3-server
synced 2024-11-12 19:38:57 +00:00
Merge branch 'master' into unstable
Conflicts: tests/modules/test_manager.py
This commit is contained in:
commit
5f6d1bfdd6
@ -98,6 +98,9 @@ Windows
|
||||
|
||||
Please use our all-in-one installer.
|
||||
|
||||
If you install it via source you need to install also:
|
||||
https://sourceforge.net/projects/pywin32/
|
||||
|
||||
Mac OS X
|
||||
--------
|
||||
|
||||
|
@ -40,7 +40,7 @@ class CrashReport:
|
||||
Report crash to a third party service
|
||||
"""
|
||||
|
||||
DSN = "sync+https://90fe04368a3e4109b584a116ee03ca87:268ef86c65cf4e8fa41d4c7fb1c70b72@app.getsentry.com/38482"
|
||||
DSN = "sync+https://22979234ab4749ceabce08e6da4c1476:1432c8c7a43d410b9b5bb33f8e55b2a6@app.getsentry.com/38482"
|
||||
if hasattr(sys, "frozen"):
|
||||
cacert = os.path.join(os.getcwd(), "cacert.pem")
|
||||
if os.path.isfile(cacert):
|
||||
@ -55,6 +55,9 @@ class CrashReport:
|
||||
def capture_exception(self, request=None):
|
||||
if not RAVEN_AVAILABLE:
|
||||
return
|
||||
if os.path.exists(".git"):
|
||||
log.warning("A .git directory exist crash report is turn off for developers")
|
||||
return
|
||||
server_config = Config.instance().get_section_config("Server")
|
||||
if server_config.getboolean("report_errors"):
|
||||
if self._client is None:
|
||||
|
@ -32,4 +32,6 @@ from gns3server.handlers.upload_handler import UploadHandler
|
||||
from gns3server.handlers.index_handler import IndexHandler
|
||||
|
||||
if sys.platform.startswith("linux") or hasattr(sys, "_called_from_test") or os.environ.get("PYTEST_BUILD_DOCUMENTATION") == "1":
|
||||
from gns3server.handlers.api.iou_handler import IOUHandler
|
||||
# IOU runs only on Linux but testsuite work on UNIX platform
|
||||
if not sys.platform.startswith("win"):
|
||||
from gns3server.handlers.api.iou_handler import IOUHandler
|
||||
|
@ -19,11 +19,11 @@ import os
|
||||
import base64
|
||||
|
||||
from ...web.route import Route
|
||||
from ...schemas.nio import NIO_SCHEMA
|
||||
from ...schemas.dynamips_vm import VM_CREATE_SCHEMA
|
||||
from ...schemas.dynamips_vm import VM_UPDATE_SCHEMA
|
||||
from ...schemas.dynamips_vm import VM_CAPTURE_SCHEMA
|
||||
from ...schemas.dynamips_vm import VM_OBJECT_SCHEMA
|
||||
from ...schemas.dynamips_vm import VM_NIO_SCHEMA
|
||||
from ...schemas.dynamips_vm import VM_CONFIGS_SCHEMA
|
||||
from ...schemas.dynamips_vm import VMS_LIST_SCHEMA
|
||||
from ...modules.dynamips import Dynamips
|
||||
@ -257,8 +257,8 @@ class DynamipsVMHandler:
|
||||
404: "Instance doesn't exist"
|
||||
},
|
||||
description="Add a NIO to a Dynamips VM instance",
|
||||
input=VM_NIO_SCHEMA,
|
||||
output=VM_NIO_SCHEMA)
|
||||
input=NIO_SCHEMA,
|
||||
output=NIO_SCHEMA)
|
||||
def create_nio(request, response):
|
||||
|
||||
dynamips_manager = Dynamips.instance()
|
||||
|
@ -19,10 +19,10 @@ import os
|
||||
from aiohttp.web import HTTPConflict
|
||||
|
||||
from ...web.route import Route
|
||||
from ...schemas.nio import NIO_SCHEMA
|
||||
from ...schemas.iou import IOU_CREATE_SCHEMA
|
||||
from ...schemas.iou import IOU_UPDATE_SCHEMA
|
||||
from ...schemas.iou import IOU_OBJECT_SCHEMA
|
||||
from ...schemas.iou import IOU_NIO_SCHEMA
|
||||
from ...schemas.iou import IOU_CAPTURE_SCHEMA
|
||||
from ...schemas.iou import IOU_INITIAL_CONFIG_SCHEMA
|
||||
from ...schemas.iou import IOU_LIST_VMS_SCHEMA
|
||||
@ -59,6 +59,8 @@ class IOUHandler:
|
||||
|
||||
for name, value in request.json.items():
|
||||
if hasattr(vm, name) and getattr(vm, name) != value:
|
||||
if name == "initial_config_content" and (vm.initial_config_content and len(vm.initial_config_content) > 0):
|
||||
continue
|
||||
setattr(vm, name, value)
|
||||
if "initial_config_content" in request.json:
|
||||
vm.initial_config = request.json.get("initial_config_content")
|
||||
@ -205,12 +207,15 @@ class IOUHandler:
|
||||
404: "Instance doesn't exist"
|
||||
},
|
||||
description="Add a NIO to a IOU instance",
|
||||
input=IOU_NIO_SCHEMA,
|
||||
output=IOU_NIO_SCHEMA)
|
||||
input=NIO_SCHEMA,
|
||||
output=NIO_SCHEMA)
|
||||
def create_nio(request, response):
|
||||
|
||||
iou_manager = IOU.instance()
|
||||
vm = iou_manager.get_vm(request.match_info["vm_id"], project_id=request.match_info["project_id"])
|
||||
nio_type = request.json["type"]
|
||||
if nio_type not in ("nio_udp", "nio_tap", "nio_generic_ethernet"):
|
||||
raise HTTPConflict(text="NIO of type {} is not supported".format(nio_type))
|
||||
nio = iou_manager.create_nio(vm.iouyap_path, request.json)
|
||||
vm.adapter_add_nio_binding(int(request.match_info["adapter_number"]), int(request.match_info["port_number"]), nio)
|
||||
response.set_status(201)
|
||||
@ -325,4 +330,3 @@ class IOUHandler:
|
||||
vms = yield from iou_manager.list_images()
|
||||
response.set_status(200)
|
||||
response.json(vms)
|
||||
|
||||
|
@ -15,11 +15,12 @@
|
||||
# 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 aiohttp.web import HTTPConflict
|
||||
from ...web.route import Route
|
||||
from ...schemas.nio import NIO_SCHEMA
|
||||
from ...schemas.qemu import QEMU_CREATE_SCHEMA
|
||||
from ...schemas.qemu import QEMU_UPDATE_SCHEMA
|
||||
from ...schemas.qemu import QEMU_OBJECT_SCHEMA
|
||||
from ...schemas.qemu import QEMU_NIO_SCHEMA
|
||||
from ...schemas.qemu import QEMU_BINARY_LIST_SCHEMA
|
||||
from ...schemas.qemu import QEMU_LIST_IMAGES_SCHEMA
|
||||
from ...modules.qemu import Qemu
|
||||
@ -240,12 +241,15 @@ class QEMUHandler:
|
||||
404: "Instance doesn't exist"
|
||||
},
|
||||
description="Add a NIO to a Qemu VM instance",
|
||||
input=QEMU_NIO_SCHEMA,
|
||||
output=QEMU_NIO_SCHEMA)
|
||||
input=NIO_SCHEMA,
|
||||
output=NIO_SCHEMA)
|
||||
def create_nio(request, response):
|
||||
|
||||
qemu_manager = Qemu.instance()
|
||||
vm = qemu_manager.get_vm(request.match_info["vm_id"], project_id=request.match_info["project_id"])
|
||||
nio_type = request.json["type"]
|
||||
if nio_type not in ("nio_udp", "nio_tap"):
|
||||
raise HTTPConflict(text="NIO of type {} is not supported".format(nio_type))
|
||||
nio = qemu_manager.create_nio(vm.qemu_path, request.json)
|
||||
yield from vm.adapter_add_nio_binding(int(request.match_info["adapter_number"]), nio)
|
||||
response.set_status(201)
|
||||
|
@ -16,10 +16,11 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import os
|
||||
from aiohttp.web import HTTPConflict
|
||||
from ...web.route import Route
|
||||
from ...schemas.nio import NIO_SCHEMA
|
||||
from ...schemas.virtualbox import VBOX_CREATE_SCHEMA
|
||||
from ...schemas.virtualbox import VBOX_UPDATE_SCHEMA
|
||||
from ...schemas.virtualbox import VBOX_NIO_SCHEMA
|
||||
from ...schemas.virtualbox import VBOX_CAPTURE_SCHEMA
|
||||
from ...schemas.virtualbox import VBOX_OBJECT_SCHEMA
|
||||
from ...modules.virtualbox import VirtualBox
|
||||
@ -286,12 +287,15 @@ class VirtualBoxHandler:
|
||||
404: "Instance doesn't exist"
|
||||
},
|
||||
description="Add a NIO to a VirtualBox VM instance",
|
||||
input=VBOX_NIO_SCHEMA,
|
||||
output=VBOX_NIO_SCHEMA)
|
||||
input=NIO_SCHEMA,
|
||||
output=NIO_SCHEMA)
|
||||
def create_nio(request, response):
|
||||
|
||||
vbox_manager = VirtualBox.instance()
|
||||
vm = vbox_manager.get_vm(request.match_info["vm_id"], project_id=request.match_info["project_id"])
|
||||
nio_type = request.json["type"]
|
||||
if nio_type != "nio_udp":
|
||||
raise HTTPConflict(text="NIO of type {} is not supported".format(nio_type))
|
||||
nio = vbox_manager.create_nio(vbox_manager.vboxmanage_path, request.json)
|
||||
yield from vm.adapter_add_nio_binding(int(request.match_info["adapter_number"]), nio)
|
||||
response.set_status(201)
|
||||
|
@ -15,11 +15,12 @@
|
||||
# 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 aiohttp.web import HTTPConflict
|
||||
from ...web.route import Route
|
||||
from ...schemas.nio import NIO_SCHEMA
|
||||
from ...schemas.vpcs import VPCS_CREATE_SCHEMA
|
||||
from ...schemas.vpcs import VPCS_UPDATE_SCHEMA
|
||||
from ...schemas.vpcs import VPCS_OBJECT_SCHEMA
|
||||
from ...schemas.vpcs import VPCS_NIO_SCHEMA
|
||||
from ...modules.vpcs import VPCS
|
||||
|
||||
|
||||
@ -191,12 +192,15 @@ class VPCSHandler:
|
||||
404: "Instance doesn't exist"
|
||||
},
|
||||
description="Add a NIO to a VPCS instance",
|
||||
input=VPCS_NIO_SCHEMA,
|
||||
output=VPCS_NIO_SCHEMA)
|
||||
input=NIO_SCHEMA,
|
||||
output=NIO_SCHEMA)
|
||||
def create_nio(request, response):
|
||||
|
||||
vpcs_manager = VPCS.instance()
|
||||
vm = vpcs_manager.get_vm(request.match_info["vm_id"], project_id=request.match_info["project_id"])
|
||||
nio_type = request.json["type"]
|
||||
if nio_type not in ("nio_udp", "nio_tap"):
|
||||
raise HTTPConflict(text="NIO of type {} is not supported".format(nio_type))
|
||||
nio = vpcs_manager.create_nio(vm.vpcs_path, request.json)
|
||||
vm.port_add_nio_binding(int(request.match_info["port_number"]), nio)
|
||||
response.set_status(201)
|
||||
|
@ -25,6 +25,8 @@ from .qemu import Qemu
|
||||
MODULES = [VPCS, VirtualBox, Dynamips, Qemu]
|
||||
|
||||
if sys.platform.startswith("linux") or hasattr(sys, "_called_from_test") or os.environ.get("PYTEST_BUILD_DOCUMENTATION") == "1":
|
||||
# IOU runs only on Linux
|
||||
from .iou import IOU
|
||||
MODULES.append(IOU)
|
||||
|
||||
# IOU runs only on Linux but testsuite work on UNIX platform
|
||||
if not sys.platform.startswith("win"):
|
||||
from .iou import IOU
|
||||
MODULES.append(IOU)
|
||||
|
@ -364,8 +364,9 @@ class BaseManager:
|
||||
nio = NIOUDP(lport, rhost, rport)
|
||||
elif nio_settings["type"] == "nio_tap":
|
||||
tap_device = nio_settings["tap_device"]
|
||||
if not self._has_privileged_access(executable):
|
||||
raise aiohttp.web.HTTPForbidden(text="{} has no privileged access to {}.".format(executable, tap_device))
|
||||
# FIXME: check for permissions on tap device
|
||||
# if not self._has_privileged_access(executable):
|
||||
# raise aiohttp.web.HTTPForbidden(text="{} has no privileged access to {}.".format(executable, tap_device))
|
||||
nio = NIOTAP(tap_device)
|
||||
elif nio_settings["type"] == "nio_generic_ethernet":
|
||||
nio = NIOGenericEthernet(nio_settings["ethernet_device"])
|
||||
@ -380,8 +381,9 @@ class BaseManager:
|
||||
:return: file path
|
||||
"""
|
||||
|
||||
if not path:
|
||||
return ""
|
||||
img_directory = self.get_images_directory()
|
||||
|
||||
if not os.path.isabs(path):
|
||||
s = os.path.split(path)
|
||||
return os.path.normpath(os.path.join(img_directory, *s))
|
||||
@ -397,6 +399,8 @@ class BaseManager:
|
||||
:return: file path
|
||||
"""
|
||||
|
||||
if not path:
|
||||
return ""
|
||||
img_directory = self.get_images_directory()
|
||||
path = self.get_abs_image_path(path)
|
||||
if os.path.dirname(path) == img_directory:
|
||||
@ -426,4 +430,5 @@ class BaseManager:
|
||||
"""
|
||||
Get the image directory on disk
|
||||
"""
|
||||
|
||||
raise NotImplementedError
|
||||
|
@ -208,7 +208,7 @@ class Dynamips(BaseManager):
|
||||
"""
|
||||
|
||||
# save the configs when the project is committed
|
||||
for vm in self._vms.values():
|
||||
for vm in self._vms.copy().values():
|
||||
if vm.project.id == project.id:
|
||||
yield from vm.save_configs()
|
||||
|
||||
|
@ -284,7 +284,7 @@ class DynamipsHypervisor:
|
||||
if not line:
|
||||
raise DynamipsError("No data returned from {host}:{port}, Dynamips process running: {run}"
|
||||
.format(host=self._host, port=self._port, run=self.is_running()))
|
||||
buf += line.decode()
|
||||
buf += line.decode("utf-8")
|
||||
except OSError as e:
|
||||
raise DynamipsError("Lost communication with {host}:{port} :{error}, Dynamips process running: {run}"
|
||||
.format(host=self._host, port=self._port, error=e, run=self.is_running()))
|
||||
|
@ -158,8 +158,8 @@ class Hypervisor(DynamipsHypervisor):
|
||||
output = ""
|
||||
if self._stdout_file and os.access(self._stdout_file, os.R_OK):
|
||||
try:
|
||||
with open(self._stdout_file, errors="replace") as file:
|
||||
output = file.read()
|
||||
with open(self._stdout_file, "rb") as file:
|
||||
output = file.read().decode("utf-8", errors="replace")
|
||||
except OSError as e:
|
||||
log.warn("could not read {}: {}".format(self._stdout_file, e))
|
||||
return output
|
||||
|
@ -120,7 +120,7 @@ class C7200(Router):
|
||||
npe-225, npe-300, npe-400 and npe-g2 (PowerPC c7200 only)
|
||||
"""
|
||||
|
||||
if self.is_running():
|
||||
if (yield from self.is_running()):
|
||||
raise DynamipsError("Cannot change NPE on running router")
|
||||
|
||||
yield from self._hypervisor.send('c7200 set_npe "{name}" {npe}'.format(name=self._name, npe=npe))
|
||||
|
@ -1396,7 +1396,7 @@ class Router(BaseVM):
|
||||
startup_config_path = os.path.join(module_workdir, "configs", "i{}_startup-config.cfg".format(self._dynamips_id))
|
||||
if os.path.isfile(startup_config_path):
|
||||
try:
|
||||
with open(startup_config_path, "r+", errors="replace") as f:
|
||||
with open(startup_config_path, "r+", encoding="utf-8", errors="replace") as f:
|
||||
old_config = f.read()
|
||||
new_config = old_config.replace(self.name, new_name)
|
||||
f.seek(0)
|
||||
@ -1409,7 +1409,7 @@ class Router(BaseVM):
|
||||
private_config_path = os.path.join(module_workdir, "configs", "i{}_private-config.cfg".format(self._dynamips_id))
|
||||
if os.path.isfile(private_config_path):
|
||||
try:
|
||||
with open(private_config_path, "r+", errors="replace") as f:
|
||||
with open(private_config_path, "r+", encoding="utf-8", errors="replace") as f:
|
||||
old_config = f.read()
|
||||
new_config = old_config.replace(self.name, new_name)
|
||||
f.seek(0)
|
||||
@ -1484,7 +1484,7 @@ class Router(BaseVM):
|
||||
startup_config_base64, private_config_base64 = yield from self.extract_config()
|
||||
if startup_config_base64:
|
||||
try:
|
||||
config = base64.b64decode(startup_config_base64).decode(errors='replace')
|
||||
config = base64.b64decode(startup_config_base64).decode("utf-8", errors="replace")
|
||||
config = "!\n" + config.replace("\r", "")
|
||||
config_path = os.path.join(module_workdir, self.startup_config)
|
||||
with open(config_path, "wb") as f:
|
||||
@ -1495,7 +1495,7 @@ class Router(BaseVM):
|
||||
|
||||
if private_config_base64:
|
||||
try:
|
||||
config = base64.b64decode(private_config_base64).decode(errors='replace')
|
||||
config = base64.b64decode(private_config_base64).decode("utf-8", errors="replace")
|
||||
config = "!\n" + config.replace("\r", "")
|
||||
config_path = os.path.join(module_workdir, self.private_config)
|
||||
with open(config_path, "wb") as f:
|
||||
|
@ -332,8 +332,8 @@ class IOUVM(BaseVM):
|
||||
def iourc_content(self):
|
||||
|
||||
try:
|
||||
with open(os.path.join(self.temporary_directory, "iourc")) as f:
|
||||
return f.read()
|
||||
with open(os.path.join(self.temporary_directory, "iourc"), "rb") as f:
|
||||
return f.read().decode("utf-8")
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
@ -343,8 +343,8 @@ class IOUVM(BaseVM):
|
||||
if value is not None:
|
||||
path = os.path.join(self.temporary_directory, "iourc")
|
||||
try:
|
||||
with open(path, "w+") as f:
|
||||
f.write(value)
|
||||
with open(path, "wb+") as f:
|
||||
f.write(value.encode("utf-8"))
|
||||
except OSError as e:
|
||||
raise IOUError("Could not write the iourc file {}: {}".format(path, e))
|
||||
|
||||
@ -378,12 +378,14 @@ class IOUVM(BaseVM):
|
||||
|
||||
config = configparser.ConfigParser()
|
||||
try:
|
||||
with open(self.iourc_path) as f:
|
||||
with open(self.iourc_path, encoding="utf-8") as f:
|
||||
config.read_file(f)
|
||||
except OSError as e:
|
||||
raise IOUError("Could not open iourc file {}: {}".format(self.iourc_path, e))
|
||||
except configparser.Error as e:
|
||||
raise IOUError("Could not parse iourc file {}: {}".format(self.iourc_path, e))
|
||||
except UnicodeDecodeError as e:
|
||||
raise IOUError("Invalid iourc file {}: {}".format(self.iourc_path, e))
|
||||
if "license" not in config:
|
||||
raise IOUError("License section not found in iourc file {}".format(self.iourc_path))
|
||||
hostname = socket.gethostname()
|
||||
@ -455,7 +457,7 @@ class IOUVM(BaseVM):
|
||||
log.info("Starting IOU: {}".format(self._command))
|
||||
self._iou_stdout_file = os.path.join(self.working_dir, "iou.log")
|
||||
log.info("Logging to {}".format(self._iou_stdout_file))
|
||||
with open(self._iou_stdout_file, "w") as fd:
|
||||
with open(self._iou_stdout_file, "w", encoding="utf-8") as fd:
|
||||
self._iou_process = yield from asyncio.create_subprocess_exec(*self._command,
|
||||
stdout=fd,
|
||||
stderr=subprocess.STDOUT,
|
||||
@ -499,7 +501,7 @@ class IOUVM(BaseVM):
|
||||
log.info("starting iouyap: {}".format(command))
|
||||
self._iouyap_stdout_file = os.path.join(self.working_dir, "iouyap.log")
|
||||
log.info("logging to {}".format(self._iouyap_stdout_file))
|
||||
with open(self._iouyap_stdout_file, "w") as fd:
|
||||
with open(self._iouyap_stdout_file, "w", encoding="utf-8") as fd:
|
||||
self._iouyap_process = yield from asyncio.create_subprocess_exec(*command,
|
||||
stdout=fd,
|
||||
stderr=subprocess.STDOUT,
|
||||
@ -565,7 +567,7 @@ class IOUVM(BaseVM):
|
||||
bay_id += 1
|
||||
|
||||
try:
|
||||
with open(iouyap_ini, "w") as config_file:
|
||||
with open(iouyap_ini, "w", encoding="utf-8") as config_file:
|
||||
config.write(config_file)
|
||||
log.info("IOU {name} [id={id}]: iouyap.ini updated".format(name=self._name,
|
||||
id=self._id))
|
||||
@ -587,7 +589,6 @@ class IOUVM(BaseVM):
|
||||
self._ioucon_thread = None
|
||||
|
||||
self._terminate_process_iou()
|
||||
|
||||
if self._iou_process.returncode is None:
|
||||
try:
|
||||
yield from gns3server.utils.asyncio.wait_for_process_termination(self._iou_process, timeout=3)
|
||||
@ -671,7 +672,7 @@ class IOUVM(BaseVM):
|
||||
|
||||
netmap_path = os.path.join(self.working_dir, "NETMAP")
|
||||
try:
|
||||
with open(netmap_path, "w") as f:
|
||||
with open(netmap_path, "w", encoding="utf-8") as f:
|
||||
for bay in range(0, 16):
|
||||
for unit in range(0, 4):
|
||||
f.write("{iouyap_id}:{bay}/{unit}{iou_id:>5d}:{bay}/{unit}\n".format(iouyap_id=str(self.application_id + 512),
|
||||
@ -741,8 +742,8 @@ class IOUVM(BaseVM):
|
||||
output = ""
|
||||
if self._iou_stdout_file:
|
||||
try:
|
||||
with open(self._iou_stdout_file, errors="replace") as file:
|
||||
output = file.read()
|
||||
with open(self._iou_stdout_file, "rb") as file:
|
||||
output = file.read().decode("utf-8", errors="replace")
|
||||
except OSError as e:
|
||||
log.warn("could not read {}: {}".format(self._iou_stdout_file, e))
|
||||
return output
|
||||
@ -756,8 +757,8 @@ class IOUVM(BaseVM):
|
||||
output = ""
|
||||
if self._iouyap_stdout_file:
|
||||
try:
|
||||
with open(self._iouyap_stdout_file, errors="replace") as file:
|
||||
output = file.read()
|
||||
with open(self._iouyap_stdout_file, "rb") as file:
|
||||
output = file.read().decode("utf-8", errors="replace")
|
||||
except OSError as e:
|
||||
log.warn("could not read {}: {}".format(self._iouyap_stdout_file, e))
|
||||
return output
|
||||
@ -949,8 +950,8 @@ class IOUVM(BaseVM):
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(config_file) as f:
|
||||
return f.read()
|
||||
with open(config_file, "rb") as f:
|
||||
return f.read().decode("utf-8", errors="replace")
|
||||
except OSError as e:
|
||||
raise IOUError("Can't read configuration file '{}': {}".format(config_file, e))
|
||||
|
||||
@ -963,23 +964,23 @@ class IOUVM(BaseVM):
|
||||
"""
|
||||
|
||||
try:
|
||||
script_file = os.path.join(self.working_dir, "initial-config.cfg")
|
||||
initial_config_path = os.path.join(self.working_dir, "initial-config.cfg")
|
||||
|
||||
if initial_config is None:
|
||||
initial_config = ''
|
||||
|
||||
# We disallow erasing the initial config file
|
||||
if len(initial_config) == 0 and os.path.exists(script_file):
|
||||
if len(initial_config) == 0 and os.path.exists(initial_config_path):
|
||||
return
|
||||
|
||||
with open(script_file, 'w+') as f:
|
||||
with open(initial_config_path, 'w+', encoding='utf-8') as f:
|
||||
if len(initial_config) == 0:
|
||||
f.write('')
|
||||
else:
|
||||
initial_config = initial_config.replace("%h", self._name)
|
||||
f.write(initial_config)
|
||||
except OSError as e:
|
||||
raise IOUError("Can't write initial configuration file '{}': {}".format(script_file, e))
|
||||
raise IOUError("Can't write initial configuration file '{}': {}".format(initial_config_path, e))
|
||||
|
||||
@property
|
||||
def initial_config_file(self):
|
||||
@ -1070,6 +1071,10 @@ class IOUVM(BaseVM):
|
||||
port_number=port_number))
|
||||
|
||||
nio = adapter.get_nio(port_number)
|
||||
if not nio:
|
||||
raise IOUError("NIO {port_number} does not exist in adapter {adapter}".format(adapter=adapter,
|
||||
port_number=port_number))
|
||||
|
||||
nio.stopPacketCapture()
|
||||
log.info('IOU "{name}" [{id}]: stopping packet capture on {adapter_number}/{port_number}'.format(name=self._name,
|
||||
id=self._id,
|
||||
|
@ -102,6 +102,7 @@ class Project:
|
||||
|
||||
server_config = Config.instance().get_section_config("Server")
|
||||
path = os.path.expanduser(server_config.get("projects_path", "~/GNS3/projects"))
|
||||
path = os.path.normpath(path)
|
||||
try:
|
||||
os.makedirs(path, exist_ok=True)
|
||||
except OSError as e:
|
||||
|
@ -30,6 +30,9 @@ from ..base_manager import BaseManager
|
||||
from .qemu_error import QemuError
|
||||
from .qemu_vm import QemuVM
|
||||
|
||||
import logging
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Qemu(BaseManager):
|
||||
|
||||
@ -44,7 +47,15 @@ class Qemu(BaseManager):
|
||||
"""
|
||||
|
||||
qemus = []
|
||||
paths = [os.getcwd()] + os.environ["PATH"].split(os.pathsep)
|
||||
paths = []
|
||||
try:
|
||||
paths.append(os.getcwd())
|
||||
except FileNotFoundError:
|
||||
log.warning("The current working directory doesn't exist")
|
||||
if "PATH" in os.environ:
|
||||
paths.extend(os.environ["PATH"].split(os.pathsep))
|
||||
else:
|
||||
log.warning("The PATH environment variable doesn't exist")
|
||||
# look for Qemu binaries in the current working directory and $PATH
|
||||
if sys.platform.startswith("win"):
|
||||
# add specific Windows paths
|
||||
|
@ -32,6 +32,7 @@ import socket
|
||||
from .qemu_error import QemuError
|
||||
from ..adapters.ethernet_adapter import EthernetAdapter
|
||||
from ..nios.nio_udp import NIOUDP
|
||||
from ..nios.nio_tap import NIOTAP
|
||||
from ..base_vm import BaseVM
|
||||
from ...schemas.qemu import QEMU_OBJECT_SCHEMA
|
||||
|
||||
@ -574,7 +575,7 @@ class QemuVM(BaseVM):
|
||||
log.info("Starting QEMU: {}".format(self._command))
|
||||
self._stdout_file = os.path.join(self.working_dir, "qemu.log")
|
||||
log.info("logging to {}".format(self._stdout_file))
|
||||
with open(self._stdout_file, "w") as fd:
|
||||
with open(self._stdout_file, "w", encoding="utf-8") as fd:
|
||||
self._process = yield from asyncio.create_subprocess_exec(*self._command,
|
||||
stdout=fd,
|
||||
stderr=subprocess.STDOUT,
|
||||
@ -644,7 +645,7 @@ class QemuVM(BaseVM):
|
||||
break
|
||||
for expect in expected:
|
||||
if expect in line:
|
||||
result = line.decode().strip()
|
||||
result = line.decode("utf-8").strip()
|
||||
break
|
||||
except EOFError as e:
|
||||
log.warn("Could not read from QEMU monitor: {}".format(e))
|
||||
@ -817,8 +818,8 @@ class QemuVM(BaseVM):
|
||||
output = ""
|
||||
if self._stdout_file:
|
||||
try:
|
||||
with open(self._stdout_file, errors="replace") as file:
|
||||
output = file.read()
|
||||
with open(self._stdout_file, "rb") as file:
|
||||
output = file.read().decode("utf-8", errors="replace")
|
||||
except OSError as e:
|
||||
log.warn("Could not read {}: {}".format(self._stdout_file, e))
|
||||
return output
|
||||
@ -995,19 +996,25 @@ class QemuVM(BaseVM):
|
||||
else:
|
||||
network_options.extend(["-device", "{},mac={},netdev=gns3-{}".format(self._adapter_type, mac, adapter_number)])
|
||||
nio = adapter.get_nio(0)
|
||||
if nio and isinstance(nio, NIOUDP):
|
||||
if self._legacy_networking:
|
||||
network_options.extend(["-net", "udp,vlan={},name=gns3-{},sport={},dport={},daddr={}".format(adapter_number,
|
||||
adapter_number,
|
||||
nio.lport,
|
||||
nio.rport,
|
||||
nio.rhost)])
|
||||
else:
|
||||
network_options.extend(["-netdev", "socket,id=gns3-{},udp={}:{},localaddr={}:{}".format(adapter_number,
|
||||
nio.rhost,
|
||||
nio.rport,
|
||||
self._host,
|
||||
nio.lport)])
|
||||
if nio:
|
||||
if isinstance(nio, NIOUDP):
|
||||
if self._legacy_networking:
|
||||
network_options.extend(["-net", "udp,vlan={},name=gns3-{},sport={},dport={},daddr={}".format(adapter_number,
|
||||
adapter_number,
|
||||
nio.lport,
|
||||
nio.rport,
|
||||
nio.rhost)])
|
||||
else:
|
||||
network_options.extend(["-netdev", "socket,id=gns3-{},udp={}:{},localaddr={}:{}".format(adapter_number,
|
||||
nio.rhost,
|
||||
nio.rport,
|
||||
self._host,
|
||||
nio.lport)])
|
||||
elif isinstance(nio, NIOTAP):
|
||||
if self._legacy_networking:
|
||||
network_options.extend(["-net", "tap,name=gns3-{},ifname={}".format(adapter_number, nio.tap_device)])
|
||||
else:
|
||||
network_options.extend(["-netdev", "tap,id=gns3-{},ifname={}".format(adapter_number, nio.tap_device)])
|
||||
else:
|
||||
if self._legacy_networking:
|
||||
network_options.extend(["-net", "user,vlan={},name=gns3-{}".format(adapter_number, adapter_number)])
|
||||
|
@ -110,7 +110,6 @@ class VirtualBox(BaseManager):
|
||||
raise VirtualBoxError("VBoxManage has timed out after {} seconds!".format(timeout))
|
||||
|
||||
if process.returncode:
|
||||
# only the first line of the output is useful
|
||||
vboxmanage_error = stderr_data.decode("utf-8", errors="ignore")
|
||||
raise VirtualBoxError("VirtualBox has returned an error: {}".format(vboxmanage_error))
|
||||
|
||||
@ -133,6 +132,7 @@ class VirtualBox(BaseManager):
|
||||
if not extra_data[0].strip() == "Value: yes":
|
||||
# get the amount of RAM
|
||||
info_results = yield from self.execute("showvminfo", [vmname, "--machinereadable"])
|
||||
ram = 0
|
||||
for info in info_results:
|
||||
try:
|
||||
name, value = info.split('=', 1)
|
||||
|
@ -209,7 +209,7 @@ class VirtualBoxVM(BaseVM):
|
||||
log.info("VirtualBox VM '{name}' [{id}] stopped".format(name=self.name, id=self.id))
|
||||
log.debug("Stop result: {}".format(result))
|
||||
|
||||
yield from asyncio.sleep(0.5) # give some time for VirtualBox to unlock the VM
|
||||
# yield from asyncio.sleep(0.5) # give some time for VirtualBox to unlock the VM
|
||||
try:
|
||||
# deactivate the first serial port
|
||||
yield from self._modify_vm("--uart1 off")
|
||||
@ -276,7 +276,7 @@ class VirtualBoxVM(BaseVM):
|
||||
|
||||
hdd_info_file = os.path.join(self.working_dir, self._vmname, "hdd_info.json")
|
||||
try:
|
||||
with open(hdd_info_file, "r") as f:
|
||||
with open(hdd_info_file, "r", encoding="utf-8") as f:
|
||||
hdd_table = json.load(f)
|
||||
except OSError as e:
|
||||
raise VirtualBoxError("Could not read HDD info file: {}".format(e))
|
||||
@ -321,6 +321,7 @@ class VirtualBoxVM(BaseVM):
|
||||
|
||||
if self._linked_clone:
|
||||
hdd_table = []
|
||||
hdd_files_to_close = []
|
||||
if os.path.exists(self.working_dir):
|
||||
hdd_files = yield from self._get_all_hdd_files()
|
||||
vm_info = yield from self._get_vm_info()
|
||||
@ -331,6 +332,7 @@ class VirtualBoxVM(BaseVM):
|
||||
port = match.group(2)
|
||||
device = match.group(3)
|
||||
if value in hdd_files:
|
||||
hdd_files_to_close.append(value)
|
||||
log.info("VirtualBox VM '{name}' [{id}] detaching HDD {controller} {port} {device}".format(name=self.name,
|
||||
id=self.id,
|
||||
controller=controller,
|
||||
@ -351,10 +353,16 @@ class VirtualBoxVM(BaseVM):
|
||||
log.info("VirtualBox VM '{name}' [{id}] unregistering".format(name=self.name, id=self.id))
|
||||
yield from self.manager.execute("unregistervm", [self._name])
|
||||
|
||||
for hdd_file in hdd_files_to_close:
|
||||
log.info("VirtualBox VM '{name}' [{id}] closing disk {disk}".format(name=self.name,
|
||||
id=self.id,
|
||||
disk=os.path.basename(hdd_file)))
|
||||
yield from self.manager.execute("closemedium", ["disk", hdd_file])
|
||||
|
||||
if hdd_table:
|
||||
try:
|
||||
hdd_info_file = os.path.join(self.working_dir, self._vmname, "hdd_info.json")
|
||||
with open(hdd_info_file, "w") as f:
|
||||
with open(hdd_info_file, "w", encoding="utf-8") as f:
|
||||
json.dump(hdd_table, f, indent=4)
|
||||
except OSError as e:
|
||||
log.warning("VirtualBox VM '{name}' [{id}] could not write HHD info file: {error}".format(name=self.name,
|
||||
@ -586,12 +594,14 @@ class VirtualBoxVM(BaseVM):
|
||||
:returns: pipe path (string)
|
||||
"""
|
||||
|
||||
p = re.compile('\s+', re.UNICODE)
|
||||
pipe_name = p.sub("_", self._vmname)
|
||||
if sys.platform.startswith("win"):
|
||||
pipe_name = r"\\.\pipe\VBOX\{}".format(pipe_name)
|
||||
pipe_name = r"\\.\pipe\gns3_vbox\{}".format(self.id)
|
||||
else:
|
||||
pipe_name = os.path.join(tempfile.gettempdir(), "pipe_{}".format(pipe_name))
|
||||
pipe_name = os.path.join(tempfile.gettempdir(), "gns3_vbox", "{}".format(self.id))
|
||||
try:
|
||||
os.makedirs(os.path.dirname(pipe_name), exist_ok=True)
|
||||
except OSError as e:
|
||||
raise VirtualBoxError("Could not create the VirtualBox pipe directory: {}".format(e))
|
||||
return pipe_name
|
||||
|
||||
@asyncio.coroutine
|
||||
|
@ -27,6 +27,7 @@ import signal
|
||||
import re
|
||||
import asyncio
|
||||
import shutil
|
||||
import gns3server.utils.asyncio
|
||||
|
||||
from pkg_resources import parse_version
|
||||
from .vpcs_error import VPCSError
|
||||
@ -166,8 +167,8 @@ class VPCSVM(BaseVM):
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(script_file) as f:
|
||||
return f.read()
|
||||
with open(script_file, "rb") as f:
|
||||
return f.read().decode("utf-8", errors="replace")
|
||||
except OSError as e:
|
||||
raise VPCSError('Cannot read the startup script file "{}": {}'.format(script_file, e))
|
||||
|
||||
@ -181,12 +182,12 @@ class VPCSVM(BaseVM):
|
||||
|
||||
try:
|
||||
script_file = os.path.join(self.working_dir, 'startup.vpc')
|
||||
with open(script_file, 'w+') as f:
|
||||
with open(script_file, "wb+") as f:
|
||||
if startup_script is None:
|
||||
f.write('')
|
||||
f.write(b'')
|
||||
else:
|
||||
startup_script = startup_script.replace("%h", self._name)
|
||||
f.write(startup_script)
|
||||
f.write(startup_script.encode("utf-8"))
|
||||
except OSError as e:
|
||||
raise VPCSError('Cannot write the startup script file "{}": {}'.format(self.script_file, e))
|
||||
|
||||
@ -226,7 +227,7 @@ class VPCSVM(BaseVM):
|
||||
flags = 0
|
||||
if sys.platform.startswith("win32"):
|
||||
flags = subprocess.CREATE_NEW_PROCESS_GROUP
|
||||
with open(self._vpcs_stdout_file, "w") as fd:
|
||||
with open(self._vpcs_stdout_file, "w", encoding="utf-8") as fd:
|
||||
self._process = yield from asyncio.create_subprocess_exec(*self._command,
|
||||
stdout=fd,
|
||||
stderr=subprocess.STDOUT,
|
||||
@ -247,12 +248,13 @@ class VPCSVM(BaseVM):
|
||||
|
||||
if self.is_running():
|
||||
self._terminate_process()
|
||||
try:
|
||||
yield from asyncio.wait_for(self._process.wait(), timeout=3)
|
||||
except asyncio.TimeoutError:
|
||||
if self._process.returncode is None:
|
||||
log.warn("VPCS process {} is still running... killing it".format(self._process.pid))
|
||||
self._process.kill()
|
||||
if self._process.returncode is None:
|
||||
try:
|
||||
yield from gns3server.utils.asyncio.wait_for_process_termination(self._process, timeout=3)
|
||||
except asyncio.TimeoutError:
|
||||
if self._process.returncode is None:
|
||||
log.warn("VPCS process {} is still running... killing it".format(self._process.pid))
|
||||
self._process.kill()
|
||||
|
||||
self._process = None
|
||||
self._started = False
|
||||
@ -290,8 +292,8 @@ class VPCSVM(BaseVM):
|
||||
output = ""
|
||||
if self._vpcs_stdout_file:
|
||||
try:
|
||||
with open(self._vpcs_stdout_file, errors="replace") as file:
|
||||
output = file.read()
|
||||
with open(self._vpcs_stdout_file, "rb") as file:
|
||||
output = file.read().decode("utf-8", errors="replace")
|
||||
except OSError as e:
|
||||
log.warn("Could not read {}: {}".format(self._vpcs_stdout_file, e))
|
||||
return output
|
||||
@ -407,7 +409,7 @@ class VPCSVM(BaseVM):
|
||||
command.extend(["-F"]) # option to avoid the daemonization of VPCS
|
||||
|
||||
if self.script_file:
|
||||
command.extend([self.script_file])
|
||||
command.extend([os.path.basename(self.script_file)])
|
||||
return command
|
||||
|
||||
@property
|
||||
|
@ -473,147 +473,6 @@ VM_UPDATE_SCHEMA = {
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
VM_NIO_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation to add a NIO for a Dynamips VM instance",
|
||||
"type": "object",
|
||||
"definitions": {
|
||||
"UDP": {
|
||||
"description": "UDP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_udp"]
|
||||
},
|
||||
"lport": {
|
||||
"description": "Local port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
},
|
||||
"rhost": {
|
||||
"description": "Remote host",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"rport": {
|
||||
"description": "Remote port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
}
|
||||
},
|
||||
"required": ["type", "lport", "rhost", "rport"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"Ethernet": {
|
||||
"description": "Generic Ethernet Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_generic_ethernet"]
|
||||
},
|
||||
"ethernet_device": {
|
||||
"description": "Ethernet device name e.g. eth0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "ethernet_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"LinuxEthernet": {
|
||||
"description": "Linux Ethernet Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_linux_ethernet"]
|
||||
},
|
||||
"ethernet_device": {
|
||||
"description": "Ethernet device name e.g. eth0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "ethernet_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"TAP": {
|
||||
"description": "TAP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_tap"]
|
||||
},
|
||||
"tap_device": {
|
||||
"description": "TAP device name e.g. tap0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "tap_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"UNIX": {
|
||||
"description": "UNIX Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_unix"]
|
||||
},
|
||||
"local_file": {
|
||||
"description": "path to the UNIX socket file (local)",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"remote_file": {
|
||||
"description": "path to the UNIX socket file (remote)",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "local_file", "remote_file"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"VDE": {
|
||||
"description": "VDE Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_vde"]
|
||||
},
|
||||
"control_file": {
|
||||
"description": "path to the VDE control file",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"local_file": {
|
||||
"description": "path to the VDE control file",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "control_file", "local_file"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"NULL": {
|
||||
"description": "NULL Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_null"]
|
||||
},
|
||||
},
|
||||
"required": ["type"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
},
|
||||
"oneOf": [
|
||||
{"$ref": "#/definitions/UDP"},
|
||||
{"$ref": "#/definitions/Ethernet"},
|
||||
{"$ref": "#/definitions/LinuxEthernet"},
|
||||
{"$ref": "#/definitions/TAP"},
|
||||
{"$ref": "#/definitions/UNIX"},
|
||||
{"$ref": "#/definitions/VDE"},
|
||||
{"$ref": "#/definitions/NULL"},
|
||||
],
|
||||
"additionalProperties": True,
|
||||
"required": ["type"]
|
||||
}
|
||||
|
||||
VM_CAPTURE_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation to start a packet capture on a Dynamips VM instance port",
|
||||
|
@ -214,78 +214,6 @@ IOU_OBJECT_SCHEMA = {
|
||||
"required": ["name", "vm_id", "console", "project_id", "path", "serial_adapters", "ethernet_adapters", "ram", "nvram", "l1_keepalives", "initial_config", "use_default_iou_values"]
|
||||
}
|
||||
|
||||
IOU_NIO_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation to add a NIO for a IOU instance",
|
||||
"type": "object",
|
||||
"definitions": {
|
||||
"UDP": {
|
||||
"description": "UDP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_udp"]
|
||||
},
|
||||
"lport": {
|
||||
"description": "Local port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
},
|
||||
"rhost": {
|
||||
"description": "Remote host",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"rport": {
|
||||
"description": "Remote port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
}
|
||||
},
|
||||
"required": ["type", "lport", "rhost", "rport"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"Ethernet": {
|
||||
"description": "Generic Ethernet Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_generic_ethernet"]
|
||||
},
|
||||
"ethernet_device": {
|
||||
"description": "Ethernet device name e.g. eth0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "ethernet_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"TAP": {
|
||||
"description": "TAP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_tap"]
|
||||
},
|
||||
"tap_device": {
|
||||
"description": "TAP device name e.g. tap0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "tap_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
},
|
||||
"oneOf": [
|
||||
{"$ref": "#/definitions/UDP"},
|
||||
{"$ref": "#/definitions/Ethernet"},
|
||||
{"$ref": "#/definitions/TAP"},
|
||||
],
|
||||
"additionalProperties": True,
|
||||
"required": ["type"]
|
||||
}
|
||||
|
||||
IOU_CAPTURE_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation to start a packet capture on a IOU instance",
|
||||
|
158
gns3server/schemas/nio.py
Normal file
158
gns3server/schemas/nio.py
Normal file
@ -0,0 +1,158 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2015 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/>.
|
||||
|
||||
|
||||
NIO_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation to add a NIO for a VM instance",
|
||||
"type": "object",
|
||||
"definitions": {
|
||||
"UDP": {
|
||||
"description": "UDP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_udp"]
|
||||
},
|
||||
"lport": {
|
||||
"description": "Local port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
},
|
||||
"rhost": {
|
||||
"description": "Remote host",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"rport": {
|
||||
"description": "Remote port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
}
|
||||
},
|
||||
"required": ["type", "lport", "rhost", "rport"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"Ethernet": {
|
||||
"description": "Generic Ethernet Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_generic_ethernet"]
|
||||
},
|
||||
"ethernet_device": {
|
||||
"description": "Ethernet device name e.g. eth0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "ethernet_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"LinuxEthernet": {
|
||||
"description": "Linux Ethernet Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_linux_ethernet"]
|
||||
},
|
||||
"ethernet_device": {
|
||||
"description": "Ethernet device name e.g. eth0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "ethernet_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"TAP": {
|
||||
"description": "TAP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_tap"]
|
||||
},
|
||||
"tap_device": {
|
||||
"description": "TAP device name e.g. tap0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "tap_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"UNIX": {
|
||||
"description": "UNIX Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_unix"]
|
||||
},
|
||||
"local_file": {
|
||||
"description": "path to the UNIX socket file (local)",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"remote_file": {
|
||||
"description": "path to the UNIX socket file (remote)",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "local_file", "remote_file"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"VDE": {
|
||||
"description": "VDE Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_vde"]
|
||||
},
|
||||
"control_file": {
|
||||
"description": "path to the VDE control file",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"local_file": {
|
||||
"description": "path to the VDE control file",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "control_file", "local_file"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"NULL": {
|
||||
"description": "NULL Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_null"]
|
||||
},
|
||||
},
|
||||
"required": ["type"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
},
|
||||
"oneOf": [
|
||||
{"$ref": "#/definitions/UDP"},
|
||||
{"$ref": "#/definitions/Ethernet"},
|
||||
{"$ref": "#/definitions/LinuxEthernet"},
|
||||
{"$ref": "#/definitions/TAP"},
|
||||
{"$ref": "#/definitions/UNIX"},
|
||||
{"$ref": "#/definitions/VDE"},
|
||||
{"$ref": "#/definitions/NULL"},
|
||||
],
|
||||
"additionalProperties": True,
|
||||
"required": ["type"]
|
||||
}
|
@ -211,62 +211,6 @@ QEMU_UPDATE_SCHEMA = {
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
QEMU_NIO_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation to add a NIO for a QEMU instance",
|
||||
"type": "object",
|
||||
"definitions": {
|
||||
"UDP": {
|
||||
"description": "UDP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_udp"]
|
||||
},
|
||||
"lport": {
|
||||
"description": "Local port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
},
|
||||
"rhost": {
|
||||
"description": "Remote host",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"rport": {
|
||||
"description": "Remote port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
}
|
||||
},
|
||||
"required": ["type", "lport", "rhost", "rport"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"Ethernet": {
|
||||
"description": "Generic Ethernet Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_generic_ethernet"]
|
||||
},
|
||||
"ethernet_device": {
|
||||
"description": "Ethernet device name e.g. eth0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "ethernet_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
},
|
||||
"oneOf": [
|
||||
{"$ref": "#/definitions/UDP"},
|
||||
{"$ref": "#/definitions/Ethernet"},
|
||||
],
|
||||
"additionalProperties": True,
|
||||
"required": ["type"]
|
||||
}
|
||||
|
||||
QEMU_OBJECT_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation for a QEMU VM instance",
|
||||
|
@ -139,46 +139,6 @@ VBOX_UPDATE_SCHEMA = {
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
VBOX_NIO_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation to add a NIO for a VirtualBox VM instance",
|
||||
"type": "object",
|
||||
"definitions": {
|
||||
"UDP": {
|
||||
"description": "UDP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_udp"]
|
||||
},
|
||||
"lport": {
|
||||
"description": "Local port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
},
|
||||
"rhost": {
|
||||
"description": "Remote host",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"rport": {
|
||||
"description": "Remote port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
}
|
||||
},
|
||||
"required": ["type", "lport", "rhost", "rport"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
},
|
||||
"oneOf": [
|
||||
{"$ref": "#/definitions/UDP"},
|
||||
],
|
||||
"additionalProperties": True,
|
||||
"required": ["type"]
|
||||
}
|
||||
|
||||
VBOX_CAPTURE_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation to start a packet capture on a VirtualBox VM instance port",
|
||||
|
@ -75,62 +75,6 @@ VPCS_UPDATE_SCHEMA = {
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
VPCS_NIO_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "Request validation to add a NIO for a VPCS instance",
|
||||
"type": "object",
|
||||
"definitions": {
|
||||
"UDP": {
|
||||
"description": "UDP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_udp"]
|
||||
},
|
||||
"lport": {
|
||||
"description": "Local port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
},
|
||||
"rhost": {
|
||||
"description": "Remote host",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
"rport": {
|
||||
"description": "Remote port",
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 65535
|
||||
}
|
||||
},
|
||||
"required": ["type", "lport", "rhost", "rport"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
"TAP": {
|
||||
"description": "TAP Network Input/Output",
|
||||
"properties": {
|
||||
"type": {
|
||||
"enum": ["nio_tap"]
|
||||
},
|
||||
"tap_device": {
|
||||
"description": "TAP device name e.g. tap0",
|
||||
"type": "string",
|
||||
"minLength": 1
|
||||
},
|
||||
},
|
||||
"required": ["type", "tap_device"],
|
||||
"additionalProperties": False
|
||||
},
|
||||
},
|
||||
"oneOf": [
|
||||
{"$ref": "#/definitions/UDP"},
|
||||
{"$ref": "#/definitions/TAP"},
|
||||
],
|
||||
"additionalProperties": True,
|
||||
"required": ["type"]
|
||||
}
|
||||
|
||||
VPCS_OBJECT_SCHEMA = {
|
||||
"$schema": "http://json-schema.org/draft-04/schema#",
|
||||
"description": "VPCS instance",
|
||||
|
@ -58,10 +58,11 @@ def get_windows_interfaces():
|
||||
|
||||
import win32com.client
|
||||
import pywintypes
|
||||
locator = win32com.client.Dispatch("WbemScripting.SWbemLocator")
|
||||
service = locator.ConnectServer(".", "root\cimv2")
|
||||
|
||||
interfaces = []
|
||||
try:
|
||||
locator = win32com.client.Dispatch("WbemScripting.SWbemLocator")
|
||||
service = locator.ConnectServer(".", "root\cimv2")
|
||||
# more info on Win32_NetworkAdapter: http://msdn.microsoft.com/en-us/library/aa394216%28v=vs.85%29.aspx
|
||||
for adapter in service.InstancesOf("Win32_NetworkAdapter"):
|
||||
if adapter.NetConnectionStatus == 2 or adapter.NetConnectionStatus == 7:
|
||||
@ -70,7 +71,7 @@ def get_windows_interfaces():
|
||||
interfaces.append({"id": npf_interface,
|
||||
"name": adapter.NetConnectionID})
|
||||
except (AttributeError, pywintypes.com_error):
|
||||
log.warn("could not use the COM service to retrieve interface info, trying using the registry...")
|
||||
log.warn("Could not use the COM service to retrieve interface info, trying using the registry...")
|
||||
return _get_windows_interfaces_from_registry()
|
||||
|
||||
return interfaces
|
||||
|
@ -30,10 +30,11 @@ renderer = jinja2.Environment(loader=jinja2.PackageLoader('gns3server', 'templat
|
||||
|
||||
class Response(aiohttp.web.Response):
|
||||
|
||||
def __init__(self, route=None, output_schema=None, headers={}, **kwargs):
|
||||
def __init__(self, request=None, route=None, output_schema=None, headers={}, **kwargs):
|
||||
|
||||
self._route = route
|
||||
self._output_schema = output_schema
|
||||
self._request = request
|
||||
headers['X-Route'] = self._route
|
||||
headers['Server'] = "Python/{0[0]}.{0[1]} GNS3/{1}".format(sys.version_info, __version__)
|
||||
super().__init__(headers=headers, **kwargs)
|
||||
@ -70,6 +71,7 @@ class Response(aiohttp.web.Response):
|
||||
"""
|
||||
template = renderer.get_template(template_filename)
|
||||
kwargs["gns3_version"] = __version__
|
||||
kwargs["gns3_host"] = self._request.host
|
||||
self.html(template.render(**kwargs))
|
||||
|
||||
def json(self, answer):
|
||||
|
@ -120,7 +120,7 @@ class Route(object):
|
||||
|
||||
# Non API call
|
||||
if api_version is None:
|
||||
response = Response(route=route, output_schema=output_schema)
|
||||
response = Response(request=request, route=route, output_schema=output_schema)
|
||||
yield from func(request, response)
|
||||
return response
|
||||
|
||||
@ -131,39 +131,39 @@ class Route(object):
|
||||
record_file = server_config.get("record")
|
||||
if record_file:
|
||||
try:
|
||||
with open(record_file, "a") as f:
|
||||
with open(record_file, "a", encoding="utf-8") as f:
|
||||
f.write("curl -X {} 'http://{}{}' -d '{}'".format(request.method, request.host, request.path_qs, json.dumps(request.json)))
|
||||
f.write("\n")
|
||||
except OSError as e:
|
||||
log.warn("Could not write to the record file {}: {}".format(record_file, e))
|
||||
response = Response(route=route, output_schema=output_schema)
|
||||
response = Response(request=request, route=route, output_schema=output_schema)
|
||||
yield from func(request, response)
|
||||
except aiohttp.web.HTTPBadRequest as e:
|
||||
response = Response(route=route)
|
||||
response = Response(request=request, route=route)
|
||||
response.set_status(e.status)
|
||||
response.json({"message": e.text, "status": e.status, "path": route, "request": request.json})
|
||||
except aiohttp.web.HTTPException as e:
|
||||
response = Response(route=route)
|
||||
response = Response(request=request, route=route)
|
||||
response.set_status(e.status)
|
||||
response.json({"message": e.text, "status": e.status})
|
||||
except VMError as e:
|
||||
log.error("VM error detected: {type}".format(type=type(e)), exc_info=1)
|
||||
response = Response(route=route)
|
||||
response = Response(request=request, route=route)
|
||||
response.set_status(409)
|
||||
response.json({"message": str(e), "status": 409})
|
||||
except asyncio.futures.CancelledError as e:
|
||||
log.error("Request canceled")
|
||||
response = Response(route=route)
|
||||
response = Response(request=request, route=route)
|
||||
response.set_status(408)
|
||||
response.json({"message": "Request canceled", "status": 408})
|
||||
except aiohttp.ClientDisconnectedError:
|
||||
log.error("Client disconnected")
|
||||
response = Response(route=route)
|
||||
response = Response(request=request, route=route)
|
||||
response.set_status(408)
|
||||
response.json({"message": "Client disconnected", "status": 408})
|
||||
except Exception as e:
|
||||
log.error("Uncaught exception detected: {type}".format(type=type(e)), exc_info=1)
|
||||
response = Response(route=route)
|
||||
response = Response(request=request, route=route)
|
||||
response.set_status(500)
|
||||
CrashReport.instance().capture_exception(request)
|
||||
exc_type, exc_value, exc_tb = sys.exc_info()
|
||||
|
@ -18,10 +18,14 @@
|
||||
import pytest
|
||||
import os
|
||||
import stat
|
||||
import sys
|
||||
import uuid
|
||||
|
||||
from tests.utils import asyncio_patch
|
||||
from unittest.mock import patch, MagicMock, PropertyMock
|
||||
|
||||
pytestmark = pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_iou_bin(tmpdir):
|
||||
@ -91,11 +95,32 @@ def test_iou_create_with_params(server, project, base_params):
|
||||
|
||||
assert "initial-config.cfg" in response.json["initial_config"]
|
||||
with open(initial_config_file(project, response.json)) as f:
|
||||
assert f.read() == params["initial_config_content"]
|
||||
assert f.read() == "hostname test"
|
||||
|
||||
assert "iourc" in response.json["iourc_path"]
|
||||
|
||||
|
||||
def test_iou_create_initial_config_already_exist(server, project, base_params):
|
||||
"""We don't erase an initial config if already exist at project creation"""
|
||||
|
||||
vm_id = str(uuid.uuid4())
|
||||
initial_config_file_path = initial_config_file(project, {'vm_id': vm_id})
|
||||
with open(initial_config_file_path, 'w+') as f:
|
||||
f.write("echo hello")
|
||||
|
||||
params = base_params
|
||||
params["vm_id"] = vm_id
|
||||
params["initial_config_content"] = "hostname test"
|
||||
|
||||
response = server.post("/projects/{project_id}/iou/vms".format(project_id=project.id), params, example=True)
|
||||
assert response.status == 201
|
||||
assert response.route == "/projects/{project_id}/iou/vms"
|
||||
|
||||
assert "initial-config.cfg" in response.json["initial_config"]
|
||||
with open(initial_config_file(project, response.json)) as f:
|
||||
assert f.read() == "echo hello"
|
||||
|
||||
|
||||
def test_iou_get(server, project, vm):
|
||||
response = server.get("/projects/{project_id}/iou/vms/{vm_id}".format(project_id=vm["project_id"], vm_id=vm["vm_id"]), example=True)
|
||||
assert response.status == 200
|
||||
|
@ -170,10 +170,7 @@ def test_qemu_nio_create_ethernet(server, vm):
|
||||
"ethernet_device": "eth0",
|
||||
},
|
||||
example=True)
|
||||
assert response.status == 201
|
||||
assert response.route == "/projects/{project_id}/qemu/vms/{vm_id}/adapters/{adapter_number:\d+}/ports/{port_number:\d+}/nio"
|
||||
assert response.json["type"] == "nio_generic_ethernet"
|
||||
assert response.json["ethernet_device"] == "eth0"
|
||||
assert response.status == 409
|
||||
|
||||
|
||||
def test_qemu_delete_nio(server, vm):
|
||||
|
@ -18,6 +18,7 @@
|
||||
|
||||
import pytest
|
||||
import tempfile
|
||||
import sys
|
||||
|
||||
from gns3server.modules.dynamips import Dynamips
|
||||
from gns3server.modules.dynamips.dynamips_error import DynamipsError
|
||||
@ -37,6 +38,7 @@ def test_vm_invalid_dynamips_path(manager):
|
||||
manager.find_dynamips()
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported by Windows")
|
||||
def test_vm_non_executable_dynamips_path(manager):
|
||||
tmpfile = tempfile.NamedTemporaryFile()
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"dynamips_path": tmpfile.name}):
|
||||
|
@ -20,10 +20,14 @@ import pytest
|
||||
from unittest.mock import patch
|
||||
import uuid
|
||||
import os
|
||||
import sys
|
||||
|
||||
pytestmark = pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
|
||||
if not sys.platform.startswith("win"):
|
||||
from gns3server.modules.iou import IOU
|
||||
from gns3server.modules.iou.iou_error import IOUError
|
||||
|
||||
from gns3server.modules.iou import IOU
|
||||
from gns3server.modules.iou.iou_error import IOUError
|
||||
from gns3server.modules.project_manager import ProjectManager
|
||||
|
||||
|
||||
|
@ -21,13 +21,19 @@ import asyncio
|
||||
import os
|
||||
import stat
|
||||
import socket
|
||||
import sys
|
||||
from tests.utils import asyncio_patch
|
||||
|
||||
|
||||
from unittest.mock import patch, MagicMock, PropertyMock
|
||||
from gns3server.modules.iou.iou_vm import IOUVM
|
||||
from gns3server.modules.iou.iou_error import IOUError
|
||||
from gns3server.modules.iou import IOU
|
||||
|
||||
pytestmark = pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
|
||||
if not sys.platform.startswith("win"):
|
||||
from gns3server.modules.iou.iou_vm import IOUVM
|
||||
from gns3server.modules.iou.iou_error import IOUError
|
||||
from gns3server.modules.iou import IOU
|
||||
|
||||
from gns3server.config import Config
|
||||
|
||||
|
||||
|
@ -18,6 +18,7 @@
|
||||
import os
|
||||
import stat
|
||||
import asyncio
|
||||
import sys
|
||||
|
||||
from gns3server.modules.qemu import Qemu
|
||||
from tests.utils import asyncio_patch
|
||||
@ -27,7 +28,10 @@ def test_get_qemu_version(loop):
|
||||
|
||||
with asyncio_patch("gns3server.modules.qemu.subprocess_check_output", return_value="QEMU emulator version 2.2.0, Copyright (c) 2003-2008 Fabrice Bellard") as mock:
|
||||
version = loop.run_until_complete(asyncio.async(Qemu._get_qemu_version("/tmp/qemu-test")))
|
||||
assert version == "2.2.0"
|
||||
if sys.platform.startswith("win"):
|
||||
assert version == ""
|
||||
else:
|
||||
assert version == "2.2.0"
|
||||
|
||||
|
||||
def test_binary_list(loop):
|
||||
@ -43,12 +47,17 @@ def test_binary_list(loop):
|
||||
with asyncio_patch("gns3server.modules.qemu.subprocess_check_output", return_value="QEMU emulator version 2.2.0, Copyright (c) 2003-2008 Fabrice Bellard") as mock:
|
||||
qemus = loop.run_until_complete(asyncio.async(Qemu.binary_list()))
|
||||
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-system-x86"), "version": "2.2.0"} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-kvm"), "version": "2.2.0"} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-system-x42"), "version": "2.2.0"} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "hello"), "version": "2.2.0"} not in qemus
|
||||
if sys.platform.startswith("win"):
|
||||
version = ""
|
||||
else:
|
||||
version = "2.2.0"
|
||||
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-system-x86"), "version": version} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-kvm"), "version": version} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "qemu-system-x42"), "version": version} in qemus
|
||||
assert {"path": os.path.join(os.environ["PATH"], "hello"), "version": version} not in qemus
|
||||
|
||||
|
||||
def test_get_legacy_vm_workdir():
|
||||
|
||||
assert Qemu.get_legacy_vm_workdir(42, "bla") == "qemu/vm-42"
|
||||
assert Qemu.get_legacy_vm_workdir(42, "bla") == os.path.join("qemu", "vm-42")
|
||||
|
@ -19,6 +19,7 @@ import pytest
|
||||
import aiohttp
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
import stat
|
||||
import re
|
||||
from tests.utils import asyncio_patch
|
||||
@ -51,7 +52,10 @@ def fake_qemu_img_binary():
|
||||
@pytest.fixture
|
||||
def fake_qemu_binary():
|
||||
|
||||
bin_path = os.path.join(os.environ["PATH"], "qemu_x42")
|
||||
if sys.platform.startswith("win"):
|
||||
bin_path = os.path.join(os.environ["PATH"], "qemu_x42.EXE")
|
||||
else:
|
||||
bin_path = os.path.join(os.environ["PATH"], "qemu_x42")
|
||||
with open(bin_path, "w+") as f:
|
||||
f.write("1")
|
||||
os.chmod(bin_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
|
||||
@ -178,8 +182,9 @@ def test_set_qemu_path(vm, tmpdir, fake_qemu_binary):
|
||||
f.write("1")
|
||||
|
||||
# Raise because file is not executable
|
||||
with pytest.raises(QemuError):
|
||||
vm.qemu_path = path
|
||||
if not sys.platform.startswith("win"):
|
||||
with pytest.raises(QemuError):
|
||||
vm.qemu_path = path
|
||||
|
||||
os.chmod(path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
|
||||
|
||||
@ -204,6 +209,7 @@ def test_disk_options(vm, loop, fake_qemu_img_binary):
|
||||
assert args == (fake_qemu_img_binary, "create", "-f", "qcow2", os.path.join(vm.working_dir, "flash.qcow2"), "256M")
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
def test_set_process_priority(vm, loop, fake_qemu_img_binary):
|
||||
|
||||
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()) as process:
|
||||
@ -274,6 +280,7 @@ def test_build_command(vm, loop, fake_qemu_binary, port_manager):
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
def test_build_command_without_display(vm, loop, fake_qemu_binary):
|
||||
|
||||
os.environ["DISPLAY"] = ""
|
||||
@ -282,7 +289,9 @@ def test_build_command_without_display(vm, loop, fake_qemu_binary):
|
||||
assert "-nographic" in cmd
|
||||
|
||||
|
||||
def test_build_command_witht_invalid_options(vm, loop, fake_qemu_binary):
|
||||
# Windows accept this kind of mistake
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
def test_build_command_with_invalid_options(vm, loop, fake_qemu_binary):
|
||||
|
||||
vm.options = "'test"
|
||||
with pytest.raises(QemuError):
|
||||
@ -306,6 +315,7 @@ def test_hdb_disk_image(vm, tmpdir):
|
||||
vm.hdb_disk_image = "test"
|
||||
assert vm.hdb_disk_image == str(tmpdir / "QEMU" / "test")
|
||||
|
||||
|
||||
def test_hdc_disk_image(vm, tmpdir):
|
||||
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"images_path": str(tmpdir)}):
|
||||
@ -314,6 +324,7 @@ def test_hdc_disk_image(vm, tmpdir):
|
||||
vm.hdc_disk_image = "test"
|
||||
assert vm.hdc_disk_image == str(tmpdir / "QEMU" / "test")
|
||||
|
||||
|
||||
def test_hdd_disk_image(vm, tmpdir):
|
||||
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"images_path": str(tmpdir)}):
|
||||
|
@ -22,7 +22,7 @@ from unittest.mock import patch
|
||||
|
||||
|
||||
from gns3server.modules.vpcs import VPCS
|
||||
from gns3server.modules.iou import IOU
|
||||
from gns3server.modules.qemu import Qemu
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
@ -34,11 +34,11 @@ def vpcs(port_manager):
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def iou(port_manager):
|
||||
IOU._instance = None
|
||||
iou = IOU.instance()
|
||||
iou.port_manager = port_manager
|
||||
return iou
|
||||
def qemu(port_manager):
|
||||
Qemu._instance = None
|
||||
qemu = Qemu.instance()
|
||||
qemu.port_manager = port_manager
|
||||
return qemu
|
||||
|
||||
|
||||
def test_create_vm_new_topology(loop, project, vpcs):
|
||||
@ -86,55 +86,55 @@ def test_create_vm_old_topology(loop, project, tmpdir, vpcs):
|
||||
assert f.read() == "1"
|
||||
|
||||
|
||||
def test_get_abs_image_path(iou, tmpdir):
|
||||
os.makedirs(str(tmpdir / "IOU"))
|
||||
def test_get_abs_image_path(qemu, tmpdir):
|
||||
os.makedirs(str(tmpdir / "QEMU"))
|
||||
path1 = str(tmpdir / "test1.bin")
|
||||
open(path1, 'w+').close()
|
||||
|
||||
path2 = str(tmpdir / "IOU" / "test2.bin")
|
||||
path2 = str(tmpdir / "QEMU" / "test2.bin")
|
||||
open(path2, 'w+').close()
|
||||
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"images_path": str(tmpdir)}):
|
||||
assert iou.get_abs_image_path(path1) == path1
|
||||
assert iou.get_abs_image_path(path2) == path2
|
||||
assert iou.get_abs_image_path("test2.bin") == path2
|
||||
assert iou.get_abs_image_path("../test1.bin") == path1
|
||||
assert qemu.get_abs_image_path(path1) == path1
|
||||
assert qemu.get_abs_image_path(path2) == path2
|
||||
assert qemu.get_abs_image_path("test2.bin") == path2
|
||||
assert qemu.get_abs_image_path("../test1.bin") == path1
|
||||
|
||||
|
||||
def test_get_relative_image_path(iou, tmpdir):
|
||||
os.makedirs(str(tmpdir / "IOU"))
|
||||
def test_get_relative_image_path(qemu, tmpdir):
|
||||
os.makedirs(str(tmpdir / "Qemu"))
|
||||
path1 = str(tmpdir / "test1.bin")
|
||||
open(path1, 'w+').close()
|
||||
|
||||
path2 = str(tmpdir / "IOU" / "test2.bin")
|
||||
path2 = str(tmpdir / "QEMU" / "test2.bin")
|
||||
open(path2, 'w+').close()
|
||||
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"images_path": str(tmpdir)}):
|
||||
assert iou.get_relative_image_path(path1) == path1
|
||||
assert iou.get_relative_image_path(path2) == "test2.bin"
|
||||
assert iou.get_relative_image_path("test2.bin") == "test2.bin"
|
||||
assert iou.get_relative_image_path("../test1.bin") == path1
|
||||
assert qemu.get_relative_image_path(path1) == path1
|
||||
assert qemu.get_relative_image_path(path2) == "test2.bin"
|
||||
assert qemu.get_relative_image_path("test2.bin") == "test2.bin"
|
||||
assert qemu.get_relative_image_path("../test1.bin") == path1
|
||||
|
||||
|
||||
def test_list_images(loop, iou, tmpdir):
|
||||
def test_list_images(loop, qemu, tmpdir):
|
||||
|
||||
fake_images = ["a.bin", "b.bin", ".blu.bin"]
|
||||
for image in fake_images:
|
||||
with open(str(tmpdir / image), "w+") as f:
|
||||
f.write("1")
|
||||
|
||||
with patch("gns3server.modules.IOU.get_images_directory", return_value=str(tmpdir)):
|
||||
assert loop.run_until_complete(iou.list_images()) == [
|
||||
with patch("gns3server.modules.Qemu.get_images_directory", return_value=str(tmpdir)):
|
||||
assert loop.run_until_complete(qemu.list_images()) == [
|
||||
{"filename": "a.bin"},
|
||||
{"filename": "b.bin"}
|
||||
]
|
||||
|
||||
|
||||
def test_list_images_empty(loop, iou, tmpdir):
|
||||
with patch("gns3server.modules.IOU.get_images_directory", return_value=str(tmpdir)):
|
||||
assert loop.run_until_complete(iou.list_images()) == []
|
||||
def test_list_images_empty(loop, qemu, tmpdir):
|
||||
with patch("gns3server.modules.Qemu.get_images_directory", return_value=str(tmpdir)):
|
||||
assert loop.run_until_complete(qemu.list_images()) == []
|
||||
|
||||
|
||||
def test_list_images_directory_not_exist(loop, iou):
|
||||
with patch("gns3server.modules.IOU.get_images_directory", return_value="/bla"):
|
||||
assert loop.run_until_complete(iou.list_images()) == []
|
||||
def test_list_images_directory_not_exist(loop, qemu):
|
||||
with patch("gns3server.modules.Qemu.get_images_directory", return_value="/bla"):
|
||||
assert loop.run_until_complete(qemu.list_images()) == []
|
||||
|
@ -19,6 +19,7 @@ import pytest
|
||||
import aiohttp
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
from tests.utils import asyncio_patch
|
||||
|
||||
|
||||
@ -97,9 +98,14 @@ def test_stop(loop, vm):
|
||||
loop.run_until_complete(asyncio.async(vm.start()))
|
||||
assert vm.is_running()
|
||||
|
||||
loop.run_until_complete(asyncio.async(vm.stop()))
|
||||
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
|
||||
loop.run_until_complete(asyncio.async(vm.stop()))
|
||||
assert vm.is_running() is False
|
||||
process.terminate.assert_called_with()
|
||||
|
||||
if sys.platform.startswith("win"):
|
||||
process.send_signal.assert_called_with(1)
|
||||
else:
|
||||
process.terminate.assert_called_with()
|
||||
|
||||
|
||||
def test_reload(loop, vm):
|
||||
@ -117,9 +123,15 @@ def test_reload(loop, vm):
|
||||
vm.port_add_nio_binding(0, nio)
|
||||
loop.run_until_complete(asyncio.async(vm.start()))
|
||||
assert vm.is_running()
|
||||
loop.run_until_complete(asyncio.async(vm.reload()))
|
||||
|
||||
with asyncio_patch("gns3server.utils.asyncio.wait_for_process_termination"):
|
||||
loop.run_until_complete(asyncio.async(vm.reload()))
|
||||
assert vm.is_running() is True
|
||||
process.terminate.assert_called_with()
|
||||
|
||||
if sys.platform.startswith("win"):
|
||||
process.send_signal.assert_called_with(1)
|
||||
else:
|
||||
process.terminate.assert_called_with()
|
||||
|
||||
|
||||
def test_add_nio_binding_udp(vm):
|
||||
@ -135,13 +147,13 @@ def test_add_nio_binding_tap(vm):
|
||||
assert nio.tap_device == "test"
|
||||
|
||||
|
||||
def test_add_nio_binding_tap_no_privileged_access(vm):
|
||||
with patch("gns3server.modules.base_manager.BaseManager._has_privileged_access", return_value=False):
|
||||
with pytest.raises(aiohttp.web.HTTPForbidden):
|
||||
nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_tap", "tap_device": "test"})
|
||||
vm.port_add_nio_binding(0, nio)
|
||||
assert vm._ethernet_adapter.ports[0] is None
|
||||
|
||||
# def test_add_nio_binding_tap_no_privileged_access(vm):
|
||||
# with patch("gns3server.modules.base_manager.BaseManager._has_privileged_access", return_value=False):
|
||||
# with pytest.raises(aiohttp.web.HTTPForbidden):
|
||||
# nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_tap", "tap_device": "test"})
|
||||
# vm.port_add_nio_binding(0, nio)
|
||||
# assert vm._ethernet_adapter.ports[0] is None
|
||||
#
|
||||
|
||||
def test_port_remove_nio_binding(vm):
|
||||
nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
|
||||
@ -190,8 +202,8 @@ def test_get_startup_script_using_default_script(vm):
|
||||
vm._script_file = None
|
||||
|
||||
filepath = os.path.join(vm.working_dir, 'startup.vpc')
|
||||
with open(filepath, 'w+') as f:
|
||||
assert f.write(content)
|
||||
with open(filepath, 'wb+') as f:
|
||||
assert f.write(content.encode("utf-8"))
|
||||
|
||||
assert vm.startup_script == content
|
||||
assert vm.script_file == filepath
|
||||
|
@ -1,143 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2015 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 locale
|
||||
|
||||
from gns3server import main
|
||||
from gns3server.config import Config
|
||||
from gns3server.version import __version__
|
||||
|
||||
|
||||
def test_locale_check():
|
||||
|
||||
try:
|
||||
locale.setlocale(locale.LC_ALL, ("fr_FR"))
|
||||
except: # Locale is not available on the server
|
||||
return
|
||||
main.locale_check()
|
||||
assert locale.getlocale() == ('fr_FR', 'UTF-8')
|
||||
|
||||
|
||||
def test_parse_arguments(capsys, tmpdir):
|
||||
|
||||
Config.reset()
|
||||
config = Config.instance(str(tmpdir / "test.cfg"))
|
||||
server_config = config.get_section_config("Server")
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
main.parse_arguments(["--fail"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert "usage" in err
|
||||
assert "fail" in err
|
||||
assert "unrecognized arguments" in err
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
main.parse_arguments(["-v"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert __version__ in "{}{}".format(out, err) # Depending of the Python version the location of the version change
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
main.parse_arguments(["--version"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert __version__ in "{}{}".format(out, err) # Depending of the Python version the location of the version change
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
main.parse_arguments(["-h"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert __version__ in out
|
||||
assert "optional arguments" in out
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
main.parse_arguments(["--help"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert __version__ in out
|
||||
assert "optional arguments" in out
|
||||
|
||||
assert main.parse_arguments(["--host", "192.168.1.1"], server_config).host == "192.168.1.1"
|
||||
assert main.parse_arguments([], server_config).host == "0.0.0.0"
|
||||
server_config["host"] = "192.168.1.2"
|
||||
assert main.parse_arguments(["--host", "192.168.1.1"], server_config).host == "192.168.1.1"
|
||||
assert main.parse_arguments([], server_config).host == "192.168.1.2"
|
||||
|
||||
assert main.parse_arguments(["--port", "8002"], server_config).port == 8002
|
||||
assert main.parse_arguments([], server_config).port == 8000
|
||||
server_config["port"] = "8003"
|
||||
assert main.parse_arguments([], server_config).port == 8003
|
||||
|
||||
assert main.parse_arguments(["--ssl"], server_config).ssl
|
||||
assert main.parse_arguments([], server_config).ssl is False
|
||||
server_config["ssl"] = "True"
|
||||
assert main.parse_arguments([], server_config).ssl
|
||||
|
||||
assert main.parse_arguments(["--certfile", "bla"], server_config).certfile == "bla"
|
||||
assert main.parse_arguments([], server_config).certfile == ""
|
||||
|
||||
assert main.parse_arguments(["--certkey", "blu"], server_config).certkey == "blu"
|
||||
assert main.parse_arguments([], server_config).certkey == ""
|
||||
|
||||
assert main.parse_arguments(["-L"], server_config).local
|
||||
assert main.parse_arguments(["--local"], server_config).local
|
||||
assert main.parse_arguments([], server_config).local is False
|
||||
server_config["local"] = "True"
|
||||
assert main.parse_arguments([], server_config).local
|
||||
|
||||
assert main.parse_arguments(["-A"], server_config).allow
|
||||
assert main.parse_arguments(["--allow"], server_config).allow
|
||||
assert main.parse_arguments([], server_config).allow is False
|
||||
server_config["allow_remote_console"] = "True"
|
||||
assert main.parse_arguments([], server_config).allow
|
||||
|
||||
assert main.parse_arguments(["-q"], server_config).quiet
|
||||
assert main.parse_arguments(["--quiet"], server_config).quiet
|
||||
assert main.parse_arguments([], server_config).quiet is False
|
||||
|
||||
assert main.parse_arguments(["-d"], server_config).debug
|
||||
assert main.parse_arguments([], server_config).debug is False
|
||||
server_config["debug"] = "True"
|
||||
assert main.parse_arguments([], server_config).debug
|
||||
|
||||
|
||||
def test_set_config_with_args():
|
||||
|
||||
config = Config.instance()
|
||||
args = main.parse_arguments(["--host",
|
||||
"192.168.1.1",
|
||||
"--local",
|
||||
"--allow",
|
||||
"--port",
|
||||
"8001",
|
||||
"--ssl",
|
||||
"--certfile",
|
||||
"bla",
|
||||
"--certkey",
|
||||
"blu",
|
||||
"--debug"],
|
||||
config.get_section_config("Server"))
|
||||
main.set_config(args)
|
||||
server_config = config.get_section_config("Server")
|
||||
|
||||
assert server_config.getboolean("local")
|
||||
assert server_config.getboolean("allow_remote_console")
|
||||
assert server_config["host"] == "192.168.1.1"
|
||||
assert server_config["port"] == "8001"
|
||||
assert server_config.getboolean("ssl")
|
||||
assert server_config["certfile"] == "bla"
|
||||
assert server_config["certkey"] == "blu"
|
||||
assert server_config.getboolean("debug")
|
143
tests/test_run.py
Normal file
143
tests/test_run.py
Normal file
@ -0,0 +1,143 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2015 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 locale
|
||||
|
||||
from gns3server import run
|
||||
from gns3server.config import Config
|
||||
from gns3server.version import __version__
|
||||
|
||||
|
||||
def test_locale_check():
|
||||
|
||||
try:
|
||||
locale.setlocale(locale.LC_ALL, ("fr_FR"))
|
||||
except: # Locale is not available on the server
|
||||
return
|
||||
run.locale_check()
|
||||
assert locale.getlocale() == ('fr_FR', 'UTF-8')
|
||||
|
||||
|
||||
def test_parse_arguments(capsys, tmpdir):
|
||||
|
||||
Config.reset()
|
||||
config = Config.instance(str(tmpdir / "test.cfg"))
|
||||
server_config = config.get_section_config("Server")
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
run.parse_arguments(["--fail"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert "usage" in err
|
||||
assert "fail" in err
|
||||
assert "unrecognized arguments" in err
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
run.parse_arguments(["-v"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert __version__ in "{}{}".format(out, err) # Depending of the Python version the location of the version change
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
run.parse_arguments(["--version"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert __version__ in "{}{}".format(out, err) # Depending of the Python version the location of the version change
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
run.parse_arguments(["-h"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert __version__ in out
|
||||
assert "optional arguments" in out
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
run.parse_arguments(["--help"], server_config)
|
||||
out, err = capsys.readouterr()
|
||||
assert __version__ in out
|
||||
assert "optional arguments" in out
|
||||
|
||||
assert run.parse_arguments(["--host", "192.168.1.1"], server_config).host == "192.168.1.1"
|
||||
assert run.parse_arguments([], server_config).host == "0.0.0.0"
|
||||
server_config["host"] = "192.168.1.2"
|
||||
assert run.parse_arguments(["--host", "192.168.1.1"], server_config).host == "192.168.1.1"
|
||||
assert run.parse_arguments([], server_config).host == "192.168.1.2"
|
||||
|
||||
assert run.parse_arguments(["--port", "8002"], server_config).port == 8002
|
||||
assert run.parse_arguments([], server_config).port == 8000
|
||||
server_config["port"] = "8003"
|
||||
assert run.parse_arguments([], server_config).port == 8003
|
||||
|
||||
assert run.parse_arguments(["--ssl"], server_config).ssl
|
||||
assert run.parse_arguments([], server_config).ssl is False
|
||||
server_config["ssl"] = "True"
|
||||
assert run.parse_arguments([], server_config).ssl
|
||||
|
||||
assert run.parse_arguments(["--certfile", "bla"], server_config).certfile == "bla"
|
||||
assert run.parse_arguments([], server_config).certfile == ""
|
||||
|
||||
assert run.parse_arguments(["--certkey", "blu"], server_config).certkey == "blu"
|
||||
assert run.parse_arguments([], server_config).certkey == ""
|
||||
|
||||
assert run.parse_arguments(["-L"], server_config).local
|
||||
assert run.parse_arguments(["--local"], server_config).local
|
||||
assert run.parse_arguments([], server_config).local is False
|
||||
server_config["local"] = "True"
|
||||
assert run.parse_arguments([], server_config).local
|
||||
|
||||
assert run.parse_arguments(["-A"], server_config).allow
|
||||
assert run.parse_arguments(["--allow"], server_config).allow
|
||||
assert run.parse_arguments([], server_config).allow is False
|
||||
server_config["allow_remote_console"] = "True"
|
||||
assert run.parse_arguments([], server_config).allow
|
||||
|
||||
assert run.parse_arguments(["-q"], server_config).quiet
|
||||
assert run.parse_arguments(["--quiet"], server_config).quiet
|
||||
assert run.parse_arguments([], server_config).quiet is False
|
||||
|
||||
assert run.parse_arguments(["-d"], server_config).debug
|
||||
assert run.parse_arguments([], server_config).debug is False
|
||||
server_config["debug"] = "True"
|
||||
assert run.parse_arguments([], server_config).debug
|
||||
|
||||
|
||||
def test_set_config_with_args():
|
||||
|
||||
config = Config.instance()
|
||||
args = run.parse_arguments(["--host",
|
||||
"192.168.1.1",
|
||||
"--local",
|
||||
"--allow",
|
||||
"--port",
|
||||
"8001",
|
||||
"--ssl",
|
||||
"--certfile",
|
||||
"bla",
|
||||
"--certkey",
|
||||
"blu",
|
||||
"--debug"],
|
||||
config.get_section_config("Server"))
|
||||
run.set_config(args)
|
||||
server_config = config.get_section_config("Server")
|
||||
|
||||
assert server_config.getboolean("local")
|
||||
assert server_config.getboolean("allow_remote_console")
|
||||
assert server_config["host"] == "192.168.1.1"
|
||||
assert server_config["port"] == "8001"
|
||||
assert server_config.getboolean("ssl")
|
||||
assert server_config["certfile"] == "bla"
|
||||
assert server_config["certkey"] == "blu"
|
||||
assert server_config.getboolean("debug")
|
@ -18,6 +18,7 @@
|
||||
|
||||
import asyncio
|
||||
import pytest
|
||||
import sys
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from gns3server.utils.asyncio import wait_run_in_executor, subprocess_check_output, wait_for_process_termination
|
||||
@ -43,6 +44,7 @@ def test_exception_wait_run_in_executor(loop):
|
||||
result = loop.run_until_complete(asyncio.async(exec))
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
|
||||
def test_subprocess_check_output(loop, tmpdir, restore_original_path):
|
||||
|
||||
path = str(tmpdir / "test")
|
||||
|
Loading…
Reference in New Issue
Block a user