Adding Save/Load FGA Resources Details MGMT APIs (#562)

Author: yosiharan

descope/management/common.py

@@ -141,6 +141,8 @@ class MgmtV1:
     fga_create_relations = "/v1/mgmt/fga/relations"
     fga_delete_relations = "/v1/mgmt/fga/relations/delete"
     fga_check = "/v1/mgmt/fga/check"
+    fga_resources_load = "/v1/mgmt/fga/resources/load"
+    fga_resources_save = "/v1/mgmt/fga/resources/save"
 
     # Project
     project_update_name = "/v1/mgmt/project/update/name"

descope/management/fga.py

@@ -138,3 +138,30 @@ def check(
                 response.json()["tuples"],
             )
         )
+
+    def load_resources_details(self, resource_identifiers: List[dict]) -> List[dict]:
+        """
+        Load details for the given resource identifiers.
+        Args:
+            resource_identifiers (List[dict]): list of dicts each containing 'resourceId' and 'resourceType'.
+        Returns:
+            List[dict]: list of resources details as returned by the server.
+        """
+        response = self._auth.do_post(
+            MgmtV1.fga_resources_load,
+            {"resourceIdentifiers": resource_identifiers},
+            pswd=self._auth.management_key,
+        )
+        return response.json().get("resourcesDetails", [])
+
+    def save_resources_details(self, resources_details: List[dict]) -> None:
+        """
+        Save details for the given resources.
+        Args:
+            resources_details (List[dict]): list of dicts each containing 'resourceId' and 'resourceType' plus optionally containing metadata fields such as 'displayName'.
+        """
+        self._auth.do_post(
+            MgmtV1.fga_resources_save,
+            {"resourcesDetails": resources_details},
+            pswd=self._auth.management_key,
+        )

tests/management/test_fga.py

@@ -211,3 +211,100 @@ def test_check(self):
                 verify=True,
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
+
+    def test_load_resources_details_success(self):
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+        )
+        response_body = {
+            "resourcesDetails": [
+                {"resourceId": "r1", "resourceType": "type1", "displayName": "Name1"},
+                {"resourceId": "r2", "resourceType": "type2", "displayName": "Name2"},
+            ]
+        }
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            mock_post.return_value.json.return_value = response_body
+            ids = [
+                {"resourceId": "r1", "resourceType": "type1"},
+                {"resourceId": "r2", "resourceType": "type2"},
+            ]
+            details = client.mgmt.fga.load_resources_details(ids)
+            self.assertEqual(details, response_body["resourcesDetails"])
+            mock_post.assert_called_with(
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.fga_resources_load}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={"resourceIdentifiers": ids},
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+    def test_load_resources_details_error(self):
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+        )
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = False
+            ids = [{"resourceId": "r1", "resourceType": "type1"}]
+            self.assertRaises(
+                AuthException,
+                client.mgmt.fga.load_resources_details,
+                ids,
+            )
+
+    def test_save_resources_details_success(self):
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+        )
+        details = [
+            {"resourceId": "r1", "resourceType": "type1", "displayName": "Name1"}
+        ]
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            client.mgmt.fga.save_resources_details(details)
+            mock_post.assert_called_with(
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.fga_resources_save}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={"resourcesDetails": details},
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+    def test_save_resources_details_error(self):
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+        )
+        details = [
+            {"resourceId": "r1", "resourceType": "type1", "displayName": "Name1"}
+        ]
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = False
+            self.assertRaises(
+                AuthException,
+                client.mgmt.fga.save_resources_details,
+                details,
+            )