Create App with Role Owners

api/operations/create_app.py

@@ -11,6 +11,7 @@
 from api.operations.create_group import CreateGroup, GroupDict
 from api.operations.modify_group_type import ModifyGroupType
 from api.operations.modify_group_users import ModifyGroupUsers
+from api.operations.modify_role_groups import ModifyRoleGroups
 from api.views.schemas import AuditLogSchema, EventType
 
 
@@ -25,6 +26,7 @@ def __init__(
         app: App | AppDict,
         tags: list[str] = [],
         owner_id: Optional[str] = None,
+        owner_role_ids: Optional[list[str]] = None,
         additional_app_groups: Optional[list[AppGroup] | list[GroupDict]] = None,
         current_user_id: Optional[str] = None
     ):
@@ -48,6 +50,15 @@ def __init__(
             .filter(OktaUser.deleted_at.is_(None))
             .filter(OktaUser.id == owner_id).first(), 'id', None)
         )
+        self.owner_role_ids = None
+        if owner_role_ids is not None:
+            self.owner_roles = (
+                RoleGroup.query.filter(RoleGroup.id.in_(owner_role_ids))
+                .filter(RoleGroup.deleted_at.is_(None))
+                .all()
+            )
+            self.owner_role_ids = [role.id for role in self.owner_roles]
+
         self.app_group_prefix = f"{AppGroup.APP_GROUP_NAME_PREFIX}{self.app.name}{AppGroup.APP_NAME_GROUP_NAME_SEPARATOR}"
         self.owner_group_name = f"{self.app_group_prefix}{AppGroup.APP_OWNERS_GROUP_NAME_SUFFIX}"
         self.additional_app_groups = []
@@ -149,6 +160,15 @@ def execute(self) -> App:
                 owners_to_add=[self.owner_id]
             ).execute()
 
+        if self.owner_role_ids is not None:
+            for role_id in self.owner_role_ids:
+                ModifyRoleGroups(
+                    role_group=role_id,
+                    current_user_id=self.current_user_id,
+                    groups_to_add=[owner_app_group.id],
+                    owner_groups_to_add=[owner_app_group.id]
+                ).execute()
+
         # Find other app groups with the same app name prefix and update them
         other_existing_app_groups = (
             db.session.query(with_polymorphic(OktaGroup, [AppGroup, RoleGroup]))

api/swagger.json

@@ -225,6 +225,14 @@
           "minLength": 1,
           "type": "string"
         },
