From e6234483bc3e80382e52f9f17d01a06bf48777ce Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Fri, 20 Feb 2026 05:34:19 -1000 Subject: [PATCH 1/3] FGA implementation pr1 --- src/workos/authorization.py | 196 ++++++++++++++++++++- src/workos/types/authorization/resource.py | 5 +- tests/test_authorization_resource_crud.py | 189 ++++++++++++++++++++ 3 files changed, 388 insertions(+), 2 deletions(-) create mode 100644 tests/test_authorization_resource_crud.py diff --git a/src/workos/authorization.py b/src/workos/authorization.py index 6e12f035..eac54b3f 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -1,6 +1,7 @@ -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, @@ -8,6 +9,7 @@ ) from workos.types.authorization.organization_role import OrganizationRole 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 +30,24 @@ ) AUTHORIZATION_PERMISSIONS_PATH = "authorization/permissions" +AUTHORIZATION_RESOURCES_PATH = "authorization/resources" + + +class ParentResourceById(TypedDict): + """Identify a parent resource by its WorkOS resource ID.""" + + resource_id: str + + +class ParentResourceByExternalId(TypedDict): + """Identify a parent resource by organization, type, and external ID.""" + + organization_id: str + resource_type: str + external_id: str + + +ParentResource = Union[ParentResourceById, ParentResourceByExternalId] _role_adapter: TypeAdapter[Role] = TypeAdapter(Role) @@ -161,6 +181,34 @@ 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: str, + organization_id: str, + external_id: Optional[str] = None, + meta: Optional[Dict[str, Any]] = None, + parent: Optional[ParentResource] = None, + ) -> SyncOrAsync[Resource]: ... + + def update_resource( + self, + resource_id: str, + *, + meta: Optional[Dict[str, Any]] = None, + ) -> SyncOrAsync[Resource]: ... + + def delete_resource( + self, + resource_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> SyncOrAsync[None]: ... + class Authorization(AuthorizationModule): _http_client: SyncHTTPClient @@ -437,6 +485,79 @@ 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: str, + organization_id: str, + external_id: Optional[str] = None, + meta: Optional[Dict[str, Any]] = None, + parent: Optional[ParentResource] = None, + ) -> Resource: + json: Dict[str, Any] = { + "resource_type": resource_type, + "organization_id": organization_id, + } + if external_id is not None: + json["external_id"] = external_id + if meta is not None: + json["meta"] = meta + if parent is not None: + json["parent"] = parent + + 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, + *, + meta: Optional[Dict[str, Any]] = None, + ) -> Resource: + json: Dict[str, Any] = {} + if meta is not None: + json["meta"] = meta + + 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, + ) + class AsyncAuthorization(AuthorizationModule): _http_client: AsyncHTTPClient @@ -712,3 +833,76 @@ 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: str, + organization_id: str, + external_id: Optional[str] = None, + meta: Optional[Dict[str, Any]] = None, + parent: Optional[ParentResource] = None, + ) -> Resource: + json: Dict[str, Any] = { + "resource_type": resource_type, + "organization_id": organization_id, + } + if external_id is not None: + json["external_id"] = external_id + if meta is not None: + json["meta"] = meta + if parent is not None: + json["parent"] = parent + + 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, + *, + meta: Optional[Dict[str, Any]] = None, + ) -> Resource: + json: Dict[str, Any] = {} + if meta is not None: + json["meta"] = meta + + 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, + ) diff --git a/src/workos/types/authorization/resource.py b/src/workos/types/authorization/resource.py index 917673c4..5b29778b 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 @@ -14,5 +14,8 @@ class Resource(WorkOSModel): resource_type_slug: str organization_id: str parent_resource_id: Optional[str] = None + # The API returns meta when set via create_resource / update_resource. + # Without this field the model would silently discard that data. + meta: Optional[Mapping[str, Any]] = None created_at: str updated_at: str 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} From a5d3ce2a1b6dea42b126d005b3dd1d6feac773ed Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Fri, 20 Feb 2026 06:52:02 -1000 Subject: [PATCH 2/3] cleanup --- src/workos/authorization.py | 89 +++++++++++----------- src/workos/types/authorization/resource.py | 3 - 2 files changed, 46 insertions(+), 46 deletions(-) diff --git a/src/workos/authorization.py b/src/workos/authorization.py index eac54b3f..eeea21bb 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -34,17 +34,12 @@ class ParentResourceById(TypedDict): - """Identify a parent resource by its WorkOS resource ID.""" - - resource_id: str + parent_resource_id: str class ParentResourceByExternalId(TypedDict): - """Identify a parent resource by organization, type, and external ID.""" - - organization_id: str - resource_type: str - external_id: str + parent_resource_external_id: str + parent_resource_type_slug: str ParentResource = Union[ParentResourceById, ParentResourceByExternalId] @@ -188,18 +183,20 @@ def get_resource(self, resource_id: str) -> SyncOrAsync[Resource]: ... def create_resource( self, *, - resource_type: str, + resource_type_slug: str, organization_id: str, - external_id: Optional[str] = None, - meta: Optional[Dict[str, Any]] = None, - parent: Optional[ParentResource] = None, + external_id: str, + name: str, + parent: ParentResource, + description: Optional[str] = None, ) -> SyncOrAsync[Resource]: ... def update_resource( self, resource_id: str, *, - meta: Optional[Dict[str, Any]] = None, + name: Optional[str] = None, + description: Optional[str] = None, ) -> SyncOrAsync[Resource]: ... def delete_resource( @@ -498,22 +495,22 @@ def get_resource(self, resource_id: str) -> Resource: def create_resource( self, *, - resource_type: str, + resource_type_slug: str, organization_id: str, - external_id: Optional[str] = None, - meta: Optional[Dict[str, Any]] = None, - parent: Optional[ParentResource] = None, + external_id: str, + name: str, + parent: ParentResource, + description: Optional[str] = None, ) -> Resource: json: Dict[str, Any] = { - "resource_type": resource_type, + "resource_type_slug": resource_type_slug, "organization_id": organization_id, + "external_id": external_id, + "name": name, + **parent, } - if external_id is not None: - json["external_id"] = external_id - if meta is not None: - json["meta"] = meta - if parent is not None: - json["parent"] = parent + if description is not None: + json["description"] = description response = self._http_client.request( AUTHORIZATION_RESOURCES_PATH, @@ -525,13 +522,16 @@ def create_resource( def update_resource( self, - resource_id: str, *, - meta: Optional[Dict[str, Any]] = None, + resource_id: str, + name: Optional[str] = None, + description: Optional[str] = None, ) -> Resource: json: Dict[str, Any] = {} - if meta is not None: - json["meta"] = meta + 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}", @@ -543,8 +543,8 @@ def update_resource( def delete_resource( self, - resource_id: str, *, + resource_id: str, cascade_delete: Optional[bool] = None, ) -> None: if cascade_delete is not None: @@ -847,22 +847,22 @@ async def get_resource(self, resource_id: str) -> Resource: async def create_resource( self, *, - resource_type: str, + resource_type_slug: str, organization_id: str, - external_id: Optional[str] = None, - meta: Optional[Dict[str, Any]] = None, - parent: Optional[ParentResource] = None, + external_id: str, + name: str, + parent: ParentResource, + description: Optional[str] = None, ) -> Resource: json: Dict[str, Any] = { - "resource_type": resource_type, + "resource_type_slug": resource_type_slug, "organization_id": organization_id, + "external_id": external_id, + "name": name, + **parent, } - if external_id is not None: - json["external_id"] = external_id - if meta is not None: - json["meta"] = meta - if parent is not None: - json["parent"] = parent + if description is not None: + json["description"] = description response = await self._http_client.request( AUTHORIZATION_RESOURCES_PATH, @@ -876,11 +876,14 @@ async def update_resource( self, resource_id: str, *, - meta: Optional[Dict[str, Any]] = None, + name: Optional[str] = None, + description: Optional[str] = None, ) -> Resource: json: Dict[str, Any] = {} - if meta is not None: - json["meta"] = meta + 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}", diff --git a/src/workos/types/authorization/resource.py b/src/workos/types/authorization/resource.py index 5b29778b..e699292b 100644 --- a/src/workos/types/authorization/resource.py +++ b/src/workos/types/authorization/resource.py @@ -14,8 +14,5 @@ class Resource(WorkOSModel): resource_type_slug: str organization_id: str parent_resource_id: Optional[str] = None - # The API returns meta when set via create_resource / update_resource. - # Without this field the model would silently discard that data. - meta: Optional[Mapping[str, Any]] = None created_at: str updated_at: str From e30a2f7bdb8ceea1aebc02caa62c0c4dbbce1b9d Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Fri, 20 Feb 2026 08:44:20 -1000 Subject: [PATCH 3/3] fga p4 --- src/workos/authorization.py | 193 +++++++++++++++++++ tests/test_authorization_role_assignments.py | 177 +++++++++++++++++ 2 files changed, 370 insertions(+) create mode 100644 tests/test_authorization_role_assignments.py diff --git a/src/workos/authorization.py b/src/workos/authorization.py index eeea21bb..3d794da3 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -1,3 +1,4 @@ +from functools import partial from typing import Any, Dict, Optional, Protocol, Sequence, Union from pydantic import TypeAdapter @@ -11,6 +12,7 @@ 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.authorization.role_assignment import RoleAssignment from workos.types.list_resource import ( ListArgs, ListMetadata, @@ -31,6 +33,10 @@ AUTHORIZATION_PERMISSIONS_PATH = "authorization/permissions" AUTHORIZATION_RESOURCES_PATH = "authorization/resources" +AUTHORIZATION_ROLE_ASSIGNMENTS_PATH = ( + "authorization/organization_memberships" + "/{organization_membership_id}/role_assignments" +) class ParentResourceById(TypedDict): @@ -56,6 +62,15 @@ class PermissionListFilters(ListArgs, total=False): ] +class RoleAssignmentListFilters(ListArgs, total=False): + pass + + +RoleAssignmentsListResource = WorkOSListResource[ + RoleAssignment, RoleAssignmentListFilters, ListMetadata +] + + class AuthorizationModule(Protocol): """Offers methods through the WorkOS Authorization service.""" @@ -206,6 +221,38 @@ def delete_resource( cascade_delete: Optional[bool] = None, ) -> SyncOrAsync[None]: ... + # Role Assignments + + def list_role_assignments( + self, + organization_membership_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> SyncOrAsync[RoleAssignmentsListResource]: ... + + def assign_role( + self, + organization_membership_id: str, + *, + role_slug: str, + ) -> SyncOrAsync[RoleAssignment]: ... + + def remove_role( + self, + organization_membership_id: str, + *, + role_slug: str, + ) -> SyncOrAsync[None]: ... + + def remove_role_assignment( + self, + organization_membership_id: str, + role_assignment_id: str, + ) -> SyncOrAsync[None]: ... + class Authorization(AuthorizationModule): _http_client: SyncHTTPClient @@ -558,6 +605,79 @@ def delete_resource( method=REQUEST_METHOD_DELETE, ) + # Role Assignments + + def list_role_assignments( + self, + organization_membership_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> RoleAssignmentsListResource: + list_params: RoleAssignmentListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + response = self._http_client.request( + AUTHORIZATION_ROLE_ASSIGNMENTS_PATH.format( + organization_membership_id=organization_membership_id + ), + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + RoleAssignment, RoleAssignmentListFilters, ListMetadata + ]( + list_method=partial(self.list_role_assignments, organization_membership_id), + list_args=list_params, + **ListPage[RoleAssignment](**response).model_dump(), + ) + + def assign_role( + self, + organization_membership_id: str, + *, + role_slug: str, + ) -> RoleAssignment: + response = self._http_client.request( + AUTHORIZATION_ROLE_ASSIGNMENTS_PATH.format( + organization_membership_id=organization_membership_id + ), + method=REQUEST_METHOD_POST, + json={"role_slug": role_slug}, + ) + + return RoleAssignment.model_validate(response) + + def remove_role( + self, + organization_membership_id: str, + *, + role_slug: str, + ) -> None: + self._http_client.delete_with_body( + AUTHORIZATION_ROLE_ASSIGNMENTS_PATH.format( + organization_membership_id=organization_membership_id + ), + json={"role_slug": role_slug}, + ) + + def remove_role_assignment( + self, + organization_membership_id: str, + role_assignment_id: str, + ) -> None: + self._http_client.request( + f"{AUTHORIZATION_ROLE_ASSIGNMENTS_PATH.format(organization_membership_id=organization_membership_id)}/{role_assignment_id}", + method=REQUEST_METHOD_DELETE, + ) + class AsyncAuthorization(AuthorizationModule): _http_client: AsyncHTTPClient @@ -909,3 +1029,76 @@ async def delete_resource( f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", method=REQUEST_METHOD_DELETE, ) + + # Role Assignments + + async def list_role_assignments( + self, + organization_membership_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> RoleAssignmentsListResource: + list_params: RoleAssignmentListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + response = await self._http_client.request( + AUTHORIZATION_ROLE_ASSIGNMENTS_PATH.format( + organization_membership_id=organization_membership_id + ), + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + RoleAssignment, RoleAssignmentListFilters, ListMetadata + ]( + list_method=partial(self.list_role_assignments, organization_membership_id), + list_args=list_params, + **ListPage[RoleAssignment](**response).model_dump(), + ) + + async def assign_role( + self, + organization_membership_id: str, + *, + role_slug: str, + ) -> RoleAssignment: + response = await self._http_client.request( + AUTHORIZATION_ROLE_ASSIGNMENTS_PATH.format( + organization_membership_id=organization_membership_id + ), + method=REQUEST_METHOD_POST, + json={"role_slug": role_slug}, + ) + + return RoleAssignment.model_validate(response) + + async def remove_role( + self, + organization_membership_id: str, + *, + role_slug: str, + ) -> None: + await self._http_client.delete_with_body( + AUTHORIZATION_ROLE_ASSIGNMENTS_PATH.format( + organization_membership_id=organization_membership_id + ), + json={"role_slug": role_slug}, + ) + + async def remove_role_assignment( + self, + organization_membership_id: str, + role_assignment_id: str, + ) -> None: + await self._http_client.request( + f"{AUTHORIZATION_ROLE_ASSIGNMENTS_PATH.format(organization_membership_id=organization_membership_id)}/{role_assignment_id}", + method=REQUEST_METHOD_DELETE, + ) diff --git a/tests/test_authorization_role_assignments.py b/tests/test_authorization_role_assignments.py new file mode 100644 index 00000000..621c5784 --- /dev/null +++ b/tests/test_authorization_role_assignments.py @@ -0,0 +1,177 @@ +from typing import Union + +import pytest +from tests.types.test_auto_pagination_function import TestAutoPaginationFunction +from tests.utils.list_resource import list_response_of +from tests.utils.syncify import syncify +from workos.authorization import AsyncAuthorization, Authorization + + +MOCK_ROLE_ASSIGNMENT = { + "object": "role_assignment", + "id": "ra_01ABC", + "role": {"slug": "admin"}, + "resource": { + "id": "res_01ABC", + "external_id": "ext_123", + "resource_type_slug": "document", + }, + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", +} + +MOCK_OM_ID = "om_01MEMBERSHIP" + + +def _mock_role_assignment(id: str) -> dict: + return { + "object": "role_assignment", + "id": id, + "role": {"slug": "admin"}, + "resource": { + "id": "res_01ABC", + "external_id": "ext_123", + "resource_type_slug": "document", + }, + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", + } + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestAuthorizationRoleAssignments: + @pytest.fixture(autouse=True) + def setup(self, module_instance: Union[Authorization, AsyncAuthorization]): + self.http_client = module_instance._http_client + self.authorization = module_instance + + # --- list_role_assignments --- + + def test_list_role_assignments(self, capture_and_mock_http_client_request): + mock_list = list_response_of( + data=[MOCK_ROLE_ASSIGNMENT], + before=None, + after=None, + ) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_list, 200 + ) + + result = syncify(self.authorization.list_role_assignments(MOCK_OM_ID)) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + f"/authorization/organization_memberships/{MOCK_OM_ID}/role_assignments" + ) + assert len(result.data) == 1 + assert result.data[0].id == "ra_01ABC" + assert result.data[0].role.slug == "admin" + + def test_list_role_assignments_empty(self, capture_and_mock_http_client_request): + mock_list = list_response_of(data=[], before=None, after=None) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_list, 200 + ) + + result = syncify(self.authorization.list_role_assignments(MOCK_OM_ID)) + + assert request_kwargs["method"] == "get" + assert len(result.data) == 0 + + def test_list_role_assignments_with_pagination_params( + self, capture_and_mock_http_client_request + ): + mock_list = list_response_of(data=[], before=None, after=None) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_list, 200 + ) + + syncify( + self.authorization.list_role_assignments( + MOCK_OM_ID, + limit=5, + before="before_cursor", + after="after_cursor", + order="asc", + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["params"]["limit"] == 5 + assert request_kwargs["params"]["before"] == "before_cursor" + assert request_kwargs["params"]["after"] == "after_cursor" + assert request_kwargs["params"]["order"] == "asc" + + @pytest.fixture + def mock_role_assignments_multiple_data_pages(self): + data = [_mock_role_assignment(f"ra_{i:03d}") for i in range(40)] + return list_response_of(data=data) + + def test_list_role_assignments_auto_pagination( + self, + mock_role_assignments_multiple_data_pages, + test_auto_pagination: TestAutoPaginationFunction, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.authorization.list_role_assignments, + expected_all_page_data=mock_role_assignments_multiple_data_pages["data"], + list_function_params={ + "organization_membership_id": MOCK_OM_ID, + }, + ) + + # --- assign_role --- + + def test_assign_role(self, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, MOCK_ROLE_ASSIGNMENT, 201 + ) + + result = syncify(self.authorization.assign_role(MOCK_OM_ID, role_slug="admin")) + + assert result.id == "ra_01ABC" + assert result.object == "role_assignment" + assert result.role.slug == "admin" + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith( + f"/authorization/organization_memberships/{MOCK_OM_ID}/role_assignments" + ) + assert request_kwargs["json"] == {"role_slug": "admin"} + + # --- remove_role --- + + def test_remove_role(self, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=204, + headers={"content-type": "text/plain; charset=utf-8"}, + ) + + result = syncify(self.authorization.remove_role(MOCK_OM_ID, role_slug="admin")) + + assert result is None + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + f"/authorization/organization_memberships/{MOCK_OM_ID}/role_assignments" + ) + assert request_kwargs["json"] == {"role_slug": "admin"} + + # --- remove_role_assignment --- + + def test_remove_role_assignment(self, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=204, + headers={"content-type": "text/plain; charset=utf-8"}, + ) + + result = syncify( + self.authorization.remove_role_assignment(MOCK_OM_ID, "ra_01ABC") + ) + + assert result is None + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + f"/authorization/organization_memberships/{MOCK_OM_ID}/role_assignments/ra_01ABC" + )