arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-12-23 23:38:21 +00:00
Code Issues Releases Wiki Activity

Move appliance and template management code in their own classes.

Browse Source
This commit is contained in:
grossmj 2019-01-14 16:09:06 +07:00
parent 5a3929f01a
commit 8360ae98b1
10 changed files with 340 additions and 224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

206
gns3server/controller/__init__.py
View File

@ -21,15 +21,14 @@ import json
import uuid
import socket
import shutil
import asyncio
import aiohttp
import jsonschema
import copy
from ..config import Config
from .project import Project
from .template import Template
from .appliance import Appliance
from .appliance_manager import ApplianceManager
from .template_manager import TemplateManager
from .compute import Compute, ComputeError
from .notification import Notification
from .symbols import Symbols
@ -37,7 +36,6 @@ from ..version import __version__
from .topology import load_topology
from .gns3vm import GNS3VM
from ..utils.get_resource import get_resource
from ..utils.asyncio import locking
from .gns3vm.gns3_vm_error import GNS3VMError
import logging
@ -52,168 +50,17 @@ class Controller:
def __init__(self):
self._computes = {}
self._projects = {}
self._notification = Notification(self)
self.gns3vm = GNS3VM(self)
self.symbols = Symbols()
self._appliance_manager = ApplianceManager()
self._template_manager = TemplateManager()
self._iou_license_settings = {"iourc_content": "",
"license_check": True}
self._config_loaded = False
self._templates = {}
self._appliances = {}
self._appliances_etag = None
self._config_file = os.path.join(Config.instance().config_dir, "gns3_controller.conf")
log.info("Load controller configuration file {}".format(self._config_file))
@locking
async def download_appliances(self):
try:
headers = {}
if self._appliances_etag:
log.info("Checking if appliances are up-to-date (ETag {})".format(self._appliances_etag))
headers["If-None-Match"] = self._appliances_etag
async with aiohttp.ClientSession() as session:
async with session.get('https://api.github.com/repos/GNS3/gns3-registry/contents/appliances', headers=headers) as response:
if response.status == 304:
log.info("Appliances are already up-to-date (ETag {})".format(self._appliances_etag))
return
elif response.status != 200:
raise aiohttp.web.HTTPConflict(text="Could not retrieve appliances on GitHub due to HTTP error code {}".format(response.status))
etag = response.headers.get("ETag")
if etag:
self._appliances_etag = etag
self.save()
json_data = await response.json()
appliances_dir = get_resource('appliances')
for appliance in json_data:
if appliance["type"] == "file":
appliance_name = appliance["name"]
log.info("Download appliance file from '{}'".format(appliance["download_url"]))
async with session.get(appliance["download_url"]) as response:
if response.status != 200:
log.warning("Could not download '{}' due to HTTP error code {}".format(appliance["download_url"], response.status))
continue
try:
appliance_data = await response.read()
except asyncio.TimeoutError:
log.warning("Timeout while downloading '{}'".format(appliance["download_url"]))
continue
path = os.path.join(appliances_dir, appliance_name)
try:
log.info("Saving {} file to {}".format(appliance_name, path))
with open(path, 'wb') as f:
f.write(appliance_data)
except OSError as e:
raise aiohttp.web.HTTPConflict(text="Could not write appliance file '{}': {}".format(path, e))
except ValueError as e:
raise aiohttp.web.HTTPConflict(text="Could not read appliances information from GitHub: {}".format(e))
def load_appliances(self):
self._appliances = {}
for directory, builtin in ((get_resource('appliances'), True,), (self.appliances_path(), False,)):
if directory and os.path.isdir(directory):
for file in os.listdir(directory):
if not file.endswith('.gns3a') and not file.endswith('.gns3appliance'):
continue
path = os.path.join(directory, file)
appliance_id = uuid.uuid3(uuid.NAMESPACE_URL, path) # Generate UUID from path to avoid change between reboots
try:
with open(path, 'r', encoding='utf-8') as f:
appliance = Appliance(appliance_id, json.load(f), builtin=builtin)
appliance.__json__() # Check if loaded without error
if appliance.status != 'broken':
self._appliances[appliance.id] = appliance
except (ValueError, OSError, KeyError) as e:
log.warning("Cannot load appliance file '%s': %s", path, str(e))
continue
def add_template(self, settings):
"""
Adds a new template.
:param settings: template settings
:returns: Template object
"""
template_id = settings.get("template_id", "")
if template_id in self._templates:
raise aiohttp.web.HTTPConflict(text="Template ID '{}' already exists".format(template_id))
else:
template_id = settings.setdefault("template_id", str(uuid.uuid4()))
try:
template = Template(template_id, settings)
except jsonschema.ValidationError as e:
message = "JSON schema error adding template with JSON data '{}': {}".format(settings, e.message)
raise aiohttp.web.HTTPBadRequest(text=message)
self._templates[template.id] = template
self.save()
self.notification.controller_emit("template.created", template.__json__())
return template
def get_template(self, template_id):
"""
Gets a template.
:param template_id: template identifier
:returns: Template object
"""
template = self._templates.get(template_id)
if not template:
raise aiohttp.web.HTTPNotFound(text="Template ID {} doesn't exist".format(template_id))
return template
def delete_template(self, template_id):
"""
Deletes a template.
:param template_id: template identifier
"""
template = self.get_template(template_id)
if template.builtin:
raise aiohttp.web.HTTPConflict(text="Template ID {} cannot be deleted because it is a builtin".format(template_id))
self._templates.pop(template_id)
self.save()
self.notification.controller_emit("template.deleted", template.__json__())
def duplicate_template(self, template_id):
"""
Duplicates a template.
:param template_id: template identifier
"""
template = self.get_template(template_id)
if template.builtin:
raise aiohttp.web.HTTPConflict(text="Template ID {} cannot be duplicated because it is a builtin".format(template_id))
template_settings = copy.deepcopy(template.settings)
del template_settings["template_id"]
return self.add_template(template_settings)
def load_templates(self):
# Add builtins
builtins = []
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "cloud"), {"template_type": "cloud", "name": "Cloud", "category": 2, "symbol": ":/symbols/cloud.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "nat"), {"template_type": "nat", "name": "NAT", "category": 2, "symbol": ":/symbols/cloud.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "vpcs"), {"template_type": "vpcs", "name": "VPCS", "default_name_format": "PC-{0}", "category": 2, "symbol": ":/symbols/vpcs_guest.svg", "properties": {"base_script_file": "vpcs_base_config.txt"}}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "ethernet_switch"), {"template_type": "ethernet_switch", "console_type": "telnet", "name": "Ethernet switch", "category": 1, "symbol": ":/symbols/ethernet_switch.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "ethernet_hub"), {"template_type": "ethernet_hub", "name": "Ethernet hub", "category": 1, "symbol": ":/symbols/hub.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "frame_relay_switch"), {"template_type": "frame_relay_switch", "name": "Frame Relay switch", "category": 1, "symbol": ":/symbols/frame_relay_switch.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "atm_switch"), {"template_type": "atm_switch", "name": "ATM switch", "category": 1, "symbol": ":/symbols/atm_switch.svg"}, builtin=True))
#FIXME: disable TraceNG
#if sys.platform.startswith("win"):
# builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "traceng"), {"template_type": "traceng", "name": "TraceNG", "default_name_format": "TraceNG-{0}", "category": 2, "symbol": ":/symbols/traceng.svg", "properties": {}}, builtin=True))
for b in builtins:
self._templates[b.id] = b
async def start(self):
log.info("Controller is starting")
@ -296,10 +143,10 @@ class Controller:
"templates": [],
"gns3vm": self.gns3vm.__json__(),
"iou_license": self._iou_license_settings,
"appliances_etag": self._appliances_etag,
"appliances_etag": self._appliance_manager.appliances_etag,
"version": __version__}
for template in self._templates.values():
for template in self._template_manager.templates.values():
if not template.builtin:
controller_settings["templates"].append(template.__json__())
@ -336,17 +183,6 @@ class Controller:
log.critical("Cannot load configuration file '{}': {}".format(self._config_file, e))
return []
# load the templates
if "templates" in controller_settings:
for template_settings in controller_settings["templates"]:
try:
template = Template(template_settings.get("template_id"), template_settings)
self._templates[template.id] = template
except jsonschema.ValidationError as e:
message = "Cannot load template with JSON data '{}': {}".format(template_settings, e.message)
log.warning(message)
continue
# load GNS3 VM settings
if "gns3vm" in controller_settings:
self.gns3vm.settings = controller_settings["gns3vm"]
@ -355,9 +191,9 @@ class Controller:
if "iou_license" in controller_settings:
self._iou_license_settings = controller_settings["iou_license"]
self._appliances_etag = controller_settings.get("appliances_etag")
self.load_appliances()
self.load_templates()
self._appliance_manager.appliances_etag = controller_settings.get("appliances_etag")
self._appliance_manager.load_appliances()
self._template_manager.load_templates(controller_settings.get("templates"))
self._config_loaded = True
return controller_settings.get("computes", [])
@ -417,16 +253,6 @@ class Controller:
os.makedirs(images_path, exist_ok=True)
return images_path
def appliances_path(self):
"""
Get the image storage directory
"""
server_config = Config.instance().get_section_config("Server")
appliances_path = os.path.expanduser(server_config.get("appliances_path", "~/GNS3/projects"))
os.makedirs(appliances_path, exist_ok=True)
return appliances_path
async def _import_gns3_gui_conf(self):
"""
Import old config from GNS3 GUI
@ -533,7 +359,7 @@ class Controller:
try:
template = Template(vm["template_id"], vm)
template.__json__() # Check if loaded without error
self._templates[template.id] = template
self.template_manager.templates[template.id] = template
except KeyError as e:
# template data is not complete (missing name or type)
log.warning("Cannot load template {} ('{}'): missing key {}".format(vm["template_id"], vm.get("name", "unknown"), e))
@ -759,20 +585,20 @@ class Controller:
return self._projects
@property
def appliances(self):
def appliance_manager(self):
"""
:returns: The dictionary of appliances managed by GNS3
:returns: Appliance Manager instance
"""
return self._appliances
return self._appliance_manager
@property
def templates(self):
def template_manager(self):
"""
:returns: The dictionary of templates managed by GNS3
:returns: Template Manager instance
"""
return self._templates
return self._template_manager
@property
def iou_license(self):

145
gns3server/controller/appliance_manager.py Normal file
View File

@ -0,0 +1,145 @@
#!/usr/bin/env python
#
# Copyright (C) 2019 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 json
import uuid
import asyncio
import aiohttp
from .appliance import Appliance
from ..config import Config
from ..utils.asyncio import locking
from ..utils.get_resource import get_resource
import logging
log = logging.getLogger(__name__)
class ApplianceManager:
"""
Manages appliances
"""
def __init__(self):
self._appliances = {}
self._appliances_etag = None
@property
def appliances_etag(self):
"""
:returns: ETag for downloaded appliances
"""
return self._appliances_etag
@appliances_etag.setter
def appliances_etag(self, etag):
"""
:param etag: ETag for downloaded appliances
"""
self._appliances_etag = etag
@property
def appliances(self):
"""
:returns: The dictionary of appliances managed by GNS3
"""
return self._appliances
def appliances_path(self):
"""
Get the image storage directory
"""
server_config = Config.instance().get_section_config("Server")
appliances_path = os.path.expanduser(server_config.get("appliances_path", "~/GNS3/projects"))
os.makedirs(appliances_path, exist_ok=True)
return appliances_path
def load_appliances(self):
"""
Loads appliance files from disk.
"""
self._appliances = {}
for directory, builtin in ((get_resource('appliances'), True,), (self.appliances_path(), False,)):
if directory and os.path.isdir(directory):
for file in os.listdir(directory):
if not file.endswith('.gns3a') and not file.endswith('.gns3appliance'):
continue
path = os.path.join(directory, file)
appliance_id = uuid.uuid3(uuid.NAMESPACE_URL, path) # Generate UUID from path to avoid change between reboots
try:
with open(path, 'r', encoding='utf-8') as f:
appliance = Appliance(appliance_id, json.load(f), builtin=builtin)
appliance.__json__() # Check if loaded without error
if appliance.status != 'broken':
self._appliances[appliance.id] = appliance
except (ValueError, OSError, KeyError) as e:
log.warning("Cannot load appliance file '%s': %s", path, str(e))
continue
@locking
async def download_appliances(self):
"""
Downloads appliance files from GitHub registry repository.
"""
try:
headers = {}
if self._appliances_etag:
log.info("Checking if appliances are up-to-date (ETag {})".format(self._appliances_etag))
headers["If-None-Match"] = self._appliances_etag
async with aiohttp.ClientSession() as session:
async with session.get('https://api.github.com/repos/GNS3/gns3-registry/contents/appliances', headers=headers) as response:
if response.status == 304:
log.info("Appliances are already up-to-date (ETag {})".format(self._appliances_etag))
return
elif response.status != 200:
raise aiohttp.web.HTTPConflict(text="Could not retrieve appliances on GitHub due to HTTP error code {}".format(response.status))
etag = response.headers.get("ETag")
if etag:
self._appliances_etag = etag
self.save()
json_data = await response.json()
appliances_dir = get_resource('appliances')
for appliance in json_data:
if appliance["type"] == "file":
appliance_name = appliance["name"]
log.info("Download appliance file from '{}'".format(appliance["download_url"]))
async with session.get(appliance["download_url"]) as response:
if response.status != 200:
log.warning("Could not download '{}' due to HTTP error code {}".format(appliance["download_url"], response.status))
continue
try:
appliance_data = await response.read()
except asyncio.TimeoutError:
log.warning("Timeout while downloading '{}'".format(appliance["download_url"]))
continue
path = os.path.join(appliances_dir, appliance_name)
try:
log.info("Saving {} file to {}".format(appliance_name, path))
with open(path, 'wb') as f:
f.write(appliance_data)
except OSError as e:
raise aiohttp.web.HTTPConflict(text="Could not write appliance file '{}': {}".format(path, e))
except ValueError as e:
raise aiohttp.web.HTTPConflict(text="Could not read appliances information from GitHub: {}".format(e))

2
gns3server/controller/project.py
View File

@ -481,7 +481,7 @@ class Project:
Create a node from a template.
"""
try:
template = copy.deepcopy(self.controller.templates[template_id].settings)
template = copy.deepcopy(self.controller.template_manager.templates[template_id].settings)
except KeyError:
msg = "Template {} doesn't exist".format(template_id)
log.error(msg)