+        "initial_owner_role_ids": {
+          "items": {
+            "maxLength": 20,
+            "minLength": 20,
+            "type": "string"
+          },
+          "type": "array"
+        },
         "is_managed": {
           "readOnly": true,
           "type": "boolean"

api/views/resources/app.py

@@ -7,14 +7,7 @@
 from api.apispec import FlaskApiSpecDecorators
 from api.authorization import AuthorizationDecorator, AuthorizationHelpers
 from api.extensions import db
-from api.models import (
-    App,
-    AppGroup,
-    AppTagMap,
-    OktaUser,
-    OktaUserGroupMember,
-    RoleGroupMap,
-)
+from api.models import App, AppGroup, AppTagMap, OktaUser, OktaUserGroupMember, RoleGroup, RoleGroupMap
 from api.operations import CreateApp, DeleteApp, ModifyAppTags
 from api.pagination import paginate
 from api.services import okta
@@ -287,6 +280,18 @@ def post(self) -> ResponseReturnValue:
         if owner_id is None:
             abort(400, "App initial_owner_id is required")
 
+        owner_role_ids = []
+        if "initial_owner_role_ids" in json_data:
+            owner_roles = (
+                RoleGroup.query.filter(RoleGroup.id.in_(json_data["initial_owner_role_ids"]))
+                .filter(RoleGroup.deleted_at.is_(None))
+                .all()
+            )
+            owner_role_ids = [role.id for role in owner_roles]
+
+            if len(owner_role_ids) != len(json_data["initial_owner_role_ids"]):
+                abort(400, "Given App initial_owner_role_ids contains invalid role ids")
+
         initial_additional_app_groups = []
         app_group_prefix = f"{AppGroup.APP_GROUP_NAME_PREFIX}{app.name}{AppGroup.APP_NAME_GROUP_NAME_SEPARATOR}"
         owner_group_name = f"{app_group_prefix}{AppGroup.APP_OWNERS_GROUP_NAME_SUFFIX}"
@@ -302,6 +307,7 @@ def post(self) -> ResponseReturnValue:
 
         app = CreateApp(
             owner_id=owner_id,
+            owner_role_ids=owner_role_ids,
             app=app,
             tags=json_data.get("tags_to_add", []),
             additional_app_groups=initial_additional_app_groups,

api/views/schemas/core_schemas.py

@@ -1161,6 +1161,7 @@ class AppSchema(SQLAlchemyAutoSchema):
     description = auto_field(validate=validate.Length(max=1024))
 
     initial_owner_id = fields.String(validate=validate.Length(min=1, max=255), load_only=True)
+    initial_owner_role_ids = fields.List(fields.String(validate=validate.Length(equal=20)), load_only=True)
     initial_additional_app_groups = fields.Nested(lambda: InitialAppGroupSchema, many=True, load_only=True)
 
     tags_to_add = fields.List(fields.String(validate=validate.Length(equal=20)), load_only=True)
@@ -1207,6 +1208,7 @@ class Meta:
             "name",
             "description",
             "initial_owner_id",
+            "initial_owner_role_ids",
             "initial_additional_app_groups",
             "tags_to_add",
             "tags_to_remove",
@@ -1231,6 +1233,7 @@ class Meta:
         )
         load_only = (
             "initial_owner_id",
+            "initial_owner_role_ids",
             "initial_additional_app_groups",
             "tags_to_add",
             "tags_to_remove",

tests/test_app.py

@@ -7,7 +7,17 @@
 from okta.models.group import Group
 from pytest_mock import MockerFixture
 
-from api.models import App, AppGroup, AppTagMap, OktaGroupTagMap, OktaUser, RoleGroup, Tag
+from api.models import (
+    App,
+    AppGroup,
+    AppTagMap,
+    OktaGroupTagMap,
+    OktaUser,
+    OktaUserGroupMember,
+    RoleGroup,
+    RoleGroupMap,
+    Tag,
+)
 from api.operations import ModifyGroupUsers, ModifyRoleGroups
 from api.services import okta
 from tests.factories import AppFactory, AppGroupFactory
@@ -258,6 +268,61 @@ def test_create_app(client: FlaskClient, db: SQLAlchemy, mocker: MockerFixture,
     assert test_app_group is not None
     assert test_app_group.is_owner is True
 
+def test_create_app_with_initial_owners(
+        client: FlaskClient,
+        db: SQLAlchemy,
+        mocker: MockerFixture,
+        faker: Faker,
+        user: OktaUser,
+        role_group: RoleGroup
+    ) -> None:
+
+    db.session.add(role_group)
+    db.session.add(user)
+    db.session.commit()
+
+    create_group_spy = mocker.patch.object(
+        okta, "create_group", return_value=Group({"id": faker.pystr()})
+    )
+    add_user_to_group_spy = mocker.patch.object(okta, "async_add_user_to_group")
+    add_owner_to_group_spy = mocker.patch.object(okta, "async_add_owner_to_group")
+
+    data = {"name": "Created", "initial_owner_id": user.id, "initial_owner_role_ids": [role_group.id]}
+
+    apps_url = url_for("api-apps.apps")
+    rep = client.post(apps_url, json=data)
+    assert rep.status_code == 201
+    assert create_group_spy.call_count == 1
+    assert add_user_to_group_spy.call_count == 1
+    assert add_owner_to_group_spy.call_count == 1
+
+    data = rep.get_json()
+    assert db.session.get(App, data["id"]) is not None
+
+    assert data["name"] == "Created"
+    assert data["description"] == ""
+
+    app_groups = AppGroup.query.filter(AppGroup.app_id == data["id"]).all()
+    assert len(app_groups) == 1
+
+    test_app_group = AppGroup.query.filter(AppGroup.name == "App-Created-Owners", AppGroup.app_id == data["id"]).first()
+    assert test_app_group is not None
+    assert test_app_group.is_owner is True
+
+    assert (
+        OktaUserGroupMember.query
+            .filter(OktaUserGroupMember.group_id == test_app_group.id)
+            .filter(OktaUserGroupMember.user_id == user.id)
+            .filter(OktaUserGroupMember.ended_at.is_(None)).count() == 2
+    )
+    assert (
+        RoleGroupMap.query
+            .filter(RoleGroupMap.group_id == test_app_group.id)
+            .filter(RoleGroupMap.role_group_id == role_group.id)
+            .filter(RoleGroupMap.ended_at.is_(None)).count() == 2
+    )
+
+
 def test_create_app_with_additional_groups(client: FlaskClient, db: SQLAlchemy, mocker: MockerFixture, faker: Faker) -> None:
     # test bad data
     apps_url = url_for("api-apps.apps")