From d730c591b332bcb8b62faf5f27e692f46a96dfc9 Mon Sep 17 00:00:00 2001 From: grossmj Date: Sun, 28 Mar 2021 11:15:08 +1030 Subject: [PATCH] Refactor template management to use database. --- gns3server/api/routes/controller/templates.py | 107 +- gns3server/controller/__init__.py | 16 - gns3server/controller/node.py | 1 - gns3server/controller/project.py | 9 +- gns3server/controller/template.py | 168 -- gns3server/controller/template_manager.py | 148 -- gns3server/db/models/__init__.py | 32 + gns3server/db/{models.py => models/base.py} | 49 +- gns3server/db/models/templates.py | 286 +++ gns3server/db/models/users.py | 48 + gns3server/db/repositories/base.py | 2 + gns3server/db/repositories/templates.py | 243 +++ gns3server/db/repositories/users.py | 12 +- gns3server/db/tasks.py | 2 +- gns3server/schemas/templates.py | 9 +- requirements.txt | 4 +- tests/api/routes/controller/test_templates.py | 1854 ++++++++--------- tests/controller/test_controller.py | 27 - tests/controller/test_project.py | 133 +- tests/controller/test_template.py | 89 - 20 files changed, 1704 insertions(+), 1535 deletions(-) delete mode 100644 gns3server/controller/template.py delete mode 100644 gns3server/controller/template_manager.py create mode 100644 gns3server/db/models/__init__.py rename gns3server/db/{models.py => models/base.py} (67%) create mode 100644 gns3server/db/models/templates.py create mode 100644 gns3server/db/models/users.py create mode 100644 gns3server/db/repositories/templates.py delete mode 100644 tests/controller/test_template.py diff --git a/gns3server/api/routes/controller/templates.py b/gns3server/api/routes/controller/templates.py index a8f014e0..46123189 100644 --- a/gns3server/api/routes/controller/templates.py +++ b/gns3server/api/routes/controller/templates.py @@ -21,18 +21,25 @@ API routes for templates. import hashlib import json +import pydantic import logging log = logging.getLogger(__name__) -from fastapi import APIRouter, Request, Response, HTTPException, status -from fastapi.encoders import jsonable_encoder +from fastapi import APIRouter, Request, Response, HTTPException, Depends, status from typing import List from uuid import UUID from gns3server import schemas from gns3server.controller import Controller +from gns3server.db.repositories.templates import TemplatesRepository +from gns3server.controller.controller_error import ( + ControllerBadRequestError, + ControllerNotFoundError, + ControllerForbiddenError +) +from .dependencies.database import get_repository router = APIRouter() @@ -41,107 +48,141 @@ responses = { } -@router.post("/templates", - status_code=status.HTTP_201_CREATED, - response_model=schemas.Template) -def create_template(template_data: schemas.TemplateCreate): +@router.post("/templates", response_model=schemas.Template, status_code=status.HTTP_201_CREATED) +async def create_template( + new_template: schemas.TemplateCreate, + template_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)) +) -> dict: """ Create a new template. """ - controller = Controller.instance() - template = controller.template_manager.add_template(jsonable_encoder(template_data, exclude_unset=True)) - # Reset the symbol list - controller.symbols.list() - return template.__json__() + try: + return await template_repo.create_template(new_template) + except pydantic.ValidationError as e: + raise ControllerBadRequestError(f"JSON schema error received while creating new template: {e}") @router.get("/templates/{template_id}", response_model=schemas.Template, response_model_exclude_unset=True, responses=responses) -def get_template(template_id: UUID, request: Request, response: Response): +async def get_template( + template_id: UUID, + request: Request, + response: Response, + template_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)) +) -> dict: """ Return a template. """ request_etag = request.headers.get("If-None-Match", "") - controller = Controller.instance() - template = controller.template_manager.get_template(str(template_id)) - data = json.dumps(template.__json__()) + template = await template_repo.get_template(template_id) + if not template: + raise ControllerNotFoundError(f"Template '{template_id}' not found") + data = json.dumps(template) template_etag = '"' + hashlib.md5(data.encode()).hexdigest() + '"' if template_etag == request_etag: raise HTTPException(status_code=status.HTTP_304_NOT_MODIFIED) else: response.headers["ETag"] = template_etag - return template.__json__() + return template @router.put("/templates/{template_id}", response_model=schemas.Template, response_model_exclude_unset=True, responses=responses) -def update_template(template_id: UUID, template_data: schemas.TemplateUpdate): +async def update_template( + template_id: UUID, + template_data: schemas.TemplateUpdate, + template_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)) +) -> dict: """ Update a template. """ - controller = Controller.instance() - template = controller.template_manager.get_template(str(template_id)) - template.update(**jsonable_encoder(template_data, exclude_unset=True)) - return template.__json__() + if template_repo.get_builtin_template(template_id): + raise ControllerForbiddenError(f"Template '{template_id}' cannot be updated because it is built-in") + template = await template_repo.update_template(template_id, template_data) + if not template: + raise ControllerNotFoundError(f"Template '{template_id}' not found") + return template @router.delete("/templates/{template_id}", status_code=status.HTTP_204_NO_CONTENT, responses=responses) -def delete_template(template_id: UUID): +async def delete_template( + template_id: UUID, + template_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)) +) -> None: """ Delete a template. """ - controller = Controller.instance() - controller.template_manager.delete_template(str(template_id)) + if template_repo.get_builtin_template(template_id): + raise ControllerForbiddenError(f"Template '{template_id}' cannot be deleted because it is built-in") + success = await template_repo.delete_template(template_id) + if not success: + raise ControllerNotFoundError(f"Template '{template_id}' not found") @router.get("/templates", response_model=List[schemas.Template], response_model_exclude_unset=True) -def get_templates(): +async def get_templates( + template_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)) +) -> List[dict]: """ Return all templates. """ - controller = Controller.instance() - return [c.__json__() for c in controller.template_manager.templates.values()] + templates = await template_repo.get_templates() + return templates @router.post("/templates/{template_id}/duplicate", response_model=schemas.Template, status_code=status.HTTP_201_CREATED, responses=responses) -async def duplicate_template(template_id: UUID): +async def duplicate_template( + template_id: UUID, + template_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)) +) -> dict: """ Duplicate a template. """ - controller = Controller.instance() - template = controller.template_manager.duplicate_template(str(template_id)) - return template.__json__() + if template_repo.get_builtin_template(template_id): + raise ControllerForbiddenError(f"Template '{template_id}' cannot be duplicated because it is built-in") + template = await template_repo.duplicate_template(template_id) + if not template: + raise ControllerNotFoundError(f"Template '{template_id}' not found") + return template @router.post("/projects/{project_id}/templates/{template_id}", response_model=schemas.Node, status_code=status.HTTP_201_CREATED, responses={404: {"model": schemas.ErrorMessage, "description": "Could not find project or template"}}) -async def create_node_from_template(project_id: UUID, template_id: UUID, template_usage: schemas.TemplateUsage): +async def create_node_from_template( + project_id: UUID, + template_id: UUID, + template_usage: schemas.TemplateUsage, + template_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)) +) -> schemas.Node: """ Create a new node from a template. """ + template = await template_repo.get_template(template_id) + if not template: + raise ControllerNotFoundError(f"Template '{template_id}' not found") controller = Controller.instance() project = controller.get_project(str(project_id)) - node = await project.add_node_from_template(str(template_id), + node = await project.add_node_from_template(template, x=template_usage.x, y=template_usage.y, compute_id=template_usage.compute_id) diff --git a/gns3server/controller/__init__.py b/gns3server/controller/__init__.py index 3c4ee31c..0ffef683 100644 --- a/gns3server/controller/__init__.py +++ b/gns3server/controller/__init__.py @@ -25,10 +25,8 @@ import asyncio 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 @@ -55,7 +53,6 @@ class Controller: 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 @@ -208,10 +205,6 @@ class Controller: "appliances_etag": self._appliance_manager.appliances_etag, "version": __version__} - for template in self._template_manager.templates.values(): - if not template.builtin: - controller_settings["templates"].append(template.__json__()) - for compute in self._computes.values(): if compute.id != "local" and compute.id != "vm": controller_settings["computes"].append({"host": compute.host, @@ -259,7 +252,6 @@ class Controller: 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", []) @@ -546,14 +538,6 @@ class Controller: return self._appliance_manager - @property - def template_manager(self): - """ - :returns: Template Manager instance - """ - - return self._template_manager - @property def iou_license(self): """ diff --git a/gns3server/controller/node.py b/gns3server/controller/node.py index 8c3bf779..76270ce4 100644 --- a/gns3server/controller/node.py +++ b/gns3server/controller/node.py @@ -612,7 +612,6 @@ class Node: if the image exists """ - print("UPLOAD MISSING IMAGE") for directory in images_directories(type): image = os.path.join(directory, img) if os.path.exists(image): diff --git a/gns3server/controller/project.py b/gns3server/controller/project.py index d80cadcd..4189b888 100644 --- a/gns3server/controller/project.py +++ b/gns3server/controller/project.py @@ -500,16 +500,11 @@ class Project: return new_name @open_required - async def add_node_from_template(self, template_id, x=0, y=0, name=None, compute_id=None): + async def add_node_from_template(self, template, x=0, y=0, name=None, compute_id=None): """ Create a node from a template. """ - try: - template = copy.deepcopy(self.controller.template_manager.templates[template_id].settings) - except KeyError: - msg = "Template {} doesn't exist".format(template_id) - log.error(msg) - raise ControllerNotFoundError(msg) + template["x"] = x template["y"] = y node_type = template.pop("template_type") diff --git a/gns3server/controller/template.py b/gns3server/controller/template.py deleted file mode 100644 index f15f14cd..00000000 --- a/gns3server/controller/template.py +++ /dev/null @@ -1,168 +0,0 @@ -#!/usr/bin/env python -# -# Copyright (C) 2020 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 . - -import copy -import uuid - -from pydantic import ValidationError -from fastapi.encoders import jsonable_encoder -from gns3server import schemas - -import logging -log = logging.getLogger(__name__) - -ID_TO_CATEGORY = { - 3: "firewall", - 2: "guest", - 1: "switch", - 0: "router" -} - -TEMPLATE_TYPE_TO_SHEMA = { - "cloud": schemas.CloudTemplate, - "ethernet_hub": schemas.EthernetHubTemplate, - "ethernet_switch": schemas.EthernetSwitchTemplate, - "docker": schemas.DockerTemplate, - "dynamips": schemas.DynamipsTemplate, - "vpcs": schemas.VPCSTemplate, - "virtualbox": schemas.VirtualBoxTemplate, - "vmware": schemas.VMwareTemplate, - "iou": schemas.IOUTemplate, - "qemu": schemas.QemuTemplate -} - -DYNAMIPS_PLATFORM_TO_SHEMA = { - "c7200": schemas.C7200DynamipsTemplate, - "c3745": schemas.C3745DynamipsTemplate, - "c3725": schemas.C3725DynamipsTemplate, - "c3600": schemas.C3600DynamipsTemplate, - "c2691": schemas.C2691DynamipsTemplate, - "c2600": schemas.C2600DynamipsTemplate, - "c1700": schemas.C1700DynamipsTemplate -} - - -class Template: - - def __init__(self, template_id, settings, builtin=False): - - if template_id is None: - self._id = str(uuid.uuid4()) - elif isinstance(template_id, uuid.UUID): - self._id = str(template_id) - else: - self._id = template_id - - self._settings = copy.deepcopy(settings) - - # Version of the gui before 2.1 use linked_base - # and the server linked_clone - if "linked_base" in self.settings: - linked_base = self._settings.pop("linked_base") - if "linked_clone" not in self._settings: - self._settings["linked_clone"] = linked_base - - # Convert old GUI category to text category - try: - self._settings["category"] = ID_TO_CATEGORY[self._settings["category"]] - except KeyError: - pass - - # The "server" setting has been replaced by "compute_id" setting in version 2.2 - if "server" in self._settings: - self._settings["compute_id"] = self._settings.pop("server") - - # The "node_type" setting has been replaced by "template_type" setting in version 2.2 - if "node_type" in self._settings: - self._settings["template_type"] = self._settings.pop("node_type") - - # Remove an old IOU setting - if self._settings["template_type"] == "iou" and "image" in self._settings: - del self._settings["image"] - - self._builtin = builtin - - if builtin is False: - try: - template_schema = TEMPLATE_TYPE_TO_SHEMA[self.template_type] - template_settings_with_defaults = template_schema.parse_obj(self.__json__()) - self._settings = jsonable_encoder(template_settings_with_defaults.dict()) - if self.template_type == "dynamips": - # special case for Dynamips to cover all platform types that contain specific settings - dynamips_template_schema = DYNAMIPS_PLATFORM_TO_SHEMA[self._settings["platform"]] - dynamips_template_settings_with_defaults = dynamips_template_schema.parse_obj(self.__json__()) - self._settings = jsonable_encoder(dynamips_template_settings_with_defaults.dict()) - except ValidationError as e: - print(e) #TODO: handle errors - raise - - log.debug('Template "{name}" [{id}] loaded'.format(name=self.name, id=self._id)) - - @property - def id(self): - return self._id - - @property - def settings(self): - return self._settings - - @settings.setter - def settings(self, settings): - self._settings.update(settings) - - @property - def name(self): - return self._settings["name"] - - @property - def compute_id(self): - return self._settings["compute_id"] - - @property - def template_type(self): - return self._settings["template_type"] - - @property - def builtin(self): - return self._builtin - - def update(self, **kwargs): - - from gns3server.controller import Controller - controller = Controller.instance() - Controller.instance().check_can_write_config() - self._settings.update(kwargs) - controller.notification.controller_emit("template.updated", self.__json__()) - controller.save() - - def __json__(self): - """ - Template settings. - """ - - settings = self._settings - settings.update({"template_id": self._id, - "builtin": self.builtin}) - - if self.builtin: - # builin templates have compute_id set to None to tell clients - # to select a compute - settings["compute_id"] = None - else: - settings["compute_id"] = self.compute_id - - return settings diff --git a/gns3server/controller/template_manager.py b/gns3server/controller/template_manager.py deleted file mode 100644 index 663d63e1..00000000 --- a/gns3server/controller/template_manager.py +++ /dev/null @@ -1,148 +0,0 @@ -#!/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 . - -import copy -import uuid -import pydantic - -from .controller_error import ControllerError, ControllerNotFoundError -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 pydantic.ValidationError as e: - message = "Cannot load template with JSON data '{}': {}".format(template_settings, e) - log.warning(message) - continue - - # Add builtins - builtins = [] - builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "cloud"), {"template_type": "cloud", "name": "Cloud", "default_name_format": "Cloud{0}", "category": 2, "symbol": ":/symbols/cloud.svg"}, builtin=True)) - builtins.append(Template(uuid.uuid3(uuid.NAMESPACE_DNS, "nat"), {"template_type": "nat", "name": "NAT", "default_name_format": "NAT{0}", "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": "none", "name": "Ethernet switch", "default_name_format": "Switch{0}", "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", "default_name_format": "Hub{0}", "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", "default_name_format": "FRSW{0}", "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", "default_name_format": "ATMSW{0}", "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 ControllerError("Template ID '{}' already exists".format(template_id)) - else: - template_id = settings.setdefault("template_id", str(uuid.uuid4())) - try: - template = Template(template_id, settings) - except pydantic.ValidationError as e: - message = "JSON schema error adding template with JSON data '{}': {}".format(settings, e) - raise ControllerError(message) - - from . import Controller - Controller.instance().check_can_write_config() - self._templates[template.id] = template - 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 ControllerNotFoundError("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 ControllerError("Template ID {} cannot be deleted because it is a builtin".format(template_id)) - from . import Controller - Controller.instance().check_can_write_config() - self._templates.pop(template_id) - 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 ControllerError("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) - - diff --git a/gns3server/db/models/__init__.py b/gns3server/db/models/__init__.py new file mode 100644 index 00000000..86756367 --- /dev/null +++ b/gns3server/db/models/__init__.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python +# +# Copyright (C) 2021 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 . + +from .base import Base +from .users import User +from .templates import ( + Template, + CloudTemplate, + DockerTemplate, + DynamipsTemplate, + EthernetHubTemplate, + EthernetSwitchTemplate, + IOUTemplate, + QemuTemplate, + VirtualBoxTemplate, + VMwareTemplate, + VPCSTemplate +) diff --git a/gns3server/db/models.py b/gns3server/db/models/base.py similarity index 67% rename from gns3server/db/models.py rename to gns3server/db/models/base.py index 24be8cc6..09ad9c46 100644 --- a/gns3server/db/models.py +++ b/gns3server/db/models/base.py @@ -17,21 +17,33 @@ import uuid -from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, DateTime, func -from sqlalchemy.orm import relationship +from fastapi.encoders import jsonable_encoder +from sqlalchemy import Column, DateTime, func, inspect from sqlalchemy.types import TypeDecorator, CHAR from sqlalchemy.dialects.postgresql import UUID +from sqlalchemy.ext.declarative import as_declarative -from sqlalchemy.orm import declarative_base -Base = declarative_base() +@as_declarative() +class Base: + + def _asdict(self): + + return {c.key: getattr(self, c.key) + for c in inspect(self).mapper.column_attrs} + + def _asjson(self): + + return jsonable_encoder(self._asdict()) class GUID(TypeDecorator): - """Platform-independent GUID type. + """ + Platform-independent GUID type. Uses PostgreSQL's UUID type, otherwise uses CHAR(32), storing as stringified hex values. """ + impl = CHAR def load_dialect_impl(self, dialect): @@ -73,30 +85,3 @@ class BaseTable(Base): def generate_uuid(): return str(uuid.uuid4()) - - -class User(BaseTable): - - __tablename__ = "users" - - user_id = Column(GUID, primary_key=True, default=generate_uuid) - username = Column(String, unique=True, index=True) - email = Column(String, unique=True, index=True) - full_name = Column(String) - hashed_password = Column(String) - is_active = Column(Boolean, default=True) - is_superuser = Column(Boolean, default=False) - - -# items = relationship("Item", back_populates="owner") -# -# -# class Item(Base): -# __tablename__ = "items" -# -# id = Column(Integer, primary_key=True, index=True) -# title = Column(String, index=True) -# description = Column(String, index=True) -# owner_id = Column(Integer, ForeignKey("users.id")) -# -# owner = relationship("User", back_populates="items") diff --git a/gns3server/db/models/templates.py b/gns3server/db/models/templates.py new file mode 100644 index 00000000..ee2a83cd --- /dev/null +++ b/gns3server/db/models/templates.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python +# +# Copyright (C) 2021 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 . + + +from sqlalchemy import Boolean, Column, String, Integer, ForeignKey, PickleType + +from .base import BaseTable, generate_uuid, GUID + + +class Template(BaseTable): + + __tablename__ = "templates" + + template_id = Column(GUID, primary_key=True, default=generate_uuid) + name = Column(String, index=True) + category = Column(String) + default_name_format = Column(String) + symbol = Column(String) + builtin = Column(Boolean, default=False) + compute_id = Column(String) + usage = Column(String) + template_type = Column(String) + + __mapper_args__ = { + "polymorphic_identity": "templates", + "polymorphic_on": template_type + } + + +class CloudTemplate(Template): + + __tablename__ = "cloud_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + ports_mapping = Column(PickleType) + remote_console_host = Column(String) + remote_console_port = Column(Integer) + remote_console_type = Column(String) + remote_console_http_path = Column(String) + + __mapper_args__ = { + "polymorphic_identity": "cloud" + } + + +class DockerTemplate(Template): + + __tablename__ = "docker_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + image = Column(String) + adapters = Column(Integer) + start_command = Column(String) + environment = Column(String) + console_type = Column(String) + aux_type = Column(String) + console_auto_start = Column(Boolean) + console_http_port = Column(Integer) + console_http_path = Column(String) + console_resolution = Column(String) + extra_hosts = Column(String) + extra_volumes = Column(PickleType) + memory = Column(Integer) + cpus = Column(Integer) + custom_adapters = Column(PickleType) + + __mapper_args__ = { + "polymorphic_identity": "docker" + } + + +class DynamipsTemplate(Template): + + __tablename__ = "dynamips_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + platform = Column(String) + chassis = Column(String) + image = Column(String) + exec_area = Column(Integer) + mmap = Column(Boolean) + mac_addr = Column(String) + system_id = Column(String) + startup_config = Column(String) + private_config = Column(String) + idlepc = Column(String) + idlemax = Column(Integer) + idlesleep = Column(Integer) + disk0 = Column(Integer) + disk1 = Column(Integer) + auto_delete_disks = Column(Boolean) + console_type = Column(String) + console_auto_start = Column(Boolean) + aux_type = Column(String) + ram = Column(Integer) + nvram = Column(Integer) + npe = Column(String) + midplane = Column(String) + sparsemem = Column(Boolean) + iomem = Column(Integer) + slot0 = Column(String) + slot1 = Column(String) + slot2 = Column(String) + slot3 = Column(String) + slot4 = Column(String) + slot5 = Column(String) + slot6 = Column(String) + wic0 = Column(String) + wic1 = Column(String) + wic2 = Column(String) + + __mapper_args__ = { + "polymorphic_identity": "dynamips" + } + + +class EthernetHubTemplate(Template): + + __tablename__ = "ethernet_hub_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + ports_mapping = Column(PickleType) + + __mapper_args__ = { + "polymorphic_identity": "ethernet_hub" + } + + +class EthernetSwitchTemplate(Template): + + __tablename__ = "ethernet_switch_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + ports_mapping = Column(PickleType) + console_type = Column(String) + + __mapper_args__ = { + "polymorphic_identity": "ethernet_switch" + } + + +class IOUTemplate(Template): + + __tablename__ = "iou_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + path = Column(String) + ethernet_adapters = Column(Integer) + serial_adapters = Column(Integer) + ram = Column(Integer) + nvram = Column(Integer) + use_default_iou_values = Column(Boolean) + startup_config = Column(String) + private_config = Column(String) + l1_keepalives = Column(Boolean) + console_type = Column(String) + console_auto_start = Column(Boolean) + + __mapper_args__ = { + "polymorphic_identity": "iou" + } + + +class QemuTemplate(Template): + + __tablename__ = "qemu_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + qemu_path = Column(String) + platform = Column(String) + linked_clone = Column(Boolean) + ram = Column(Integer) + cpus = Column(Integer) + maxcpus = Column(Integer) + adapters = Column(Integer) + adapter_type = Column(String) + mac_address = Column(String) + first_port_name = Column(String) + port_name_format = Column(String) + port_segment_size = Column(Integer) + console_type = Column(String) + console_auto_start = Column(Boolean) + aux_type = Column(String) + boot_priority = Column(String) + hda_disk_image = Column(String) + hda_disk_interface = Column(String) + hdb_disk_image = Column(String) + hdb_disk_interface = Column(String) + hdc_disk_image = Column(String) + hdc_disk_interface = Column(String) + hdd_disk_image = Column(String) + hdd_disk_interface = Column(String) + cdrom_image = Column(String) + initrd = Column(String) + kernel_image = Column(String) + bios_image = Column(String) + kernel_command_line = Column(String) + legacy_networking = Column(Boolean) + replicate_network_connection_state = Column(Boolean) + create_config_disk = Column(Boolean) + on_close = Column(String) + cpu_throttling = Column(Integer) + process_priority = Column(String) + options = Column(String) + custom_adapters = Column(PickleType) + + __mapper_args__ = { + "polymorphic_identity": "qemu" + } + + +class VirtualBoxTemplate(Template): + + __tablename__ = "virtualbox_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + vmname = Column(String) + ram = Column(Integer) + linked_clone = Column(Boolean) + adapters = Column(Integer) + use_any_adapter = Column(Boolean) + adapter_type = Column(String) + first_port_name = Column(String) + port_name_format = Column(String) + port_segment_size = Column(Integer) + headless = Column(Boolean) + on_close = Column(String) + console_type = Column(String) + console_auto_start = Column(Boolean) + custom_adapters = Column(PickleType) + + __mapper_args__ = { + "polymorphic_identity": "virtualbox" + } + + +class VMwareTemplate(Template): + + __tablename__ = "vmware_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + vmx_path = Column(String) + linked_clone = Column(Boolean) + first_port_name = Column(String) + port_name_format = Column(String) + port_segment_size = Column(Integer) + adapters = Column(Integer) + adapter_type = Column(String) + use_any_adapter = Column(Boolean) + headless = Column(Boolean) + on_close = Column(String) + console_type = Column(String) + console_auto_start = Column(Boolean) + custom_adapters = Column(PickleType) + + __mapper_args__ = { + "polymorphic_identity": "vmware" + } + + +class VPCSTemplate(Template): + + __tablename__ = "vpcs_templates" + + template_id = Column(GUID, ForeignKey("templates.template_id"), primary_key=True) + base_script_file = Column(String) + console_type = Column(String) + console_auto_start = Column(Boolean, default=False) + + __mapper_args__ = { + "polymorphic_identity": "vpcs" + } diff --git a/gns3server/db/models/users.py b/gns3server/db/models/users.py new file mode 100644 index 00000000..d02db8ba --- /dev/null +++ b/gns3server/db/models/users.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python +# +# Copyright (C) 2021 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 . + +from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, DateTime, func +from sqlalchemy.orm import relationship + +from .base import BaseTable, generate_uuid, GUID + + +class User(BaseTable): + + __tablename__ = "users" + + user_id = Column(GUID, primary_key=True, default=generate_uuid) + username = Column(String, unique=True, index=True) + email = Column(String, unique=True, index=True) + full_name = Column(String) + hashed_password = Column(String) + is_active = Column(Boolean, default=True) + is_superuser = Column(Boolean, default=False) + + +# items = relationship("Item", back_populates="owner") +# +# +# class Item(Base): +# __tablename__ = "items" +# +# id = Column(Integer, primary_key=True, index=True) +# title = Column(String, index=True) +# description = Column(String, index=True) +# owner_id = Column(Integer, ForeignKey("users.id")) +# +# owner = relationship("User", back_populates="items") \ No newline at end of file diff --git a/gns3server/db/repositories/base.py b/gns3server/db/repositories/base.py index ab7c5ca3..e4e8179b 100644 --- a/gns3server/db/repositories/base.py +++ b/gns3server/db/repositories/base.py @@ -16,6 +16,7 @@ # along with this program. If not, see . from sqlalchemy.ext.asyncio import AsyncSession +from gns3server.controller import Controller class BaseRepository: @@ -23,3 +24,4 @@ class BaseRepository: def __init__(self, db_session: AsyncSession) -> None: self._db_session = db_session + self._controller = Controller.instance() diff --git a/gns3server/db/repositories/templates.py b/gns3server/db/repositories/templates.py new file mode 100644 index 00000000..7122625a --- /dev/null +++ b/gns3server/db/repositories/templates.py @@ -0,0 +1,243 @@ +#!/usr/bin/env python +# +# Copyright (C) 2021 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 . + +import uuid + +from uuid import UUID +from typing import List +from fastapi.encoders import jsonable_encoder +from sqlalchemy import select, update, delete +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm.session import make_transient + +from .base import BaseRepository + +import gns3server.db.models as models +from gns3server import schemas + +TEMPLATE_TYPE_TO_SHEMA = { + "cloud": schemas.CloudTemplate, + "ethernet_hub": schemas.EthernetHubTemplate, + "ethernet_switch": schemas.EthernetSwitchTemplate, + "docker": schemas.DockerTemplate, + "dynamips": schemas.DynamipsTemplate, + "vpcs": schemas.VPCSTemplate, + "virtualbox": schemas.VirtualBoxTemplate, + "vmware": schemas.VMwareTemplate, + "iou": schemas.IOUTemplate, + "qemu": schemas.QemuTemplate +} + +DYNAMIPS_PLATFORM_TO_SHEMA = { + "c7200": schemas.C7200DynamipsTemplate, + "c3745": schemas.C3745DynamipsTemplate, + "c3725": schemas.C3725DynamipsTemplate, + "c3600": schemas.C3600DynamipsTemplate, + "c2691": schemas.C2691DynamipsTemplate, + "c2600": schemas.C2600DynamipsTemplate, + "c1700": schemas.C1700DynamipsTemplate +} + +TEMPLATE_TYPE_TO_MODEL = { + "cloud": models.CloudTemplate, + "docker": models.DockerTemplate, + "dynamips": models.DynamipsTemplate, + "ethernet_hub": models.EthernetHubTemplate, + "ethernet_switch": models.EthernetSwitchTemplate, + "iou": models.IOUTemplate, + "qemu": models.QemuTemplate, + "virtualbox": models.VirtualBoxTemplate, + "vmware": models.VMwareTemplate, + "vpcs": models.VPCSTemplate +} + +# built-in templates have their compute_id set to None to tell clients to select a compute +BUILTIN_TEMPLATES = [ + { + "template_id": uuid.uuid3(uuid.NAMESPACE_DNS, "cloud"), + "template_type": "cloud", + "name": "Cloud", + "default_name_format": "Cloud{0}", + "category": "guest", + "symbol": ":/symbols/cloud.svg", + "compute_id": None, + "builtin": True + }, + { + "template_id": uuid.uuid3(uuid.NAMESPACE_DNS, "nat"), + "template_type": "nat", + "name": "NAT", + "default_name_format": "NAT{0}", + "category": "guest", + "symbol": ":/symbols/cloud.svg", + "compute_id": None, + "builtin": True + }, + { + "template_id": uuid.uuid3(uuid.NAMESPACE_DNS, "vpcs"), + "template_type": "vpcs", + "name": "VPCS", + "default_name_format": "PC{0}", + "category": "guest", + "symbol": ":/symbols/vpcs_guest.svg", + "base_script_file": "vpcs_base_config.txt", + "compute_id": None, + "builtin": True + }, + { + "template_id": uuid.uuid3(uuid.NAMESPACE_DNS, "ethernet_switch"), + "template_type": "ethernet_switch", + "name": "Ethernet switch", + "console_type": "none", + "default_name_format": "Switch{0}", + "category": "switch", + "symbol": ":/symbols/ethernet_switch.svg", + "compute_id": None, + "builtin": True + }, + { + "template_id": uuid.uuid3(uuid.NAMESPACE_DNS, "ethernet_hub"), + "template_type": "ethernet_hub", + "name": "Ethernet hub", + "default_name_format": "Hub{0}", + "category": "switch", + "symbol": ":/symbols/hub.svg", + "compute_id": None, + "builtin": True + }, + { + "template_id": uuid.uuid3(uuid.NAMESPACE_DNS, "frame_relay_switch"), + "template_type": "frame_relay_switch", + "name": "Frame Relay switch", + "default_name_format": "FRSW{0}", + "category": "switch", + "symbol": ":/symbols/frame_relay_switch.svg", + "compute_id": None, + "builtin": True + }, + { + "template_id": uuid.uuid3(uuid.NAMESPACE_DNS, "atm_switch"), + "template_type": "atm_switch", + "name": "ATM switch", + "default_name_format": "ATMSW{0}", + "category": "switch", + "symbol": ":/symbols/atm_switch.svg", + "compute_id": None, + "builtin": True + }, +] + + +class TemplatesRepository(BaseRepository): + + def __init__(self, db_session: AsyncSession) -> None: + + super().__init__(db_session) + + def get_builtin_template(self, template_id: UUID) -> dict: + + for builtin_template in BUILTIN_TEMPLATES: + if builtin_template["template_id"] == template_id: + return jsonable_encoder(builtin_template) + + async def get_template(self, template_id: UUID) -> dict: + + query = select(models.Template).where(models.Template.template_id == template_id) + result = (await self._db_session.execute(query)).scalars().first() + if result: + return result._asjson() + else: + return self.get_builtin_template(template_id) + + async def get_templates(self) -> List[dict]: + + templates = [] + query = select(models.Template) + result = await self._db_session.execute(query) + for db_template in result.scalars().all(): + templates.append(db_template._asjson()) + for builtin_template in BUILTIN_TEMPLATES: + templates.append(jsonable_encoder(builtin_template)) + return templates + + async def create_template(self, template_create: schemas.TemplateCreate) -> dict: + + # get the default template settings + template_settings = jsonable_encoder(template_create, exclude_unset=True) + template_schema = TEMPLATE_TYPE_TO_SHEMA[template_create.template_type] + template_settings_with_defaults = template_schema.parse_obj(template_settings) + settings = template_settings_with_defaults.dict() + if template_create.template_type == "dynamips": + # special case for Dynamips to cover all platform types that contain specific settings + dynamips_template_schema = DYNAMIPS_PLATFORM_TO_SHEMA[settings["platform"]] + dynamips_template_settings_with_defaults = dynamips_template_schema.parse_obj(template_settings) + settings = dynamips_template_settings_with_defaults.dict() + + model = TEMPLATE_TYPE_TO_MODEL[template_create.template_type] + db_template = model(**settings) + self._db_session.add(db_template) + await self._db_session.commit() + await self._db_session.refresh(db_template) + template = db_template._asjson() + self._controller.notification.controller_emit("template.created", template) + return template + + async def update_template( + self, + template_id: UUID, + template_update: schemas.TemplateUpdate) -> dict: + + update_values = template_update.dict(exclude_unset=True) + + query = update(models.Template) \ + .where(models.Template.template_id == template_id) \ + .values(update_values) + + await self._db_session.execute(query) + await self._db_session.commit() + template = await self.get_template(template_id) + if template: + self._controller.notification.controller_emit("template.updated", template) + return template + + async def delete_template(self, template_id: UUID) -> bool: + + query = delete(models.Template).where(models.Template.template_id == template_id) + result = await self._db_session.execute(query) + await self._db_session.commit() + if result.rowcount > 0: + self._controller.notification.controller_emit("template.deleted", {"template_id": str(template_id)}) + return True + return False + + async def duplicate_template(self, template_id: UUID) -> dict: + + query = select(models.Template).where(models.Template.template_id == template_id) + db_template = (await self._db_session.execute(query)).scalars().first() + if not db_template: + return db_template + + # duplicate db object with new primary key (template_id) + self._db_session.expunge(db_template) + make_transient(db_template) + db_template.template_id = None + self._db_session.add(db_template) + await self._db_session.commit() + await self._db_session.refresh(db_template) + template = db_template._asjson() + self._controller.notification.controller_emit("template.created", template) + return template diff --git a/gns3server/db/repositories/users.py b/gns3server/db/repositories/users.py index 587d12a5..ab620973 100644 --- a/gns3server/db/repositories/users.py +++ b/gns3server/db/repositories/users.py @@ -36,22 +36,26 @@ class UsersRepository(BaseRepository): async def get_user(self, user_id: UUID) -> Optional[models.User]: - result = await self._db_session.execute(select(models.User).where(models.User.user_id == user_id)) + query = select(models.User).where(models.User.user_id == user_id) + result = await self._db_session.execute(query) return result.scalars().first() async def get_user_by_username(self, username: str) -> Optional[models.User]: - result = await self._db_session.execute(select(models.User).where(models.User.username == username)) + query = select(models.User).where(models.User.username == username) + result = await self._db_session.execute(query) return result.scalars().first() async def get_user_by_email(self, email: str) -> Optional[models.User]: - result = await self._db_session.execute(select(models.User).where(models.User.email == email)) + query = select(models.User).where(models.User.email == email) + result = await self._db_session.execute(query) return result.scalars().first() async def get_users(self) -> List[models.User]: - result = await self._db_session.execute(select(models.User)) + query = select(models.User) + result = await self._db_session.execute(query) return result.scalars().all() async def create_user(self, user: schemas.UserCreate) -> models.User: diff --git a/gns3server/db/tasks.py b/gns3server/db/tasks.py index 97749ee5..2a6d0787 100644 --- a/gns3server/db/tasks.py +++ b/gns3server/db/tasks.py @@ -31,7 +31,7 @@ log = logging.getLogger(__name__) async def connect_to_db(app: FastAPI) -> None: db_path = os.path.join(Config.instance().config_dir, "gns3_controller.db") - db_url = os.environ.get("GNS3_DATABASE_URI", f"sqlite:///{db_path}") + db_url = os.environ.get("GNS3_DATABASE_URI", f"sqlite+pysqlite:///{db_path}") engine = create_async_engine(db_url, connect_args={"check_same_thread": False}, future=True) try: async with engine.begin() as conn: diff --git a/gns3server/schemas/templates.py b/gns3server/schemas/templates.py index af3dcec4..fe270a25 100644 --- a/gns3server/schemas/templates.py +++ b/gns3server/schemas/templates.py @@ -18,8 +18,10 @@ from pydantic import BaseModel, Field from typing import Optional, Union from enum import Enum +from uuid import UUID from .nodes import NodeType +from .base import DateTimeModelMixin class Category(str, Enum): @@ -38,7 +40,7 @@ class TemplateBase(BaseModel): Common template properties. """ - template_id: Optional[str] = None + template_id: Optional[UUID] = None name: Optional[str] = None category: Optional[Category] = None default_name_format: Optional[str] = None @@ -50,6 +52,7 @@ class TemplateBase(BaseModel): class Config: extra = "allow" + orm_mode = True class TemplateCreate(TemplateBase): @@ -67,9 +70,9 @@ class TemplateUpdate(TemplateBase): pass -class Template(TemplateBase): +class Template(DateTimeModelMixin, TemplateBase): - template_id: str + template_id: UUID name: str category: Category symbol: str diff --git a/requirements.txt b/requirements.txt index 9e9f9ef4..6e3c36c9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ uvicorn==0.13.3 -fastapi==0.62.0 +fastapi==0.63.0 websockets==8.1 python-multipart==0.0.5 aiohttp==3.7.2 @@ -10,7 +10,7 @@ psutil==5.7.3 async-timeout==3.0.1 distro==1.5.0 py-cpuinfo==7.0.0 -sqlalchemy==1.4.0b1 # beta version with asyncio support +sqlalchemy==1.4.0b2 # beta version with asyncio support passlib[bcrypt]==1.7.2 python-jose==3.2.0 email-validator==1.1.2 diff --git a/tests/api/routes/controller/test_templates.py b/tests/api/routes/controller/test_templates.py index 639e4aae..2c2ef2f3 100644 --- a/tests/api/routes/controller/test_templates.py +++ b/tests/api/routes/controller/test_templates.py @@ -23,944 +23,924 @@ from fastapi import FastAPI, status from httpx import AsyncClient from gns3server.controller import Controller -from gns3server.controller.template import Template +from gns3server.db.repositories.templates import BUILTIN_TEMPLATES pytestmark = pytest.mark.asyncio -async def test_template_list(app: FastAPI, client: AsyncClient, controller: Controller) -> None: - - id = str(uuid.uuid4()) - controller.template_manager.load_templates() - controller.template_manager._templates[id] = Template(id, { - "template_type": "qemu", - "category": 0, - "name": "test", - "symbol": "guest.svg", - "default_name_format": "{name}-{0}", - "compute_id": "local" - }) - response = await client.get(app.url_path_for("get_templates")) - assert response.status_code == status.HTTP_200_OK - assert len(response.json()) > 0 - - -async def test_template_create_without_id(app: FastAPI, client: AsyncClient, controller: Controller) -> None: - - params = {"base_script_file": "vpcs_base_config.txt", - "category": "guest", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "PC{0}", - "name": "VPCS_TEST", - "compute_id": "local", - "symbol": ":/symbols/vpcs_guest.svg", - "template_type": "vpcs"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - assert len(controller.template_manager.templates) == 1 - - -async def test_template_create_with_id(app: FastAPI, client: AsyncClient, controller: Controller): - - params = {"template_id": str(uuid.uuid4()), - "base_script_file": "vpcs_base_config.txt", - "category": "guest", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "PC{0}", - "name": "VPCS_TEST", - "compute_id": "local", - "symbol": ":/symbols/vpcs_guest.svg", - "template_type": "vpcs"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - assert len(controller.template_manager.templates) == 1 - - -async def test_template_create_wrong_type(app: FastAPI, client: AsyncClient, controller: Controller) -> None: - - params = {"template_id": str(uuid.uuid4()), - "base_script_file": "vpcs_base_config.txt", - "category": "guest", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "PC{0}", - "name": "VPCS_TEST", - "compute_id": "local", - "symbol": ":/symbols/vpcs_guest.svg", - "template_type": "invalid_template_type"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY - assert len(controller.template_manager.templates) == 0 - - -async def test_template_get(app: FastAPI, client: AsyncClient) -> None: - - template_id = str(uuid.uuid4()) - params = {"template_id": template_id, - "base_script_file": "vpcs_base_config.txt", - "category": "guest", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "PC{0}", - "name": "VPCS_TEST", - "compute_id": "local", - "symbol": ":/symbols/vpcs_guest.svg", - "template_type": "vpcs"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - - response = await client.get(app.url_path_for("get_template", template_id=template_id)) - assert response.status_code == status.HTTP_200_OK - assert response.json()["template_id"] == template_id - - -async def test_template_update(app: FastAPI, client: AsyncClient) -> None: - - template_id = str(uuid.uuid4()) - params = {"template_id": template_id, - "base_script_file": "vpcs_base_config.txt", - "category": "guest", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "PC{0}", - "name": "VPCS_TEST", - "compute_id": "local", - "symbol": ":/symbols/vpcs_guest.svg", - "template_type": "vpcs"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - - response = await client.get(app.url_path_for("get_template", template_id=template_id)) - assert response.status_code == status.HTTP_200_OK - assert response.json()["template_id"] == template_id - - params["name"] = "VPCS_TEST_RENAMED" - response = await client.put(app.url_path_for("update_template", template_id=template_id), json=params) - - assert response.status_code == status.HTTP_200_OK - assert response.json()["name"] == "VPCS_TEST_RENAMED" - - -async def test_template_delete(app: FastAPI, client: AsyncClient, controller: Controller) -> None: - - template_id = str(uuid.uuid4()) - params = {"template_id": template_id, - "base_script_file": "vpcs_base_config.txt", - "category": "guest", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "PC{0}", - "name": "VPCS_TEST", - "compute_id": "local", - "symbol": ":/symbols/vpcs_guest.svg", - "template_type": "vpcs"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - - response = await client.get(app.url_path_for("get_templates")) - assert len(response.json()) == 1 - assert len(controller.template_manager._templates) == 1 - - response = await client.delete(app.url_path_for("delete_template", template_id=template_id)) - assert response.status_code == status.HTTP_204_NO_CONTENT - - response = await client.get(app.url_path_for("get_templates")) - assert len(response.json()) == 0 - assert len(controller.template_manager.templates) == 0 - - -async def test_template_duplicate(app: FastAPI, client: AsyncClient, controller: Controller) -> None: - - template_id = str(uuid.uuid4()) - params = {"template_id": template_id, - "base_script_file": "vpcs_base_config.txt", - "category": "guest", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "PC{0}", - "name": "VPCS_TEST", - "compute_id": "local", - "symbol": ":/symbols/vpcs_guest.svg", - "template_type": "vpcs"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - - response = await client.post(app.url_path_for("duplicate_template", template_id=template_id)) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] != template_id - params.pop("template_id") - for param, value in params.items(): - assert response.json()[param] == value - - response = await client.get(app.url_path_for("get_templates")) - assert len(response.json()) == 2 - assert len(controller.template_manager.templates) == 2 - - -async def test_c7200_dynamips_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c7200 template", - "platform": "c7200", - "compute_id": "local", - "image": "c7200-adventerprisek9-mz.124-24.T5.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "dynamips", - "auto_delete_disks": False, - "builtin": False, - "category": "router", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "R{0}", - "disk0": 0, - "disk1": 0, - "exec_area": 64, - "idlemax": 500, - "idlepc": "", - "idlesleep": 30, - "image": "c7200-adventerprisek9-mz.124-24.T5.image", - "mac_addr": "", - "midplane": "vxr", - "mmap": True, - "name": "Cisco c7200 template", - "npe": "npe-400", - "nvram": 512, - "platform": "c7200", - "private_config": "", - "ram": 512, - "sparsemem": True, - "startup_config": "ios_base_startup-config.txt", - "symbol": ":/symbols/router.svg", - "system_id": "FTX0945W0MY"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_c3745_dynamips_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c3745 template", - "platform": "c3745", - "compute_id": "local", - "image": "c3745-adventerprisek9-mz.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "dynamips", - "auto_delete_disks": False, - "builtin": False, - "category": "router", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "R{0}", - "disk0": 0, - "disk1": 0, - "exec_area": 64, - "idlemax": 500, - "idlepc": "", - "idlesleep": 30, - "image": "c3745-adventerprisek9-mz.124-25d.image", - "mac_addr": "", - "mmap": True, - "name": "Cisco c3745 template", - "iomem": 5, - "nvram": 256, - "platform": "c3745", - "private_config": "", - "ram": 256, - "sparsemem": True, - "startup_config": "ios_base_startup-config.txt", - "symbol": ":/symbols/router.svg", - "system_id": "FTX0945W0MY"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_c3725_dynamips_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c3725 template", - "platform": "c3725", - "compute_id": "local", - "image": "c3725-adventerprisek9-mz.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "dynamips", - "auto_delete_disks": False, - "builtin": False, - "category": "router", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "R{0}", - "disk0": 0, - "disk1": 0, - "exec_area": 64, - "idlemax": 500, - "idlepc": "", - "idlesleep": 30, - "image": "c3725-adventerprisek9-mz.124-25d.image", - "mac_addr": "", - "mmap": True, - "name": "Cisco c3725 template", - "iomem": 5, - "nvram": 256, - "platform": "c3725", - "private_config": "", - "ram": 128, - "sparsemem": True, - "startup_config": "ios_base_startup-config.txt", - "symbol": ":/symbols/router.svg", - "system_id": "FTX0945W0MY"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_c3600_dynamips_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c3600 template", - "platform": "c3600", - "chassis": "3660", - "compute_id": "local", - "image": "c3660-a3jk9s-mz.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "dynamips", - "auto_delete_disks": False, - "builtin": False, - "category": "router", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "R{0}", - "disk0": 0, - "disk1": 0, - "exec_area": 64, - "idlemax": 500, - "idlepc": "", - "idlesleep": 30, - "image": "c3660-a3jk9s-mz.124-25d.image", - "mac_addr": "", - "mmap": True, - "name": "Cisco c3600 template", - "iomem": 5, - "nvram": 128, - "platform": "c3600", - "chassis": "3660", - "private_config": "", - "ram": 192, - "sparsemem": True, - "startup_config": "ios_base_startup-config.txt", - "symbol": ":/symbols/router.svg", - "system_id": "FTX0945W0MY"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_c3600_dynamips_template_create_wrong_chassis(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c3600 template", - "platform": "c3600", - "chassis": "3650", - "compute_id": "local", - "image": "c3660-a3jk9s-mz.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_409_CONFLICT - - -async def test_c2691_dynamips_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c2691 template", - "platform": "c2691", - "compute_id": "local", - "image": "c2691-adventerprisek9-mz.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "dynamips", - "auto_delete_disks": False, - "builtin": False, - "category": "router", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "R{0}", - "disk0": 0, - "disk1": 0, - "exec_area": 64, - "idlemax": 500, - "idlepc": "", - "idlesleep": 30, - "image": "c2691-adventerprisek9-mz.124-25d.image", - "mac_addr": "", - "mmap": True, - "name": "Cisco c2691 template", - "iomem": 5, - "nvram": 256, - "platform": "c2691", - "private_config": "", - "ram": 192, - "sparsemem": True, - "startup_config": "ios_base_startup-config.txt", - "symbol": ":/symbols/router.svg", - "system_id": "FTX0945W0MY"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_c2600_dynamips_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c2600 template", - "platform": "c2600", - "chassis": "2651XM", - "compute_id": "local", - "image": "c2600-adventerprisek9-mz.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "dynamips", - "auto_delete_disks": False, - "builtin": False, - "category": "router", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "R{0}", - "disk0": 0, - "disk1": 0, - "exec_area": 64, - "idlemax": 500, - "idlepc": "", - "idlesleep": 30, - "image": "c2600-adventerprisek9-mz.124-25d.image", - "mac_addr": "", - "mmap": True, - "name": "Cisco c2600 template", - "iomem": 15, - "nvram": 128, - "platform": "c2600", - "chassis": "2651XM", - "private_config": "", - "ram": 160, - "sparsemem": True, - "startup_config": "ios_base_startup-config.txt", - "symbol": ":/symbols/router.svg", - "system_id": "FTX0945W0MY"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_c2600_dynamips_template_create_wrong_chassis(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c2600 template", - "platform": "c2600", - "chassis": "2660XM", - "compute_id": "local", - "image": "c2600-adventerprisek9-mz.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_409_CONFLICT - - -async def test_c1700_dynamips_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c1700 template", - "platform": "c1700", - "chassis": "1760", - "compute_id": "local", - "image": "c1700-adventerprisek9-mz.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "dynamips", - "auto_delete_disks": False, - "builtin": False, - "category": "router", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "R{0}", - "disk0": 0, - "disk1": 0, - "exec_area": 64, - "idlemax": 500, - "idlepc": "", - "idlesleep": 30, - "image": "c1700-adventerprisek9-mz.124-25d.image", - "mac_addr": "", - "mmap": True, - "name": "Cisco c1700 template", - "iomem": 15, - "nvram": 128, - "platform": "c1700", - "chassis": "1760", - "private_config": "", - "ram": 160, - "sparsemem": False, - "startup_config": "ios_base_startup-config.txt", - "symbol": ":/symbols/router.svg", - "system_id": "FTX0945W0MY"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_c1700_dynamips_template_create_wrong_chassis(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c1700 template", - "platform": "c1700", - "chassis": "1770", - "compute_id": "local", - "image": "c1700-adventerprisek9-mz.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_409_CONFLICT - - -async def test_dynamips_template_create_wrong_platform(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cisco c3900 template", - "platform": "c3900", - "compute_id": "local", - "image": "c3900-test.124-25d.image", - "template_type": "dynamips"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_409_CONFLICT - - -async def test_iou_template_create(app: FastAPI, client: AsyncClient) -> None: - - image_path = str(Path("/path/to/i86bi_linux-ipbase-ms-12.4.bin")) - params = {"name": "IOU template", - "compute_id": "local", - "path": image_path, - "template_type": "iou"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "iou", - "builtin": False, - "category": "router", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "IOU{0}", - "ethernet_adapters": 2, - "name": "IOU template", - "nvram": 128, - "path": image_path, - "private_config": "", - "ram": 256, - "serial_adapters": 2, - "startup_config": "iou_l3_base_startup-config.txt", - "symbol": ":/symbols/multilayer_switch.svg", - "use_default_iou_values": True, - "l1_keepalives": False} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_docker_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Docker template", - "compute_id": "local", - "image": "gns3/endhost:latest", - "template_type": "docker"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"adapters": 1, - "template_type": "docker", - "builtin": False, - "category": "guest", - "compute_id": "local", - "console_auto_start": False, - "console_http_path": "/", - "console_http_port": 80, - "console_resolution": "1024x768", - "console_type": "telnet", - "default_name_format": "{name}-{0}", - "environment": "", - "extra_hosts": "", - "image": "gns3/endhost:latest", - "name": "Docker template", - "start_command": "", - "symbol": ":/symbols/docker_guest.svg", - "custom_adapters": []} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_qemu_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Qemu template", - "compute_id": "local", - "platform": "i386", - "hda_disk_image": "IOSvL2-15.2.4.0.55E.qcow2", - "ram": 512, - "template_type": "qemu"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"adapter_type": "e1000", - "adapters": 1, - "template_type": "qemu", - "bios_image": "", - "boot_priority": "c", - "builtin": False, - "category": "guest", - "cdrom_image": "", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "cpu_throttling": 0, - "cpus": 1, - "default_name_format": "{name}-{0}", - "first_port_name": "", - "hda_disk_image": "IOSvL2-15.2.4.0.55E.qcow2", - "hda_disk_interface": "none", - "hdb_disk_image": "", - "hdb_disk_interface": "none", - "hdc_disk_image": "", - "hdc_disk_interface": "none", - "hdd_disk_image": "", - "hdd_disk_interface": "none", - "initrd": "", - "kernel_command_line": "", - "kernel_image": "", - "legacy_networking": False, - "linked_clone": True, - "mac_address": "", - "name": "Qemu template", - "on_close": "power_off", - "options": "", - "platform": "i386", - "port_name_format": "Ethernet{0}", - "port_segment_size": 0, - "process_priority": "normal", - "qemu_path": "", - "ram": 512, - "symbol": ":/symbols/qemu_guest.svg", - "usage": "", - "custom_adapters": []} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_vmware_template_create(app: FastAPI, client: AsyncClient) -> None: - - vmx_path = str(Path("/path/to/vm.vmx")) - params = {"name": "VMware template", - "compute_id": "local", - "template_type": "vmware", - "vmx_path": vmx_path} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"adapter_type": "e1000", - "adapters": 1, - "template_type": "vmware", - "builtin": False, - "category": "guest", - "compute_id": "local", - "console_auto_start": False, - "console_type": "none", - "default_name_format": "{name}-{0}", - "first_port_name": "", - "headless": False, - "linked_clone": False, - "name": "VMware template", - "on_close": "power_off", - "port_name_format": "Ethernet{0}", - "port_segment_size": 0, - "symbol": ":/symbols/vmware_guest.svg", - "use_any_adapter": False, - "vmx_path": vmx_path, - "custom_adapters": []} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_virtualbox_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "VirtualBox template", - "compute_id": "local", - "template_type": "virtualbox", - "vmname": "My VirtualBox VM"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"adapter_type": "Intel PRO/1000 MT Desktop (82540EM)", - "adapters": 1, - "template_type": "virtualbox", - "builtin": False, - "category": "guest", - "compute_id": "local", - "console_auto_start": False, - "console_type": "none", - "default_name_format": "{name}-{0}", - "first_port_name": "", - "headless": False, - "linked_clone": False, - "name": "VirtualBox template", - "on_close": "power_off", - "port_name_format": "Ethernet{0}", - "port_segment_size": 0, - "ram": 256, - "symbol": ":/symbols/vbox_guest.svg", - "use_any_adapter": False, - "vmname": "My VirtualBox VM", - "custom_adapters": []} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_vpcs_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "VPCS template", - "compute_id": "local", - "template_type": "vpcs"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "vpcs", - "base_script_file": "vpcs_base_config.txt", - "builtin": False, - "category": "guest", - "compute_id": "local", - "console_auto_start": False, - "console_type": "telnet", - "default_name_format": "PC{0}", - "name": "VPCS template", - "symbol": ":/symbols/vpcs_guest.svg"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_ethernet_switch_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Ethernet switch template", - "compute_id": "local", - "template_type": "ethernet_switch"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "ethernet_switch", - "builtin": False, - "category": "switch", - "compute_id": "local", - "console_type": "none", - "default_name_format": "Switch{0}", - "name": "Ethernet switch template", - "ports_mapping": [{"ethertype": "0x8100", - "name": "Ethernet0", - "port_number": 0, - "type": "access", - "vlan": 1 - }, - {"ethertype": "0x8100", - "name": "Ethernet1", - "port_number": 1, - "type": "access", - "vlan": 1 - }, - {"ethertype": "0x8100", - "name": "Ethernet2", - "port_number": 2, - "type": "access", - "vlan": 1 - }, - {"ethertype": "0x8100", - "name": "Ethernet3", - "port_number": 3, - "type": "access", - "vlan": 1 - }, - {"ethertype": "0x8100", - "name": "Ethernet4", - "port_number": 4, - "type": "access", - "vlan": 1 - }, - {"ethertype": "0x8100", - "name": "Ethernet5", - "port_number": 5, - "type": "access", - "vlan": 1 - }, - {"ethertype": "0x8100", - "name": "Ethernet6", - "port_number": 6, - "type": "access", - "vlan": 1 - }, - {"ethertype": "0x8100", - "name": "Ethernet7", - "port_number": 7, - "type": "access", - "vlan": 1 - }], - "symbol": ":/symbols/ethernet_switch.svg"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_cloud_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Cloud template", - "compute_id": "local", - "template_type": "cloud"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"template_type": "cloud", - "builtin": False, - "category": "guest", - "compute_id": "local", - "default_name_format": "Cloud{0}", - "name": "Cloud template", - "ports_mapping": [], - "symbol": ":/symbols/cloud.svg", - "remote_console_host": "127.0.0.1", - "remote_console_port": 23, - "remote_console_type": "none", - "remote_console_http_path": "/"} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -async def test_ethernet_hub_template_create(app: FastAPI, client: AsyncClient) -> None: - - params = {"name": "Ethernet hub template", - "compute_id": "local", - "template_type": "ethernet_hub"} - - response = await client.post(app.url_path_for("create_template"), json=params) - assert response.status_code == status.HTTP_201_CREATED - assert response.json()["template_id"] is not None - - expected_response = {"ports_mapping": [{"port_number": 0, - "name": "Ethernet0" - }, - {"port_number": 1, - "name": "Ethernet1" - }, - {"port_number": 2, - "name": "Ethernet2" - }, - {"port_number": 3, - "name": "Ethernet3" - }, - {"port_number": 4, - "name": "Ethernet4" - }, - {"port_number": 5, - "name": "Ethernet5" - }, - {"port_number": 6, - "name": "Ethernet6" - }, - {"port_number": 7, - "name": "Ethernet7" - }], - "compute_id": "local", - "name": "Ethernet hub template", - "symbol": ":/symbols/hub.svg", - "default_name_format": "Hub{0}", - "template_type": "ethernet_hub", - "category": "switch", - "builtin": False} - - for item, value in expected_response.items(): - assert response.json().get(item) == value - - -# @pytest.mark.asyncio -# async def test_create_node_from_template(controller_api, controller, project): -# -# id = str(uuid.uuid4()) -# controller.template_manager._templates = {id: Template(id, { -# "template_type": "qemu", -# "category": 0, -# "name": "test", -# "symbol": "guest.svg", -# "default_name_format": "{name}-{0}", -# "compute_id": "example.com" -# })} -# with asyncio_patch("gns3server.controller.project.Project.add_node_from_template", return_value={"name": "test", "node_type": "qemu", "compute_id": "example.com"}) as mock: -# response = await client.post("/projects/{}/templates/{}".format(project.id, id), { -# "x": 42, -# "y": 12 -# }) -# mock.assert_called_with(id, x=42, y=12, compute_id=None) -# assert response.status_code == status.HTTP_201_CREATED +class TestTemplateRoutes: + + async def test_route_exist(self, app: FastAPI, client: AsyncClient) -> None: + + new_template = {"base_script_file": "vpcs_base_config.txt", + "category": "guest", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "PC{0}", + "name": "VPCS_TEST", + "compute_id": "local", + "symbol": ":/symbols/vpcs_guest.svg", + "template_type": "vpcs"} + + response = await client.post(app.url_path_for("create_template"), json=new_template) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + async def test_template_list(self, app: FastAPI, client: AsyncClient) -> None: + + response = await client.get(app.url_path_for("get_templates")) + assert response.status_code == status.HTTP_200_OK + assert len(response.json()) > 0 + + async def test_template_get(self, app: FastAPI, client: AsyncClient) -> None: + + template_id = str(uuid.uuid4()) + params = {"template_id": template_id, + "name": "VPCS_TEST", + "compute_id": "local", + "template_type": "vpcs"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + + response = await client.get(app.url_path_for("get_template", template_id=template_id)) + assert response.status_code == status.HTTP_200_OK + assert response.json()["template_id"] == template_id + + async def test_template_create_wrong_type(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None: + + params = {"name": "VPCS_TEST", + "compute_id": "local", + "template_type": "invalid_template_type"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY + + async def test_template_update(self, app: FastAPI, client: AsyncClient) -> None: + + template_id = str(uuid.uuid4()) + params = {"template_id": template_id, + "name": "VPCS_TEST", + "compute_id": "local", + "template_type": "vpcs"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + + response = await client.get(app.url_path_for("get_template", template_id=template_id)) + assert response.status_code == status.HTTP_200_OK + assert response.json()["template_id"] == template_id + + params["name"] = "VPCS_TEST_RENAMED" + response = await client.put(app.url_path_for("update_template", template_id=template_id), json=params) + + assert response.status_code == status.HTTP_200_OK + assert response.json()["name"] == "VPCS_TEST_RENAMED" + + async def test_template_delete(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None: + + template_id = str(uuid.uuid4()) + params = {"template_id": template_id, + "name": "VPCS_TEST", + "compute_id": "local", + "template_type": "vpcs"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + + response = await client.delete(app.url_path_for("delete_template", template_id=template_id)) + assert response.status_code == status.HTTP_204_NO_CONTENT + + # async def test_create_node_from_template(self, controller_api, controller, project): + # + # id = str(uuid.uuid4()) + # controller.template_manager._templates = {id: Template(id, { + # "template_type": "qemu", + # "category": 0, + # "name": "test", + # "symbol": "guest.svg", + # "default_name_format": "{name}-{0}", + # "compute_id": "example.com" + # })} + # with asyncio_patch("gns3server.controller.project.Project.add_node_from_template", return_value={"name": "test", "node_type": "qemu", "compute_id": "example.com"}) as mock: + # response = await client.post("/projects/{}/templates/{}".format(project.id, id), { + # "x": 42, + # "y": 12 + # }) + # mock.assert_called_with(id, x=42, y=12, compute_id=None) + # assert response.status_code == status.HTTP_201_CREATED + + +class TestDuplicateTemplates: + + async def test_template_duplicate(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None: + + template_id = str(uuid.uuid4()) + params = {"template_id": template_id, + "name": "VPCS_TEST", + "compute_id": "local", + "template_type": "vpcs"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + + response = await client.post(app.url_path_for("duplicate_template", template_id=template_id)) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] != template_id + params.pop("template_id") + for param, value in params.items(): + assert response.json()[param] == value + + response = await client.get(app.url_path_for("get_templates")) + assert len(response.json()) == 9 # includes builtin templates + + async def test_template_duplicate_invalid_template_id( + self, + app: FastAPI, + client: AsyncClient, + controller: Controller + ) -> None: + + template_id = str(uuid.uuid4()) + response = await client.post(app.url_path_for("duplicate_template", template_id=template_id)) + assert response.status_code == status.HTTP_404_NOT_FOUND + + +class TestBuiltinTemplates: + + async def test_list_builtin_templates(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None: + + response = await client.get(app.url_path_for("get_templates")) + assert len(response.json()) == 7 # there currently are 7 built-in templates + + async def test_get_builtin_template(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None: + + template_id = str(BUILTIN_TEMPLATES[0]["template_id"]) # take the first built-in template + response = await client.get(app.url_path_for("get_template", template_id=template_id)) + assert response.status_code == status.HTTP_200_OK + assert response.json()["template_id"] == template_id + + async def test_update_builtin_template(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None: + + template_id = str(BUILTIN_TEMPLATES[0]["template_id"]) # take the first built-in template + params = {"name": "RENAME_BUILTIN_TEMPLATE"} + response = await client.put(app.url_path_for("update_template", template_id=template_id), json=params) + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_duplicate_builtin_template(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None: + + template_id = str(BUILTIN_TEMPLATES[0]["template_id"]) # take the first built-in template + response = await client.post(app.url_path_for("duplicate_template", template_id=template_id)) + assert response.status_code == status.HTTP_403_FORBIDDEN + + async def test_delete_builtin_template(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None: + + template_id = str(BUILTIN_TEMPLATES[0]["template_id"]) # take the first built-in template + response = await client.delete(app.url_path_for("delete_template", template_id=template_id)) + assert response.status_code == status.HTTP_403_FORBIDDEN + + +class TestDynamipsTemplate: + + async def test_c7200_dynamips_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c7200 template", + "platform": "c7200", + "compute_id": "local", + "image": "c7200-adventerprisek9-mz.124-24.T5.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "dynamips", + "auto_delete_disks": False, + "builtin": False, + "category": "router", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "R{0}", + "disk0": 0, + "disk1": 0, + "exec_area": 64, + "idlemax": 500, + "idlepc": "", + "idlesleep": 30, + "image": "c7200-adventerprisek9-mz.124-24.T5.image", + "mac_addr": "", + "midplane": "vxr", + "mmap": True, + "name": "Cisco c7200 template", + "npe": "npe-400", + "nvram": 512, + "platform": "c7200", + "private_config": "", + "ram": 512, + "sparsemem": True, + "startup_config": "ios_base_startup-config.txt", + "symbol": ":/symbols/router.svg", + "system_id": "FTX0945W0MY"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + + async def test_c3745_dynamips_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c3745 template", + "platform": "c3745", + "compute_id": "local", + "image": "c3745-adventerprisek9-mz.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "dynamips", + "auto_delete_disks": False, + "builtin": False, + "category": "router", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "R{0}", + "disk0": 0, + "disk1": 0, + "exec_area": 64, + "idlemax": 500, + "idlepc": "", + "idlesleep": 30, + "image": "c3745-adventerprisek9-mz.124-25d.image", + "mac_addr": "", + "mmap": True, + "name": "Cisco c3745 template", + "iomem": 5, + "nvram": 256, + "platform": "c3745", + "private_config": "", + "ram": 256, + "sparsemem": True, + "startup_config": "ios_base_startup-config.txt", + "symbol": ":/symbols/router.svg", + "system_id": "FTX0945W0MY"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + async def test_c3725_dynamips_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c3725 template", + "platform": "c3725", + "compute_id": "local", + "image": "c3725-adventerprisek9-mz.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "dynamips", + "auto_delete_disks": False, + "builtin": False, + "category": "router", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "R{0}", + "disk0": 0, + "disk1": 0, + "exec_area": 64, + "idlemax": 500, + "idlepc": "", + "idlesleep": 30, + "image": "c3725-adventerprisek9-mz.124-25d.image", + "mac_addr": "", + "mmap": True, + "name": "Cisco c3725 template", + "iomem": 5, + "nvram": 256, + "platform": "c3725", + "private_config": "", + "ram": 128, + "sparsemem": True, + "startup_config": "ios_base_startup-config.txt", + "symbol": ":/symbols/router.svg", + "system_id": "FTX0945W0MY"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + async def test_c3600_dynamips_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c3600 template", + "platform": "c3600", + "chassis": "3660", + "compute_id": "local", + "image": "c3660-a3jk9s-mz.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "dynamips", + "auto_delete_disks": False, + "builtin": False, + "category": "router", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "R{0}", + "disk0": 0, + "disk1": 0, + "exec_area": 64, + "idlemax": 500, + "idlepc": "", + "idlesleep": 30, + "image": "c3660-a3jk9s-mz.124-25d.image", + "mac_addr": "", + "mmap": True, + "name": "Cisco c3600 template", + "iomem": 5, + "nvram": 128, + "platform": "c3600", + "chassis": "3660", + "private_config": "", + "ram": 192, + "sparsemem": True, + "startup_config": "ios_base_startup-config.txt", + "symbol": ":/symbols/router.svg", + "system_id": "FTX0945W0MY"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + async def test_c3600_dynamips_template_create_wrong_chassis(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c3600 template", + "platform": "c3600", + "chassis": "3650", + "compute_id": "local", + "image": "c3660-a3jk9s-mz.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_400_BAD_REQUEST + + async def test_c2691_dynamips_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c2691 template", + "platform": "c2691", + "compute_id": "local", + "image": "c2691-adventerprisek9-mz.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "dynamips", + "auto_delete_disks": False, + "builtin": False, + "category": "router", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "R{0}", + "disk0": 0, + "disk1": 0, + "exec_area": 64, + "idlemax": 500, + "idlepc": "", + "idlesleep": 30, + "image": "c2691-adventerprisek9-mz.124-25d.image", + "mac_addr": "", + "mmap": True, + "name": "Cisco c2691 template", + "iomem": 5, + "nvram": 256, + "platform": "c2691", + "private_config": "", + "ram": 192, + "sparsemem": True, + "startup_config": "ios_base_startup-config.txt", + "symbol": ":/symbols/router.svg", + "system_id": "FTX0945W0MY"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + async def test_c2600_dynamips_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c2600 template", + "platform": "c2600", + "chassis": "2651XM", + "compute_id": "local", + "image": "c2600-adventerprisek9-mz.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "dynamips", + "auto_delete_disks": False, + "builtin": False, + "category": "router", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "R{0}", + "disk0": 0, + "disk1": 0, + "exec_area": 64, + "idlemax": 500, + "idlepc": "", + "idlesleep": 30, + "image": "c2600-adventerprisek9-mz.124-25d.image", + "mac_addr": "", + "mmap": True, + "name": "Cisco c2600 template", + "iomem": 15, + "nvram": 128, + "platform": "c2600", + "chassis": "2651XM", + "private_config": "", + "ram": 160, + "sparsemem": True, + "startup_config": "ios_base_startup-config.txt", + "symbol": ":/symbols/router.svg", + "system_id": "FTX0945W0MY"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + async def test_c2600_dynamips_template_create_wrong_chassis(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c2600 template", + "platform": "c2600", + "chassis": "2660XM", + "compute_id": "local", + "image": "c2600-adventerprisek9-mz.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_400_BAD_REQUEST + + async def test_c1700_dynamips_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c1700 template", + "platform": "c1700", + "chassis": "1760", + "compute_id": "local", + "image": "c1700-adventerprisek9-mz.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "dynamips", + "auto_delete_disks": False, + "builtin": False, + "category": "router", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "R{0}", + "disk0": 0, + "disk1": 0, + "exec_area": 64, + "idlemax": 500, + "idlepc": "", + "idlesleep": 30, + "image": "c1700-adventerprisek9-mz.124-25d.image", + "mac_addr": "", + "mmap": True, + "name": "Cisco c1700 template", + "iomem": 15, + "nvram": 128, + "platform": "c1700", + "chassis": "1760", + "private_config": "", + "ram": 160, + "sparsemem": False, + "startup_config": "ios_base_startup-config.txt", + "symbol": ":/symbols/router.svg", + "system_id": "FTX0945W0MY"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + async def test_c1700_dynamips_template_create_wrong_chassis(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c1700 template", + "platform": "c1700", + "chassis": "1770", + "compute_id": "local", + "image": "c1700-adventerprisek9-mz.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_400_BAD_REQUEST + + async def test_dynamips_template_create_wrong_platform(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cisco c3900 template", + "platform": "c3900", + "compute_id": "local", + "image": "c3900-test.124-25d.image", + "template_type": "dynamips"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_400_BAD_REQUEST + + +class TestIOUTemplate: + + async def test_iou_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + image_path = str(Path("/path/to/i86bi_linux-ipbase-ms-12.4.bin")) + params = {"name": "IOU template", + "compute_id": "local", + "path": image_path, + "template_type": "iou"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "iou", + "builtin": False, + "category": "router", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "IOU{0}", + "ethernet_adapters": 2, + "name": "IOU template", + "nvram": 128, + "path": image_path, + "private_config": "", + "ram": 256, + "serial_adapters": 2, + "startup_config": "iou_l3_base_startup-config.txt", + "symbol": ":/symbols/multilayer_switch.svg", + "use_default_iou_values": True, + "l1_keepalives": False} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + +class TestDockerTemplate: + + async def test_docker_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Docker template", + "compute_id": "local", + "image": "gns3/endhost:latest", + "template_type": "docker"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"adapters": 1, + "template_type": "docker", + "builtin": False, + "category": "guest", + "compute_id": "local", + "console_auto_start": False, + "console_http_path": "/", + "console_http_port": 80, + "console_resolution": "1024x768", + "console_type": "telnet", + "default_name_format": "{name}-{0}", + "environment": "", + "extra_hosts": "", + "image": "gns3/endhost:latest", + "name": "Docker template", + "start_command": "", + "symbol": ":/symbols/docker_guest.svg", + "custom_adapters": []} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + +class TestQemuTemplate: + + async def test_qemu_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Qemu template", + "compute_id": "local", + "platform": "i386", + "hda_disk_image": "IOSvL2-15.2.4.0.55E.qcow2", + "ram": 512, + "template_type": "qemu"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"adapter_type": "e1000", + "adapters": 1, + "template_type": "qemu", + "bios_image": "", + "boot_priority": "c", + "builtin": False, + "category": "guest", + "cdrom_image": "", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "cpu_throttling": 0, + "cpus": 1, + "default_name_format": "{name}-{0}", + "first_port_name": "", + "hda_disk_image": "IOSvL2-15.2.4.0.55E.qcow2", + "hda_disk_interface": "none", + "hdb_disk_image": "", + "hdb_disk_interface": "none", + "hdc_disk_image": "", + "hdc_disk_interface": "none", + "hdd_disk_image": "", + "hdd_disk_interface": "none", + "initrd": "", + "kernel_command_line": "", + "kernel_image": "", + "legacy_networking": False, + "linked_clone": True, + "mac_address": "", + "name": "Qemu template", + "on_close": "power_off", + "options": "", + "platform": "i386", + "port_name_format": "Ethernet{0}", + "port_segment_size": 0, + "process_priority": "normal", + "qemu_path": "", + "ram": 512, + "symbol": ":/symbols/qemu_guest.svg", + "usage": "", + "custom_adapters": []} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + +class TestVMwareTemplate: + + async def test_vmware_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + vmx_path = str(Path("/path/to/vm.vmx")) + params = {"name": "VMware template", + "compute_id": "local", + "template_type": "vmware", + "vmx_path": vmx_path} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"adapter_type": "e1000", + "adapters": 1, + "template_type": "vmware", + "builtin": False, + "category": "guest", + "compute_id": "local", + "console_auto_start": False, + "console_type": "none", + "default_name_format": "{name}-{0}", + "first_port_name": "", + "headless": False, + "linked_clone": False, + "name": "VMware template", + "on_close": "power_off", + "port_name_format": "Ethernet{0}", + "port_segment_size": 0, + "symbol": ":/symbols/vmware_guest.svg", + "use_any_adapter": False, + "vmx_path": vmx_path, + "custom_adapters": []} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + +class TestVirtualBoxTemplate: + + async def test_virtualbox_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "VirtualBox template", + "compute_id": "local", + "template_type": "virtualbox", + "vmname": "My VirtualBox VM"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"adapter_type": "Intel PRO/1000 MT Desktop (82540EM)", + "adapters": 1, + "template_type": "virtualbox", + "builtin": False, + "category": "guest", + "compute_id": "local", + "console_auto_start": False, + "console_type": "none", + "default_name_format": "{name}-{0}", + "first_port_name": "", + "headless": False, + "linked_clone": False, + "name": "VirtualBox template", + "on_close": "power_off", + "port_name_format": "Ethernet{0}", + "port_segment_size": 0, + "ram": 256, + "symbol": ":/symbols/vbox_guest.svg", + "use_any_adapter": False, + "vmname": "My VirtualBox VM", + "custom_adapters": []} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + +class TestVPCSTemplate: + + async def test_vpcs_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "VPCS template", + "compute_id": "local", + "template_type": "vpcs"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "vpcs", + "base_script_file": "vpcs_base_config.txt", + "builtin": False, + "category": "guest", + "compute_id": "local", + "console_auto_start": False, + "console_type": "telnet", + "default_name_format": "PC{0}", + "name": "VPCS template", + "symbol": ":/symbols/vpcs_guest.svg"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + +class TestEthernetSwitchTemplate: + + async def test_ethernet_switch_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Ethernet switch template", + "compute_id": "local", + "template_type": "ethernet_switch"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "ethernet_switch", + "builtin": False, + "category": "switch", + "compute_id": "local", + "console_type": "none", + "default_name_format": "Switch{0}", + "name": "Ethernet switch template", + "ports_mapping": [{"ethertype": "0x8100", + "name": "Ethernet0", + "port_number": 0, + "type": "access", + "vlan": 1 + }, + {"ethertype": "0x8100", + "name": "Ethernet1", + "port_number": 1, + "type": "access", + "vlan": 1 + }, + {"ethertype": "0x8100", + "name": "Ethernet2", + "port_number": 2, + "type": "access", + "vlan": 1 + }, + {"ethertype": "0x8100", + "name": "Ethernet3", + "port_number": 3, + "type": "access", + "vlan": 1 + }, + {"ethertype": "0x8100", + "name": "Ethernet4", + "port_number": 4, + "type": "access", + "vlan": 1 + }, + {"ethertype": "0x8100", + "name": "Ethernet5", + "port_number": 5, + "type": "access", + "vlan": 1 + }, + {"ethertype": "0x8100", + "name": "Ethernet6", + "port_number": 6, + "type": "access", + "vlan": 1 + }, + {"ethertype": "0x8100", + "name": "Ethernet7", + "port_number": 7, + "type": "access", + "vlan": 1 + }], + "symbol": ":/symbols/ethernet_switch.svg"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + +class TestHubTemplate: + + async def test_ethernet_hub_template_create(self, app: FastAPI, client: AsyncClient) -> None: + params = {"name": "Ethernet hub template", + "compute_id": "local", + "template_type": "ethernet_hub"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"ports_mapping": [{"port_number": 0, + "name": "Ethernet0" + }, + {"port_number": 1, + "name": "Ethernet1" + }, + {"port_number": 2, + "name": "Ethernet2" + }, + {"port_number": 3, + "name": "Ethernet3" + }, + {"port_number": 4, + "name": "Ethernet4" + }, + {"port_number": 5, + "name": "Ethernet5" + }, + {"port_number": 6, + "name": "Ethernet6" + }, + {"port_number": 7, + "name": "Ethernet7" + }], + "compute_id": "local", + "name": "Ethernet hub template", + "symbol": ":/symbols/hub.svg", + "default_name_format": "Hub{0}", + "template_type": "ethernet_hub", + "category": "switch", + "builtin": False} + + for item, value in expected_response.items(): + assert response.json().get(item) == value + + +class TestCloudTemplate: + + async def test_cloud_template_create(self, app: FastAPI, client: AsyncClient) -> None: + + params = {"name": "Cloud template", + "compute_id": "local", + "template_type": "cloud"} + + response = await client.post(app.url_path_for("create_template"), json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["template_id"] is not None + + expected_response = {"template_type": "cloud", + "builtin": False, + "category": "guest", + "compute_id": "local", + "default_name_format": "Cloud{0}", + "name": "Cloud template", + "ports_mapping": [], + "symbol": ":/symbols/cloud.svg", + "remote_console_host": "127.0.0.1", + "remote_console_port": 23, + "remote_console_type": "none", + "remote_console_http_path": "/"} + + for item, value in expected_response.items(): + assert response.json().get(item) == value diff --git a/tests/controller/test_controller.py b/tests/controller/test_controller.py index 763f5efb..a7b69e35 100644 --- a/tests/controller/test_controller.py +++ b/tests/controller/test_controller.py @@ -445,33 +445,6 @@ def test_appliances(controller, tmpdir): elif j["name"] == "My Appliance": assert not j["builtin"] - -def test_load_templates(controller): - - controller._settings = {} - controller.template_manager.load_templates() - - 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.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.template_manager.templates.values(): - if template.name == "Test": - qemu_uuid = template.id - elif template.name == "Cloud": - cloud_uuid = template.id - 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": - assert cloud_uuid == template.id - - @pytest.mark.asyncio async def test_autoidlepc(controller): diff --git a/tests/controller/test_project.py b/tests/controller/test_project.py index 862fca76..0e4b81c6 100644 --- a/tests/controller/test_project.py +++ b/tests/controller/test_project.py @@ -26,7 +26,6 @@ from unittest.mock import patch from uuid import uuid4 from gns3server.controller.project import Project -from gns3server.controller.template import Template from gns3server.controller.node import Node from gns3server.controller.ports.ethernet_port import EthernetPort from gns3server.controller.controller_error import ControllerError, ControllerNotFoundError, ControllerForbiddenError @@ -343,72 +342,72 @@ async def test_add_node_iou_no_id_available(controller): await project.add_node(compute, "test1", None, node_type="iou") -@pytest.mark.asyncio -async def test_add_node_from_template(controller): - """ - For a local server we send the project path - """ - - compute = MagicMock() - compute.id = "local" - project = Project(controller=controller, name="Test") - project.emit_notification = MagicMock() - template = Template(str(uuid.uuid4()), { - "compute_id": "local", - "name": "Test", - "template_type": "vpcs", - "builtin": False, - }) - controller.template_manager.templates[template.id] = template - controller._computes["local"] = compute - - response = MagicMock() - response.json = {"console": 2048} - compute.post = AsyncioMagicMock(return_value=response) - - node = await project.add_node_from_template(template.id, x=23, y=12) - compute.post.assert_any_call('/projects', data={ - "name": project._name, - "project_id": project._id, - "path": project._path - }) - - assert compute in project._project_created_on_compute - project.emit_notification.assert_any_call("node.created", node.__json__()) - - -@pytest.mark.asyncio -async def test_add_builtin_node_from_template(controller): - """ - For a local server we send the project path - """ - - compute = MagicMock() - compute.id = "local" - project = Project(controller=controller, name="Test") - project.emit_notification = MagicMock() - template = Template(str(uuid.uuid4()), { - "name": "Builtin-switch", - "template_type": "ethernet_switch", - }, builtin=True) - - controller.template_manager.templates[template.id] = template - template.__json__() - controller._computes["local"] = compute - - response = MagicMock() - response.json = {"console": 2048} - compute.post = AsyncioMagicMock(return_value=response) - - node = await project.add_node_from_template(template.id, x=23, y=12, compute_id="local") - compute.post.assert_any_call('/projects', data={ - "name": project._name, - "project_id": project._id, - "path": project._path - }) - - assert compute in project._project_created_on_compute - project.emit_notification.assert_any_call("node.created", node.__json__()) +# @pytest.mark.asyncio +# async def test_add_node_from_template(controller): +# """ +# For a local server we send the project path +# """ +# +# compute = MagicMock() +# compute.id = "local" +# project = Project(controller=controller, name="Test") +# project.emit_notification = MagicMock() +# template = Template(str(uuid.uuid4()), { +# "compute_id": "local", +# "name": "Test", +# "template_type": "vpcs", +# "builtin": False, +# }) +# controller.template_manager.templates[template.id] = template +# controller._computes["local"] = compute +# +# response = MagicMock() +# response.json = {"console": 2048} +# compute.post = AsyncioMagicMock(return_value=response) +# +# node = await project.add_node_from_template(template.id, x=23, y=12) +# compute.post.assert_any_call('/projects', data={ +# "name": project._name, +# "project_id": project._id, +# "path": project._path +# }) +# +# assert compute in project._project_created_on_compute +# project.emit_notification.assert_any_call("node.created", node.__json__()) +# +# +# @pytest.mark.asyncio +# async def test_add_builtin_node_from_template(controller): +# """ +# For a local server we send the project path +# """ +# +# compute = MagicMock() +# compute.id = "local" +# project = Project(controller=controller, name="Test") +# project.emit_notification = MagicMock() +# template = Template(str(uuid.uuid4()), { +# "name": "Builtin-switch", +# "template_type": "ethernet_switch", +# }, builtin=True) +# +# controller.template_manager.templates[template.id] = template +# template.__json__() +# controller._computes["local"] = compute +# +# response = MagicMock() +# response.json = {"console": 2048} +# compute.post = AsyncioMagicMock(return_value=response) +# +# node = await project.add_node_from_template(template.id, x=23, y=12, compute_id="local") +# compute.post.assert_any_call('/projects', data={ +# "name": project._name, +# "project_id": project._id, +# "path": project._path +# }) +# +# assert compute in project._project_created_on_compute +# project.emit_notification.assert_any_call("node.created", node.__json__()) @pytest.mark.asyncio diff --git a/tests/controller/test_template.py b/tests/controller/test_template.py deleted file mode 100644 index 7c5039c0..00000000 --- a/tests/controller/test_template.py +++ /dev/null @@ -1,89 +0,0 @@ -#!/usr/bin/env python -# -# Copyright (C) 2016 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 . - -import pytest -import pydantic - -from gns3server.controller.template import Template - - -def test_template_json(): - a = Template(None, { - "node_type": "qemu", - "name": "Test", - "default_name_format": "{name}-{0}", - "category": 0, - "symbol": "qemu.svg", - "server": "local", - "platform": "i386" - }) - settings = a.__json__() - assert settings["template_id"] == a.id - assert settings["template_type"] == "qemu" - assert settings["builtin"] == False - - -def test_template_json_with_not_known_category(): - - with pytest.raises(pydantic.ValidationError): - Template(None, { - "node_type": "qemu", - "name": "Test", - "default_name_format": "{name}-{0}", - "category": 'Not known', - "symbol": "qemu.svg", - "server": "local", - "platform": "i386" - }) - - -def test_template_json_with_platform(): - - a = Template(None, { - "node_type": "dynamips", - "name": "Test", - "default_name_format": "{name}-{0}", - "category": 0, - "symbol": "dynamips.svg", - "image": "IOS_image.bin", - "server": "local", - "platform": "c3725" - }) - settings = a.__json__() - assert settings["template_id"] == a.id - assert settings["template_type"] == "dynamips" - assert settings["builtin"] == False - assert settings["platform"] == "c3725" - - -def test_template_fix_linked_base(): - """ - Version of the gui before 2.1 use linked_base and the server - linked_clone - """ - - a = Template(None, { - "node_type": "qemu", - "name": "Test", - "default_name_format": "{name}-{0}", - "category": 0, - "symbol": "qemu.svg", - "server": "local", - "linked_base": True - }) - assert a.settings["linked_clone"] - assert "linked_base" not in a.settings