145
gns3server/controller/template_manager.py Normal file
View File

@ -0,0 +1,145 @@
#!/usr/bin/env python
#
# Copyright (C) 2019 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 copy
import uuid
import aiohttp
import jsonschema
from .template import Template
import logging
log = logging.getLogger(__name__)
class TemplateManager:
"""
Manages templates.
"""
def __init__(self):
self._templates = {}
@property
def templates(self):
"""
:returns: The dictionary of templates managed by GNS3
"""
return self._templates
def load_templates(self, template_settings=None):
"""
Loads templates from controller settings.
"""
if template_settings:
for template_settings in template_settings:
try:
template = Template(template_settings.get("template_id"), template_settings)
self._templates[template.id] = template
except jsonschema.ValidationError as e:
message = "Cannot load template with JSON data '{}': {}".format(template_settings, e.message)
log.warning(message)
continue
# Add builtins
builtins = []
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "cloud"), {"template_type": "cloud", "name": "Cloud", "category": 2, "symbol": ":/symbols/cloud.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "nat"), {"template_type": "nat", "name": "NAT", "category": 2, "symbol": ":/symbols/cloud.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "vpcs"), {"template_type": "vpcs", "name": "VPCS", "default_name_format": "PC-{0}", "category": 2, "symbol": ":/symbols/vpcs_guest.svg", "properties": {"base_script_file": "vpcs_base_config.txt"}}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "ethernet_switch"), {"template_type": "ethernet_switch", "console_type": "telnet", "name": "Ethernet switch", "category": 1, "symbol": ":/symbols/ethernet_switch.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "ethernet_hub"), {"template_type": "ethernet_hub", "name": "Ethernet hub", "category": 1, "symbol": ":/symbols/hub.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "frame_relay_switch"), {"template_type": "frame_relay_switch", "name": "Frame Relay switch", "category": 1, "symbol": ":/symbols/frame_relay_switch.svg"}, builtin=True))
builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "atm_switch"), {"template_type": "atm_switch", "name": "ATM switch", "category": 1, "symbol": ":/symbols/atm_switch.svg"}, builtin=True))
#FIXME: disable TraceNG
#if sys.platform.startswith("win"):
# builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "traceng"), {"template_type": "traceng", "name": "TraceNG", "default_name_format": "TraceNG-{0}", "category": 2, "symbol": ":/symbols/traceng.svg", "properties": {}}, builtin=True))
for b in builtins:
self._templates[b.id] = b
def add_template(self, settings):
"""
Adds a new template.
:param settings: template settings
:returns: Template object
"""
template_id = settings.get("template_id", "")
if template_id in self._templates:
raise aiohttp.web.HTTPConflict(text="Template ID '{}' already exists".format(template_id))
else:
template_id = settings.setdefault("template_id", str(uuid.uuid4()))
try:
template = Template(template_id, settings)
except jsonschema.ValidationError as e:
message = "JSON schema error adding template with JSON data '{}': {}".format(settings, e.message)
raise aiohttp.web.HTTPBadRequest(text=message)
self._templates[template.id] = template
from . import Controller
Controller.instance().save()
Controller.instance().notification.controller_emit("template.created", template.__json__())
return template
def get_template(self, template_id):
"""
Gets a template.
:param template_id: template identifier
:returns: Template object
"""
template = self._templates.get(template_id)
if not template:
raise aiohttp.web.HTTPNotFound(text="Template ID {} doesn't exist".format(template_id))
return template
def delete_template(self, template_id):
"""
Deletes a template.
:param template_id: template identifier
"""
template = self.get_template(template_id)
if template.builtin:
raise aiohttp.web.HTTPConflict(text="Template ID {} cannot be deleted because it is a builtin".format(template_id))
self._templates.pop(template_id)
from . import Controller
Controller.instance().save()
Controller.instance().notification.controller_emit("template.deleted", template.__json__())
def duplicate_template(self, template_id):
"""
Duplicates a template.
:param template_id: template identifier
"""
template = self.get_template(template_id)
if template.builtin:
raise aiohttp.web.HTTPConflict(text="Template ID {} cannot be duplicated because it is a builtin".format(template_id))
template_settings = copy.deepcopy(template.settings)
del template_settings["template_id"]
return self.add_template(template_settings)

