diff --git a/src/workos/authorization.py b/src/workos/authorization.py index 6e12f035..13261570 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -1,13 +1,18 @@ -from typing import Any, Dict, Optional, Protocol, Sequence +from typing import Any, Dict, Optional, Protocol, Sequence, Union from pydantic import TypeAdapter +from typing_extensions import TypedDict from workos.types.authorization.environment_role import ( EnvironmentRole, EnvironmentRoleList, ) from workos.types.authorization.organization_role import OrganizationRole +from workos.types.authorization.organization_membership import ( + AuthorizationOrganizationMembership, +) from workos.types.authorization.permission import Permission +from workos.types.authorization.resource import Resource from workos.types.authorization.role import Role, RoleList from workos.types.list_resource import ( ListArgs, @@ -28,6 +33,19 @@ ) AUTHORIZATION_PERMISSIONS_PATH = "authorization/permissions" +AUTHORIZATION_RESOURCES_PATH = "authorization/resources" + + +class ParentResourceById(TypedDict): + parent_resource_id: str + + +class ParentResourceByExternalId(TypedDict): + parent_resource_external_id: str + parent_resource_type_slug: str + + +ParentResource = Union[ParentResourceById, ParentResourceByExternalId] _role_adapter: TypeAdapter[Role] = TypeAdapter(Role) @@ -41,6 +59,34 @@ class PermissionListFilters(ListArgs, total=False): ] +class ResourcesForMembershipListFilters(ListArgs, total=False): + organization_membership_id: str + resource_type: Optional[str] + parent_resource_id: Optional[str] + parent_organization_id: Optional[str] + parent_resource_type: Optional[str] + parent_external_id: Optional[str] + + +ResourcesForMembershipListResource = WorkOSListResource[ + Resource, ResourcesForMembershipListFilters, ListMetadata +] + + +class MembershipsForResourceListFilters(ListArgs, total=False): + resource_id: str + organization_id: str + resource_type: str + external_id: str + + +MembershipsForResourceListResource = WorkOSListResource[ + AuthorizationOrganizationMembership, + MembershipsForResourceListFilters, + ListMetadata, +] + + class AuthorizationModule(Protocol): """Offers methods through the WorkOS Authorization service.""" @@ -161,6 +207,75 @@ def add_environment_role_permission( permission_slug: str, ) -> SyncOrAsync[EnvironmentRole]: ... + # Resources + + def get_resource(self, resource_id: str) -> SyncOrAsync[Resource]: ... + + def create_resource( + self, + *, + resource_type_slug: str, + organization_id: str, + external_id: str, + name: str, + parent: ParentResource, + description: Optional[str] = None, + ) -> SyncOrAsync[Resource]: ... + + def update_resource( + self, + resource_id: str, + *, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> SyncOrAsync[Resource]: ... + + def delete_resource( + self, + resource_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> SyncOrAsync[None]: ... + + # Resource-Membership Relationships + + def list_resources_for_membership( + self, + organization_membership_id: str, + *, + resource_type: Optional[str] = None, + parent_resource_id: Optional[str] = None, + parent_organization_id: Optional[str] = None, + parent_resource_type: Optional[str] = None, + parent_external_id: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + order: PaginationOrder = "desc", + before: Optional[str] = None, + after: Optional[str] = None, + ) -> SyncOrAsync[ResourcesForMembershipListResource]: ... + + def list_memberships_for_resource( + self, + resource_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + order: PaginationOrder = "desc", + before: Optional[str] = None, + after: Optional[str] = None, + ) -> SyncOrAsync[MembershipsForResourceListResource]: ... + + def list_memberships_for_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + order: PaginationOrder = "desc", + before: Optional[str] = None, + after: Optional[str] = None, + ) -> SyncOrAsync[MembershipsForResourceListResource]: ... + class Authorization(AuthorizationModule): _http_client: SyncHTTPClient @@ -437,6 +552,204 @@ def add_environment_role_permission( return EnvironmentRole.model_validate(response) + # Resources + + def get_resource(self, resource_id: str) -> Resource: + response = self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_GET, + ) + + return Resource.model_validate(response) + + def create_resource( + self, + *, + resource_type_slug: str, + organization_id: str, + external_id: str, + name: str, + parent: ParentResource, + description: Optional[str] = None, + ) -> Resource: + json: Dict[str, Any] = { + "resource_type_slug": resource_type_slug, + "organization_id": organization_id, + "external_id": external_id, + "name": name, + **parent, + } + if description is not None: + json["description"] = description + + response = self._http_client.request( + AUTHORIZATION_RESOURCES_PATH, + method=REQUEST_METHOD_POST, + json=json, + ) + + return Resource.model_validate(response) + + def update_resource( + self, + *, + resource_id: str, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> Resource: + json: Dict[str, Any] = {} + if name is not None: + json["name"] = name + if description is not None: + json["description"] = description + + response = self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_PATCH, + json=json, + ) + + return Resource.model_validate(response) + + def delete_resource( + self, + *, + resource_id: str, + cascade_delete: Optional[bool] = None, + ) -> None: + if cascade_delete is not None: + self._http_client.delete_with_body( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + json={"cascade_delete": cascade_delete}, + ) + else: + self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_DELETE, + ) + + # Resource-Membership Relationships + + def list_resources_for_membership( + self, + organization_membership_id: str, + *, + resource_type: Optional[str] = None, + parent_resource_id: Optional[str] = None, + parent_organization_id: Optional[str] = None, + parent_resource_type: Optional[str] = None, + parent_external_id: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + order: PaginationOrder = "desc", + before: Optional[str] = None, + after: Optional[str] = None, + ) -> ResourcesForMembershipListResource: + list_params: ResourcesForMembershipListFilters = { + "limit": limit, + "order": order, + "before": before, + "after": after, + } + + if resource_type is not None: + list_params["resource_type"] = resource_type + + # Parent by internal ID and by external ID are mutually exclusive + if parent_resource_id is not None and parent_organization_id is not None: + raise ValueError( + "parent_resource_id and parent_organization_id are mutually exclusive" + ) + + if parent_resource_id is not None: + list_params["parent_resource_id"] = parent_resource_id + elif parent_organization_id is not None: + list_params["parent_organization_id"] = parent_organization_id + if parent_resource_type is not None: + list_params["parent_resource_type"] = parent_resource_type + if parent_external_id is not None: + list_params["parent_external_id"] = parent_external_id + + response = self._http_client.request( + f"authorization/organization_memberships/{organization_membership_id}/resources", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + # Add path param to list_args for pagination iterator + list_params["organization_membership_id"] = organization_membership_id + + return ResourcesForMembershipListResource( + list_method=self.list_resources_for_membership, + list_args=list_params, + **ListPage[Resource](**response).model_dump(), + ) + + def list_memberships_for_resource( + self, + resource_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + order: PaginationOrder = "desc", + before: Optional[str] = None, + after: Optional[str] = None, + ) -> MembershipsForResourceListResource: + list_params: MembershipsForResourceListFilters = { + "limit": limit, + "order": order, + "before": before, + "after": after, + } + + response = self._http_client.request( + f"authorization/resources/{resource_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + # Add path param to list_args for pagination iterator + list_params["resource_id"] = resource_id + + return MembershipsForResourceListResource( + list_method=self.list_memberships_for_resource, + list_args=list_params, + **ListPage[AuthorizationOrganizationMembership](**response).model_dump(), + ) + + def list_memberships_for_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + order: PaginationOrder = "desc", + before: Optional[str] = None, + after: Optional[str] = None, + ) -> MembershipsForResourceListResource: + list_params: MembershipsForResourceListFilters = { + "limit": limit, + "order": order, + "before": before, + "after": after, + } + + response = self._http_client.request( + f"authorization/organizations/{organization_id}/resources/{resource_type}/{external_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + # Add path params to list_args for pagination iterator + list_params["organization_id"] = organization_id + list_params["resource_type"] = resource_type + list_params["external_id"] = external_id + + return MembershipsForResourceListResource( + list_method=self.list_memberships_for_resource_by_external_id, + list_args=list_params, + **ListPage[AuthorizationOrganizationMembership](**response).model_dump(), + ) + class AsyncAuthorization(AuthorizationModule): _http_client: AsyncHTTPClient @@ -712,3 +1025,201 @@ async def add_environment_role_permission( ) return EnvironmentRole.model_validate(response) + + # Resources + + async def get_resource(self, resource_id: str) -> Resource: + response = await self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_GET, + ) + + return Resource.model_validate(response) + + async def create_resource( + self, + *, + resource_type_slug: str, + organization_id: str, + external_id: str, + name: str, + parent: ParentResource, + description: Optional[str] = None, + ) -> Resource: + json: Dict[str, Any] = { + "resource_type_slug": resource_type_slug, + "organization_id": organization_id, + "external_id": external_id, + "name": name, + **parent, + } + if description is not None: + json["description"] = description + + response = await self._http_client.request( + AUTHORIZATION_RESOURCES_PATH, + method=REQUEST_METHOD_POST, + json=json, + ) + + return Resource.model_validate(response) + + async def update_resource( + self, + resource_id: str, + *, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> Resource: + json: Dict[str, Any] = {} + if name is not None: + json["name"] = name + if description is not None: + json["description"] = description + + response = await self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_PATCH, + json=json, + ) + + return Resource.model_validate(response) + + async def delete_resource( + self, + resource_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> None: + if cascade_delete is not None: + await self._http_client.delete_with_body( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + json={"cascade_delete": cascade_delete}, + ) + else: + await self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_DELETE, + ) + + # Resource-Membership Relationships + + async def list_resources_for_membership( + self, + organization_membership_id: str, + *, + resource_type: Optional[str] = None, + parent_resource_id: Optional[str] = None, + parent_organization_id: Optional[str] = None, + parent_resource_type: Optional[str] = None, + parent_external_id: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + order: PaginationOrder = "desc", + before: Optional[str] = None, + after: Optional[str] = None, + ) -> ResourcesForMembershipListResource: + list_params: ResourcesForMembershipListFilters = { + "limit": limit, + "order": order, + "before": before, + "after": after, + } + + if resource_type is not None: + list_params["resource_type"] = resource_type + + # Parent by internal ID and by external ID are mutually exclusive + if parent_resource_id is not None and parent_organization_id is not None: + raise ValueError( + "parent_resource_id and parent_organization_id are mutually exclusive" + ) + + if parent_resource_id is not None: + list_params["parent_resource_id"] = parent_resource_id + elif parent_organization_id is not None: + list_params["parent_organization_id"] = parent_organization_id + if parent_resource_type is not None: + list_params["parent_resource_type"] = parent_resource_type + if parent_external_id is not None: + list_params["parent_external_id"] = parent_external_id + + response = await self._http_client.request( + f"authorization/organization_memberships/{organization_membership_id}/resources", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + # Add path param to list_args for pagination iterator + list_params["organization_membership_id"] = organization_membership_id + + return ResourcesForMembershipListResource( + list_method=self.list_resources_for_membership, + list_args=list_params, + **ListPage[Resource](**response).model_dump(), + ) + + async def list_memberships_for_resource( + self, + resource_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + order: PaginationOrder = "desc", + before: Optional[str] = None, + after: Optional[str] = None, + ) -> MembershipsForResourceListResource: + list_params: MembershipsForResourceListFilters = { + "limit": limit, + "order": order, + "before": before, + "after": after, + } + + response = await self._http_client.request( + f"authorization/resources/{resource_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + # Add path param to list_args for pagination iterator + list_params["resource_id"] = resource_id + + return MembershipsForResourceListResource( + list_method=self.list_memberships_for_resource, + list_args=list_params, + **ListPage[AuthorizationOrganizationMembership](**response).model_dump(), + ) + + async def list_memberships_for_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + order: PaginationOrder = "desc", + before: Optional[str] = None, + after: Optional[str] = None, + ) -> MembershipsForResourceListResource: + list_params: MembershipsForResourceListFilters = { + "limit": limit, + "order": order, + "before": before, + "after": after, + } + + response = await self._http_client.request( + f"authorization/organizations/{organization_id}/resources/{resource_type}/{external_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + # Add path params to list_args for pagination iterator + list_params["organization_id"] = organization_id + list_params["resource_type"] = resource_type + list_params["external_id"] = external_id + + return MembershipsForResourceListResource( + list_method=self.list_memberships_for_resource_by_external_id, + list_args=list_params, + **ListPage[AuthorizationOrganizationMembership](**response).model_dump(), + ) diff --git a/src/workos/types/authorization/resource.py b/src/workos/types/authorization/resource.py index 917673c4..e699292b 100644 --- a/src/workos/types/authorization/resource.py +++ b/src/workos/types/authorization/resource.py @@ -1,4 +1,4 @@ -from typing import Literal, Optional +from typing import Any, Literal, Mapping, Optional from workos.types.workos_model import WorkOSModel diff --git a/tests/test_authorization_resource_crud.py b/tests/test_authorization_resource_crud.py new file mode 100644 index 00000000..6cd9ea68 --- /dev/null +++ b/tests/test_authorization_resource_crud.py @@ -0,0 +1,189 @@ +from typing import Union + +import pytest +from tests.utils.fixtures.mock_resource import MockResource +from tests.utils.syncify import syncify +from workos.authorization import AsyncAuthorization, Authorization + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestAuthorizationResourceCRUD: + @pytest.fixture(autouse=True) + def setup(self, module_instance: Union[Authorization, AsyncAuthorization]): + self.http_client = module_instance._http_client + self.authorization = module_instance + + @pytest.fixture + def mock_resource(self): + return MockResource(id="res_01ABC").dict() + + # --- get_resource --- + + def test_get_resource(self, mock_resource, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 200 + ) + + resource = syncify(self.authorization.get_resource("res_01ABC")) + + assert resource.id == "res_01ABC" + assert resource.object == "authorization_resource" + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + + # --- create_resource --- + + def test_create_resource_required_fields_only( + self, mock_resource, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 201 + ) + + resource = syncify( + self.authorization.create_resource( + resource_type="document", + organization_id="org_01EHT88Z8J8795GZNQ4ZP1J81T", + ) + ) + + assert resource.id == "res_01ABC" + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith("/authorization/resources") + assert request_kwargs["json"] == { + "resource_type": "document", + "organization_id": "org_01EHT88Z8J8795GZNQ4ZP1J81T", + } + + def test_create_resource_with_all_optional_fields( + self, mock_resource, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 201 + ) + + syncify( + self.authorization.create_resource( + resource_type="document", + organization_id="org_01EHT88Z8J8795GZNQ4ZP1J81T", + external_id="ext_123", + meta={"key": "value"}, + parent={"resource_id": "res_01PARENT"}, + ) + ) + + assert request_kwargs["json"] == { + "resource_type": "document", + "organization_id": "org_01EHT88Z8J8795GZNQ4ZP1J81T", + "external_id": "ext_123", + "meta": {"key": "value"}, + "parent": {"resource_id": "res_01PARENT"}, + } + + def test_create_resource_with_parent_by_id( + self, mock_resource, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 201 + ) + + syncify( + self.authorization.create_resource( + resource_type="document", + organization_id="org_01EHT88Z8J8795GZNQ4ZP1J81T", + parent={"resource_id": "res_01PARENT"}, + ) + ) + + assert request_kwargs["json"]["parent"] == {"resource_id": "res_01PARENT"} + + def test_create_resource_with_parent_by_external_id( + self, mock_resource, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 201 + ) + + syncify( + self.authorization.create_resource( + resource_type="document", + organization_id="org_01EHT88Z8J8795GZNQ4ZP1J81T", + parent={ + "organization_id": "org_01EHT88Z8J8795GZNQ4ZP1J81T", + "resource_type": "folder", + "external_id": "ext_parent_456", + }, + ) + ) + + assert request_kwargs["json"]["parent"] == { + "organization_id": "org_01EHT88Z8J8795GZNQ4ZP1J81T", + "resource_type": "folder", + "external_id": "ext_parent_456", + } + + # --- update_resource --- + + def test_update_resource_with_meta( + self, mock_resource, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 200 + ) + + resource = syncify( + self.authorization.update_resource( + "res_01ABC", + meta={"updated_key": "updated_value"}, + ) + ) + + assert resource.id == "res_01ABC" + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + assert request_kwargs["json"] == {"meta": {"updated_key": "updated_value"}} + + def test_update_resource_without_meta( + self, mock_resource, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 200 + ) + + syncify(self.authorization.update_resource("res_01ABC")) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["json"] == {} + + # --- delete_resource --- + + def test_delete_resource_without_cascade( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=202, + headers={"content-type": "text/plain; charset=utf-8"}, + ) + + response = syncify(self.authorization.delete_resource("res_01ABC")) + + assert response is None + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + + def test_delete_resource_with_cascade(self, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=202, + headers={"content-type": "text/plain; charset=utf-8"}, + ) + + response = syncify( + self.authorization.delete_resource("res_01ABC", cascade_delete=True) + ) + + assert response is None + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + assert request_kwargs["json"] == {"cascade_delete": True} diff --git a/tests/test_authorization_resource_memberships.py b/tests/test_authorization_resource_memberships.py new file mode 100644 index 00000000..fce9b4c6 --- /dev/null +++ b/tests/test_authorization_resource_memberships.py @@ -0,0 +1,260 @@ +from typing import Union + +import pytest +from tests.utils.fixtures.mock_resource import MockResource +from tests.utils.list_resource import list_response_of +from tests.utils.syncify import syncify +from workos.authorization import AsyncAuthorization, Authorization + + +def _mock_membership( + id: str = "om_01ABC", + user_id: str = "user_01ABC", + organization_id: str = "org_01ABC", +) -> dict: + """Build a minimal organization-membership dict for test responses. + + Includes ``custom_attributes: None`` so that the dict round-trips + identically through the Pydantic model's ``.dict()`` serialisation + (the model defines a default of ``None`` for that field). + """ + return { + "object": "organization_membership", + "id": id, + "user_id": user_id, + "organization_id": organization_id, + "organization_name": "Test Org", + "status": "active", + "custom_attributes": None, + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", + } + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestAuthorizationResourceMemberships: + @pytest.fixture(autouse=True) + def setup(self, module_instance: Union[Authorization, AsyncAuthorization]): + self.http_client = module_instance._http_client + self.authorization = module_instance + + # ------------------------------------------------------------------ + # list_resources_for_membership + # ------------------------------------------------------------------ + + @pytest.fixture + def mock_resources_list(self): + resources = [MockResource(id=f"res_{i}").dict() for i in range(3)] + return { + "data": resources, + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + + @pytest.fixture + def mock_resources_multiple_pages(self): + resources = [MockResource(id=f"res_{i}").dict() for i in range(40)] + return list_response_of(data=resources) + + def test_list_resources_for_membership( + self, mock_resources_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + result = syncify(self.authorization.list_resources_for_membership("om_01ABC")) + + assert len(result.data) == 3 + assert result.data[0].id == "res_0" + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/resources" + ) + + def test_list_resources_for_membership_with_resource_type_filter( + self, mock_resources_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", resource_type="document" + ) + ) + + assert request_kwargs["params"]["resource_type"] == "document" + + def test_list_resources_for_membership_with_parent_by_id( + self, mock_resources_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", parent_resource_id="res_01PARENT" + ) + ) + + assert request_kwargs["params"]["parent_resource_id"] == "res_01PARENT" + assert "parent_organization_id" not in request_kwargs["params"] + + def test_list_resources_for_membership_with_parent_by_external_id( + self, mock_resources_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + parent_organization_id="org_01ABC", + parent_resource_type="folder", + parent_external_id="ext_parent_456", + ) + ) + + assert request_kwargs["params"]["parent_organization_id"] == "org_01ABC" + assert request_kwargs["params"]["parent_resource_type"] == "folder" + assert request_kwargs["params"]["parent_external_id"] == "ext_parent_456" + assert "parent_resource_id" not in request_kwargs["params"] + + def test_list_resources_for_membership_mutual_exclusivity(self): + with pytest.raises(ValueError, match="mutually exclusive"): + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + parent_resource_id="res_01PARENT", + parent_organization_id="org_01ABC", + ) + ) + + def test_list_resources_for_membership_pagination( + self, + mock_resources_multiple_pages, + test_auto_pagination, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.authorization.list_resources_for_membership, + expected_all_page_data=mock_resources_multiple_pages["data"], + list_function_params={ + "organization_membership_id": "om_01ABC", + }, + url_path_keys=["organization_membership_id"], + ) + + # ------------------------------------------------------------------ + # list_memberships_for_resource + # ------------------------------------------------------------------ + + @pytest.fixture + def mock_memberships_list(self): + memberships = [_mock_membership(id=f"om_{i}") for i in range(3)] + return { + "data": memberships, + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + + @pytest.fixture + def mock_memberships_empty(self): + return { + "data": [], + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + + @pytest.fixture + def mock_memberships_multiple_pages(self): + memberships = [_mock_membership(id=f"om_{i}") for i in range(40)] + return list_response_of(data=memberships) + + def test_list_memberships_for_resource( + self, mock_memberships_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list, 200 + ) + + result = syncify(self.authorization.list_memberships_for_resource("res_01ABC")) + + assert len(result.data) == 3 + assert result.data[0].id == "om_0" + assert result.data[0].object == "organization_membership" + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/resources/res_01ABC/organization_memberships" + ) + + def test_list_memberships_for_resource_empty( + self, mock_memberships_empty, capture_and_mock_http_client_request + ): + capture_and_mock_http_client_request( + self.http_client, mock_memberships_empty, 200 + ) + + result = syncify(self.authorization.list_memberships_for_resource("res_01ABC")) + + assert len(result.data) == 0 + + def test_list_memberships_for_resource_pagination( + self, + mock_memberships_multiple_pages, + test_auto_pagination, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.authorization.list_memberships_for_resource, + expected_all_page_data=mock_memberships_multiple_pages["data"], + list_function_params={ + "resource_id": "res_01ABC", + }, + url_path_keys=["resource_id"], + ) + + # ------------------------------------------------------------------ + # list_memberships_for_resource_by_external_id + # ------------------------------------------------------------------ + + def test_list_memberships_for_resource_by_external_id( + self, mock_memberships_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list, 200 + ) + + result = syncify( + self.authorization.list_memberships_for_resource_by_external_id( + "org_01ABC", "document", "ext_123" + ) + ) + + assert len(result.data) == 3 + assert result.data[0].id == "om_0" + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organizations/org_01ABC/resources/document/ext_123/organization_memberships" + ) + + def test_list_memberships_for_resource_by_external_id_pagination( + self, + mock_memberships_multiple_pages, + test_auto_pagination, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.authorization.list_memberships_for_resource_by_external_id, + expected_all_page_data=mock_memberships_multiple_pages["data"], + list_function_params={ + "organization_id": "org_01ABC", + "resource_type": "document", + "external_id": "ext_123", + }, + url_path_keys=["organization_id", "resource_type", "external_id"], + )