mirror of https://github.com/GNS3/gns3-server
github-actions
9 months ago
commit
91ca94c798
@ -0,0 +1,228 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# Copyright (C) 2023 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/>.
|
||||
|
||||
"""
|
||||
API routes for resource pools.
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, Depends, status
|
||||
from uuid import UUID
|
||||
from typing import List
|
||||
|
||||
from gns3server import schemas
|
||||
from gns3server.controller.controller_error import (
|
||||
ControllerError,
|
||||
ControllerBadRequestError,
|
||||
ControllerNotFoundError
|
||||
)
|
||||
|
||||
from gns3server.controller import Controller
|
||||
from gns3server.db.repositories.rbac import RbacRepository
|
||||
from gns3server.db.repositories.pools import ResourcePoolsRepository
|
||||
|
||||
from .dependencies.rbac import has_privilege
|
||||
from .dependencies.database import get_repository
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get(
|
||||
"",
|
||||
response_model=List[schemas.ResourcePool],
|
||||
dependencies=[Depends(has_privilege("Pool.Audit"))]
|
||||
)
|
||||
async def get_resource_pools(
|
||||
pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository))
|
||||
) -> List[schemas.ResourcePool]:
|
||||
"""
|
||||
Get all resource pools.
|
||||
|
||||
Required privilege: Pool.Audit
|
||||
"""
|
||||
|
||||
return await pools_repo.get_resource_pools()
|
||||
|
||||
|
||||
@router.post(
|
||||
"",
|
||||
response_model=schemas.ResourcePool,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
dependencies=[Depends(has_privilege("Pool.Allocate"))]
|
||||
)
|
||||
async def create_resource_pool(
|
||||
resource_pool_create: schemas.ResourcePoolCreate,
|
||||
pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository))
|
||||
) -> schemas.ResourcePool:
|
||||
"""
|
||||
Create a new resource pool
|
||||
|
||||
Required privilege: Pool.Allocate
|
||||
"""
|
||||
|
||||
if await pools_repo.get_resource_pool_by_name(resource_pool_create.name):
|
||||
raise ControllerBadRequestError(f"Resource pool '{resource_pool_create.name}' already exists")
|
||||
|
||||
return await pools_repo.create_resource_pool(resource_pool_create)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{resource_pool_id}",
|
||||
response_model=schemas.ResourcePool,
|
||||
dependencies=[Depends(has_privilege("Pool.Audit"))]
|
||||
)
|
||||
async def get_resource_pool(
|
||||
resource_pool_id: UUID,
|
||||
pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository))
|
||||
) -> schemas.ResourcePool:
|
||||
"""
|
||||
Get a resource pool.
|
||||
|
||||
Required privilege: Pool.Audit
|
||||
"""
|
||||
|
||||
resource_pool = await pools_repo.get_resource_pool(resource_pool_id)
|
||||
if not resource_pool:
|
||||
raise ControllerNotFoundError(f"Resource pool '{resource_pool_id}' not found")
|
||||
return resource_pool
|
||||
|
||||
|
||||
@router.put(
|
||||
"/{resource_pool_id}",
|
||||
response_model=schemas.ResourcePool,
|
||||
dependencies=[Depends(has_privilege("Pool.Modify"))]
|
||||
)
|
||||
async def update_resource_pool(
|
||||
resource_pool_id: UUID,
|
||||
resource_pool_update: schemas.ResourcePoolUpdate,
|
||||
pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository))
|
||||
) -> schemas.ResourcePool:
|
||||
"""
|
||||
Update a resource pool.
|
||||
|
||||
Required privilege: Pool.Modify
|
||||
"""
|
||||
|
||||
resource_pool = await pools_repo.get_resource_pool(resource_pool_id)
|
||||
if not resource_pool:
|
||||
raise ControllerNotFoundError(f"Resource pool '{resource_pool_id}' not found")
|
||||
|
||||
return await pools_repo.update_resource_pool(resource_pool_id, resource_pool_update)
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/{resource_pool_id}",
|
||||
status_code=status.HTTP_204_NO_CONTENT,
|
||||
dependencies=[Depends(has_privilege("Pool.Allocate"))]
|
||||
)
|
||||
async def delete_resource_pool(
|
||||
resource_pool_id: UUID,
|
||||
pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository)),
|
||||
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
|
||||
) -> None:
|
||||
"""
|
||||
Delete a resource pool.
|
||||
|
||||
Required privilege: Pool.Allocate
|
||||
"""
|
||||
|
||||
resource_pool = await pools_repo.get_resource_pool(resource_pool_id)
|
||||
if not resource_pool:
|
||||
raise ControllerNotFoundError(f"Resource pool '{resource_pool_id}' not found")
|
||||
|
||||
success = await pools_repo.delete_resource_pool(resource_pool_id)
|
||||
if not success:
|
||||
raise ControllerError(f"Resource pool '{resource_pool_id}' could not be deleted")
|
||||
await rbac_repo.delete_all_ace_starting_with_path(f"/pools/{resource_pool_id}")
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{resource_pool_id}/resources",
|
||||
response_model=List[schemas.Resource],
|
||||
dependencies=[Depends(has_privilege("Pool.Audit"))]
|
||||
)
|
||||
async def get_pool_resources(
|
||||
resource_pool_id: UUID,
|
||||
pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository)),
|
||||
) -> List[schemas.Resource]:
|
||||
"""
|
||||
Get all resource in a pool.
|
||||
|
||||
Required privilege: Pool.Audit
|
||||
"""
|
||||
|
||||
return await pools_repo.get_pool_resources(resource_pool_id)
|
||||
|
||||
|
||||
@router.put(
|
||||
"/{resource_pool_id}/resources/{resource_id}",
|
||||
status_code=status.HTTP_204_NO_CONTENT,
|
||||
dependencies=[Depends(has_privilege("Pool.Modify"))]
|
||||
)
|
||||
async def add_resource_to_pool(
|
||||
resource_pool_id: UUID,
|
||||
resource_id: UUID,
|
||||
pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository)),
|
||||
) -> None:
|
||||
"""
|
||||
Add resource to a resource pool.
|
||||
|
||||
Required privilege: Pool.Modify
|
||||
"""
|
||||
|
||||
resource_pool = await pools_repo.get_resource_pool(resource_pool_id)
|
||||
if not resource_pool:
|
||||
raise ControllerNotFoundError(f"Resource pool '{resource_pool_id}' not found")
|
||||
|
||||
resources = await pools_repo.get_pool_resources(resource_pool_id)
|
||||
for resource in resources:
|
||||
if resource.resource_id == resource_id:
|
||||
raise ControllerBadRequestError(f"Resource '{resource_id}' is already in '{resource_pool.name}'")
|
||||
|
||||
# we only support projects in resource pools for now
|
||||
project = Controller.instance().get_project(str(resource_id))
|
||||
resource_create = schemas.ResourceCreate(resource_id=resource_id, resource_type="project", name=project.name)
|
||||
resource = await pools_repo.create_resource(resource_create)
|
||||
await pools_repo.add_resource_to_pool(resource_pool_id, resource)
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/{resource_pool_id}/resources/{resource_id}",
|
||||
status_code=status.HTTP_204_NO_CONTENT,
|
||||
dependencies=[Depends(has_privilege("Pool.Modify"))]
|
||||
)
|
||||
async def remove_resource_from_pool(
|
||||
resource_pool_id: UUID,
|
||||
resource_id: UUID,
|
||||
pools_repo: ResourcePoolsRepository = Depends(get_repository(ResourcePoolsRepository)),
|
||||
) -> None:
|
||||
"""
|
||||
Remove resource from a resource pool.
|
||||
|
||||
Required privilege: Pool.Modify
|
||||
"""
|
||||
|
||||
resource = await pools_repo.get_resource(resource_id)
|
||||
if not resource:
|
||||
raise ControllerNotFoundError(f"Resource '{resource_id}' not found")
|
||||
|
||||
resource_pool = await pools_repo.remove_resource_from_pool(resource_pool_id, resource)
|
||||
if not resource_pool:
|
||||
raise ControllerNotFoundError(f"Resource pool '{resource_pool_id}' not found")
|
||||
@ -0,0 +1,43 @@
|
||||
#
|
||||
# Software Name : GNS3 server
|
||||
# Version: 3
|
||||
# SPDX-FileCopyrightText: Copyright (c) 2023 Orange Business Services
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
#
|
||||
# This software is distributed under the GPL-3.0 or any later version,
|
||||
# the text of which is available at https://www.gnu.org/licenses/gpl-3.0.txt
|
||||
# or see the "LICENSE" file for more details.
|
||||
#
|
||||
# Author: Sylvain MATHIEU
|
||||
#
|
||||
|
||||
"""
|
||||
API route for privileges
|
||||
"""
|
||||
|
||||
from typing import List
|
||||
from gns3server.db.repositories.rbac import RbacRepository
|
||||
from .dependencies.database import get_repository
|
||||
from fastapi import APIRouter, Depends
|
||||
import logging
|
||||
|
||||
from gns3server import schemas
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get(
|
||||
"",
|
||||
response_model=List[schemas.Privilege],
|
||||
)
|
||||
async def get_privileges(
|
||||
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository))
|
||||
) -> List[schemas.Privilege]:
|
||||
"""
|
||||
Get all privileges.
|
||||
|
||||
Required privilege: None
|
||||
"""
|
||||
|
||||
return await rbac_repo.get_privileges()
|
||||
@ -0,0 +1,206 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# Copyright (C) 2023 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/>.
|
||||
|
||||
from uuid import UUID
|
||||
from typing import Optional, List, Union
|
||||
from sqlalchemy import select, update, delete
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from .base import BaseRepository
|
||||
|
||||
import gns3server.db.models as models
|
||||
from gns3server import schemas
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ResourcePoolsRepository(BaseRepository):
|
||||
|
||||
def __init__(self, db_session: AsyncSession) -> None:
|
||||
|
||||
super().__init__(db_session)
|
||||
|
||||
async def get_resource(self, resource_id: UUID) -> Optional[models.Resource]:
|
||||
"""
|
||||
Get a resource by its ID.
|
||||
"""
|
||||
|
||||
query = select(models.Resource).where(models.Resource.resource_id == resource_id)
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().first()
|
||||
|
||||
async def get_resources(self) -> List[models.Resource]:
|
||||
"""
|
||||
Get all resources.
|
||||
"""
|
||||
|
||||
query = select(models.Resource)
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().all()
|
||||
|
||||
async def create_resource(self, resource: schemas.ResourceCreate) -> models.Resource:
|
||||
"""
|
||||
Create a new resource.
|
||||
"""
|
||||
|
||||
db_resource = models.Resource(
|
||||
resource_id=resource.resource_id,
|
||||
resource_type=resource.resource_type,
|
||||
name=resource.name
|
||||
)
|
||||
self._db_session.add(db_resource)
|
||||
await self._db_session.commit()
|
||||
await self._db_session.refresh(db_resource)
|
||||
return db_resource
|
||||
|
||||
async def delete_resource(self, resource_id: UUID) -> bool:
|
||||
"""
|
||||
Delete a resource.
|
||||
"""
|
||||
|
||||
query = delete(models.Resource).where(models.Resource.resource_id == resource_id)
|
||||
result = await self._db_session.execute(query)
|
||||
await self._db_session.commit()
|
||||
return result.rowcount > 0
|
||||
|
||||
async def get_resource_pool(self, resource_pool_id: UUID) -> Optional[models.ResourcePool]:
|
||||
"""
|
||||
Get a resource pool by its ID.
|
||||
"""
|
||||
|
||||
query = select(models.ResourcePool).where(models.ResourcePool.resource_pool_id == resource_pool_id)
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().first()
|
||||
|
||||
async def get_resource_pool_by_name(self, name: str) -> Optional[models.ResourcePool]:
|
||||
"""
|
||||
Get a resource pool by its name.
|
||||
"""
|
||||
|
||||
query = select(models.ResourcePool).where(models.ResourcePool.name == name)
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().first()
|
||||
|
||||
async def get_resource_pools(self) -> List[models.ResourcePool]:
|
||||
"""
|
||||
Get all resource pools.
|
||||
"""
|
||||
|
||||
query = select(models.ResourcePool)
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().all()
|
||||
|
||||
async def create_resource_pool(self, resource_pool: schemas.ResourcePoolCreate) -> models.ResourcePool:
|
||||
"""
|
||||
Create a new resource pool.
|
||||
"""
|
||||
|
||||
db_resource_pool = models.ResourcePool(name=resource_pool.name)
|
||||
self._db_session.add(db_resource_pool)
|
||||
await self._db_session.commit()
|
||||
await self._db_session.refresh(db_resource_pool)
|
||||
return db_resource_pool
|
||||
|
||||
async def update_resource_pool(
|
||||
self,
|
||||
resource_pool_id: UUID,
|
||||
resource_pool_update: schemas.ResourcePoolUpdate
|
||||
) -> Optional[models.ResourcePool]:
|
||||
"""
|
||||
Update a resource pool.
|
||||
"""
|
||||
|
||||
update_values = resource_pool_update.model_dump(exclude_unset=True)
|
||||
query = update(models.ResourcePool).\
|
||||
where(models.ResourcePool.resource_pool_id == resource_pool_id).\
|
||||
values(update_values)
|
||||
|
||||
await self._db_session.execute(query)
|
||||
await self._db_session.commit()
|
||||
resource_pool_db = await self.get_resource_pool(resource_pool_id)
|
||||
if resource_pool_db:
|
||||
await self._db_session.refresh(resource_pool_db) # force refresh of updated_at value
|
||||
return resource_pool_db
|
||||
|
||||
async def delete_resource_pool(self, resource_pool_id: UUID) -> bool:
|
||||
"""
|
||||
Delete a resource pool.
|
||||
"""
|
||||
|
||||
query = delete(models.ResourcePool).where(models.ResourcePool.resource_pool_id == resource_pool_id)
|
||||
result = await self._db_session.execute(query)
|
||||
await self._db_session.commit()
|
||||
return result.rowcount > 0
|
||||
|
||||
async def add_resource_to_pool(
|
||||
self,
|
||||
resource_pool_id: UUID,
|
||||
resource: models.Resource
|
||||
) -> Union[None, models.ResourcePool]:
|
||||
"""
|
||||
Add a resource to a resource pool.
|
||||
"""
|
||||
|
||||
query = select(models.ResourcePool).\
|
||||
options(selectinload(models.ResourcePool.resources)).\
|
||||
where(models.ResourcePool.resource_pool_id == resource_pool_id)
|
||||
result = await self._db_session.execute(query)
|
||||
resource_pool_db = result.scalars().first()
|
||||
if not resource_pool_db:
|
||||
return None
|
||||
|
||||
resource_pool_db.resources.append(resource)
|
||||
await self._db_session.commit()
|
||||
await self._db_session.refresh(resource_pool_db)
|
||||
return resource_pool_db
|
||||
|
||||
async def remove_resource_from_pool(
|
||||
self,
|
||||
resource_pool_id: UUID,
|
||||
resource: models.Resource
|
||||
) -> Union[None, models.ResourcePool]:
|
||||
"""
|
||||
Remove a resource from a resource pool.
|
||||
"""
|
||||
|
||||
query = select(models.ResourcePool).\
|
||||
options(selectinload(models.ResourcePool.resources)).\
|
||||
where(models.ResourcePool.resource_pool_id == resource_pool_id)
|
||||
result = await self._db_session.execute(query)
|
||||
resource_pool_db = result.scalars().first()
|
||||
if not resource_pool_db:
|
||||
return None
|
||||
|
||||
resource_pool_db.resources.remove(resource)
|
||||
await self._db_session.commit()
|
||||
await self._db_session.refresh(resource_pool_db)
|
||||
return resource_pool_db
|
||||
|
||||
async def get_pool_resources(self, resource_pool_id: UUID) -> List[models.Resource]:
|
||||
"""
|
||||
Get all resources from a resource pool.
|
||||
"""
|
||||
|
||||
query = select(models.Resource).\
|
||||
join(models.Resource.resource_pools).\
|
||||
filter(models.ResourcePool.resource_pool_id == resource_pool_id)
|
||||
|
||||
result = await self._db_session.execute(query)
|
||||
return result.scalars().all()
|
||||
@ -0,0 +1,81 @@
|
||||
#
|
||||
# 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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
from typing import Optional
|
||||
from pydantic import ConfigDict, BaseModel, Field
|
||||
from uuid import UUID
|
||||
from enum import Enum
|
||||
|
||||
from .base import DateTimeModelMixin
|
||||
|
||||
|
||||
class ResourceType(str, Enum):
|
||||
|
||||
project = "project"
|
||||
|
||||
|
||||
class ResourceBase(BaseModel):
|
||||
"""
|
||||
Common resource properties.
|
||||
"""
|
||||
|
||||
resource_id: UUID
|
||||
resource_type: ResourceType = Field(..., description="Type of the resource")
|
||||
name: Optional[str] = None
|
||||
model_config = ConfigDict(use_enum_values=True)
|
||||
|
||||
|
||||
class ResourceCreate(ResourceBase):
|
||||
"""
|
||||
Properties to create a resource.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class Resource(DateTimeModelMixin, ResourceBase):
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
|
||||
class ResourcePoolBase(BaseModel):
|
||||
"""
|
||||
Common resource pool properties.
|
||||
"""
|
||||
|
||||
name: str
|
||||
|
||||
|
||||
class ResourcePoolCreate(ResourcePoolBase):
|
||||
"""
|
||||
Properties to create a resource pool.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class ResourcePoolUpdate(ResourcePoolBase):
|
||||
"""
|
||||
Properties to update a resource pool.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class ResourcePool(DateTimeModelMixin, ResourcePoolBase):
|
||||
|
||||
resource_pool_id: UUID
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
@ -0,0 +1,183 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# Copyright (C) 2023 GNS3 Technologies Inc.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import uuid
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
|
||||
from fastapi import FastAPI, status
|
||||
from httpx import AsyncClient
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from gns3server.db.repositories.pools import ResourcePoolsRepository
|
||||
from gns3server.controller import Controller
|
||||
from gns3server.controller.project import Project
|
||||
from gns3server.schemas.controller.pools import ResourceCreate, ResourcePoolCreate
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
class TestPoolRoutes:
|
||||
|
||||
async def test_resource_pool(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
new_group = {"name": "pool1"}
|
||||
response = await client.post(app.url_path_for("create_resource_pool"), json=new_group)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
async def test_get_resource_pool(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
|
||||
|
||||
pools_repo = ResourcePoolsRepository(db_session)
|
||||
pool_in_db = await pools_repo.get_resource_pool_by_name("pool1")
|
||||
response = await client.get(app.url_path_for("get_resource_pool", resource_pool_id=pool_in_db.resource_pool_id))
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["resource_pool_id"] == str(pool_in_db.resource_pool_id)
|
||||
|
||||
async def test_list_resource_pools(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
response = await client.get(app.url_path_for("get_resource_pools"))
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) == 1
|
||||
|
||||
async def test_update_resource_pool(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
|
||||
|
||||
pools_repo = ResourcePoolsRepository(db_session)
|
||||
pool_in_db = await pools_repo.get_resource_pool_by_name("pool1")
|
||||
|
||||
update_pool = {"name": "pool42"}
|
||||
response = await client.put(
|
||||
app.url_path_for("update_resource_pool", resource_pool_id=pool_in_db.resource_pool_id),
|
||||
json=update_pool
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
updated_pool_in_db = await pools_repo.get_resource_pool(pool_in_db.resource_pool_id)
|
||||
assert updated_pool_in_db.name == "pool42"
|
||||
|
||||
async def test_resource_group(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession
|
||||
) -> None:
|
||||
|
||||
pools_repo = ResourcePoolsRepository(db_session)
|
||||
pool_in_db = await pools_repo.get_resource_pool_by_name("pool42")
|
||||
response = await client.delete(app.url_path_for("delete_resource_pool", resource_pool_id=pool_in_db.resource_pool_id))
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
|
||||
class TestResourcesPoolRoutes:
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def project(self, app: FastAPI, client: AsyncClient, controller: Controller) -> Project:
|
||||
project_id = str(uuid.uuid4())
|
||||
params = {"name": "test", "project_id": project_id}
|
||||
await client.post(app.url_path_for("create_project"), json=params)
|
||||
return controller.get_project(project_id)
|
||||
|
||||
async def test_add_resource_to_pool(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession,
|
||||
project: Project
|
||||
) -> None:
|
||||
|
||||
pools_repo = ResourcePoolsRepository(db_session)
|
||||
new_resource_pool = ResourcePoolCreate(
|
||||
name="pool1",
|
||||
)
|
||||
pool_in_db = await pools_repo.create_resource_pool(new_resource_pool)
|
||||
response = await client.put(
|
||||
app.url_path_for(
|
||||
"add_resource_to_pool",
|
||||
resource_pool_id=pool_in_db.resource_pool_id,
|
||||
resource_id=str(project.id)
|
||||
)
|
||||
)
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
resources = await pools_repo.get_pool_resources(pool_in_db.resource_pool_id)
|
||||
assert len(resources) == 1
|
||||
assert str(resources[0].resource_id) == project.id
|
||||
|
||||
async def test_add_to_resource_already_in_resource_pool(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession,
|
||||
project: Project
|
||||
) -> None:
|
||||
|
||||
pools_repo = ResourcePoolsRepository(db_session)
|
||||
pool_in_db = await pools_repo.get_resource_pool_by_name("pool1")
|
||||
resource_create = ResourceCreate(resource_id=project.id, resource_type="project")
|
||||
resource = await pools_repo.create_resource(resource_create)
|
||||
await pools_repo.add_resource_to_pool(pool_in_db.resource_pool_id, resource)
|
||||
|
||||
response = await client.put(
|
||||
app.url_path_for(
|
||||
"add_resource_to_pool",
|
||||
resource_pool_id=pool_in_db.resource_pool_id,
|
||||
resource_id=str(resource.resource_id)
|
||||
)
|
||||
)
|
||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||
|
||||
async def test_get_pool_resources(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession
|
||||
) -> None:
|
||||
|
||||
pools_repo = ResourcePoolsRepository(db_session)
|
||||
pool_in_db = await pools_repo.get_resource_pool_by_name("pool1")
|
||||
response = await client.get(
|
||||
app.url_path_for(
|
||||
"get_pool_resources",
|
||||
resource_pool_id=pool_in_db.resource_pool_id)
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) == 2
|
||||
|
||||
async def test_remove_resource_from_pool(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession,
|
||||
project: Project
|
||||
) -> None:
|
||||
|
||||
pools_repo = ResourcePoolsRepository(db_session)
|
||||
pool_in_db = await pools_repo.get_resource_pool_by_name("pool1")
|
||||
resource_create = ResourceCreate(resource_id=project.id, resource_type="project")
|
||||
resource = await pools_repo.create_resource(resource_create)
|
||||
await pools_repo.add_resource_to_pool(pool_in_db.resource_pool_id, resource)
|
||||
|
||||
resources = await pools_repo.get_pool_resources(pool_in_db.resource_pool_id)
|
||||
assert len(resources) == 3
|
||||
|
||||
response = await client.delete(
|
||||
app.url_path_for(
|
||||
"remove_resource_from_pool",
|
||||
resource_pool_id=pool_in_db.resource_pool_id,
|
||||
resource_id=str(project.id)
|
||||
),
|
||||
)
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
resources = await pools_repo.get_pool_resources(pool_in_db.resource_pool_id)
|
||||
assert len(resources) == 2
|
||||
@ -0,0 +1,25 @@
|
||||
#
|
||||
# Software Name : GNS3 server
|
||||
# Version: 3
|
||||
# SPDX-FileCopyrightText: Copyright (c) 2023 Orange Business Services
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
#
|
||||
# This software is distributed under the GPL-3.0 or any later version,
|
||||
# the text of which is available at https://www.gnu.org/licenses/gpl-3.0.txt
|
||||
# or see the "LICENSE" file for more details.
|
||||
#
|
||||
# Author: Sylvain MATHIEU
|
||||
#
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI, status
|
||||
from httpx import AsyncClient
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
class TestPrivilegesRoute:
|
||||
|
||||
async def test_get_privileges(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
response = await client.get(app.url_path_for("get_privileges"))
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
Loading…
Reference in new issue