mirror of
https://github.com/GNS3/gns3-server
synced 2024-11-28 11:18:11 +00:00
Merge pull request #1985 from GNS3/appliances-refactoring
Appliance management refactoring
This commit is contained in:
commit
52d4804e03
@ -18,22 +18,106 @@
|
||||
API routes for appliances.
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends, Response, status
|
||||
from typing import Optional, List
|
||||
from uuid import UUID
|
||||
|
||||
from gns3server import schemas
|
||||
from gns3server.controller import Controller
|
||||
from gns3server.controller.controller_error import (
|
||||
ControllerError,
|
||||
ControllerBadRequestError,
|
||||
ControllerNotFoundError
|
||||
)
|
||||
|
||||
from gns3server.db.repositories.images import ImagesRepository
|
||||
from gns3server.db.repositories.templates import TemplatesRepository
|
||||
from gns3server.db.repositories.rbac import RbacRepository
|
||||
|
||||
from .dependencies.authentication import get_current_active_user
|
||||
from .dependencies.database import get_repository
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("")
|
||||
async def get_appliances(update: Optional[bool] = False, symbol_theme: Optional[str] = "Classic") -> List[dict]:
|
||||
async def get_appliances(
|
||||
update: Optional[bool] = False,
|
||||
symbol_theme: Optional[str] = "Classic"
|
||||
) -> List[schemas.Appliance]:
|
||||
"""
|
||||
Return all appliances known by the controller.
|
||||
"""
|
||||
|
||||
from gns3server.controller import Controller
|
||||
|
||||
controller = Controller.instance()
|
||||
if update:
|
||||
await controller.appliance_manager.download_appliances()
|
||||
controller.appliance_manager.load_appliances(symbol_theme=symbol_theme)
|
||||
return [c.asdict() for c in controller.appliance_manager.appliances.values()]
|
||||
|
||||
|
||||
@router.get("/{appliance_id}")
|
||||
def get_appliance(appliance_id: UUID) -> schemas.Appliance:
|
||||
"""
|
||||
Get an appliance file.
|
||||
"""
|
||||
|
||||
controller = Controller.instance()
|
||||
appliance = controller.appliance_manager.appliances.get(str(appliance_id))
|
||||
if not appliance:
|
||||
raise ControllerNotFoundError(message=f"Could not find appliance '{appliance_id}'")
|
||||
return appliance.asdict()
|
||||
|
||||
|
||||
@router.post("/{appliance_id}/version", status_code=status.HTTP_201_CREATED)
|
||||
def add_appliance_version(appliance_id: UUID, appliance_version: schemas.ApplianceVersion) -> schemas.Appliance:
|
||||
"""
|
||||
Add a version to an appliance
|
||||
"""
|
||||
|
||||
controller = Controller.instance()
|
||||
appliance = controller.appliance_manager.appliances.get(str(appliance_id))
|
||||
if not appliance:
|
||||
raise ControllerNotFoundError(message=f"Could not find appliance '{appliance_id}'")
|
||||
|
||||
if not appliance.versions:
|
||||
raise ControllerBadRequestError(message=f"Appliance '{appliance_id}' do not have versions")
|
||||
|
||||
if not appliance_version.images:
|
||||
raise ControllerBadRequestError(message=f"Version '{appliance_version.name}' must contain images")
|
||||
|
||||
for version in appliance.versions:
|
||||
if version.get("name") == appliance_version.name:
|
||||
raise ControllerError(message=f"Appliance '{appliance_id}' already has version '{appliance_version.name}'")
|
||||
|
||||
appliance.versions.append(appliance_version.dict(exclude_unset=True))
|
||||
return appliance.asdict()
|
||||
|
||||
|
||||
@router.post("/{appliance_id}/install", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def install_appliance(
|
||||
appliance_id: UUID,
|
||||
version: Optional[str] = None,
|
||||
images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
|
||||
templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
|
||||
current_user: schemas.User = Depends(get_current_active_user),
|
||||
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
|
||||
) -> Response:
|
||||
"""
|
||||
Install an appliance.
|
||||
"""
|
||||
|
||||
controller = Controller.instance()
|
||||
await controller.appliance_manager.install_appliance(
|
||||
appliance_id,
|
||||
version,
|
||||
images_repo,
|
||||
templates_repo,
|
||||
rbac_repo,
|
||||
current_user
|
||||
)
|
||||
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||
|
@ -24,14 +24,12 @@ import urllib.parse
|
||||
|
||||
from fastapi import APIRouter, Request, Response, Depends, status
|
||||
from sqlalchemy.orm.exc import MultipleResultsFound
|
||||
from typing import List
|
||||
from typing import List, Optional
|
||||
from gns3server import schemas
|
||||
from pydantic import ValidationError
|
||||
|
||||
from gns3server.utils.images import InvalidImageError, default_images_directory, write_image
|
||||
from gns3server.db.repositories.images import ImagesRepository
|
||||
from gns3server.db.repositories.templates import TemplatesRepository
|
||||
from gns3server.services.templates import TemplatesService
|
||||
from gns3server.db.repositories.rbac import RbacRepository
|
||||
from gns3server.controller import Controller
|
||||
from gns3server.controller.controller_error import (
|
||||
@ -68,7 +66,8 @@ async def upload_image(
|
||||
images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
|
||||
templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
|
||||
current_user: schemas.User = Depends(get_current_active_user),
|
||||
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
|
||||
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
|
||||
install_appliances: Optional[bool] = True
|
||||
) -> schemas.Image:
|
||||
"""
|
||||
Upload an image.
|
||||
@ -92,25 +91,18 @@ async def upload_image(
|
||||
except (OSError, InvalidImageError) as e:
|
||||
raise ControllerError(f"Could not save {image_type} image '{image_path}': {e}")
|
||||
|
||||
try:
|
||||
# attempt to automatically create a template based on image checksum
|
||||
template = await Controller.instance().appliance_manager.install_appliance_from_image(
|
||||
if install_appliances:
|
||||
# attempt to automatically create templates based on image checksum
|
||||
await Controller.instance().appliance_manager.install_appliances_from_image(
|
||||
image_path,
|
||||
image.checksum,
|
||||
images_repo,
|
||||
templates_repo,
|
||||
rbac_repo,
|
||||
current_user,
|
||||
directory
|
||||
)
|
||||
|
||||
if template:
|
||||
template_create = schemas.TemplateCreate(**template)
|
||||
template = await TemplatesService(templates_repo).create_template(template_create)
|
||||
template_id = template.get("template_id")
|
||||
await rbac_repo.add_permission_to_user_with_path(current_user.user_id, f"/templates/{template_id}/*")
|
||||
log.info(f"Template '{template.get('name')}' version {template.get('version')} "
|
||||
f"has been created using image '{image_name}'")
|
||||
|
||||
except (ControllerError, ValidationError, InvalidImageError) as e:
|
||||
log.warning(f"Could not automatically create template using image '{image_path}': {e}")
|
||||
|
||||
return image
|
||||
|
||||
|
||||
@ -150,8 +142,10 @@ async def delete_image(
|
||||
if not image:
|
||||
raise ControllerNotFoundError(f"Image '{image_path}' not found")
|
||||
|
||||
if await images_repo.get_image_templates(image.image_id):
|
||||
raise ControllerError(f"Image '{image_path}' is used by one or more templates")
|
||||
templates = await images_repo.get_image_templates(image.image_id)
|
||||
if templates:
|
||||
template_names = ", ".join([template.name for template in templates])
|
||||
raise ControllerError(f"Image '{image_path}' is used by one or more templates: {template_names}")
|
||||
|
||||
try:
|
||||
os.remove(image.path)
|
||||
|
@ -16,7 +16,6 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import copy
|
||||
import uuid
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@ -24,19 +23,12 @@ log = logging.getLogger(__name__)
|
||||
|
||||
class Appliance:
|
||||
|
||||
def __init__(self, appliance_id, data, builtin=True):
|
||||
def __init__(self, path, data, builtin=True):
|
||||
|
||||
if appliance_id is None:
|
||||
self._id = str(uuid.uuid4())
|
||||
elif isinstance(appliance_id, uuid.UUID):
|
||||
self._id = str(appliance_id)
|
||||
else:
|
||||
self._id = appliance_id
|
||||
self._data = data.copy()
|
||||
self._id = self._data.get("appliance_id")
|
||||
self._path = path
|
||||
self._builtin = builtin
|
||||
if "appliance_id" in self._data:
|
||||
del self._data["appliance_id"]
|
||||
|
||||
if self.status != "broken":
|
||||
log.debug(f'Appliance "{self.name}" [{self._id}] loaded')
|
||||
|
||||
@ -44,6 +36,10 @@ class Appliance:
|
||||
def id(self):
|
||||
return self._id
|
||||
|
||||
@property
|
||||
def path(self):
|
||||
return self._path
|
||||
|
||||
@property
|
||||
def status(self):
|
||||
return self._data["status"]
|
||||
@ -75,6 +71,8 @@ class Appliance:
|
||||
return "iou"
|
||||
elif "dynamips" in self._data:
|
||||
return "dynamips"
|
||||
elif "docker" in self._data:
|
||||
return "docker"
|
||||
else:
|
||||
return "qemu"
|
||||
|
||||
@ -82,6 +80,7 @@ class Appliance:
|
||||
"""
|
||||
Appliance data (a hash)
|
||||
"""
|
||||
|
||||
data = copy.deepcopy(self._data)
|
||||
data["builtin"] = self._builtin
|
||||
return data
|
||||
|
@ -17,22 +17,32 @@
|
||||
|
||||
import os
|
||||
import json
|
||||
import uuid
|
||||
import asyncio
|
||||
import aiofiles
|
||||
|
||||
from typing import Tuple, List
|
||||
from aiohttp.client_exceptions import ClientError
|
||||
|
||||
from uuid import UUID
|
||||
from pydantic import ValidationError
|
||||
|
||||
from .appliance import Appliance
|
||||
from ..config import Config
|
||||
from ..utils.asyncio import locking
|
||||
from ..utils.get_resource import get_resource
|
||||
from ..utils.http_client import HTTPClient
|
||||
from .controller_error import ControllerError
|
||||
from .controller_error import ControllerBadRequestError, ControllerNotFoundError, ControllerError
|
||||
from .appliance_to_template import ApplianceToTemplate
|
||||
from ..utils.images import InvalidImageError, write_image, md5sum
|
||||
from ..utils.asyncio import wait_run_in_executor
|
||||
|
||||
from gns3server import schemas
|
||||
from gns3server.utils.images import default_images_directory
|
||||
from gns3server.db.repositories.images import ImagesRepository
|
||||
from gns3server.db.repositories.templates import TemplatesRepository
|
||||
from gns3server.services.templates import TemplatesService
|
||||
from gns3server.db.repositories.rbac import RbacRepository
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@ -49,7 +59,7 @@ class ApplianceManager:
|
||||
self._appliances_etag = None
|
||||
|
||||
@property
|
||||
def appliances_etag(self):
|
||||
def appliances_etag(self) -> str:
|
||||
"""
|
||||
:returns: ETag for downloaded appliances
|
||||
"""
|
||||
@ -65,14 +75,14 @@ class ApplianceManager:
|
||||
self._appliances_etag = etag
|
||||
|
||||
@property
|
||||
def appliances(self):
|
||||
def appliances(self) -> dict:
|
||||
"""
|
||||
:returns: The dictionary of appliances managed by GNS3
|
||||
"""
|
||||
|
||||
return self._appliances
|
||||
|
||||
def appliances_path(self):
|
||||
def appliances_path(self) -> str:
|
||||
"""
|
||||
Get the image storage directory
|
||||
"""
|
||||
@ -82,18 +92,27 @@ class ApplianceManager:
|
||||
os.makedirs(appliances_path, exist_ok=True)
|
||||
return appliances_path
|
||||
|
||||
def _find_appliance_from_image_checksum(self, image_checksum):
|
||||
def _find_appliances_from_image_checksum(self, image_checksum: str) -> List[Tuple[Appliance, str]]:
|
||||
"""
|
||||
Find an appliance and version that matches an image checksum.
|
||||
Find appliances that matches an image checksum.
|
||||
"""
|
||||
|
||||
appliances = []
|
||||
for appliance in self._appliances.values():
|
||||
if appliance.images:
|
||||
for image in appliance.images:
|
||||
if image.get("md5sum") == image_checksum:
|
||||
return appliance, image.get("version")
|
||||
appliances.append((appliance, image.get("version")))
|
||||
return appliances
|
||||
|
||||
async def _download_image(self, image_dir, image_name, image_type, image_url, images_repo):
|
||||
async def _download_image(
|
||||
self,
|
||||
image_dir: str,
|
||||
image_name: str,
|
||||
image_type: str,
|
||||
image_url: str,
|
||||
images_repo: ImagesRepository
|
||||
) -> None:
|
||||
"""
|
||||
Download an image.
|
||||
"""
|
||||
@ -112,7 +131,13 @@ class ApplianceManager:
|
||||
except asyncio.TimeoutError:
|
||||
raise ControllerError(f"Timeout while downloading '{image_name}' from '{image_url}'")
|
||||
|
||||
async def _find_appliance_version_images(self, appliance, version, images_repo, image_dir):
|
||||
async def _find_appliance_version_images(
|
||||
self,
|
||||
appliance: Appliance,
|
||||
version: dict,
|
||||
images_repo: ImagesRepository,
|
||||
image_dir: str
|
||||
) -> None:
|
||||
"""
|
||||
Find all the images belonging to a specific appliance version.
|
||||
"""
|
||||
@ -145,28 +170,112 @@ class ApplianceManager:
|
||||
else:
|
||||
raise ControllerError(f"Could not find '{appliance_file}'")
|
||||
|
||||
async def install_appliance_from_image(self, image_checksum, images_repo, image_dir):
|
||||
async def _create_template(self, template_data, templates_repo, rbac_repo, current_user):
|
||||
"""
|
||||
Find the image checksum in appliance files
|
||||
Create a new template
|
||||
"""
|
||||
|
||||
try:
|
||||
template_create = schemas.TemplateCreate(**template_data)
|
||||
except ValidationError as e:
|
||||
raise ControllerError(message=f"Could not validate template data: {e}")
|
||||
template = await TemplatesService(templates_repo).create_template(template_create)
|
||||
template_id = template.get("template_id")
|
||||
await rbac_repo.add_permission_to_user_with_path(current_user.user_id, f"/templates/{template_id}/*")
|
||||
log.info(f"Template '{template.get('name')}' has been created")
|
||||
|
||||
async def _appliance_to_template(self, appliance: Appliance, version: str = None) -> dict:
|
||||
"""
|
||||
Get template data from appliance
|
||||
"""
|
||||
|
||||
from . import Controller
|
||||
|
||||
appliance_info = self._find_appliance_from_image_checksum(image_checksum)
|
||||
if appliance_info:
|
||||
appliance, image_version = appliance_info
|
||||
# downloading missing custom symbol for this appliance
|
||||
if appliance.symbol and not appliance.symbol.startswith(":/symbols/"):
|
||||
destination_path = os.path.join(Controller.instance().symbols.symbols_path(), appliance.symbol)
|
||||
if not os.path.exists(destination_path):
|
||||
await self._download_symbol(appliance.symbol, destination_path)
|
||||
return ApplianceToTemplate().new_template(appliance.asdict(), version, "local") # FIXME: "local"
|
||||
|
||||
async def install_appliances_from_image(
|
||||
self,
|
||||
image_path: str,
|
||||
image_checksum: str,
|
||||
images_repo: ImagesRepository,
|
||||
templates_repo: TemplatesRepository,
|
||||
rbac_repo: RbacRepository,
|
||||
current_user: schemas.User,
|
||||
image_dir: str
|
||||
) -> None:
|
||||
"""
|
||||
Install appliances using an image checksum
|
||||
"""
|
||||
|
||||
appliances_info = self._find_appliances_from_image_checksum(image_checksum)
|
||||
for appliance, image_version in appliances_info:
|
||||
try:
|
||||
schemas.Appliance.parse_obj(appliance.asdict())
|
||||
except ValidationError as e:
|
||||
log.warning(message=f"Could not validate appliance '{appliance.id}': {e}")
|
||||
if appliance.versions:
|
||||
for version in appliance.versions:
|
||||
if version.get("name") == image_version:
|
||||
await self._find_appliance_version_images(appliance, version, images_repo, image_dir)
|
||||
# downloading missing custom symbol for this appliance
|
||||
if appliance.symbol and not appliance.symbol.startswith(":/symbols/"):
|
||||
destination_path = os.path.join(Controller.instance().symbols.symbols_path(), appliance.symbol)
|
||||
if not os.path.exists(destination_path):
|
||||
await self._download_symbol(appliance.symbol, destination_path)
|
||||
return ApplianceToTemplate().new_template(appliance.asdict(), version, "local") # FIXME: "local"
|
||||
try:
|
||||
await self._find_appliance_version_images(appliance, version, images_repo, image_dir)
|
||||
template_data = await self._appliance_to_template(appliance, version)
|
||||
await self._create_template(template_data, templates_repo, rbac_repo, current_user)
|
||||
except (ControllerError, InvalidImageError) as e:
|
||||
log.warning(f"Could not automatically create template using image '{image_path}': {e}")
|
||||
|
||||
def load_appliances(self, symbol_theme="Classic"):
|
||||
async def install_appliance(
|
||||
self,
|
||||
appliance_id: UUID,
|
||||
version: str,
|
||||
images_repo: ImagesRepository,
|
||||
templates_repo: TemplatesRepository,
|
||||
rbac_repo: RbacRepository,
|
||||
current_user: schemas.User
|
||||
) -> None:
|
||||
"""
|
||||
Install a new appliance
|
||||
"""
|
||||
|
||||
appliance = self._appliances.get(str(appliance_id))
|
||||
if not appliance:
|
||||
raise ControllerNotFoundError(message=f"Could not find appliance '{appliance_id}'")
|
||||
|
||||
try:
|
||||
schemas.Appliance.parse_obj(appliance.asdict())
|
||||
except ValidationError as e:
|
||||
raise ControllerError(message=f"Could not validate appliance '{appliance_id}': {e}")
|
||||
|
||||
if version:
|
||||
if not appliance.versions:
|
||||
raise ControllerBadRequestError(message=f"Appliance '{appliance_id}' do not have versions")
|
||||
|
||||
image_dir = default_images_directory(appliance.type)
|
||||
for appliance_version_info in appliance.versions:
|
||||
if appliance_version_info.get("name") == version:
|
||||
try:
|
||||
await self._find_appliance_version_images(appliance, appliance_version_info, images_repo, image_dir)
|
||||
except InvalidImageError as e:
|
||||
raise ControllerError(message=f"Image error: {e}")
|
||||
template_data = await self._appliance_to_template(appliance, appliance_version_info)
|
||||
return await self._create_template(template_data, templates_repo, rbac_repo, current_user)
|
||||
|
||||
raise ControllerNotFoundError(message=f"Could not find version '{version}' in appliance '{appliance_id}'")
|
||||
|
||||
else:
|
||||
if appliance.versions:
|
||||
# TODO: install appliance versions based on available images
|
||||
raise ControllerBadRequestError(message=f"Selecting a version is required to install "
|
||||
f"appliance '{appliance_id}'")
|
||||
|
||||
template_data = await self._appliance_to_template(appliance)
|
||||
await self._create_template(template_data, templates_repo, rbac_repo, current_user)
|
||||
|
||||
def load_appliances(self, symbol_theme: str = "Classic") -> None:
|
||||
"""
|
||||
Loads appliance files from disk.
|
||||
"""
|
||||
@ -187,27 +296,23 @@ class ApplianceManager:
|
||||
if not file.endswith(".gns3a") and not file.endswith(".gns3appliance"):
|
||||
continue
|
||||
path = os.path.join(directory, file)
|
||||
# Generate UUID from path to avoid change between reboots
|
||||
appliance_id = uuid.uuid5(
|
||||
uuid.NAMESPACE_X500,
|
||||
path
|
||||
)
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
appliance = Appliance(appliance_id, json.load(f), builtin=builtin)
|
||||
appliance = Appliance(path, json.load(f), builtin=builtin)
|
||||
json_data = appliance.asdict() # Check if loaded without error
|
||||
if appliance.status != "broken":
|
||||
self._appliances[appliance.id] = appliance
|
||||
schemas.Appliance.parse_obj(json_data)
|
||||
self._appliances[appliance.id] = appliance
|
||||
if not appliance.symbol or appliance.symbol.startswith(":/symbols/"):
|
||||
# apply a default symbol if the appliance has none or a default symbol
|
||||
default_symbol = self._get_default_symbol(json_data, symbol_theme)
|
||||
if default_symbol:
|
||||
appliance.symbol = default_symbol
|
||||
except (ValueError, OSError, KeyError) as e:
|
||||
log.warning("Cannot load appliance file '%s': %s", path, str(e))
|
||||
except (ValueError, OSError, KeyError, ValidationError) as e:
|
||||
print(f"Cannot load appliance file '{path}': {e}")
|
||||
continue
|
||||
|
||||
def _get_default_symbol(self, appliance, symbol_theme):
|
||||
def _get_default_symbol(self, appliance: dict, symbol_theme: str) -> str:
|
||||
"""
|
||||
Returns the default symbol for a given appliance.
|
||||
"""
|
||||
@ -223,7 +328,7 @@ class ApplianceManager:
|
||||
return controller.symbols.get_default_symbol("qemu_guest", symbol_theme)
|
||||
return controller.symbols.get_default_symbol(category, symbol_theme)
|
||||
|
||||
async def download_custom_symbols(self):
|
||||
async def download_custom_symbols(self) -> None:
|
||||
"""
|
||||
Download custom appliance symbols from our GitHub registry repository.
|
||||
"""
|
||||
@ -242,7 +347,7 @@ class ApplianceManager:
|
||||
# refresh the symbol cache
|
||||
Controller.instance().symbols.list()
|
||||
|
||||
async def _download_symbol(self, symbol, destination_path):
|
||||
async def _download_symbol(self, symbol: str, destination_path: str) -> None:
|
||||
"""
|
||||
Download a custom appliance symbol from our GitHub registry repository.
|
||||
"""
|
||||
@ -266,7 +371,7 @@ class ApplianceManager:
|
||||
log.warning(f"Could not write appliance symbol '{destination_path}': {e}")
|
||||
|
||||
@locking
|
||||
async def download_appliances(self):
|
||||
async def download_appliances(self) -> None:
|
||||
"""
|
||||
Downloads appliance files from GitHub registry repository.
|
||||
"""
|
||||
|
@ -34,9 +34,11 @@ class ApplianceToTemplate:
|
||||
new_template = {
|
||||
"compute_id": server,
|
||||
"name": appliance_config["name"],
|
||||
"version": version.get("name")
|
||||
}
|
||||
|
||||
if version:
|
||||
new_template["version"] = version.get("name")
|
||||
|
||||
if "usage" in appliance_config:
|
||||
new_template["usage"] = appliance_config["usage"]
|
||||
|
||||
|
@ -57,6 +57,14 @@ class TemplatesRepository(BaseRepository):
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().first()
|
||||
|
||||
async def get_template_by_name_and_version(self, name: str, version: str) -> Union[None, models.Template]:
|
||||
|
||||
query = select(models.Template).\
|
||||
options(selectinload(models.Template.images)).\
|
||||
where(models.Template.name == name, models.Template.version == version)
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().first()
|
||||
|
||||
async def get_templates(self) -> List[models.Template]:
|
||||
|
||||
query = select(models.Template).options(selectinload(models.Template.images))
|
||||
|
@ -24,6 +24,7 @@ from .controller.links import LinkCreate, LinkUpdate, Link
|
||||
from .controller.computes import ComputeCreate, ComputeUpdate, AutoIdlePC, Compute
|
||||
from .controller.templates import TemplateCreate, TemplateUpdate, TemplateUsage, Template
|
||||
from .controller.images import Image, ImageType
|
||||
from .controller.appliances import ApplianceVersion, Appliance
|
||||
from .controller.drawings import Drawing
|
||||
from .controller.gns3vm import GNS3VM
|
||||
from .controller.nodes import NodeCreate, NodeUpdate, NodeDuplicate, NodeCapture, Node
|
||||
|
464
gns3server/schemas/controller/appliances.py
Normal file
464
gns3server/schemas/controller/appliances.py
Normal file
@ -0,0 +1,464 @@
|
||||
#
|
||||
# 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
# Generated from JSON schema using https://github.com/koxudaxi/datamodel-code-generator
|
||||
|
||||
from enum import Enum
|
||||
from typing import List, Optional, Union
|
||||
from uuid import UUID
|
||||
from pydantic import AnyUrl, BaseModel, EmailStr, Field, confloat, conint, constr
|
||||
|
||||
|
||||
class Category(Enum):
|
||||
|
||||
router = 'router'
|
||||
multilayer_switch = 'multilayer_switch'
|
||||
switch = 'switch'
|
||||
firewall = 'firewall'
|
||||
guest = 'guest'
|
||||
|
||||
|
||||
class RegistryVersion(Enum):
|
||||
|
||||
version1 = 1
|
||||
version2 = 2
|
||||
version3 = 3
|
||||
version4 = 4
|
||||
version5 = 5
|
||||
version6 = 6
|
||||
|
||||
|
||||
class Status(Enum):
|
||||
|
||||
stable = 'stable'
|
||||
experimental = 'experimental'
|
||||
broken = 'broken'
|
||||
|
||||
|
||||
class Availability(Enum):
|
||||
|
||||
free = 'free'
|
||||
with_registration = 'with-registration'
|
||||
free_to_try = 'free-to-try'
|
||||
service_contract = 'service-contract'
|
||||
|
||||
|
||||
class ConsoleType(Enum):
|
||||
|
||||
telnet = 'telnet'
|
||||
vnc = 'vnc'
|
||||
http = 'http'
|
||||
https = 'https'
|
||||
none = 'none'
|
||||
|
||||
|
||||
class Docker(BaseModel):
|
||||
|
||||
adapters: int = Field(..., title='Number of ethernet adapters')
|
||||
image: str = Field(..., title='Docker image in the Docker Hub')
|
||||
start_command: Optional[str] = Field(
|
||||
None,
|
||||
title='Command executed when the container start. Empty will use the default',
|
||||
)
|
||||
environment: Optional[str] = Field(None, title='One KEY=VAR environment by line')
|
||||
console_type: Optional[ConsoleType] = Field(
|
||||
None, title='Type of console connection for the administration of the appliance'
|
||||
)
|
||||
console_http_port: Optional[int] = Field(
|
||||
None, description='Internal port in the container of the HTTP server'
|
||||
)
|
||||
console_http_path: Optional[str] = Field(
|
||||
None, description='Path of the web interface'
|
||||
)
|
||||
extra_hosts: Optional[str] = Field(
|
||||
None, description='Hosts which will be written to /etc/hosts into container'
|
||||
)
|
||||
extra_volumes: Optional[List[str]] = Field(
|
||||
None,
|
||||
description='Additional directories to make persistent that are not included in the images VOLUME directive',
|
||||
)
|
||||
|
||||
|
||||
class Iou(BaseModel):
|
||||
|
||||
ethernet_adapters: int = Field(..., title='Number of ethernet adapters')
|
||||
serial_adapters: int = Field(..., title='Number of serial adapters')
|
||||
nvram: int = Field(..., title='Host NVRAM')
|
||||
ram: int = Field(..., title='Host RAM')
|
||||
startup_config: str = Field(..., title='Config loaded at startup')
|
||||
|
||||
|
||||
class Chassis(Enum):
|
||||
|
||||
chassis_1720 = '1720'
|
||||
chassis_1721 = '1721'
|
||||
chassis_1750 = '1750'
|
||||
chassis_1751 = '1751'
|
||||
chassis_1760 = '1760'
|
||||
chassis_2610 = '2610'
|
||||
chassis_2620 = '2620'
|
||||
chassis_2610XM = '2610XM'
|
||||
chassis_2620XM = '2620XM'
|
||||
chassis_2650XM = '2650XM'
|
||||
chassis_2621 = '2621'
|
||||
chassis_2611XM = '2611XM'
|
||||
chassis_2621XM = '2621XM'
|
||||
chassis_2651XM = '2651XM'
|
||||
chassis_3620 = '3620'
|
||||
chassis_3640 = '3640'
|
||||
chassis_3660 = '3660'
|
||||
|
||||
|
||||
class Platform(Enum):
|
||||
|
||||
c1700 = 'c1700'
|
||||
c2600 = 'c2600'
|
||||
c2691 = 'c2691'
|
||||
c3725 = 'c3725'
|
||||
c3745 = 'c3745'
|
||||
c3600 = 'c3600'
|
||||
c7200 = 'c7200'
|
||||
|
||||
|
||||
class Midplane(Enum):
|
||||
|
||||
std = 'std'
|
||||
vxr = 'vxr'
|
||||
|
||||
|
||||
class Npe(Enum):
|
||||
|
||||
npe_100 = 'npe-100'
|
||||
npe_150 = 'npe-150'
|
||||
npe_175 = 'npe-175'
|
||||
npe_200 = 'npe-200'
|
||||
npe_225 = 'npe-225'
|
||||
npe_300 = 'npe-300'
|
||||
npe_400 = 'npe-400'
|
||||
npe_g2 = 'npe-g2'
|
||||
|
||||
|
||||
class AdapterType(Enum):
|
||||
|
||||
e1000 = 'e1000'
|
||||
e1000_82544gc = 'e1000-82544gc'
|
||||
e1000_82545em = 'e1000-82545em'
|
||||
e1000e = 'e1000e'
|
||||
i82550 = 'i82550'
|
||||
i82551 = 'i82551'
|
||||
i82557a = 'i82557a'
|
||||
i82557b = 'i82557b'
|
||||
i82557c = 'i82557c'
|
||||
i82558a = 'i82558a'
|
||||
i82558b = 'i82558b'
|
||||
i82559a = 'i82559a'
|
||||
i82559b = 'i82559b'
|
||||
i82559c = 'i82559c'
|
||||
i82559er = 'i82559er'
|
||||
i82562 = 'i82562'
|
||||
i82801 = 'i82801'
|
||||
ne2k_pci = 'ne2k_pci'
|
||||
pcnet = 'pcnet'
|
||||
rocker = 'rocker'
|
||||
rtl8139 = 'rtl8139'
|
||||
virtio = 'virtio'
|
||||
virtio_net_pci = 'virtio-net-pci'
|
||||
vmxnet3 = 'vmxnet3'
|
||||
|
||||
|
||||
class DiskInterface(Enum):
|
||||
|
||||
ide = 'ide'
|
||||
sata = 'sata'
|
||||
nvme = 'nvme'
|
||||
scsi = 'scsi'
|
||||
sd = 'sd'
|
||||
mtd = 'mtd'
|
||||
floppy = 'floppy'
|
||||
pflash = 'pflash'
|
||||
virtio = 'virtio'
|
||||
none = 'none'
|
||||
|
||||
|
||||
class Arch(Enum):
|
||||
|
||||
aarch64 = 'aarch64'
|
||||
alpha = 'alpha'
|
||||
arm = 'arm'
|
||||
cris = 'cris'
|
||||
i386 = 'i386'
|
||||
lm32 = 'lm32'
|
||||
m68k = 'm68k'
|
||||
microblaze = 'microblaze'
|
||||
microblazeel = 'microblazeel'
|
||||
mips = 'mips'
|
||||
mips64 = 'mips64'
|
||||
mips64el = 'mips64el'
|
||||
mipsel = 'mipsel'
|
||||
moxie = 'moxie'
|
||||
or32 = 'or32'
|
||||
ppc = 'ppc'
|
||||
ppc64 = 'ppc64'
|
||||
ppcemb = 'ppcemb'
|
||||
s390x = 's390x'
|
||||
sh4 = 'sh4'
|
||||
sh4eb = 'sh4eb'
|
||||
sparc = 'sparc'
|
||||
sparc64 = 'sparc64'
|
||||
tricore = 'tricore'
|
||||
unicore32 = 'unicore32'
|
||||
x86_64 = 'x86_64'
|
||||
xtensa = 'xtensa'
|
||||
xtensaeb = 'xtensaeb'
|
||||
|
||||
|
||||
class ConsoleType1(Enum):
|
||||
|
||||
telnet = 'telnet'
|
||||
vnc = 'vnc'
|
||||
spice = 'spice'
|
||||
spice_agent = 'spice+agent'
|
||||
none = 'none'
|
||||
|
||||
|
||||
class BootPriority(Enum):
|
||||
|
||||
c = 'c'
|
||||
d = 'd'
|
||||
n = 'n'
|
||||
cn = 'cn'
|
||||
cd = 'cd'
|
||||
dn = 'dn'
|
||||
dc = 'dc'
|
||||
nc = 'nc'
|
||||
nd = 'nd'
|
||||
|
||||
|
||||
class Kvm(Enum):
|
||||
|
||||
require = 'require'
|
||||
allow = 'allow'
|
||||
disable = 'disable'
|
||||
|
||||
|
||||
class ProcessPriority(Enum):
|
||||
|
||||
realtime = 'realtime'
|
||||
very_high = 'very high'
|
||||
high = 'high'
|
||||
normal = 'normal'
|
||||
low = 'low'
|
||||
very_low = 'very low'
|
||||
null = 'null'
|
||||
|
||||
|
||||
class Qemu(BaseModel):
|
||||
|
||||
adapter_type: AdapterType = Field(..., title='Type of network adapter')
|
||||
adapters: int = Field(..., title='Number of adapters')
|
||||
ram: int = Field(..., title='Ram allocated to the appliance (MB)')
|
||||
cpus: Optional[int] = Field(None, title='Number of Virtual CPU')
|
||||
hda_disk_interface: Optional[DiskInterface] = Field(
|
||||
None, title='Disk interface for the installed hda_disk_image'
|
||||
)
|
||||
hdb_disk_interface: Optional[DiskInterface] = Field(
|
||||
None, title='Disk interface for the installed hdb_disk_image'
|
||||
)
|
||||
hdc_disk_interface: Optional[DiskInterface] = Field(
|
||||
None, title='Disk interface for the installed hdc_disk_image'
|
||||
)
|
||||
hdd_disk_interface: Optional[DiskInterface] = Field(
|
||||
None, title='Disk interface for the installed hdd_disk_image'
|
||||
)
|
||||
arch: Arch = Field(..., title='Architecture emulated')
|
||||
console_type: ConsoleType1 = Field(
|
||||
..., title='Type of console connection for the administration of the appliance'
|
||||
)
|
||||
boot_priority: Optional[BootPriority] = Field(
|
||||
None,
|
||||
title='Disk boot priority. Refer to -boot option in qemu manual for more details.',
|
||||
)
|
||||
kernel_command_line: Optional[str] = Field(
|
||||
None, title='Command line parameters send to the kernel'
|
||||
)
|
||||
kvm: Kvm = Field(..., title='KVM requirements')
|
||||
options: Optional[str] = Field(
|
||||
None, title='Optional additional qemu command line options'
|
||||
)
|
||||
cpu_throttling: Optional[confloat(ge=0.0, le=100.0)] = Field(
|
||||
None, title='Throttle the CPU'
|
||||
)
|
||||
process_priority: Optional[ProcessPriority] = Field(
|
||||
None, title='Process priority for QEMU'
|
||||
)
|
||||
|
||||
|
||||
class Compression(Enum):
|
||||
|
||||
bzip2 = 'bzip2'
|
||||
gzip = 'gzip'
|
||||
lzma = 'lzma'
|
||||
xz = 'xz'
|
||||
rar = 'rar'
|
||||
zip = 'zip'
|
||||
field_7z = '7z'
|
||||
|
||||
|
||||
class ApplianceImage(BaseModel):
|
||||
|
||||
filename: str = Field(..., title='Filename')
|
||||
version: str = Field(..., title='Version of the file')
|
||||
md5sum: str = Field(..., title='md5sum of the file', regex='^[a-f0-9]{32}$')
|
||||
filesize: int = Field(..., title='File size in bytes')
|
||||
download_url: Optional[Union[AnyUrl, constr(max_length=0)]] = Field(
|
||||
None, title='Download url where you can download the appliance from a browser'
|
||||
)
|
||||
direct_download_url: Optional[Union[AnyUrl, constr(max_length=0)]] = Field(
|
||||
None,
|
||||
title='Optional. Non authenticated url to the image file where you can download the image.',
|
||||
)
|
||||
compression: Optional[Compression] = Field(
|
||||
None, title='Optional, compression type of direct download url image.'
|
||||
)
|
||||
|
||||
|
||||
class ApplianceVersionImages(BaseModel):
|
||||
|
||||
kernel_image: Optional[str] = Field(None, title='Kernel image')
|
||||
initrd: Optional[str] = Field(None, title='Initrd disk image')
|
||||
image: Optional[str] = Field(None, title='OS image')
|
||||
bios_image: Optional[str] = Field(None, title='Bios image')
|
||||
hda_disk_image: Optional[str] = Field(None, title='Hda disk image')
|
||||
hdb_disk_image: Optional[str] = Field(None, title='Hdc disk image')
|
||||
hdc_disk_image: Optional[str] = Field(None, title='Hdd disk image')
|
||||
hdd_disk_image: Optional[str] = Field(None, title='Hdd diskimage')
|
||||
cdrom_image: Optional[str] = Field(None, title='cdrom image')
|
||||
|
||||
|
||||
class ApplianceVersion(BaseModel):
|
||||
|
||||
name: str = Field(..., title='Name of the version')
|
||||
idlepc: Optional[str] = Field(None, regex='^0x[0-9a-f]{8}')
|
||||
images: Optional[ApplianceVersionImages] = Field(None, title='Images used for this version')
|
||||
|
||||
|
||||
class DynamipsSlot(Enum):
|
||||
|
||||
C7200_IO_2FE = 'C7200-IO-2FE'
|
||||
C7200_IO_FE = 'C7200-IO-FE'
|
||||
C7200_IO_GE_E = 'C7200-IO-GE-E'
|
||||
NM_16ESW = 'NM-16ESW'
|
||||
NM_1E = 'NM-1E'
|
||||
NM_1FE_TX = 'NM-1FE-TX'
|
||||
NM_4E = 'NM-4E'
|
||||
NM_4T = 'NM-4T'
|
||||
PA_2FE_TX = 'PA-2FE-TX'
|
||||
PA_4E = 'PA-4E'
|
||||
PA_4T_ = 'PA-4T+'
|
||||
PA_8E = 'PA-8E'
|
||||
PA_8T = 'PA-8T'
|
||||
PA_A1 = 'PA-A1'
|
||||
PA_FE_TX = 'PA-FE-TX'
|
||||
PA_GE = 'PA-GE'
|
||||
PA_POS_OC3 = 'PA-POS-OC3'
|
||||
C2600_MB_2FE = 'C2600-MB-2FE'
|
||||
C2600_MB_1E = 'C2600-MB-1E'
|
||||
C1700_MB_1FE = 'C1700-MB-1FE'
|
||||
C2600_MB_2E = 'C2600-MB-2E'
|
||||
C2600_MB_1FE = 'C2600-MB-1FE'
|
||||
C1700_MB_WIC1 = 'C1700-MB-WIC1'
|
||||
GT96100_FE = 'GT96100-FE'
|
||||
Leopard_2FE = 'Leopard-2FE'
|
||||
_ = ''
|
||||
|
||||
|
||||
class DynamipsWic(Enum):
|
||||
|
||||
WIC_1ENET = 'WIC-1ENET'
|
||||
WIC_1T = 'WIC-1T'
|
||||
WIC_2T = 'WIC-2T'
|
||||
|
||||
|
||||
class Dynamips(BaseModel):
|
||||
|
||||
chassis: Optional[Chassis] = Field(None, title='Chassis type')
|
||||
platform: Platform = Field(..., title='Platform type')
|
||||
ram: conint(ge=1) = Field(..., title='Amount of ram')
|
||||
nvram: conint(ge=1) = Field(..., title='Amount of nvram')
|
||||
startup_config: Optional[str] = Field(None, title='Config loaded at startup')
|
||||
wic0: Optional[DynamipsWic] = None
|
||||
wic1: Optional[DynamipsWic] = None
|
||||
wic2: Optional[DynamipsWic] = None
|
||||
slot0: Optional[DynamipsSlot] = None
|
||||
slot1: Optional[DynamipsSlot] = None
|
||||
slot2: Optional[DynamipsSlot] = None
|
||||
slot3: Optional[DynamipsSlot] = None
|
||||
slot4: Optional[DynamipsSlot] = None
|
||||
slot5: Optional[DynamipsSlot] = None
|
||||
slot6: Optional[DynamipsSlot] = None
|
||||
midplane: Optional[Midplane] = None
|
||||
npe: Optional[Npe] = None
|
||||
|
||||
|
||||
class Appliance(BaseModel):
|
||||
|
||||
appliance_id: UUID = Field(..., title='Appliance ID')
|
||||
name: str = Field(..., title='Appliance name')
|
||||
category: Category = Field(..., title='Category of the appliance')
|
||||
description: str = Field(
|
||||
..., title='Description of the appliance. Could be a marketing description'
|
||||
)
|
||||
vendor_name: str = Field(..., title='Name of the vendor')
|
||||
vendor_url: Union[AnyUrl, constr(max_length=0)] = Field(..., title='Website of the vendor')
|
||||
documentation_url: Optional[Union[AnyUrl, constr(max_length=0)]] = Field(
|
||||
None,
|
||||
title='An optional documentation for using the appliance on vendor website',
|
||||
)
|
||||
product_name: str = Field(..., title='Product name')
|
||||
product_url: Optional[Union[AnyUrl, constr(max_length=0)]] = Field(
|
||||
None, title='An optional product url on vendor website'
|
||||
)
|
||||
registry_version: RegistryVersion = Field(
|
||||
..., title='Version of the registry compatible with this appliance'
|
||||
)
|
||||
status: Status = Field(..., title='Document if the appliance is working or not')
|
||||
availability: Optional[Availability] = Field(
|
||||
None,
|
||||
title='About image availability: can be downloaded directly; download requires a free registration; paid but a trial version (time or feature limited) is available; not available publicly',
|
||||
)
|
||||
maintainer: str = Field(..., title='Maintainer name')
|
||||
maintainer_email: Union[EmailStr, constr(max_length=0)] = Field(..., title='Maintainer email')
|
||||
usage: Optional[str] = Field(None, title='How to use the appliance')
|
||||
symbol: Optional[str] = Field(None, title='An optional symbol for the appliance')
|
||||
first_port_name: Optional[str] = Field(
|
||||
None, title='Optional name of the first networking port example: eth0'
|
||||
)
|
||||
port_name_format: Optional[str] = Field(
|
||||
None, title='Optional formating of the networking port example: eth{0}'
|
||||
)
|
||||
port_segment_size: Optional[int] = Field(
|
||||
None,
|
||||
title='Optional port segment size. A port segment is a block of port. For example Ethernet0/0 Ethernet0/1 is the module 0 with a port segment size of 2',
|
||||
)
|
||||
linked_clone: Optional[bool] = Field(
|
||||
None, title="False if you don't want to use a single image for all nodes"
|
||||
)
|
||||
docker: Optional[Docker] = Field(None, title='Docker specific options')
|
||||
iou: Optional[Iou] = Field(None, title='IOU specific options')
|
||||
dynamips: Optional[Dynamips] = Field(None, title='Dynamips specific options')
|
||||
qemu: Optional[Qemu] = Field(None, title='Qemu specific options')
|
||||
images: Optional[List[ApplianceImage]] = Field(None, title='Images for this appliance')
|
||||
versions: Optional[List[ApplianceVersion]] = Field(None, title='Versions of the appliance')
|
@ -25,8 +25,8 @@ from typing import List
|
||||
from gns3server import schemas
|
||||
import gns3server.db.models as models
|
||||
from gns3server.db.repositories.templates import TemplatesRepository
|
||||
from gns3server.controller import Controller
|
||||
from gns3server.controller.controller_error import (
|
||||
ControllerError,
|
||||
ControllerBadRequestError,
|
||||
ControllerNotFoundError,
|
||||
ControllerForbiddenError,
|
||||
@ -137,6 +137,7 @@ class TemplatesService:
|
||||
def __init__(self, templates_repo: TemplatesRepository):
|
||||
|
||||
self._templates_repo = templates_repo
|
||||
from gns3server.controller import Controller
|
||||
self._controller = Controller.instance()
|
||||
|
||||
def get_builtin_template(self, template_id: UUID) -> dict:
|
||||
@ -195,6 +196,13 @@ class TemplatesService:
|
||||
|
||||
async def create_template(self, template_create: schemas.TemplateCreate) -> dict:
|
||||
|
||||
if await self._templates_repo.get_template_by_name_and_version(template_create.name, template_create.version):
|
||||
if template_create.version:
|
||||
raise ControllerError(f"A template with name '{template_create.name}' and "
|
||||
f"version {template_create.version} already exists")
|
||||
else:
|
||||
raise ControllerError(f"A template with name '{template_create.name}' already exists")
|
||||
|
||||
try:
|
||||
# get the default template settings
|
||||
template_settings = jsonable_encoder(template_create, exclude_unset=True)
|
||||
|
@ -1,6 +1,6 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# Copyright (C) 2020 GNS3 Technologies Inc.
|
||||
# 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
|
||||
@ -15,7 +15,9 @@
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import os
|
||||
import pytest
|
||||
import shutil
|
||||
|
||||
from fastapi import FastAPI, status
|
||||
from httpx import AsyncClient
|
||||
@ -23,8 +25,69 @@ from httpx import AsyncClient
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
async def test_appliances_list(app: FastAPI, client: AsyncClient) -> None:
|
||||
class TestApplianceRoutes:
|
||||
|
||||
response = await client.get(app.url_path_for("get_appliances"))
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) > 0
|
||||
async def test_appliances_list(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
response = await client.get(app.url_path_for("get_appliances"))
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) > 0
|
||||
|
||||
async def test_get_appliance(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
appliance_id = "3bf492b6-5717-4257-9bfd-b34617c6f133" # Cisco IOSv appliance
|
||||
response = await client.get(app.url_path_for("get_appliance", appliance_id=appliance_id))
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["appliance_id"] == appliance_id
|
||||
|
||||
async def test_docker_appliance_install(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
appliance_id = "fc520ae2-a4e5-48c3-9a13-516bb2e94668" # Alpine Linux appliance
|
||||
response = await client.post(app.url_path_for("install_appliance", appliance_id=appliance_id))
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
async def test_docker_appliance_install_with_version(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
appliance_id = "fc520ae2-a4e5-48c3-9a13-516bb2e94668" # Alpine Linux appliance
|
||||
params = {"version": "123"}
|
||||
response = await client.post(app.url_path_for("install_appliance", appliance_id=appliance_id), params=params)
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
async def test_qemu_appliance_install_with_version(self, app: FastAPI, client: AsyncClient, images_dir: str) -> None:
|
||||
|
||||
shutil.copy("tests/resources/empty8G.qcow2", os.path.join(images_dir, "QEMU", "empty8G.qcow2"))
|
||||
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
|
||||
params = {"version": "8G"}
|
||||
response = await client.post(app.url_path_for("install_appliance", appliance_id=appliance_id), params=params)
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
async def test_qemu_appliance_install_without_version(self, app: FastAPI, client: AsyncClient, images_dir: str) -> None:
|
||||
|
||||
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
|
||||
response = await client.post(app.url_path_for("install_appliance", appliance_id=appliance_id))
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
async def test_add_version_appliance(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
|
||||
new_version = {
|
||||
"name": "99G",
|
||||
"images": {
|
||||
"hda_disk_image": "empty99G.qcow2"
|
||||
}
|
||||
}
|
||||
response = await client.post(app.url_path_for("add_appliance_version", appliance_id=appliance_id), json=new_version)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
assert new_version in response.json()["versions"]
|
||||
|
||||
async def test_add_existing_version_appliance(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance
|
||||
new_version = {
|
||||
"name": "8G",
|
||||
"images": {
|
||||
"hda_disk_image": "empty8G.qcow2"
|
||||
}
|
||||
}
|
||||
response = await client.post(app.url_path_for("add_appliance_version", appliance_id=appliance_id), json=new_version)
|
||||
assert response.status_code == status.HTTP_409_CONFLICT
|
||||
|
@ -23,7 +23,9 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from fastapi import FastAPI, status
|
||||
from httpx import AsyncClient
|
||||
|
||||
from gns3server.controller import Controller
|
||||
from gns3server.db.repositories.images import ImagesRepository
|
||||
from gns3server.db.repositories.templates import TemplatesRepository
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
@ -256,3 +258,27 @@ class TestImageRoutes:
|
||||
images_repo = ImagesRepository(db_session)
|
||||
images_in_db = await images_repo.get_images()
|
||||
assert len(images_in_db) == 0
|
||||
|
||||
async def test_image_upload_create_appliance(
|
||||
self, app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession,
|
||||
controller: Controller
|
||||
) -> None:
|
||||
|
||||
controller.appliance_manager.load_appliances() # make sure appliances are loaded
|
||||
image_path = "tests/resources/empty30G.qcow2"
|
||||
image_name = os.path.basename(image_path)
|
||||
with open(image_path, "rb") as f:
|
||||
image_data = f.read()
|
||||
response = await client.post(
|
||||
app.url_path_for("upload_image", image_path=image_name),
|
||||
params={"image_type": "qemu"},
|
||||
content=image_data)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
templates_repo = TemplatesRepository(db_session)
|
||||
templates = await templates_repo.get_templates()
|
||||
assert len(templates) == 1
|
||||
assert templates[0].name == "Empty VM"
|
||||
assert templates[0].version == "30G"
|
||||
|
@ -62,6 +62,7 @@ class TestTemplateRoutes:
|
||||
template_id = str(uuid.uuid4())
|
||||
params = {"template_id": template_id,
|
||||
"name": "VPCS_TEST",
|
||||
"version": "1.0",
|
||||
"compute_id": "local",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
@ -72,9 +73,25 @@ class TestTemplateRoutes:
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["template_id"] == template_id
|
||||
|
||||
async def test_template_create_same_name_and_version(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
controller: Controller
|
||||
) -> None:
|
||||
|
||||
params = {"name": "VPCS_TEST",
|
||||
"version": "1.0",
|
||||
"compute_id": "local",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
response = await client.post(app.url_path_for("create_template"), json=params)
|
||||
assert response.status_code == status.HTTP_409_CONFLICT
|
||||
|
||||
async def test_template_create_wrong_type(self, app: FastAPI, client: AsyncClient, controller: Controller) -> None:
|
||||
|
||||
params = {"name": "VPCS_TEST",
|
||||
"version": "2.0",
|
||||
"compute_id": "local",
|
||||
"template_type": "invalid_template_type"}
|
||||
|
||||
@ -86,6 +103,7 @@ class TestTemplateRoutes:
|
||||
template_id = str(uuid.uuid4())
|
||||
params = {"template_id": template_id,
|
||||
"name": "VPCS_TEST",
|
||||
"version": "3.0",
|
||||
"compute_id": "local",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
@ -107,6 +125,7 @@ class TestTemplateRoutes:
|
||||
template_id = str(uuid.uuid4())
|
||||
params = {"template_id": template_id,
|
||||
"name": "VPCS_TEST",
|
||||
"version": "4.0",
|
||||
"compute_id": "local",
|
||||
"template_type": "vpcs"}
|
||||
|
||||
@ -426,7 +445,7 @@ class TestDynamipsTemplate:
|
||||
|
||||
async def test_c3600_dynamips_template_create_wrong_chassis(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
params = {"name": "Cisco c3600 template",
|
||||
params = {"name": "Cisco c3600 template with wrong chassis",
|
||||
"platform": "c3600",
|
||||
"chassis": "3650",
|
||||
"compute_id": "local",
|
||||
@ -530,7 +549,7 @@ class TestDynamipsTemplate:
|
||||
|
||||
async def test_c2600_dynamips_template_create_wrong_chassis(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
params = {"name": "Cisco c2600 template",
|
||||
params = {"name": "Cisco c2600 template with wrong chassis",
|
||||
"platform": "c2600",
|
||||
"chassis": "2660XM",
|
||||
"compute_id": "local",
|
||||
@ -589,7 +608,7 @@ class TestDynamipsTemplate:
|
||||
|
||||
async def test_c1700_dynamips_template_create_wrong_chassis(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
params = {"name": "Cisco c1700 template",
|
||||
params = {"name": "Cisco c1700 template with wrong chassis",
|
||||
"platform": "c1700",
|
||||
"chassis": "1770",
|
||||
"compute_id": "local",
|
||||
@ -1200,6 +1219,7 @@ class TestImageAssociationWithTemplate:
|
||||
) -> None:
|
||||
|
||||
params = {"name": "Qemu template",
|
||||
"version": "1.0",
|
||||
"compute_id": "local",
|
||||
"platform": "i386",
|
||||
"hda_disk_image": "subdir/image.qcow2",
|
||||
@ -1224,7 +1244,7 @@ class TestImageAssociationWithTemplate:
|
||||
|
||||
async def test_template_create_with_non_existing_image(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
params = {"name": "Qemu template",
|
||||
params = {"name": "Qemu template with non existing image",
|
||||
"compute_id": "local",
|
||||
"platform": "i386",
|
||||
"hda_disk_image": "unkown_image.qcow2",
|
||||
|
@ -385,14 +385,13 @@ def test_appliances(controller, config, tmpdir):
|
||||
for appliance in controller.appliance_manager.appliances.values():
|
||||
assert appliance.asdict()["status"] != "broken"
|
||||
assert "Alpine Linux" in [c.asdict()["name"] for c in controller.appliance_manager.appliances.values()]
|
||||
assert "My Appliance" in [c.asdict()["name"] for c in controller.appliance_manager.appliances.values()]
|
||||
assert "My Appliance" not in [c.asdict()["name"] for c in controller.appliance_manager.appliances.values()]
|
||||
|
||||
for c in controller.appliance_manager.appliances.values():
|
||||
j = c.asdict()
|
||||
if j["name"] == "Alpine Linux":
|
||||
assert j["builtin"]
|
||||
elif j["name"] == "My Appliance":
|
||||
assert not j["builtin"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_autoidlepc(controller):
|
||||
|
BIN
tests/resources/empty30G.qcow2
Normal file
BIN
tests/resources/empty30G.qcow2
Normal file
Binary file not shown.
Loading…
Reference in New Issue
Block a user