fix(directory): expose directory_group_id filter in list_directory_users

scalekit/directory.py

@@ -72,7 +72,8 @@ def list_directory_users(
             page_size: Optional[int] = None,
             page_token: Optional[str] = None,
             include_detail: Optional[bool] = None,
-            updated_after: Optional[str] = None
+            updated_after: Optional[str] = None,
+            directory_group_id: Optional[str] = None
     ) -> tuple[ListDirUsersResponse, Any]:
         """
         Method to fetch list of directory users based on given organization and directory id
@@ -89,6 +90,8 @@ def list_directory_users(
         :type                    :     ``` bool ```
         :param updated_after      :     param to get updated after detail
         :type                    :     ``` str ```
+        :param directory_group_id :     filter users by membership in a specific directory group
+        :type                    :     ``` str ```
 
         :returns:
              list of directory users
@@ -101,11 +104,13 @@ def list_directory_users(
                 page_size=page_size,
                 page_token=page_token,
                 include_detail=include_detail,
+                directory_group_id=directory_group_id,
                 updated_after=updated_after
             ),
         )
 
         user_response = (ListDirUsersResponse(), response[1])
+        user_response[0].users = []
         for user in response[0].users:
             dir_user = DirUser()
             dir_user.id = user.id
@@ -166,6 +171,7 @@ def list_directory_groups(
         )
 
         group_response = (ListDirGroupsResponse(), response[1])
+        group_response[0].groups = []
         for group in response[0].groups:
             dir_group = DirGroup()
             dir_group.id = group.id

tests/test_directory.py

@@ -1,10 +1,13 @@
 
+import os
+import requests
+from urllib.parse import urlparse
 from faker import Faker
 from basetest import BaseTest
 from scalekit.common.exceptions import ScalekitNotFoundException
 
 from scalekit.v1.organizations.organizations_pb2 import CreateOrganization
-from scalekit.v1.directories.directories_pb2 import CreateDirectory
+from scalekit.v1.directories.directories_pb2 import CreateDirectory, CreateDirectorySecretRequest
 from scalekit.v1.directories.directories_pb2 import DirectoryProvider, DirectoryType
 
 
@@ -124,6 +127,95 @@ def test_get_primary_directory_by_organization_id(self):
         self.assertEqual(response.id, self.dir_id)
         self.assertEqual(response.organization_id, self.org_id)
 
+    def test_list_directory_users(self):
+        """ Method to test list directory users """
+        organization = CreateOrganization(display_name=Faker().company(), external_id=Faker().uuid4())
+        org_response = self.scalekit_client.organization.create_organization(organization=organization)
+        self.org_id = org_response[0].organization.id
+
+        directory = CreateDirectory(directory_provider=DirectoryProvider.OKTA, directory_type=DirectoryType.SCIM)
+        create_response = self.scalekit_client.directory.create_directory(organization_id=self.org_id, directory=directory)
+        self.dir_id = create_response[0].directory.id
+
+        response = self.scalekit_client.directory.list_directory_users(
+            organization_id=self.org_id,
+            directory_id=self.dir_id
+        )
+        self.assertEqual(response[1].code().name, "OK")
+        self.assertTrue(response[0] is not None)
+
+    def test_list_directory_users_filter_by_group(self):
+        """ Method to test list directory users filtered by directory_group_id """
+        fake = Faker()
+        organization = CreateOrganization(display_name=fake.company(), external_id=fake.uuid4())
+        org_response = self.scalekit_client.organization.create_organization(organization=organization)
+        self.org_id = org_response[0].organization.id
+
+        directory = CreateDirectory(directory_provider=DirectoryProvider.OKTA, directory_type=DirectoryType.SCIM)
+        create_response = self.scalekit_client.directory.create_directory(organization_id=self.org_id, directory=directory)
+        self.dir_id = create_response[0].directory.id
+
+        # get SCIM endpoint and bearer token
+        env_url = os.environ['SCALEKIT_ENV_URL'].rstrip('/')
+        raw_endpoint = create_response[0].directory.directory_endpoint
+        scim_base = env_url + urlparse(raw_endpoint).path
+        secret_resp = self.scalekit_client.directory.core_client.grpc_exec(
+            self.scalekit_client.directory.directory_service.CreateDirectorySecret.with_call,
+            CreateDirectorySecretRequest(organization_id=self.org_id, directory_id=self.dir_id)
+        )
+        headers = {
+            'Authorization': f'Bearer {secret_resp[0].plain_secret}',
+            'Content-Type': 'application/scim+json'
+        }
+
+        # seed a user via SCIM
+        email = fake.email()
+        user_r = requests.post(f'{scim_base}/Users', json={
+            'schemas': ['urn:ietf:params:scim:schemas:core:2.0:User'],
+            'userName': email,
+            'name': {'givenName': fake.first_name(), 'familyName': fake.last_name()},
+            'emails': [{'value': email, 'primary': True}],
+            'active': True
+        }, headers=headers)
+        self.assertEqual(user_r.status_code, 201)
+        scim_user_id = user_r.json()['id']
+
+        # seed a group containing that user via SCIM
+        group_r = requests.post(f'{scim_base}/Groups', json={
+            'schemas': ['urn:ietf:params:scim:schemas:core:2.0:Group'],
+            'displayName': fake.bs(),
+            'members': [{'value': scim_user_id}]
+        }, headers=headers)
+        self.assertEqual(group_r.status_code, 201)
+        group_id = group_r.json()['id']
+
+        # verify filter returns only the user in that group
+        response = self.scalekit_client.directory.list_directory_users(
+            organization_id=self.org_id,
+            directory_id=self.dir_id,
+            directory_group_id=group_id
+        )
+        self.assertEqual(response[1].code().name, "OK")
+        self.assertEqual(response[0].total_size, 1)
+        self.assertEqual(response[0].users[0].id, scim_user_id)
+
+    def test_list_directory_groups(self):
+        """ Method to test list directory groups """
+        organization = CreateOrganization(display_name=Faker().company(), external_id=Faker().uuid4())
+        org_response = self.scalekit_client.organization.create_organization(organization=organization)
+        self.org_id = org_response[0].organization.id
+
+        directory = CreateDirectory(directory_provider=DirectoryProvider.OKTA, directory_type=DirectoryType.SCIM)
+        create_response = self.scalekit_client.directory.create_directory(organization_id=self.org_id, directory=directory)
+        self.dir_id = create_response[0].directory.id
+
+        response = self.scalekit_client.directory.list_directory_groups(
+            organization_id=self.org_id,
+            directory_id=self.dir_id
+        )
+        self.assertEqual(response[1].code().name, "OK")
+        self.assertTrue(response[0] is not None)
+
     def tearDown(self):
         """ Method to clean up """
         if self.dir_id: