mirror of https://github.com/GNS3/gns3-server
parent
956b9056c1
commit
8810249d36
@ -0,0 +1,184 @@
|
||||
#!/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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
"""
|
||||
API routes for user groups.
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, Depends, status
|
||||
from uuid import UUID
|
||||
from typing import List
|
||||
|
||||
from gns3server import schemas
|
||||
from gns3server.controller.controller_error import (
|
||||
ControllerBadRequestError,
|
||||
ControllerNotFoundError,
|
||||
ControllerForbiddenError,
|
||||
)
|
||||
|
||||
from gns3server.db.repositories.users import UsersRepository
|
||||
from .dependencies.database import get_repository
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("", response_model=List[schemas.UserGroup])
|
||||
async def get_user_groups(
|
||||
users_repo: UsersRepository = Depends(get_repository(UsersRepository))
|
||||
) -> List[schemas.UserGroup]:
|
||||
"""
|
||||
Get all user groups.
|
||||
"""
|
||||
|
||||
return await users_repo.get_user_groups()
|
||||
|
||||
|
||||
@router.post(
|
||||
"",
|
||||
response_model=schemas.UserGroup,
|
||||
status_code=status.HTTP_201_CREATED
|
||||
)
|
||||
async def create_user_group(
|
||||
user_group_create: schemas.UserGroupCreate,
|
||||
users_repo: UsersRepository = Depends(get_repository(UsersRepository))
|
||||
) -> schemas.UserGroup:
|
||||
"""
|
||||
Create a new user group.
|
||||
"""
|
||||
|
||||
if await users_repo.get_user_group_by_name(user_group_create.name):
|
||||
raise ControllerBadRequestError(f"User group '{user_group_create.name}' already exists")
|
||||
|
||||
return await users_repo.create_user_group(user_group_create)
|
||||
|
||||
|
||||
@router.get("/{user_group_id}", response_model=schemas.UserGroup)
|
||||
async def get_user_group(
|
||||
user_group_id: UUID,
|
||||
users_repo: UsersRepository = Depends(get_repository(UsersRepository)),
|
||||
) -> schemas.UserGroup:
|
||||
"""
|
||||
Get an user group.
|
||||
"""
|
||||
|
||||
user_group = await users_repo.get_user_group(user_group_id)
|
||||
if not user_group:
|
||||
raise ControllerNotFoundError(f"User group '{user_group_id}' not found")
|
||||
return user_group
|
||||
|
||||
|
||||
@router.put("/{user_group_id}", response_model=schemas.UserGroup)
|
||||
async def update_user_group(
|
||||
user_group_id: UUID,
|
||||
user_group_update: schemas.UserGroupUpdate,
|
||||
users_repo: UsersRepository = Depends(get_repository(UsersRepository))
|
||||
) -> schemas.UserGroup:
|
||||
"""
|
||||
Update an user group.
|
||||
"""
|
||||
user_group = await users_repo.get_user_group(user_group_id)
|
||||
if not user_group:
|
||||
raise ControllerNotFoundError(f"User group '{user_group_id}' not found")
|
||||
|
||||
if not user_group.is_updatable:
|
||||
raise ControllerForbiddenError(f"User group '{user_group_id}' cannot be updated")
|
||||
|
||||
return await users_repo.update_user_group(user_group_id, user_group_update)
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/{user_group_id}",
|
||||
status_code=status.HTTP_204_NO_CONTENT
|
||||
)
|
||||
async def delete_user_group(
|
||||
user_group_id: UUID,
|
||||
users_repo: UsersRepository = Depends(get_repository(UsersRepository)),
|
||||
) -> None:
|
||||
"""
|
||||
Delete an user group
|
||||
"""
|
||||
|
||||
user_group = await users_repo.get_user_group(user_group_id)
|
||||
if not user_group:
|
||||
raise ControllerNotFoundError(f"User group '{user_group_id}' not found")
|
||||
|
||||
if not user_group.is_updatable:
|
||||
raise ControllerForbiddenError(f"User group '{user_group_id}' cannot be deleted")
|
||||
|
||||
success = await users_repo.delete_user_group(user_group_id)
|
||||
if not success:
|
||||
raise ControllerNotFoundError(f"User group '{user_group_id}' could not be deleted")
|
||||
|
||||
|
||||
@router.get("/{user_group_id}/members", response_model=List[schemas.User])
|
||||
async def get_user_group_members(
|
||||
user_group_id: UUID,
|
||||
users_repo: UsersRepository = Depends(get_repository(UsersRepository))
|
||||
) -> List[schemas.User]:
|
||||
"""
|
||||
Get all user group members.
|
||||
"""
|
||||
|
||||
return await users_repo.get_user_group_members(user_group_id)
|
||||
|
||||
|
||||
@router.put(
|
||||
"/{user_group_id}/members/{user_id}",
|
||||
status_code=status.HTTP_204_NO_CONTENT
|
||||
)
|
||||
async def add_member_to_group(
|
||||
user_group_id: UUID,
|
||||
user_id: UUID,
|
||||
users_repo: UsersRepository = Depends(get_repository(UsersRepository))
|
||||
) -> None:
|
||||
"""
|
||||
Add member to an user group.
|
||||
"""
|
||||
|
||||
user = await users_repo.get_user(user_id)
|
||||
if not user:
|
||||
raise ControllerNotFoundError(f"User '{user_id}' not found")
|
||||
|
||||
user_group = await users_repo.add_member_to_user_group(user_group_id, user)
|
||||
if not user_group:
|
||||
raise ControllerNotFoundError(f"User group '{user_group_id}' not found")
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/{user_group_id}/members/{user_id}",
|
||||
status_code=status.HTTP_204_NO_CONTENT
|
||||
)
|
||||
async def remove_member_from_group(
|
||||
user_group_id: UUID,
|
||||
user_id: UUID,
|
||||
users_repo: UsersRepository = Depends(get_repository(UsersRepository)),
|
||||
) -> None:
|
||||
"""
|
||||
Remove member from an user group.
|
||||
"""
|
||||
|
||||
user = await users_repo.get_user(user_id)
|
||||
if not user:
|
||||
raise ControllerNotFoundError(f"User '{user_id}' not found")
|
||||
|
||||
user_group = await users_repo.remove_member_from_user_group(user_group_id, user)
|
||||
if not user_group:
|
||||
raise ControllerNotFoundError(f"User group '{user_group_id}' not found")
|
@ -0,0 +1,165 @@
|
||||
#!/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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
import pytest
|
||||
|
||||
from fastapi import FastAPI, status
|
||||
from httpx import AsyncClient
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from gns3server.db.repositories.users import UsersRepository
|
||||
from gns3server.schemas.controller.users import User
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
class TestGroupRoutes:
|
||||
|
||||
async def test_create_group(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
new_group = {"name": "group1"}
|
||||
response = await client.post(app.url_path_for("create_user_group"), json=new_group)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
|
||||
async def test_get_group(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
|
||||
|
||||
user_repo = UsersRepository(db_session)
|
||||
group_in_db = await user_repo.get_user_group_by_name("group1")
|
||||
response = await client.get(app.url_path_for("get_user_group", user_group_id=group_in_db.user_group_id))
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["user_group_id"] == str(group_in_db.user_group_id)
|
||||
|
||||
async def test_list_groups(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
response = await client.get(app.url_path_for("get_user_groups"))
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) == 4 # 3 default groups + group1
|
||||
|
||||
async def test_update_group(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
|
||||
|
||||
user_repo = UsersRepository(db_session)
|
||||
group_in_db = await user_repo.get_user_group_by_name("group1")
|
||||
|
||||
update_group = {"name": "group42"}
|
||||
response = await client.put(
|
||||
app.url_path_for("update_user_group", user_group_id=group_in_db.user_group_id),
|
||||
json=update_group
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
updated_group_in_db = await user_repo.get_user_group(group_in_db.user_group_id)
|
||||
assert updated_group_in_db.name == "group42"
|
||||
|
||||
async def test_cannot_update_admin_group(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession
|
||||
) -> None:
|
||||
|
||||
user_repo = UsersRepository(db_session)
|
||||
group_in_db = await user_repo.get_user_group_by_name("Administrators")
|
||||
update_group = {"name": "Hackers"}
|
||||
response = await client.put(
|
||||
app.url_path_for("update_user_group", user_group_id=group_in_db.user_group_id),
|
||||
json=update_group
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
async def test_delete_group(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession
|
||||
) -> None:
|
||||
|
||||
user_repo = UsersRepository(db_session)
|
||||
group_in_db = await user_repo.get_user_group_by_name("group42")
|
||||
response = await client.delete(app.url_path_for("delete_user_group", user_group_id=group_in_db.user_group_id))
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
async def test_cannot_delete_admin_group(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession
|
||||
) -> None:
|
||||
|
||||
user_repo = UsersRepository(db_session)
|
||||
group_in_db = await user_repo.get_user_group_by_name("Administrators")
|
||||
response = await client.delete(app.url_path_for("delete_user_group", user_group_id=group_in_db.user_group_id))
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
async def test_add_member_to_group(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
test_user: User,
|
||||
db_session: AsyncSession
|
||||
) -> None:
|
||||
|
||||
user_repo = UsersRepository(db_session)
|
||||
group_in_db = await user_repo.get_user_group_by_name("Users")
|
||||
response = await client.put(
|
||||
app.url_path_for(
|
||||
"add_member_to_group",
|
||||
user_group_id=group_in_db.user_group_id,
|
||||
user_id=str(test_user.user_id)
|
||||
)
|
||||
)
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
members = await user_repo.get_user_group_members(group_in_db.user_group_id)
|
||||
assert len(members) == 1
|
||||
assert members[0].username == test_user.username
|
||||
|
||||
async def test_get_user_group_members(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
db_session: AsyncSession
|
||||
) -> None:
|
||||
|
||||
user_repo = UsersRepository(db_session)
|
||||
group_in_db = await user_repo.get_user_group_by_name("Users")
|
||||
response = await client.get(
|
||||
app.url_path_for(
|
||||
"get_user_group_members",
|
||||
user_group_id=group_in_db.user_group_id)
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert len(response.json()) == 1
|
||||
|
||||
async def test_remove_member_from_group(
|
||||
self,
|
||||
app: FastAPI,
|
||||
client: AsyncClient,
|
||||
test_user: User,
|
||||
db_session: AsyncSession
|
||||
) -> None:
|
||||
|
||||
user_repo = UsersRepository(db_session)
|
||||
group_in_db = await user_repo.get_user_group_by_name("Users")
|
||||
|
||||
response = await client.delete(
|
||||
app.url_path_for(
|
||||
"remove_member_from_group",
|
||||
user_group_id=group_in_db.user_group_id,
|
||||
user_id=str(test_user.user_id)
|
||||
),
|
||||
)
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
members = await user_repo.get_user_group_members(group_in_db.user_group_id)
|
||||
assert len(members) == 0
|
Loading…
Reference in new issue