6
gns3server/handlers/api/controller/appliance_handler.py
View File

@ -37,6 +37,6 @@ class ApplianceHandler:
controller = Controller.instance()
if request.query.get("update", "no") == "yes":
await controller.download_appliances()
controller.load_appliances()
response.json([c for c in controller.appliances.values()])
await controller.appliance_manager.download_appliances()
controller.appliance_manager.load_appliances()
response.json([c for c in controller.appliance_manager.appliances.values()])

12
gns3server/handlers/api/controller/template_handler.py
View File

@ -50,7 +50,7 @@ class TemplateHandler:
def create(request, response):
controller = Controller.instance()
template = controller.add_template(request.json)
template = controller.template_manager.add_template(request.json)
response.set_status(201)
response.json(template)
@ -67,7 +67,7 @@ class TemplateHandler:
request_etag = request.headers.get("If-None-Match", "")
controller = Controller.instance()
template = controller.get_template(request.match_info["template_id"])
template = controller.template_manager.get_template(request.match_info["template_id"])
data = json.dumps(template.__json__())
template_etag = '"' + hashlib.md5(data.encode()).hexdigest() + '"'
if template_etag == request_etag:
@ -90,7 +90,7 @@ class TemplateHandler:
def update(request, response):
controller = Controller.instance()
template = controller.get_template(request.match_info["template_id"])
template = controller.template_manager.get_template(request.match_info["template_id"])
# Ignore these because we only use them when creating a template
request.json.pop("template_id", None)
request.json.pop("template_type", None)
@ -114,7 +114,7 @@ class TemplateHandler:
def delete(request, response):
controller = Controller.instance()
controller.delete_template(request.match_info["template_id"])
controller.template_manager.delete_template(request.match_info["template_id"])
response.set_status(204)
@Route.get(
@ -126,7 +126,7 @@ class TemplateHandler:
def list(request, response):
controller = Controller.instance()
response.json([c for c in controller.templates.values()])
response.json([c for c in controller.template_manager.templates.values()])
@Route.post(
r"/templates/{template_id}/duplicate",
@ -143,7 +143,7 @@ class TemplateHandler:
async def duplicate(request, response):
controller = Controller.instance()
template = controller.duplicate_template(request.match_info["template_id"])
template = controller.template_manager.duplicate_template(request.match_info["template_id"])
response.set_status(201)
response.json(template)

26
tests/controller/test_controller.py
View File

@ -481,14 +481,14 @@ def test_appliances(controller, async_run, tmpdir):
json.dump(my_appliance, f)
with patch("gns3server.config.Config.get_section_config", return_value={"appliances_path": str(tmpdir)}):
controller.load_appliances()
assert len(controller.appliances) > 0
for appliance in controller.appliances.values():
controller.appliance_manager.load_appliances()
assert len(controller.appliance_manager.appliances) > 0
for appliance in controller.appliance_manager.appliances.values():
assert appliance.__json__()["status"] != "broken"
assert "Alpine Linux" in [c.__json__()["name"] for c in controller.appliances.values()]
assert "My Appliance" in [c.__json__()["name"] for c in controller.appliances.values()]
assert "Alpine Linux" in [c.__json__()["name"] for c in controller.appliance_manager.appliances.values()]
assert "My Appliance" in [c.__json__()["name"] for c in controller.appliance_manager.appliances.values()]
for c in controller.appliances.values():
for c in controller.appliance_manager.appliances.values():
j = c.__json__()
if j["name"] == "Alpine Linux":
assert j["builtin"]
@ -498,23 +498,23 @@ def test_appliances(controller, async_run, tmpdir):
def test_load_templates(controller):
controller._settings = {}
controller.load_templates()
controller.template_manager.load_templates()
assert "Cloud" in [template.name for template in controller.templates.values()]
assert "VPCS" in [template.name for template in controller.templates.values()]
assert "Cloud" in [template.name for template in controller.template_manager.templates.values()]
assert "VPCS" in [template.name for template in controller.template_manager.templates.values()]
for template in controller.templates.values():
for template in controller.template_manager.templates.values():
if template.name == "VPCS":
assert template._settings["properties"] == {"base_script_file": "vpcs_base_config.txt"}
# UUID should not change when you run again the function
for template in controller.templates.values():
for template in controller.template_manager.templates.values():
if template.name == "Test":
qemu_uuid = template.id
elif template.name == "Cloud":
cloud_uuid = template.id
controller.load_templates()
for template in controller.templates.values():
controller.template_manager.load_templates()
for template in controller.template_manager.templates.values():
if template.name == "Test":
assert qemu_uuid == template.id
elif template.name == "Cloud":

2
tests/controller/test_project.py
View File

@ -219,7 +219,7 @@ def test_add_node_from_template(async_run, controller):
"template_type": "vpcs",
"builtin": False,
})
controller._templates[template.id] = template
controller.template_manager.templates[template.id] = template
controller._computes["local"] = compute
response = MagicMock()

