Rename endpoints when fetching Directory Groups and Users (#23)

Author: Rohan Jadvani

tests/test_directory_sync.py

@@ -95,7 +95,7 @@ def mock_directories(self):
         return {
             "data": [
                 {
-                    "id": "directory_edp_id",
+                    "id": "directory_id",
                     "external_key": "fried-chicken",
                     "state": "linked",
                     "type": "gsuite directory",
@@ -108,51 +108,52 @@ def mock_directories(self):
             "listMetadata": {"before": None, "after": None},
         }
 
-    def test_list_users(self, mock_users, mock_request_method):
+    def test_list_users_with_directory(self, mock_users, mock_request_method):
         mock_response = Response()
         mock_response.status_code = 200
         mock_response.response_dict = mock_users
         mock_request_method("get", mock_response, 200)
-        response = self.directory_sync.list_users(
-            directory_endpoint_id="directory_edp_id"
-        )
+        response = self.directory_sync.list_users(directory="directory_id")
+        assert response.status_code == 200
+        assert response.response_dict == mock_users
+
+    def test_list_users_with_group(self, mock_users, mock_request_method):
+        mock_response = Response()
+        mock_response.status_code = 200
+        mock_response.response_dict = mock_users
+        mock_request_method("get", mock_response, 200)
+        response = self.directory_sync.list_users(group="directory_grp_id")
         assert response.status_code == 200
         assert response.response_dict == mock_users
 
-    def test_list_groups(self, mock_groups, mock_request_method):
+    def test_list_groups_with_directory(self, mock_groups, mock_request_method):
         mock_response = Response()
         mock_response.status_code = 200
         mock_response.response_dict = mock_groups
         mock_request_method("get", mock_response, 200)
-        response = self.directory_sync.list_groups(
-            directory_endpoint_id="directory_edp_id"
-        )
+        response = self.directory_sync.list_groups(directory="directory_id")
         assert response.status_code == 200
         assert response.response_dict == mock_groups
 
-    def test_get_user(self, mock_user, mock_request_method):
+    def test_list_groups_with_user(self, mock_groups, mock_request_method):
         mock_response = Response()
         mock_response.status_code = 200
-        mock_response.response_dict = mock_user
+        mock_response.response_dict = mock_groups
         mock_request_method("get", mock_response, 200)
-        response = self.directory_sync.get_user(
-            directory_endpoint_id="directory_edp_id",
-            directory_user_id="directory_usr_id",
-        )
+        response = self.directory_sync.list_groups(user="directory_usr_id")
         assert response.status_code == 200
-        assert response.response_dict == mock_user
+        assert response.response_dict == mock_groups
 
-    def test_list_user_groups(self, mock_user_groups, mock_request_method):
+    def test_get_user(self, mock_user, mock_request_method):
         mock_response = Response()
         mock_response.status_code = 200
-        mock_response.response_dict = mock_user_groups
+        mock_response.response_dict = mock_user
         mock_request_method("get", mock_response, 200)
-        response = self.directory_sync.list_user_groups(
-            directory_endpoint_id="directory_edp_id",
-            directory_user_id="directory_usr_id",
+        response = self.directory_sync.get_user(
+            directory="directory_id", directory_user="directory_usr_id",
         )
         assert response.status_code == 200
-        assert response.response_dict == mock_user_groups
+        assert response.response_dict == mock_user
 
     def test_list_directories(self, mock_directories, mock_request_method):
         mock_response = Response()

workos/directory_sync.py

@@ -19,86 +19,76 @@ def request_helper(self):
         return self._request_helper
 
     def list_users(
-        self, directory_endpoint_id, limit=RESPONSE_LIMIT, before=None, after=None
+        self, directory=None, group=None, limit=RESPONSE_LIMIT, before=None, after=None
     ):
-        """Gets a list of provisioned Users for a Directory Endpoint.
+        """Gets a list of provisioned Users for a Directory.
+
+        Note, either 'directory' or 'group' must be provided.
 
         Args:
-            directory_endpoint_id (str): Directory Endpoint unique identifier.
+            directory (str): Directory unique identifier.
+            group (str): Directory Group unique identifier.
             limit (int): Maximum number of records to return.
-            before (str): Pagination cursor to receive records before a provided Directory Endpoint ID.
-            after (str): Pagination cursor to receive records after a provided Directory Endpoint ID.
+            before (str): Pagination cursor to receive records before a provided Directory ID.
+            after (str): Pagination cursor to receive records after a provided Directory ID.
 
         Returns:
             dict: Directory Users response from WorkOS.
         """
         params = {"limit": limit, "before": before, "after": after}
+        if group is not None:
+            params["group"] = group
+        if directory is not None:
+            params["directory"] = directory
         return self.request_helper.request(
-            "directories/{directory_endpoint_id}/users".format(
-                directory_endpoint_id=directory_endpoint_id
-            ),
+            "directory_users",
             method=REQUEST_METHOD_GET,
             params=params,
             token=workos.api_key,
         )
 
     def list_groups(
-        self, directory_endpoint_id, limit=RESPONSE_LIMIT, before=None, after=None
+        self, directory=None, user=None, limit=RESPONSE_LIMIT, before=None, after=None
     ):
-        """Gets a list of provisioned Groups for a Directory Endpoint.
+        """Gets a list of provisioned Groups for a Directory .
+
+        Note, either 'directory' or 'user' must be provided.
 
         Args:
-            directory_endpoint_id (str): Directory Endpoint unique identifier.
+            directory (str): Directory unique identifier.
+            user (str): Directory User unique identifier.
             limit (int): Maximum number of records to return.
-            before (str): Pagination cursor to receive records before a provided Directory Endpoint ID.
-            after (str): Pagination cursor to receive records after a provided Directory Endpoint ID.
+            before (str): Pagination cursor to receive records before a provided Directory ID.
+            after (str): Pagination cursor to receive records after a provided Directory ID.
 
         Returns:
             dict: Directory Groups response from WorkOS.
         """
         params = {"limit": limit, "before": before, "after": after}
+        if user is not None:
+            params["user"] = user
+        if directory is not None:
+            params["directory"] = directory
         return self.request_helper.request(
-            "directories/{directory_endpoint_id}/groups".format(
-                directory_endpoint_id=directory_endpoint_id
-            ),
+            "directory_groups",
             method=REQUEST_METHOD_GET,
             params=params,
             token=workos.api_key,
         )
 
-    def get_user(self, directory_endpoint_id, directory_user_id):
+    def get_user(self, directory, directory_user):
         """Gets details for a single provisioned Directory User.
 
         Args:
-            directory_endpoint_id (str): Directory Endpoint unique identifier.
-            directory_user_id (str): Directory User unique identifier.
+            directory (str): Directory unique identifier.
+            directory_user (str): Directory User unique identifier.
 
         Returns:
             dict: Directory user response from WorkOS.
         """
         return self.request_helper.request(
-            "directories/{directory_endpoint_id}/users/{directory_user_id}".format(
-                directory_endpoint_id=directory_endpoint_id,
-                directory_user_id=directory_user_id,
-            ),
-            method=REQUEST_METHOD_GET,
-            token=workos.api_key,
-        )
-
-    def list_user_groups(self, directory_endpoint_id, directory_user_id):
-        """Gets details for a Directory User's provisioned Groups.
-
-        Args:
-            directory_endpoint_id (str): Directory Endpoint unique identifier.
-            directory_user_id (str): Directory User unique identifier.
-
-        Returns:
-            dict: Directory User's Groups response from WorkOS.
-        """
-        return self.request_helper.request(
-            "directories/{directory_endpoint_id}/users/{directory_user_id}/groups".format(
-                directory_endpoint_id=directory_endpoint_id,
-                directory_user_id=directory_user_id,
+            "directories/{directory}/users/{directory_user}".format(
+                directory=directory, directory_user=directory_user,
             ),
             method=REQUEST_METHOD_GET,
             token=workos.api_key,
@@ -110,11 +100,11 @@ def list_directories(
         """Gets details for existing Directories.
 
         Args:
-            domain (str): Domain of a Directory Endpoint. (Optional)
-            search (str): Searchable text for a Directory Endpoint. (Optional)
+            domain (str): Domain of a Directory. (Optional)
+            search (str): Searchable text for a Directory. (Optional)
             limit (int): Maximum number of records to return. (Optional)
-            before (str): Pagination cursor to receive records before a provided Directory Endpoint ID. (Optional)
-            after (str): Pagination cursor to receive records after a provided Directory Endpoint ID. (Optional)
+            before (str): Pagination cursor to receive records before a provided Directory ID. (Optional)
+            after (str): Pagination cursor to receive records after a provided Directory ID. (Optional)
 
         Returns:
             dict: Directories response from WorkOS.