Model and controller updates for portals

Portal does not have relationship to runs, results, etc. only
dashboards.

DB upgrade_6, dashboard and widget_config alterations

New controller for portal and portal admin

Updates to widget_config controller and dashboard controller

Update openapi spec and add unit test

Restructure test_widget_config_controller
Increase coverage with subtests

Define common headers and http response assertion failure message for
tests

Author: mshriver

backend/ibutsu_server/controllers/admin/portal_controller.py

@@ -0,0 +1,153 @@
+from http import HTTPStatus
+
+import connexion
+from flask import abort
+
+from ibutsu_server.constants import RESPONSE_JSON_REQ
+from ibutsu_server.db.base import session
+from ibutsu_server.db.models import Portal, User
+from ibutsu_server.filters import convert_filter
+from ibutsu_server.util.admin import check_user_is_admin
+from ibutsu_server.util.query import get_offset
+from ibutsu_server.util.uuid import convert_objectid_to_uuid, is_uuid, validate_uuid
+
+
+def admin_add_portal(portal=None, token_info=None, user=None) -> tuple[dict, int]:
+    """Create a portal
+
+    :param body: Portal
+    :type body: dict | bytes
+
+    :rtype: Portal
+    """
+    check_user_is_admin(user)
+    if not connexion.request.is_json:
+        return RESPONSE_JSON_REQ
+    portal = Portal.from_dict(**connexion.request.get_json())
+    # check if portal already exists
+    if portal.id and Portal.query.get(portal.id):
+        return f"Portal id {portal.id} already exist", HTTPStatus.BAD_REQUEST
+    user = User.query.get(user)
+    if user:
+        portal.owner = user
+    session.add(portal)
+    session.commit()
+    return portal.to_dict(), HTTPStatus.CREATED
+
+
+@validate_uuid
+def admin_get_portal(id_, token_info=None, user=None) -> dict:
+    """Get a single portal by ID
+
+    :param id: ID of test portal
+    :type id: str
+
+    :rtype: Portal
+    """
+    check_user_is_admin(user)
+    portal = Portal.query.get(id_)
+    if not portal:
+        portal = Portal.query.filter(Portal.name == id_).first()
+    if not portal:
+        abort(HTTPStatus.NOT_FOUND)
+    return portal.to_dict(with_owner=True)
+
+
+def admin_get_portal_list(
+    filter_=None,
+    owner_id=None,
+    group_id=None,
+    page=1,
+    page_size=25,
+    token_info=None,
+    user=None,
+) -> dict[list[dict], dict]:
+    """Get a list of portals
+
+    :param owner_id: Filter portals by owner ID
+    :type owner_id: str
+    :param group_id: Filter portals by group ID
+    :type group_id: str
+    :param limit: Limit the portals
+    :type limit: int
+    :param offset: Offset the portals
+    :type offset: int
+
+    :rtype: List[Portal]
+    """
+    check_user_is_admin(user)
+    query = Portal.query
+
+    if filter_:
+        for filter_string in filter_:
+            filter_clause = convert_filter(filter_string, Portal)
+            if filter_clause is not None:
+                query = query.filter(filter_clause)
+    if owner_id:
+        query = query.filter(Portal.owner_id == owner_id)
+    if group_id:
+        query = query.filter(Portal.group_id == group_id)
+
+    offset = get_offset(page, page_size)
+    total_items = query.count()
+    total_pages = (total_items // page_size) + (1 if total_items % page_size > 0 else 0)
+    if offset > 9223372036854775807:  # max value of bigint
+        return "The page number is too big.", HTTPStatus.BAD_REQUEST
+    portals = query.offset(offset).limit(page_size).all()
+    return {
+        "portals": [portal.to_dict(with_owner=True) for portal in portals],
+        "pagination": {
+            "page": page,
+            "pageSize": page_size,
+            "totalItems": total_items,
+            "totalPages": total_pages,
+        },
+    }
+
+
+@validate_uuid
+def admin_update_portal(id_, portal=None, body=None, token_info=None, user=None):
+    """Update a portal
+
+    :param id: ID of portal
+    :type id: str
+    :param body: Portal
+    :type body: dict | bytes
+
+    :rtype: Portal
+    """
+    check_user_is_admin(user)
+    if not connexion.request.is_json:
+        return RESPONSE_JSON_REQ
+    if not is_uuid(id_):
+        id_ = convert_objectid_to_uuid(id_)
+    portal = Portal.query.get(id_)
+
+    if not portal:
+        abort(HTTPStatus.NOT_FOUND)
+
+    # Grab the fields from the request
+    portal_dict = connexion.request.get_json()
+
+    # If the "owner" field is set, ignore it
+    portal_dict.pop("owner", None)
+
+    # update the portal info
+    portal.update(portal_dict)
+    session.add(portal)
+    session.commit()
+    return portal.to_dict()
+
+
+@validate_uuid
+def admin_delete_portal(id_, token_info=None, user=None):
+    """Delete a single portal"""
+    check_user_is_admin(user)
+    if not is_uuid(id_):
+        return f"Portal ID {id_} is not in UUID format", HTTPStatus.BAD_REQUEST
+    portal = Portal.query.get(id_)
+    if not portal:
+        abort(HTTPStatus.NOT_FOUND)
+    session.delete(portal)
+    session.commit()
+    return HTTPStatus.OK.phrase, HTTPStatus.OK

backend/ibutsu_server/controllers/dashboard_controller.py

@@ -11,7 +11,7 @@
 from ibutsu_server.util.uuid import validate_uuid
 
 
-def add_dashboard(dashboard=None, token_info=None, user=None):
+def add_dashboard(dashboard=None, token_info=None, user=None) -> tuple[dict, int]:
     """Create a dashboard
 
     :param body: Dashboard
@@ -22,17 +22,30 @@ def add_dashboard(dashboard=None, token_info=None, user=None):
     if not connexion.request.is_json:
         return RESPONSE_JSON_REQ
     dashboard = Dashboard.from_dict(**connexion.request.get_json())
+
+    if dashboard.portal_id and dashboard.project_id:
+        return "Dashboard can only have one of project_id or portal_id", HTTPStatus.BAD_REQUEST
+
+    if not (dashboard.portal_id or dashboard.project_id):
+        return "Dashboard needs either project_id or portal_id", HTTPStatus.BAD_REQUEST
+
     if dashboard.project_id and not project_has_user(dashboard.project_id, user):
         return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
+
+    # TODO utility function to resolve all users with at least one project set
+    # compare to projects assigned to the portal? or portals are open to all projects
+    # otherwise, limit to admin users or new portal_admin permission
+
     if dashboard.user_id and not User.query.get(dashboard.user_id):
         return f"User with ID {dashboard.user_id} doesn't exist", HTTPStatus.BAD_REQUEST
+
     session.add(dashboard)
     session.commit()
     return dashboard.to_dict(), HTTPStatus.CREATED
 
 
 @validate_uuid
-def get_dashboard(id_, token_info=None, user=None):
+def get_dashboard(id_, token_info=None, user=None) -> dict:
     """Get a single dashboard by ID
 
     :param id: ID of test dashboard
@@ -45,16 +58,19 @@ def get_dashboard(id_, token_info=None, user=None):
         return "Dashboard not found", HTTPStatus.NOT_FOUND
     if dashboard and dashboard.project and not project_has_user(dashboard.project, user):
         return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
+    # TODO test against dashboard with only portal set
     return dashboard.to_dict()
 
 
 def get_dashboard_list(
-    filter_=None, project_id=None, page=1, page_size=25, token_info=None, user=None
-):
+    filter_=None, project_id=None, portal_id=None, page=1, page_size=25, token_info=None, user=None
+) -> dict[list[dict], dict]:
     """Get a list of dashboards
 
     :param project_id: Filter dashboards by project ID
     :type project_id: str
+    :param portal_id: Filter dashboards by portal ID
+    :type portal_id: str
     :param user_id: Filter dashboards by user ID
     :type user_id: str
     :param limit: Limit the dashboards
@@ -66,13 +82,24 @@ def get_dashboard_list(
     """
     query = Dashboard.query
     project = None