2
tests/handlers/api/controller/test_appliance.py
View File

@ -18,7 +18,7 @@
def test_appliances_list(http_controller, controller, async_run):
controller.load_appliances()
controller.appliance_manager.load_appliances()
response = http_controller.get("/appliances", example=True)
assert response.status == 200
assert len(response.json) > 0

18
tests/handlers/api/controller/test_template.py
View File

@ -28,8 +28,8 @@ from gns3server.controller.template import Template
def test_template_list(http_controller, controller):
id = str(uuid.uuid4())
controller.load_templates()
controller._templates[id] = Template(id, {
controller.template_manager.load_templates()
controller.template_manager._templates[id] = Template(id, {
"template_type": "qemu",
"category": 0,
"name": "test",
@ -59,7 +59,7 @@ def test_template_create_without_id(http_controller, controller):
assert response.status == 201
assert response.route == "/templates"
assert response.json["template_id"] is not None
assert len(controller.templates) == 1
assert len(controller.template_manager._templates) == 1
def test_template_create_with_id(http_controller, controller):
@ -79,7 +79,7 @@ def test_template_create_with_id(http_controller, controller):
assert response.status == 201
assert response.route == "/templates"
assert response.json["template_id"] is not None
assert len(controller.templates) == 1
assert len(controller.template_manager._templates) == 1
def test_template_create_wrong_type(http_controller, controller):
@ -97,7 +97,7 @@ def test_template_create_wrong_type(http_controller, controller):
response = http_controller.post("/templates", params)
assert response.status == 400
assert len(controller.templates) == 0
assert len(controller.template_manager._templates) == 0
def test_template_get(http_controller, controller):
@ -169,14 +169,14 @@ def test_template_delete(http_controller, controller):
response = http_controller.get("/templates")
assert len(response.json) == 1
assert len(controller.templates) == 1
assert len(controller.template_manager._templates) == 1
response = http_controller.delete("/templates/{}".format(template_id), example=True)
assert response.status == 204
response = http_controller.get("/templates")
assert len(response.json) == 0
assert len(controller.templates) == 0
assert len(controller.template_manager._templates) == 0
def test_template_duplicate(http_controller, controller):
@ -205,7 +205,7 @@ def test_template_duplicate(http_controller, controller):
response = http_controller.get("/templates")
assert len(response.json) == 2
assert len(controller.templates) == 2
assert len(controller.template_manager._templates) == 2
def test_c7200_dynamips_template_create(http_controller):
@ -953,7 +953,7 @@ def project(http_controller, async_run):
def test_create_node_from_template(http_controller, controller, project, compute):
id = str(uuid.uuid4())
controller._templates = {id: Template(id, {
controller.template_manager._templates = {id: Template(id, {
"template_type": "qemu",
"category": 0,
"name": "test",