Merge branch '3.0' into image-management-refactoring

# Conflicts:
#	gns3server/api/routes/controller/permissions.py
This commit is contained in:
grossmj committed 2021-09-20 17:06:46 +09:30
commit d8bceaad5d
618 files changed
+4841 -13708

No files matched your search

+14
View File
@@ -181,6 +181,20 @@ async def test_reload_node(app: FastAPI, client: AsyncClient, project: Project,
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_isolate_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node):
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("isolate_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_unisolate_node(app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node):
compute.post = AsyncioMagicMock()
response = await client.post(app.url_path_for("unisolate_node", project_id=project.id, node_id=node.id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_duplicate_node(
app: FastAPI,
client: AsyncClient,
+63 -11
View File
@@ -16,32 +16,74 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import pytest
import uuid
from fastapi import FastAPI, status
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from gns3server.db.repositories.rbac import RbacRepository
from gns3server.controller import Controller
from gns3server.controller.project import Project
pytestmark = pytest.mark.asyncio
class TestPermissionRoutes:
async def test_create_permission(self, app: FastAPI, client: AsyncClient) -> None:
@pytest.fixture()
async def project(self, app: FastAPI, client: AsyncClient, controller: Controller) -> Project:
project_uuid = str(uuid.uuid4())
params = {"name": "test", "project_id": project_uuid}
await client.post(app.url_path_for("create_project"), json=params)
return controller.get_project(project_uuid)
async def test_create_permission(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
new_permission = {
"methods": ["GET"],
"path": "/templates",
"path": f"/projects/{project.id}",
"action": "ALLOW"
}
response = await client.post(app.url_path_for("create_permission"), json=new_permission)
assert response.status_code == status.HTTP_201_CREATED
async def test_get_permission(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
async def test_create_wildcard_permission(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
new_permission = {
"methods": ["POST"],
"path": f"/projects/{project.id}/*",
"action": "ALLOW"
}
response = await client.post(app.url_path_for("create_permission"), json=new_permission)
assert response.status_code == status.HTTP_201_CREATED
async def test_create_permission_not_existing_endpoint(self, app: FastAPI, client: AsyncClient) -> None:
new_permission = {
"methods": ["GET"],
"path": "/projects/invalid",
"action": "ALLOW"
}
response = await client.post(app.url_path_for("create_permission"), json=new_permission)
assert response.status_code == status.HTTP_400_BAD_REQUEST
async def test_create_permission_not_existing_object(self, app: FastAPI, client: AsyncClient) -> None:
new_permission = {
"methods": ["GET"],
"path": f"/projects/{str(uuid.uuid4())}/*",
"action": "ALLOW"
}
response = await client.post(app.url_path_for("create_permission"), json=new_permission)
assert response.status_code == status.HTTP_403_FORBIDDEN
async def test_get_permission(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession, project: Project) -> None:
rbac_repo = RbacRepository(db_session)
permission_in_db = await rbac_repo.get_permission_by_path("/templates")
permission_in_db = await rbac_repo.get_permission_by_path(f"/projects/{project.id}/*")
response = await client.get(app.url_path_for("get_permission", permission_id=permission_in_db.permission_id))
assert response.status_code == status.HTTP_200_OK
assert response.json()["permission_id"] == str(permission_in_db.permission_id)
@@ -50,16 +92,16 @@ class TestPermissionRoutes:
response = await client.get(app.url_path_for("get_permissions"))
assert response.status_code == status.HTTP_200_OK
assert len(response.json()) == 6 # 5 default permissions + 1 custom permission
assert len(response.json()) == 10 # 5 default permissions + 5 custom permissions
async def test_update_permission(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
async def test_update_permission(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession, project: Project) -> None:
rbac_repo = RbacRepository(db_session)
permission_in_db = await rbac_repo.get_permission_by_path("/templates")
permission_in_db = await rbac_repo.get_permission_by_path(f"/projects/{project.id}/*")
update_permission = {
"methods": ["GET"],
"path": "/appliances",
"path": f"/projects/{project.id}/*",
"action": "ALLOW"
}
response = await client.put(
@@ -68,16 +110,26 @@ class TestPermissionRoutes:
)
assert response.status_code == status.HTTP_200_OK
updated_permission_in_db = await rbac_repo.get_permission(permission_in_db.permission_id)
assert updated_permission_in_db.path == "/appliances"
assert updated_permission_in_db.path == f"/projects/{project.id}/*"
async def test_delete_permission(
self,
app: FastAPI,
client: AsyncClient,
db_session: AsyncSession
db_session: AsyncSession,
project: Project,
) -> None:
rbac_repo = RbacRepository(db_session)
permission_in_db = await rbac_repo.get_permission_by_path("/appliances")
permission_in_db = await rbac_repo.get_permission_by_path(f"/projects/{project.id}/*")
response = await client.delete(app.url_path_for("delete_permission", permission_id=permission_in_db.permission_id))
assert response.status_code == status.HTTP_204_NO_CONTENT
async def test_prune_permissions(self, app: FastAPI, client: AsyncClient, db_session: AsyncSession) -> None:
response = await client.post(app.url_path_for("prune_permissions"))
assert response.status_code == status.HTTP_204_NO_CONTENT
rbac_repo = RbacRepository(db_session)
permissions_in_db = await rbac_repo.get_permissions()
assert len(permissions_in_db) == 9 # 5 default permissions + 4 custom permissions
@@ -358,6 +358,7 @@ class TestUserMe:
)
assert response.status_code == status_code
class TestSuperAdmin:
async def test_super_admin_exists(