+    if portal_id is not None and project_id is not None:
+        return "Dashboard list can only have one of project_id or portal_id", HTTPStatus.BAD_REQUEST
+
+    # Project filter injection
     if "project_id" in connexion.request.args:
         project = Project.query.get(connexion.request.args["project_id"])
     if project:
         if not project_has_user(project, user):
             return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
         query = query.filter(Dashboard.project_id == project_id)
 
+    # Portal filter injection
+    if "portal_id" in connexion.request.args:
+        portal = Project.query.get(connexion.request.args["portal_id"])
+    if portal:
+        query = query.filter(Dashboard.portal_id == portal_id)
+
+    # Other filters follow
     if filter_:
         for filter_string in filter_:
             filter_clause = convert_filter(filter_string, Dashboard)
@@ -96,10 +123,10 @@ def get_dashboard_list(
 
 
 @validate_uuid
-def update_dashboard(id_, dashboard=None, token_info=None, user=None):
+def update_dashboard(id_, dashboard=None, token_info=None, user=None) -> dict:
     """Update a dashboard
 
-    :param id: ID of test dashboard
+    :param id: ID of dashboard
     :type id: str
     :param body: Dashboard
     :type body: dict | bytes
@@ -113,19 +140,25 @@ def update_dashboard(id_, dashboard=None, token_info=None, user=None):
         dashboard_dict["metadata"]["project"], user
     ):
         return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
+
+    # TODO user/admin check for portal ref
+
     dashboard = Dashboard.query.get(id_)
     if not dashboard:
         return "Dashboard not found", HTTPStatus.NOT_FOUND
-    if project_has_user(dashboard.project, user):
+    if dashboard.project_id is not None and project_has_user(dashboard.project, user):
         return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
+
+    # TODO user/admin check for portal ref
+
     dashboard.update(connexion.request.get_json())
     session.add(dashboard)
     session.commit()
     return dashboard.to_dict()
 
 
 @validate_uuid
-def delete_dashboard(id_, token_info=None, user=None):
+def delete_dashboard(id_, token_info=None, user=None) -> tuple[str, int]:
     """Deletes a dashboard
 
     :param id: ID of the dashboard to delete
@@ -136,8 +169,9 @@ def delete_dashboard(id_, token_info=None, user=None):
     dashboard = Dashboard.query.get(id_)
     if not dashboard:
         return HTTPStatus.NOT_FOUND.phrase, HTTPStatus.NOT_FOUND
-    if not project_has_user(dashboard.project, user):
+    if dashboard.project_id is not None and not project_has_user(dashboard.project, user):
         return HTTPStatus.FORBIDDEN.phrase, HTTPStatus.FORBIDDEN
+    # TODO user/admin check for portal ref
     widget_configs = WidgetConfig.query.filter(WidgetConfig.dashboard_id == dashboard.id).all()
     for widget_config in widget_configs:
         session.delete(widget_config)