rms/gns3-gui
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-gui.git synced 2026-05-17 00:46:01 +03:00
Code Issues Packages Projects Releases Wiki Activity

Template creation from an appliance.

Browse Source
This commit is contained in:
grossmj
2019-01-14 16:14:31 +07:00
parent b6e5a588bf
commit 3462109ef2
4 changed files with 580 additions and 567 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
gns3/dialogs/appliance_wizard.py
View File

@@ -17,9 +17,7 @@
import os
import sip
import urllib
import shutil
from ssl import CertificateError
from ..qt import QtWidgets, QtCore, QtGui, qpartial, qslot
from ..ui.appliance_wizard_ui import Ui_ApplianceWizard
@@ -28,7 +26,8 @@ from ..template import Template
from ..modules import Qemu
from ..registry.appliance import Appliance, ApplianceError
from ..registry.registry import Registry
from ..registry.config import Config, ConfigException
from ..registry.config import Config
from ..registry.appliance_to_template import ApplianceToTemplate
from ..registry.image import Image
from ..utils import human_filesize
from ..utils.wait_for_lambda_worker import WaitForLambdaWorker
@@ -392,9 +391,11 @@ class ApplianceWizard(QtWidgets.QWizard, Ui_ApplianceWizard):
for version in self._appliance["versions"]:
for image in version["images"].values():
img = self._registry.search_image_file(
self._appliance.emulator(), image["filename"], image.get("md5sum"), image.get("filesize"),
strict_md5_check=not self.allowCustomFiles.isChecked())
img = self._registry.search_image_file(self._appliance.emulator(),
image["filename"],
image.get("md5sum"),
image.get("filesize"),
strict_md5_check=not self.allowCustomFiles.isChecked())
if img:
image["status"] = "Found"
image["md5sum"] = img.md5sum
@@ -552,7 +553,9 @@ class ApplianceWizard(QtWidgets.QWizard, Ui_ApplianceWizard):
if "qemu" in appliance_configuration:
appliance_configuration["qemu"]["path"] = self.uiQemuListComboBox.currentData()
self._create_template(appliance_configuration, self._compute_id)
#TODO: check for errors if template could not be created
new_template = ApplianceToTemplate().new_template(appliance_configuration, self._compute_id, self._symbols)
TemplateManager.instance().createTemplate(Template(new_template))
return True
#worker = WaitForLambdaWorker(lambda: self._create_template(appliance_configuration, self._compute_id), allowed_exceptions=[ConfigException, OSError])
@@ -570,177 +573,6 @@ class ApplianceWizard(QtWidgets.QWizard, Ui_ApplianceWizard):
# QtWidgets.QMessageBox.information(self.parent(), "Add appliance", "{} installed!".format(appliance_configuration["name"]))
# return True
def _create_template(self, appliance_config, server):
"""
Creates a new template from an appliance.
:param appliance_config: Dictionary with appliance configuration
:param server
"""
new_template = {
"compute_id": server,
"name": appliance_config["name"]
}
if "usage" in appliance_config:
new_template["usage"] = appliance_config["usage"]
if appliance_config["category"] == "multilayer_switch":
new_template["category"] = "switch"
else:
new_template["category"] = appliance_config["category"]
if "symbol" in appliance_config:
new_template["symbol"] = self._set_symbol(appliance_config["symbol"], self._symbols)
if new_template.get("symbol") is None:
if appliance_config["category"] == "guest":
if "docker" in appliance_config:
new_template["symbol"] = ":/symbols/docker_guest.svg"
else:
new_template["symbol"] = ":/symbols/qemu_guest.svg"
elif appliance_config["category"] == "router":
new_template["symbol"] = ":/symbols/router.svg"
elif appliance_config["category"] == "switch":
new_template["symbol"] = ":/symbols/ethernet_switch.svg"
elif appliance_config["category"] == "multilayer_switch":
new_template["symbol"] = ":/symbols/multilayer_switch.svg"
elif appliance_config["category"] == "firewall":
new_template["symbol"] = ":/symbols/firewall.svg"
if "qemu" in appliance_config:
new_template["template_type"] = "qemu"
self._add_qemu_config(new_template, appliance_config)
elif "iou" in appliance_config:
new_template["template_type"] = "iou"
self._add_iou_config(new_template, appliance_config)
elif "dynamips" in appliance_config:
new_template["template_type"] = "dynamips"
self._add_dynamips_config(new_template, appliance_config)
elif "docker" in appliance_config:
new_template["template_type"] = "docker"
self._add_docker_config(new_template, appliance_config)
else:
raise ConfigException("{} no configuration found for known emulators".format(new_template["name"]))
TemplateManager.instance().createTemplate(Template(new_template))
def _add_qemu_config(self, new_config, appliance_config):
new_config.update(appliance_config["qemu"])
# the following properties are not valid for a template
new_config.pop("kvm")
new_config.pop("path")
new_config.pop("arch")
options = appliance_config["qemu"].get("options", "")
if "-nographic" not in options:
options += " -nographic"
if appliance_config["qemu"].get("kvm", "allow") == "disable" and "-no-kvm" not in options:
options += " -no-kvm"
new_config["options"] = options.strip()
for image in appliance_config["images"]:
if image.get("path"):
new_config[image["type"]] = self._relative_image_path("QEMU", image["path"])
if "path" in appliance_config["qemu"]:
new_config["qemu_path"] = appliance_config["qemu"]["path"]
else:
new_config["qemu_path"] = "qemu-system-{}".format(appliance_config["qemu"]["arch"])
if "first_port_name" in appliance_config:
new_config["first_port_name"] = appliance_config["first_port_name"]
if "port_name_format" in appliance_config:
new_config["port_name_format"] = appliance_config["port_name_format"]
if "port_segment_size" in appliance_config:
new_config["port_segment_size"] = appliance_config["port_segment_size"]
if "custom_adapters" in appliance_config:
new_config["custom_adapters"] = appliance_config["custom_adapters"]
if "linked_clone" in appliance_config:
new_config["linked_clone"] = appliance_config["linked_clone"]
def _add_docker_config(self, new_config, appliance_config):
new_config.update(appliance_config["docker"])
if "custom_adapters" in appliance_config:
new_config["custom_adapters"] = appliance_config["custom_adapters"]
def _add_dynamips_config(self, new_config, appliance_config):
new_config.update(appliance_config["dynamips"])
for image in appliance_config["images"]:
new_config[image["type"]] = self._relative_image_path("IOS", image["path"])
new_config["idlepc"] = image.get("idlepc", "")
def _add_iou_config(self, new_config, appliance_config):
new_config.update(appliance_config["iou"])
for image in appliance_config["images"]:
if "path" not in image:
raise ConfigException("Disk image is missing")
new_config[image["type"]] = self._relative_image_path("IOU", image["path"])
new_config["path"] = new_config["image"]
def _relative_image_path(self, image_dir_type, path):
"""
:param image_dir_type: Type of image directory
:param filename: Filename at the end of the process
:param path: Full path to the file
:returns: Path relative to image directory.
Copy the image to the directory if not already in the directory
"""
images_dir = os.path.join(Config().images_dir, image_dir_type)
path = os.path.abspath(path)
if os.path.commonprefix([images_dir, path]) == images_dir:
return path.replace(images_dir, '').strip('/\\')
return os.path.basename(path)
def _set_symbol(self, symbol, controller_symbols):
"""
Check if exists on controller or download symbol from the web if needed
"""
# GNS3 builtin symbol
if symbol.startswith(":/symbols/"):
return symbol
path = os.path.join(self.symbols_dir, symbol)
if os.path.exists(path):
return os.path.basename(path)
is_symbol_on_controller = len([s for s in controller_symbols
if s['symbol_id'] == symbol]) > 0
if is_symbol_on_controller:
cached = Controller.instance().getStaticCachedPath(symbol)
if os.path.exists(cached):
try:
shutil.copy(cached, path)
except IOError as e:
log.warning("Cannot copy cached symbol from `{}` to `{}` due `{}`".format(
cached, path, str(e)
))
return symbol
url = "https://raw.githubusercontent.com/GNS3/gns3-registry/master/symbols/{}".format(symbol)
try:
urllib.request.urlretrieve(url, path)
return os.path.basename(path)
except (OSError, CertificateError):
return None
def _uploadImages(self, version):
"""
Upload an image the compute.

206
gns3/registry/appliance_to_template.py Normal file
View File

@@ -0,0 +1,206 @@
#!/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 os
import urllib.request
import shutil
from ssl import CertificateError
from ..controller import Controller
from .config import Config, ConfigException
import logging
log = logging.getLogger(__name__)
class ApplianceToTemplate:
"""
Appliance installation.
"""
def new_template(self, appliance_config, server, controller_symbols=None):
"""
Creates a new template from an appliance.
:param appliance_config: Dictionary with appliance configuration
:param server
"""
new_template = {
"compute_id": server,
"name": appliance_config["name"]
}
if "usage" in appliance_config:
new_template["usage"] = appliance_config["usage"]
if appliance_config["category"] == "multilayer_switch":
new_template["category"] = "switch"
else:
new_template["category"] = appliance_config["category"]
if "symbol" in appliance_config:
new_template["symbol"] = self._set_symbol(appliance_config["symbol"], controller_symbols)
if new_template.get("symbol") is None:
if appliance_config["category"] == "guest":
if "docker" in appliance_config:
new_template["symbol"] = ":/symbols/docker_guest.svg"
else:
new_template["symbol"] = ":/symbols/qemu_guest.svg"
elif appliance_config["category"] == "router":
new_template["symbol"] = ":/symbols/router.svg"
elif appliance_config["category"] == "switch":
new_template["symbol"] = ":/symbols/ethernet_switch.svg"
elif appliance_config["category"] == "multilayer_switch":
new_template["symbol"] = ":/symbols/multilayer_switch.svg"
elif appliance_config["category"] == "firewall":
new_template["symbol"] = ":/symbols/firewall.svg"
if "qemu" in appliance_config:
new_template["template_type"] = "qemu"
self._add_qemu_config(new_template, appliance_config)
elif "iou" in appliance_config:
new_template["template_type"] = "iou"
self._add_iou_config(new_template, appliance_config)
elif "dynamips" in appliance_config:
new_template["template_type"] = "dynamips"
self._add_dynamips_config(new_template, appliance_config)
elif "docker" in appliance_config:
new_template["template_type"] = "docker"
self._add_docker_config(new_template, appliance_config)
else:
raise ConfigException("{} no configuration found for known emulators".format(new_template["name"]))
return new_template
def _add_qemu_config(self, new_config, appliance_config):
new_config.update(appliance_config["qemu"])
# the following properties are not valid for a template
new_config.pop("kvm", None)
new_config.pop("path", None)
new_config.pop("arch", None)
options = appliance_config["qemu"].get("options", "")
if "-nographic" not in options:
options += " -nographic"
if appliance_config["qemu"].get("kvm", "allow") == "disable" and "-no-kvm" not in options:
options += " -no-kvm"
new_config["options"] = options.strip()
for image in appliance_config["images"]:
if image.get("path"):
new_config[image["type"]] = self._relative_image_path("QEMU", image["path"])
if "path" in appliance_config["qemu"]:
new_config["qemu_path"] = appliance_config["qemu"]["path"]
else:
new_config["qemu_path"] = "qemu-system-{}".format(appliance_config["qemu"]["arch"])
if "first_port_name" in appliance_config:
new_config["first_port_name"] = appliance_config["first_port_name"]
if "port_name_format" in appliance_config:
new_config["port_name_format"] = appliance_config["port_name_format"]
if "port_segment_size" in appliance_config:
new_config["port_segment_size"] = appliance_config["port_segment_size"]
if "custom_adapters" in appliance_config:
new_config["custom_adapters"] = appliance_config["custom_adapters"]
if "linked_clone" in appliance_config:
new_config["linked_clone"] = appliance_config["linked_clone"]
def _add_docker_config(self, new_config, appliance_config):
new_config.update(appliance_config["docker"])
if "custom_adapters" in appliance_config:
new_config["custom_adapters"] = appliance_config["custom_adapters"]
def _add_dynamips_config(self, new_config, appliance_config):
new_config.update(appliance_config["dynamips"])
for image in appliance_config["images"]:
new_config[image["type"]] = self._relative_image_path("IOS", image["path"])
new_config["idlepc"] = image.get("idlepc", "")
def _add_iou_config(self, new_config, appliance_config):
new_config.update(appliance_config["iou"])
for image in appliance_config["images"]:
if "path" not in image:
raise ConfigException("Disk image is missing")
new_config[image["type"]] = self._relative_image_path("IOU", image["path"])
new_config["path"] = new_config["image"]
def _relative_image_path(self, image_dir_type, path):
"""
:param image_dir_type: Type of image directory
:param filename: Filename at the end of the process
:param path: Full path to the file
:returns: Path relative to image directory.
Copy the image to the directory if not already in the directory
"""
images_dir = os.path.join(Config().images_dir, image_dir_type)
path = os.path.abspath(path)
if os.path.commonprefix([images_dir, path]) == images_dir:
return path.replace(images_dir, '').strip('/\\')
return os.path.basename(path)
def _set_symbol(self, symbol, controller_symbols):
"""
Check if exists on controller or download symbol from the web if needed
"""
# GNS3 builtin symbol
if symbol.startswith(":/symbols/"):
return symbol
path = os.path.join(Config().symbols_dir, symbol)
if os.path.exists(path):
return os.path.basename(path)
if controller_symbols:
is_symbol_on_controller = len([s for s in controller_symbols
if s['symbol_id'] == symbol]) > 0
if is_symbol_on_controller:
cached = Controller.instance().getStaticCachedPath(symbol)
if os.path.exists(cached):
try:
shutil.copy(cached, path)
except IOError as e:
log.warning("Cannot copy cached symbol from `{}` to `{}` due `{}`".format(
cached, path, str(e)
))
return symbol
url = "https://raw.githubusercontent.com/GNS3/gns3-registry/master/symbols/{}".format(symbol)
try:
urllib.request.urlretrieve(url, path)
return os.path.basename(path)
except (OSError, CertificateError):
return None

363
tests/registry/test_appliance_to_template.py Normal file
View File

@@ -0,0 +1,363 @@
#!/usr/bin/env python
#
# 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 json
import os
from unittest.mock import patch
from gns3.registry.config import Config
from gns3.settings import LOCAL_SERVER_SETTINGS
from gns3.registry.appliance_to_template import ApplianceToTemplate
@pytest.fixture(scope="function")
def empty_config(tmpdir, images_dir, symbols_dir, local_server_config):
settings = local_server_config.loadSettings("Server", LOCAL_SERVER_SETTINGS)
settings["images_path"] = images_dir
settings["symbols_path"] = symbols_dir
local_server_config.saveSettings("Server", settings)
config = {
"Servers": {
},
"Dynamips": {
"allocate_aux_console_ports": False,
"dynamips_path": "/Applications/GNS3.app/Contents/Resources/dynamips",
"ghost_ios_support": True,
"mmap_support": True,
"routers": [
],
"sparse_memory_support": True,
"use_local_server": True
},
"IOU": {
"devices": [
],
"license_check": True,
"use_local_server": False
},
"Qemu": {
"use_local_server": True,
"vms": [
]
},
"Docker": {
"containers": []
}
}
path = str(tmpdir / "config")
with open(path, "w+", encoding="utf-8") as f:
json.dump(config, f)
return Config(path)
def test_add_appliance_iou(iou_l3):
with open("tests/registry/appliances/cisco-iou-l3.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "image",
"filename": "i86bi-linux-l3-adventerprisek9-15.4.1T.bin",
"path": iou_l3
}
]
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template == {
"category": "router",
"template_type": "iou",
"symbol": ":/symbols/router.svg",
"compute_id": "local",
"name": "Cisco IOU L3",
"nvram": 128,
"ram": 256,
"serial_adapters": 2,
"ethernet_adapters": 2,
"startup_config": "iou_l3_base_startup-config.txt",
"image": os.path.basename(iou_l3),
"path": os.path.basename(iou_l3)
}
def test_add_appliance_docker():
with open("tests/registry/appliances/openvswitch.gns3a", encoding="utf-8") as f:
config = json.load(f)
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template == {
"name": "Open vSwitch",
"template_type": "docker",
"image": "gns3/openvswitch:latest",
"category": "switch",
"symbol": ":/symbols/multilayer_switch.svg",
"compute_id": "local",
"adapters": 16,
"usage": "By default all interfaces are connected to the br0"
}
def test_add_appliance_dynamips(cisco_3745):
with open("tests/registry/appliances/cisco-3745.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "image",
"filename": "c3745-adventerprisek9-mz.124-25d.image",
"path": cisco_3745,
"idlepc": "0x60aa1da0"
}
]
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template == {
"template_type": "dynamips",
"category": "router",
"chassis": "",
"idlepc": "0x60aa1da0",
"image": "c3745-adventerprisek9-mz.124-25d.image",
"name": "Cisco 3745",
"nvram": 256,
"platform": "c3745",
"ram": 256,
"compute_id": "local",
"slot0": "GT96100-FE",
"slot1": "NM-1FE-TX",
"slot2": "NM-4T",
"slot3": "",
"slot4": "",
"startup_config": "ios_base_startup-config.txt",
"symbol": ":/symbols/router.svg",
"wic0": "WIC-1T",
"wic1": "WIC-1T",
"wic2": "WIC-1T"
}
def test_add_appliance_guest(linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template == {
"template_type": "qemu",
"adapter_type": "e1000",
"adapters": 1,
"category": "guest",
"console_type": "telnet",
"symbol": ":/symbols/qemu_guest.svg",
"hda_disk_image": "linux-microcore-3.4.1.img",
"name": "Micro Core Linux",
"options": "-nographic",
"qemu_path": "qemu-system-i386",
"usage": "Just start the appliance",
"ram": 32,
"compute_id": "local"
}
def test_add_appliance_with_symbol(linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["symbol"] = ":/symbols/asa.svg"
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template["symbol"] == ":/symbols/asa.svg"
def test_add_appliance_with_symbol_from_symbols_dir(linux_microcore_img, symbols_dir):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["symbol"] = "linux_guest.svg"
symbol_path = os.path.join(symbols_dir, "linux_guest.svg")
open(symbol_path, 'w+').close()
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template["symbol"] == "linux_guest.svg"
def test_add_appliance_with_symbol_from_web(empty_config, linux_microcore_img, symbols_dir):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["symbol"] = "linux_guest.svg"
symbol_path = os.path.join(symbols_dir, "linux_guest.svg")
with patch("urllib.request.urlretrieve") as mock:
new_template = ApplianceToTemplate().new_template(config, "local")
mock.assert_called_with("https://raw.githubusercontent.com/GNS3/gns3-registry/master/symbols/linux_guest.svg", symbol_path)
assert new_template["symbol"] == "linux_guest.svg"
def test_add_appliance_with_symbol_from_web_error(empty_config, linux_microcore_img, symbols_dir):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["symbol"] = "linux_guest.svg"
symbol_path = os.path.join(symbols_dir, "linux_guest.svg")
with patch("urllib.request.urlretrieve") as mock:
mock.side_effect = OSError
new_template = ApplianceToTemplate().new_template(config, "local")
mock.assert_called_with("https://raw.githubusercontent.com/GNS3/gns3-registry/master/symbols/linux_guest.svg", symbol_path)
# In case of error we fallback on default symbol
assert new_template["symbol"] == ":/symbols/qemu_guest.svg"
def test_add_appliance_with_port_name_format(linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["port_name_format"] = "eth{0}"
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template["port_name_format"] == "eth{0}"
def test_add_appliance_with_boot_priority(linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["qemu"]["boot_priority"] = "dc"
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template["boot_priority"] == "dc"
def test_add_appliance_router_two_disk(images_dir):
with open("tests/registry/appliances/arista-veos.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "a",
"path": os.path.join(images_dir, "QEMU", "a")
},
{
"type": "hdb_disk_image",
"filename": "b",
"path": os.path.join(images_dir, "QEMU", "b")
}
]
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template == {
"template_type": "qemu",
"adapter_type": "e1000",
"adapters": 8,
"category": "router",
"symbol": ":/symbols/router.svg",
"hda_disk_image": "a",
"hdb_disk_image": "b",
"name": "Arista vEOS",
"options": "-nographic",
"qemu_path": "qemu-system-x86_64",
"ram": 2048,
"console_type": "telnet",
"compute_id": "local"
}
def test_add_appliance_path_relative_to_images_dir(tmpdir, linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template["hda_disk_image"] == "linux-microcore-3.4.1.img"
def test_add_appliance_path_non_relative_to_images_dir(tmpdir, images_dir):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "a.img",
"path": str(tmpdir / "a.img")
}
]
with open(str(tmpdir / "a.img"), "w+") as f:
f.write("a")
new_template = ApplianceToTemplate().new_template(config, "local")
assert new_template["hda_disk_image"] == "a.img"
def test_relative_image_path(images_dir, tmpdir):
# Image in image directory no need to copy it
open(os.path.join(images_dir, "QEMU", "a"), "w+").close()
assert ApplianceToTemplate()._relative_image_path("QEMU", os.path.join(images_dir, "QEMU", "a")) == "a"
# Image in image directory no need to copy it but with a different file name
open(os.path.join(images_dir, "QEMU", "a"), "w+").close()
assert ApplianceToTemplate()._relative_image_path("QEMU", os.path.join(images_dir, "QEMU", "a")) == "a"
# Image outside image directory we need to copy it
open(str(tmpdir / "b"), "w+").close()
assert ApplianceToTemplate()._relative_image_path("QEMU", str(tmpdir / "b")) == "b"

390
tests/registry/test_config.py
View File

@@ -19,10 +19,8 @@
import pytest
import json
import os
from unittest.mock import MagicMock, patch
from gns3.registry.config import Config, ConfigException
from gns3.registry.config import Config
from gns3.settings import LOCAL_SERVER_SETTINGS
@@ -100,389 +98,3 @@ def test_list_servers_remote_servers(tmpdir):
json.dump(config, f)
config = Config(path)
assert config.servers == ["local", "http://darkside.moon:4242"]
def test_add_appliance_iou(empty_config, iou_l3):
with open("tests/registry/appliances/cisco-iou-l3.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "image",
"filename": "i86bi-linux-l3-adventerprisek9-15.4.1T.bin",
"path": iou_l3
}
]
empty_config.add_appliance(config, "local")
assert empty_config._config["IOU"]["devices"][0] == {
"category": 0,
"symbol": ":/symbols/router.svg",
"compute_id": "local",
"name": "Cisco IOU L3",
"l1_keepalives": False,
"nvram": 128,
"private_config": "",
"ram": 256,
"serial_adapters": 2,
"ethernet_adapters": 2,
"use_default_iou_values": True,
"console_type": "telnet",
"startup_config": "iou_l3_base_startup-config.txt",
"image": os.path.basename(iou_l3),
"path": os.path.basename(iou_l3)
}
def test_add_appliance_docker(empty_config, iou_l3):
with open("tests/registry/appliances/openvswitch.gns3a", encoding="utf-8") as f:
config = json.load(f)
empty_config.add_appliance(config, "local")
assert empty_config._config["Docker"]["containers"][0] == {
"name": "Open vSwitch",
"image": "gns3/openvswitch:latest",
"category": 1,
"symbol": ":/symbols/multilayer_switch.svg",
"compute_id": "local",
"adapters": 16,
"start_command": "",
"environment": "",
'extra_hosts': "",
"usage": "By default all interfaces are connected to the br0",
"console_type": "telnet",
"console_http_port": 80,
"console_http_path": "/"
}
def test_add_appliance_dynamips(empty_config, cisco_3745):
with open("tests/registry/appliances/cisco-3745.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "image",
"filename": "c3745-adventerprisek9-mz.124-25d.image",
"path": cisco_3745,
"idlepc": "0x60aa1da0"
}
]
empty_config.add_appliance(config, "local")
assert empty_config._config["Dynamips"]["routers"][0] == {
"auto_delete_disks": True,
"category": 0,
"chassis": "",
"disk0": 0,
"disk1": 0,
"exec_area": 64,
"idlemax": 500,
"idlepc": "0x60aa1da0",
"idlesleep": 30,
"image": "c3745-adventerprisek9-mz.124-25d.image",
"iomem": 5,
"mac_addr": "",
"mmap": True,
"name": "Cisco 3745",
"nvram": 256,
"platform": "c3745",
"private_config": "",
"ram": 256,
"compute_id": "local",
"slot0": "GT96100-FE",
"slot1": "NM-1FE-TX",
"slot2": "NM-4T",
"slot3": "",
"slot4": "",
"sparsemem": True,
"startup_config": "ios_base_startup-config.txt",
"symbol": ":/symbols/router.svg",
"system_id": "FTX0945W0MY",
"wic0": "WIC-1T",
"wic1": "WIC-1T",
"wic2": "WIC-1T"
}
def test_add_appliance_guest(empty_config, linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
empty_config.add_appliance(config, "local")
assert empty_config._config["Qemu"]["vms"][0] == {
"adapter_type": "e1000",
"adapters": 1,
"category": 2,
"cpu_throttling": 0,
"cpus": 1,
"console_type": "telnet",
"symbol": ":/symbols/qemu_guest.svg",
"hda_disk_image": "linux-microcore-3.4.1.img",
"hdb_disk_image": "",
"hdc_disk_image": "",
"hdd_disk_image": "",
"cdrom_image": "",
"bios_image": "",
"initrd": "",
"kernel_command_line": "",
"kernel_image": "",
"legacy_networking": False,
"name": "Micro Core Linux",
"options": "-nographic",
"process_priority": "normal",
"qemu_path": "qemu-system-i386",
"usage": "Just start the appliance",
"ram": 32,
"compute_id": "local",
"hda_disk_interface": "ide",
"hdb_disk_interface": "ide",
"hdc_disk_interface": "ide",
"hdd_disk_interface": "ide"
}
def test_add_appliance_with_symbol(empty_config, linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["symbol"] = ":/symbols/asa.svg"
empty_config.add_appliance(config, "local")
assert empty_config._config["Qemu"]["vms"][0]["symbol"] == ":/symbols/asa.svg"
def test_add_appliance_with_symbol_from_symbols_dir(empty_config, linux_microcore_img, symbols_dir):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["symbol"] = "linux_guest.svg"
symbol_path = os.path.join(symbols_dir, "linux_guest.svg")
open(symbol_path, 'w+').close()
empty_config.add_appliance(config, "local")
assert empty_config._config["Qemu"]["vms"][0]["symbol"] == "linux_guest.svg"
def test_add_appliance_with_symbol_from_web(empty_config, linux_microcore_img, symbols_dir):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["symbol"] = "linux_guest.svg"
symbol_path = os.path.join(symbols_dir, "linux_guest.svg")
with patch("urllib.request.urlretrieve") as mock:
empty_config.add_appliance(config, "local")
mock.assert_called_with("https://raw.githubusercontent.com/GNS3/gns3-registry/master/symbols/linux_guest.svg", symbol_path)
assert empty_config._config["Qemu"]["vms"][0]["symbol"] == "linux_guest.svg"
def test_add_appliance_with_symbol_from_web_error(empty_config, linux_microcore_img, symbols_dir):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["symbol"] = "linux_guest.svg"
symbol_path = os.path.join(symbols_dir, "linux_guest.svg")
with patch("urllib.request.urlretrieve") as mock:
mock.side_effect = OSError
empty_config.add_appliance(config, "local")
mock.assert_called_with("https://raw.githubusercontent.com/GNS3/gns3-registry/master/symbols/linux_guest.svg", symbol_path)
# In case of error we fallback on default symbol
assert empty_config._config["Qemu"]["vms"][0]["symbol"] == ":/symbols/qemu_guest.svg"
def test_add_appliance_with_port_name_format(empty_config, linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["port_name_format"] = "eth{0}"
empty_config.add_appliance(config, "local")
assert empty_config._config["Qemu"]["vms"][0]["port_name_format"] == "eth{0}"
def test_add_appliance_with_boot_priority(empty_config, linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
config["qemu"]["boot_priority"] = "dc"
empty_config.add_appliance(config, "local")
assert empty_config._config["Qemu"]["vms"][0]["boot_priority"] == "dc"
def test_add_appliance_router_two_disk(empty_config, images_dir):
with open("tests/registry/appliances/arista-veos.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "a",
"path": os.path.join(images_dir, "QEMU", "a")
},
{
"type": "hdb_disk_image",
"filename": "b",
"path": os.path.join(images_dir, "QEMU", "b")
}
]
empty_config.add_appliance(config, "local")
assert empty_config._config["Qemu"]["vms"][0] == {
"adapter_type": "e1000",
"adapters": 8,
"bios_image": "",
"category": 0,
"cpu_throttling": 0,
"cpus": 1,
"symbol": ":/symbols/router.svg",
"hda_disk_image": "a",
"hdb_disk_image": "b",
"hdc_disk_image": "",
"hdd_disk_image": "",
"cdrom_image": "",
"initrd": "",
"kernel_command_line": "",
"kernel_image": "",
"legacy_networking": False,
"name": "Arista vEOS",
"options": "-nographic",
"process_priority": "normal",
"qemu_path": "qemu-system-x86_64",
"ram": 2048,
"console_type": "telnet",
"compute_id": "local",
"hda_disk_interface": "ide",
"hdb_disk_interface": "ide",
"hdc_disk_interface": "ide",
"hdd_disk_interface": "ide"
}
def test_add_appliance_uniq(empty_config, linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
empty_config.add_appliance(config, "local")
config["qemu"]["adapters"] = 2
with pytest.raises(ConfigException):
empty_config.add_appliance(config, "local")
assert len(empty_config._config["Qemu"]["vms"]) == 1
assert empty_config._config["Qemu"]["vms"][0]["adapters"] == 1
def test_add_appliance_path_relative_to_images_dir(empty_config, tmpdir, linux_microcore_img):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "linux-microcore-3.4.1.img",
"path": linux_microcore_img
}
]
empty_config.add_appliance(config, "local")
assert empty_config._config["Qemu"]["vms"][0]["hda_disk_image"] == "linux-microcore-3.4.1.img"
def test_add_appliance_ova(empty_config, tmpdir, images_dir):
with open("tests/registry/appliances/juniper-vsrx.gns3a", encoding="utf-8") as f:
config = json.load(f)
os.makedirs(os.path.join(images_dir, "QEMU", "junos-vsrx-12.1X47-D10.4-domestic.ova"))
open(os.path.join(images_dir, "QEMU", "junos-vsrx-12.1X47-D10.4-domestic.ova", "junos-vsrx-12.1X47-D10.4-domestic-disk1.vmdk"), "w+").close()
config["images"] = [
{
"type": "hda_disk_image",
"filename": "junos-vsrx-12.1X47-D10.4-domestic.ova/junos-vsrx-12.1X47-D10.4-domestic-disk1.vmdk",
"path": os.path.join(images_dir, "QEMU", "junos-vsrx-12.1X47-D10.4-domestic.ova", "junos-vsrx-12.1X47-D10.4-domestic-disk1.vmdk")
}
]
empty_config.add_appliance(config, "local")
assert empty_config._config["Qemu"]["vms"][0]["hda_disk_image"] == os.path.join("junos-vsrx-12.1X47-D10.4-domestic.ova", "junos-vsrx-12.1X47-D10.4-domestic-disk1.vmdk")
def test_add_appliance_path_non_relative_to_images_dir(empty_config, tmpdir, images_dir):
with open("tests/registry/appliances/microcore-linux.gns3a", encoding="utf-8") as f:
config = json.load(f)
config["images"] = [
{
"type": "hda_disk_image",
"filename": "a.img",
"path": str(tmpdir / "a.img")
}
]
with open(str(tmpdir / "a.img"), "w+") as f:
f.write("a")
empty_config.add_appliance(config, "local")
assert empty_config._config["Qemu"]["vms"][0]["hda_disk_image"] == "a.img"
def test_relative_image_path(empty_config, images_dir, tmpdir):
# Image in image directory no need to copy it
open(os.path.join(images_dir, "QEMU", "a"), "w+").close()
assert empty_config._relative_image_path("QEMU", os.path.join(images_dir, "QEMU", "a")) == "a"
# Image in image directory no need to copy it but with a different file name
open(os.path.join(images_dir, "QEMU", "a"), "w+").close()
assert empty_config._relative_image_path("QEMU", os.path.join(images_dir, "QEMU", "a")) == "a"
# Image outside image directory we need to copy it
open(str(tmpdir / "b"), "w+").close()
assert empty_config._relative_image_path("QEMU", str(tmpdir / "b")) == "b"