Add tenant-aware user deletion, access request creation for users without roles

Author: NatSquared

api/src/api/resources/staff_user.py

@@ -111,8 +111,8 @@ def delete(user_id):
         """Permanently delete an inactive user."""
         try:
             user_data = TokenInfo.get_user_data()
-            StaffUserService.delete_deactivated_user(user_id, user_data.get('external_id'))
-            return {'message': 'User deleted successfully.'}, HTTPStatus.OK
+            response = StaffUserService.delete_deactivated_user(user_id, user_data.get('external_id'))
+            return response, HTTPStatus.OK
         except BusinessException as err:
             return {'message': err.error}, err.status_code
 

api/src/api/services/staff_user_service.py

@@ -9,6 +9,7 @@
 from api.models.membership import Membership
 from api.models.pagination_options import PaginationOptions
 from api.models.staff_user import StaffUser as StaffUserModel
+from api.models.tenant import Tenant as TenantModel
 from api.models.user_group_membership import UserGroupMembership
 from api.schemas.staff_user import StaffUserSchema
 from api.services.user_group_membership_service import UserGroupMembershipService
@@ -44,25 +45,32 @@ def create_or_update_user(self, user: dict):
         self.validate_fields(user)
 
         external_id = user.get('external_id')
+        tenant_id = getattr(g, 'tenant_id', None)
+        has_no_tenant_roles = len(user.get('roles', [])) == 0
         db_user = StaffUserModel.get_user_by_external_id(external_id, include_inactive=True)
 
         if db_user is None:
             new_user = StaffUserModel.create_user(user)
-            if len(user.get('roles', [])) == 0:
+            if has_no_tenant_roles and tenant_id:
+                UserGroupMembershipService.ensure_access_request_membership(external_id, tenant_id)
+            if has_no_tenant_roles:
                 self._send_access_request_email(new_user, getattr(g, 'tenant_id', None))
             return new_user
 
+        if has_no_tenant_roles and tenant_id:
+            UserGroupMembershipService.ensure_access_request_membership(external_id, tenant_id)
+
         return StaffUserModel.update_user(db_user.id, user)
 
     @classmethod
     def sync_super_admin_membership(cls, token_info: dict, token_roles, tenant_id=None):
         """Sync SUPER_ADMIN tenant membership from keycloak roles at login/request time."""
-        if getattr(g, '_syncing_super_admin_membership', False):
+        if getattr(g, 'syncing_super_admin_membership', False):
             return
 
-        g._syncing_super_admin_membership = True
+        g.syncing_super_admin_membership = True
         if not token_info:
-            g._syncing_super_admin_membership = False
+            g.syncing_super_admin_membership = False
             return
 
         try:
@@ -94,23 +102,41 @@ def sync_super_admin_membership(cls, token_info: dict, token_roles, tenant_id=No
             if not has_super_admin_role:
                 UserGroupMembershipService.remove_group_memberships_by_group_name(external_id, 'SUPER_ADMIN')
         finally:
-            g._syncing_super_admin_membership = False
+            g.syncing_super_admin_membership = False
 
     @staticmethod
     def _send_access_request_email(user: StaffUserModel, tenant_id: Optional[int] = None) -> None:
         """Send a new user email.Throws error if fails."""
         templates = current_app.config['EMAIL_TEMPLATES']
-        to_email_address = templates['ACCESS_REQUEST']['DEST_EMAIL_ADDRESS']
-        if to_email_address is None:
-            return
         template_id = templates['ACCESS_REQUEST']['ID']
         subject, body, args = StaffUserService._render_email_template(user, tenant_id)
+
+        resolved_tenant_id = tenant_id if tenant_id is not None else getattr(g, 'tenant_id', None)
+        recipient_emails = []
+
+        global_contact_email = templates['ACCESS_REQUEST']['DEST_EMAIL_ADDRESS']
+        if global_contact_email:
+            recipient_emails.append(global_contact_email)
+
+        if resolved_tenant_id:
+            tenant = TenantModel.find_by_id(resolved_tenant_id)
+            if tenant and tenant.contact_email:
+                recipient_emails.append(tenant.contact_email)
+
+        # Preserve order and avoid duplicate sends when contacts are the same.
+        recipient_emails = list(dict.fromkeys(recipient_emails))
+        if not recipient_emails:
+            return
+
         try:
-            notification.send_email(subject=subject,
-                                    email=to_email_address,
-                                    html_body=body,
-                                    args=args,
-                                    template_id=template_id)
+            for email in recipient_emails:
+                notification.send_email(
+                    subject=subject,
+                    email=email,
+                    html_body=body,
+                    args=args,
+                    template_id=template_id,
+                )
         except Exception as exc:  # noqa: B902
             current_app.logger.error('<Notification for new user registration failed', exc)
             raise BusinessException(
@@ -153,9 +179,11 @@ def attach_roles(user_collection):
             composite_role = UserGroupMembershipService.get_user_group_within_tenant(user.get('external_id'),
                                                                                      g.tenant_id)
             user['composite_roles'] = ''
+            user['main_role'] = ''
             if composite_role:
                 user['composite_roles'] = composite_role
-                user['main_role'] = CompositeRoles[composite_role].value
+                if composite_role in CompositeRoles.__members__:
+                    user['main_role'] = CompositeRoles[composite_role].value
 
     @classmethod
     def find_users(
@@ -221,7 +249,12 @@ def validate_user(db_user: StaffUserModel):
 
     @classmethod
     def delete_deactivated_user(cls, user_id, actor_external_id: Optional[str] = None):
-        """Permanently delete a user after required safeguards are satisfied."""
+        """Delete a deactivated user or only remove current tenant membership.
+
+        Behavior is tenant-aware:
+        - If the user has memberships in other tenants, only remove membership in the current tenant.
+        - If the user only belongs to the current tenant, fully delete the user.
+        """
         db_user = StaffUserModel.get_by_id(user_id, include_inactive=True)
         if db_user is None:
             raise BusinessException(error='User not found.', status_code=HTTPStatus.NOT_FOUND)
@@ -238,9 +271,48 @@ def delete_deactivated_user(cls, user_id, actor_external_id: Optional[str] = Non
                 status_code=HTTPStatus.CONFLICT,
             )
 
+        current_tenant_id = getattr(g, 'tenant_id', None)
+        if not current_tenant_id:
+            raise BusinessException(
+                error='Tenant context is required for this operation.',
+                status_code=HTTPStatus.BAD_REQUEST,
+            )
+
+        user_group_memberships = UserGroupMembership.query.filter(
+            UserGroupMembership.staff_user_external_id == db_user.external_id,
+        ).all()
+
+        has_current_tenant_membership = any(
+            membership.tenant_id == current_tenant_id for membership in user_group_memberships
+        )
+        if not has_current_tenant_membership:
+            raise BusinessException(
+                error='User does not have membership in the current tenant.',
+                status_code=HTTPStatus.BAD_REQUEST,
+            )
+
+        has_other_tenant_memberships = any(
+            membership.tenant_id != current_tenant_id for membership in user_group_memberships
+        )
+
+        if has_other_tenant_memberships:
+            UserGroupMembership.query.filter(
+                UserGroupMembership.staff_user_external_id == db_user.external_id,
+                UserGroupMembership.tenant_id == current_tenant_id,
+            ).delete(synchronize_session=False)
+            db.session.commit()
+            return {
+                'message': 'User removed from current tenant successfully.',
+                'action': 'removed_current_tenant_membership',
+            }
+
         Membership.query.filter(Membership.user_id == db_user.id).delete(synchronize_session=False)
         UserGroupMembership.query.filter(
             UserGroupMembership.staff_user_external_id == db_user.external_id,
         ).delete(synchronize_session=False)
         db.session.delete(db_user)
         db.session.commit()
+        return {
+            'message': 'User deleted successfully.',
+            'action': 'deleted_user',
+        }

api/src/api/services/user_group_membership_service.py

@@ -1,12 +1,17 @@
 """Service for user group membership management."""
 from typing import List, Tuple
 
+from sqlalchemy import func
+
 from api.models.db import db
 from api.models.user_group import UserGroup
 from api.models.user_group_membership import UserGroupMembership
 from api.models.user_role import UserRole
 
 
+ACCESS_REQUEST_GROUP_NAME = 'ACCESS_REQUEST'
+
+
 class UserGroupMembershipService:
     """User group membership management service."""
 
@@ -86,6 +91,39 @@ def ensure_group_membership(external_id: str, tenant_id: int, group_name: str):
         membership.save()
         return membership
 
+    @staticmethod
+    def ensure_access_request_membership(external_id: str, tenant_id: int):
+        """Ensure a tenant-scoped access-request membership exists for a user.
+
+        The access-request group intentionally has no role mappings, so users
+        become visible in tenant user management without receiving permissions.
+        """
+        if not external_id or not tenant_id:
+            return None
+
+        existing_membership = UserGroupMembership.query.filter(
+            UserGroupMembership.staff_user_external_id == external_id,
+            UserGroupMembership.tenant_id == tenant_id,
+        ).first()
+        if existing_membership:
+            return existing_membership
+
+        group = UserGroup.query.filter(UserGroup.name == ACCESS_REQUEST_GROUP_NAME).first()
+        if not group:
+            next_group_id = (db.session.query(func.max(UserGroup.id)).scalar() or 0) + 1
+            group = UserGroup(name=ACCESS_REQUEST_GROUP_NAME)
+            group.id = next_group_id
+            group.save()
+
+        membership = UserGroupMembership(
+            staff_user_external_id=external_id,
+            group_id=group.id,
+            tenant_id=tenant_id,
+            is_active=True,
+        )
+        membership.save()
+        return membership
+
     @staticmethod
     def remove_group_memberships_by_group_name(external_id: str, group_name: str) -> int:
         """Remove all memberships for a user that belong to the given group name."""

api/tests/unit/api/test_user.py

@@ -81,6 +81,56 @@ def test_create_staff_user(client, jwt, session, side_effect, expected_status,
     assert rv.status_code == expected_status
 
 
+def test_no_role_login_creates_access_request_membership_for_new_user(client, jwt, session, notify_mock):
+    """Assert no-role login creates a tenant-scoped ACCESS_REQUEST membership for new users."""
+    set_global_tenant(tenant_id=1)
+    claims = copy.deepcopy(TestJwtClaims.staff_admin_role.value)
+    claims['sub'] = 'b2f9b95e-3e6d-4b25-9b58-611201fce100'
+    claims['email'] = 'new-requester@gov.bc.ca'
+    claims['client_roles'] = []
+
+    headers = factory_auth_header(jwt=jwt, claims=claims)
+    rv = client.put('/api/user/', headers=headers, content_type=ContentType.JSON.value)
+
+    assert rv.status_code == HTTPStatus.OK
+
+    access_request_group = UserGroup.query.filter_by(name='ACCESS_REQUEST').first()
+    assert access_request_group is not None
+
+    membership = UserGroupMembership.get_group_by_user_and_tenant_id(claims['sub'], 1)
+    assert membership is not None
+    assert membership.group_id == access_request_group.id
+
+
+def test_no_role_login_creates_access_request_membership_for_existing_user_in_new_tenant(client, jwt, session):
+    """Assert no-role login adds ACCESS_REQUEST membership when an existing user enters a new tenant."""
+    set_global_tenant(tenant_id=1)
+    existing_user = factory_staff_user_model(external_id='6fdde698-238f-4d65-8a4d-b71715ea99ab')
+    factory_user_group_membership_model(str(existing_user.external_id), tenant_id=1)
+
+    tenant_data = dict(TestTenantInfo.tenant2)
+    tenant_data['short_name'] = 'GDXT'
+    tenant_2 = factory_tenant_model(tenant_data)
+
+    claims = copy.deepcopy(TestJwtClaims.staff_admin_role.value)
+    claims['sub'] = str(existing_user.external_id)
+    claims['email'] = existing_user.email_address
+    claims['client_roles'] = []
+
+    headers = factory_auth_header(jwt=jwt, claims=claims)
+    headers[TENANT_ID_HEADER] = tenant_2.short_name
+    rv = client.put('/api/user/', headers=headers, content_type=ContentType.JSON.value)
+
+    assert rv.status_code == HTTPStatus.OK
+
+    access_request_group = UserGroup.query.filter_by(name='ACCESS_REQUEST').first()
+    assert access_request_group is not None
+
+    membership = UserGroupMembership.get_group_by_user_and_tenant_id(str(existing_user.external_id), tenant_2.id)
+    assert membership is not None
+    assert membership.group_id == access_request_group.id
+
+
 def test_super_admin_login_assigns_super_admin_group_in_tenant(client, jwt, session):
     """Assert super admin users are persisted in tenant membership on login."""
     set_global_tenant()