diff --git a/src/workos/authorization.py b/src/workos/authorization.py index 6e12f035..c470c147 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -1,14 +1,27 @@ -from typing import Any, Dict, Optional, Protocol, Sequence +from enum import Enum +from functools import partial +from typing import Any, Dict, Optional, Protocol, Sequence, Union from pydantic import TypeAdapter +from workos.types.authorization.access_check_response import AccessCheckResponse +from workos.types.authorization.assignment import Assignment from workos.types.authorization.environment_role import ( EnvironmentRole, EnvironmentRoleList, ) +from workos.types.authorization.organization_membership import ( + AuthorizationOrganizationMembership, +) from workos.types.authorization.organization_role import OrganizationRole +from workos.types.authorization.parent_resource_identifier import ( + ParentResourceIdentifier, +) from workos.types.authorization.permission import Permission +from workos.types.authorization.resource_identifier import ResourceIdentifier +from workos.types.authorization.authorization_resource import AuthorizationResource from workos.types.authorization.role import Role, RoleList +from workos.types.authorization.role_assignment import RoleAssignment from workos.types.list_resource import ( ListArgs, ListMetadata, @@ -27,11 +40,65 @@ REQUEST_METHOD_PUT, ) + +class _Unset(Enum): + TOKEN = 0 + + +UNSET: _Unset = _Unset.TOKEN + AUTHORIZATION_PERMISSIONS_PATH = "authorization/permissions" +AUTHORIZATION_RESOURCES_PATH = "authorization/resources" +AUTHORIZATION_ORGANIZATIONS_PATH = "authorization/organizations" +AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH = "authorization/organization_memberships" + + +class ResourceListFilters(ListArgs, total=False): + organization_id: Optional[str] + resource_type_slug: Optional[str] + parent_resource_id: Optional[str] + parent_resource_type_slug: Optional[str] + parent_external_id: Optional[str] + search: Optional[str] + + +AuthorizationResourcesList = WorkOSListResource[ + AuthorizationResource, ResourceListFilters, ListMetadata +] + + +class ResourcesForMembershipListFilters(ListArgs, total=False): + permission_slug: str + + +AuthorizationResourcesForMembershipList = WorkOSListResource[ + AuthorizationResource, ResourcesForMembershipListFilters, ListMetadata +] + + +class AuthorizationOrganizationMembershipListFilters(ListArgs, total=False): + permission_slug: str + assignment: Optional[Assignment] + + +AuthorizationOrganizationMembershipList = WorkOSListResource[ + AuthorizationOrganizationMembership, + AuthorizationOrganizationMembershipListFilters, + ListMetadata, +] _role_adapter: TypeAdapter[Role] = TypeAdapter(Role) +class RoleAssignmentListFilters(ListArgs, total=False): + organization_membership_id: str + + +RoleAssignmentsListResource = WorkOSListResource[ + RoleAssignment, RoleAssignmentListFilters, ListMetadata +] + + class PermissionListFilters(ListArgs, total=False): pass @@ -161,6 +228,155 @@ def add_environment_role_permission( permission_slug: str, ) -> SyncOrAsync[EnvironmentRole]: ... + # Resources + + def get_resource(self, resource_id: str) -> SyncOrAsync[AuthorizationResource]: ... + + def create_resource( + self, + *, + external_id: str, + name: str, + description: Optional[str] = None, + resource_type_slug: str, + organization_id: str, + parent: Optional[ParentResourceIdentifier] = None, + ) -> SyncOrAsync[AuthorizationResource]: ... + + def update_resource( + self, + resource_id: str, + *, + name: Optional[str] = None, + description: Union[str, None, _Unset] = UNSET, + ) -> SyncOrAsync[AuthorizationResource]: ... + + def delete_resource( + self, + resource_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> SyncOrAsync[None]: ... + + def list_resources( + self, + *, + organization_id: Optional[str] = None, + resource_type_slug: Optional[str] = None, + parent_resource_id: Optional[str] = None, + parent_resource_type_slug: Optional[str] = None, + parent_external_id: Optional[str] = None, + search: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> SyncOrAsync[AuthorizationResourcesList]: ... + + def get_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + ) -> SyncOrAsync[AuthorizationResource]: ... + + def update_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + name: Optional[str] = None, + description: Union[str, None, _Unset] = UNSET, + ) -> SyncOrAsync[AuthorizationResource]: ... + + def delete_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> SyncOrAsync[None]: ... + + def check( + self, + organization_membership_id: str, + *, + permission_slug: str, + resource: ResourceIdentifier, + ) -> SyncOrAsync[AccessCheckResponse]: ... + + def assign_role( + self, + organization_membership_id: str, + *, + role_slug: str, + resource_identifier: ResourceIdentifier, + ) -> SyncOrAsync[RoleAssignment]: ... + + def remove_role( + self, + organization_membership_id: str, + *, + role_slug: str, + resource_identifier: ResourceIdentifier, + ) -> SyncOrAsync[None]: ... + + def remove_role_assignment( + self, + organization_membership_id: str, + role_assignment_id: str, + ) -> SyncOrAsync[None]: ... + + 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 list_resources_for_membership( + self, + organization_membership_id: str, + *, + permission_slug: str, + parent_resource: ParentResourceIdentifier, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> SyncOrAsync[AuthorizationResourcesForMembershipList]: ... + + def list_memberships_for_resource( + self, + resource_id: str, + *, + permission_slug: str, + assignment: Optional[Assignment] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> SyncOrAsync[AuthorizationOrganizationMembershipList]: ... + + def list_memberships_for_resource_by_external_id( + self, + organization_id: str, + resource_type_slug: str, + external_id: str, + *, + permission_slug: str, + assignment: Optional[Assignment] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> SyncOrAsync[AuthorizationOrganizationMembershipList]: ... + class Authorization(AuthorizationModule): _http_client: SyncHTTPClient @@ -437,179 +653,579 @@ def add_environment_role_permission( return EnvironmentRole.model_validate(response) + def get_resource(self, resource_id: str) -> AuthorizationResource: + response = self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_GET, + ) -class AsyncAuthorization(AuthorizationModule): - _http_client: AsyncHTTPClient - - def __init__(self, http_client: AsyncHTTPClient): - self._http_client = http_client + return AuthorizationResource.model_validate(response) - async def create_permission( + def create_resource( self, *, - slug: str, + external_id: str, name: str, description: Optional[str] = None, - ) -> Permission: - json: Dict[str, Any] = {"slug": slug, "name": name} + resource_type_slug: str, + organization_id: str, + parent: Optional[ParentResourceIdentifier] = None, + ) -> AuthorizationResource: + json: Dict[str, Any] = { + "resource_type_slug": resource_type_slug, + "organization_id": organization_id, + "external_id": external_id, + "name": name, + } + if parent is not None: + json.update(parent) if description is not None: json["description"] = description - response = await self._http_client.request( - AUTHORIZATION_PERMISSIONS_PATH, + response = self._http_client.request( + AUTHORIZATION_RESOURCES_PATH, method=REQUEST_METHOD_POST, json=json, ) - return Permission.model_validate(response) + return AuthorizationResource.model_validate(response) - async def list_permissions( + def update_resource( self, + resource_id: str, *, + name: Optional[str] = None, + description: Union[str, None, _Unset] = UNSET, + ) -> AuthorizationResource: + json: Dict[str, Any] = {} + if name is not None: + json["name"] = name + if not isinstance(description, _Unset): + json["description"] = description + + response = self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_PATCH, + json=json, + exclude_none=False, + ) + + return AuthorizationResource.model_validate(response) + + def delete_resource( + self, + resource_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> None: + params = ( + {"cascade_delete": str(cascade_delete).lower()} + if cascade_delete is not None + else None + ) + self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_DELETE, + params=params, + ) + + def list_resources( + self, + *, + organization_id: Optional[str] = None, + resource_type_slug: Optional[str] = None, + parent_resource_id: Optional[str] = None, + parent_resource_type_slug: Optional[str] = None, + parent_external_id: Optional[str] = None, + search: Optional[str] = None, limit: int = DEFAULT_LIST_RESPONSE_LIMIT, before: Optional[str] = None, after: Optional[str] = None, order: PaginationOrder = "desc", - ) -> PermissionsListResource: - list_params: PermissionListFilters = { + ) -> AuthorizationResourcesList: + list_params: ResourceListFilters = { "limit": limit, "before": before, "after": after, "order": order, } + if organization_id is not None: + list_params["organization_id"] = organization_id + if resource_type_slug is not None: + list_params["resource_type_slug"] = resource_type_slug + if parent_resource_id is not None: + list_params["parent_resource_id"] = parent_resource_id + if parent_resource_type_slug is not None: + list_params["parent_resource_type_slug"] = parent_resource_type_slug + if parent_external_id is not None: + list_params["parent_external_id"] = parent_external_id + if search is not None: + list_params["search"] = search - response = await self._http_client.request( - AUTHORIZATION_PERMISSIONS_PATH, + response = self._http_client.request( + AUTHORIZATION_RESOURCES_PATH, method=REQUEST_METHOD_GET, params=list_params, ) - return WorkOSListResource[Permission, PermissionListFilters, ListMetadata]( - list_method=self.list_permissions, + return WorkOSListResource[ + AuthorizationResource, ResourceListFilters, ListMetadata + ]( + list_method=self.list_resources, list_args=list_params, - **ListPage[Permission](**response).model_dump(), + **ListPage[AuthorizationResource](**response).model_dump(), ) - async def get_permission(self, slug: str) -> Permission: - response = await self._http_client.request( - f"{AUTHORIZATION_PERMISSIONS_PATH}/{slug}", + def get_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + ) -> AuthorizationResource: + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}", method=REQUEST_METHOD_GET, ) - return Permission.model_validate(response) + return AuthorizationResource.model_validate(response) - async def update_permission( + def update_resource_by_external_id( self, - slug: str, + organization_id: str, + resource_type: str, + external_id: str, *, name: Optional[str] = None, - description: Optional[str] = None, - ) -> Permission: + description: Union[str, None, _Unset] = UNSET, + ) -> AuthorizationResource: json: Dict[str, Any] = {} if name is not None: json["name"] = name - if description is not None: + if not isinstance(description, _Unset): json["description"] = description - response = await self._http_client.request( - f"{AUTHORIZATION_PERMISSIONS_PATH}/{slug}", + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}", method=REQUEST_METHOD_PATCH, json=json, + exclude_none=False, ) - return Permission.model_validate(response) + return AuthorizationResource.model_validate(response) - async def delete_permission(self, slug: str) -> None: - await self._http_client.request( - f"{AUTHORIZATION_PERMISSIONS_PATH}/{slug}", + def delete_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> None: + path = f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}" + params = ( + {"cascade_delete": str(cascade_delete).lower()} + if cascade_delete is not None + else None + ) + self._http_client.request( + path, method=REQUEST_METHOD_DELETE, + params=params, ) - # Organization Roles - - async def create_organization_role( + def check( self, - organization_id: str, + organization_membership_id: str, *, - slug: str, - name: str, - description: Optional[str] = None, - ) -> OrganizationRole: - json: Dict[str, Any] = {"slug": slug, "name": name} - if description is not None: - json["description"] = description + permission_slug: str, + resource: ResourceIdentifier, + ) -> AccessCheckResponse: + json: Dict[str, Any] = {"permission_slug": permission_slug} + json.update(resource) - response = await self._http_client.request( - f"authorization/organizations/{organization_id}/roles", + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/check", method=REQUEST_METHOD_POST, json=json, ) - return OrganizationRole.model_validate(response) + return AccessCheckResponse.model_validate(response) - async def list_organization_roles(self, organization_id: str) -> RoleList: - response = await self._http_client.request( - f"authorization/organizations/{organization_id}/roles", - method=REQUEST_METHOD_GET, - ) + # Role Assignments - return RoleList.model_validate(response) + def assign_role( + self, + organization_membership_id: str, + *, + role_slug: str, + resource_identifier: ResourceIdentifier, + ) -> RoleAssignment: + json: Dict[str, Any] = {"role_slug": role_slug} + json.update(resource_identifier) - async def get_organization_role(self, organization_id: str, slug: str) -> Role: - response = await self._http_client.request( - f"authorization/organizations/{organization_id}/roles/{slug}", - method=REQUEST_METHOD_GET, + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/role_assignments", + method=REQUEST_METHOD_POST, + json=json, ) - return _role_adapter.validate_python(response) + return RoleAssignment.model_validate(response) - async def update_organization_role( + def remove_role( self, - organization_id: str, - slug: str, + organization_membership_id: str, *, - name: Optional[str] = None, - description: Optional[str] = None, - ) -> OrganizationRole: - json: Dict[str, Any] = {} - if name is not None: - json["name"] = name - if description is not None: - json["description"] = description + role_slug: str, + resource_identifier: ResourceIdentifier, + ) -> None: + json: Dict[str, Any] = {"role_slug": role_slug} + json.update(resource_identifier) - response = await self._http_client.request( - f"authorization/organizations/{organization_id}/roles/{slug}", - method=REQUEST_METHOD_PATCH, + self._http_client.delete_with_body( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/role_assignments", json=json, ) - return OrganizationRole.model_validate(response) - - async def set_organization_role_permissions( + def remove_role_assignment( self, - organization_id: str, - slug: str, - *, - permissions: Sequence[str], - ) -> OrganizationRole: - response = await self._http_client.request( - f"authorization/organizations/{organization_id}/roles/{slug}/permissions", - method=REQUEST_METHOD_PUT, - json={"permissions": list(permissions)}, + organization_membership_id: str, + role_assignment_id: str, + ) -> None: + self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/role_assignments/{role_assignment_id}", + method=REQUEST_METHOD_DELETE, ) - return OrganizationRole.model_validate(response) - - async def add_organization_role_permission( + def list_role_assignments( self, - organization_id: str, - slug: str, *, - permission_slug: str, - ) -> OrganizationRole: - response = await self._http_client.request( - f"authorization/organizations/{organization_id}/roles/{slug}/permissions", - method=REQUEST_METHOD_POST, + 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 = { + "organization_membership_id": organization_membership_id, + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + query_params: ListArgs = { + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/role_assignments", + method=REQUEST_METHOD_GET, + params=query_params, + ) + + return WorkOSListResource[ + RoleAssignment, RoleAssignmentListFilters, ListMetadata + ]( + list_method=self.list_role_assignments, + list_args=list_params, + **ListPage[RoleAssignment](**response).model_dump(), + ) + + def list_resources_for_membership( + self, + organization_membership_id: str, + *, + permission_slug: str, + parent_resource: ParentResourceIdentifier, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> AuthorizationResourcesForMembershipList: + list_params: ResourcesForMembershipListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + "permission_slug": permission_slug, + } + + http_params: Dict[str, Any] = {**list_params} + http_params.update(parent_resource) + + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/resources", + method=REQUEST_METHOD_GET, + params=http_params, + ) + + return AuthorizationResourcesForMembershipList( + list_method=partial( + self.list_resources_for_membership, + organization_membership_id, + parent_resource=parent_resource, + ), + list_args=list_params, + **ListPage[AuthorizationResource](**response).model_dump(), + ) + + def list_memberships_for_resource( + self, + resource_id: str, + *, + permission_slug: str, + assignment: Optional[Assignment] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> AuthorizationOrganizationMembershipList: + list_params: AuthorizationOrganizationMembershipListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + "permission_slug": permission_slug, + } + if assignment is not None: + list_params["assignment"] = assignment + + response = self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + AuthorizationOrganizationMembership, + AuthorizationOrganizationMembershipListFilters, + ListMetadata, + ]( + list_method=partial(self.list_memberships_for_resource, resource_id), + list_args=list_params, + **ListPage[AuthorizationOrganizationMembership](**response).model_dump(), + ) + + def list_memberships_for_resource_by_external_id( + self, + organization_id: str, + resource_type_slug: str, + external_id: str, + *, + permission_slug: str, + assignment: Optional[Assignment] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> AuthorizationOrganizationMembershipList: + list_params: AuthorizationOrganizationMembershipListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + "permission_slug": permission_slug, + } + if assignment is not None: + list_params["assignment"] = assignment + + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type_slug}/{external_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + AuthorizationOrganizationMembership, + AuthorizationOrganizationMembershipListFilters, + ListMetadata, + ]( + list_method=partial( + self.list_memberships_for_resource_by_external_id, + organization_id, + resource_type_slug, + external_id, + ), + list_args=list_params, + **ListPage[AuthorizationOrganizationMembership](**response).model_dump(), + ) + + +class AsyncAuthorization(AuthorizationModule): + _http_client: AsyncHTTPClient + + def __init__(self, http_client: AsyncHTTPClient): + self._http_client = http_client + + async def create_permission( + self, + *, + slug: str, + name: str, + description: Optional[str] = None, + ) -> Permission: + json: Dict[str, Any] = {"slug": slug, "name": name} + if description is not None: + json["description"] = description + + response = await self._http_client.request( + AUTHORIZATION_PERMISSIONS_PATH, + method=REQUEST_METHOD_POST, + json=json, + ) + + return Permission.model_validate(response) + + async def list_permissions( + self, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> PermissionsListResource: + list_params: PermissionListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + response = await self._http_client.request( + AUTHORIZATION_PERMISSIONS_PATH, + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[Permission, PermissionListFilters, ListMetadata]( + list_method=self.list_permissions, + list_args=list_params, + **ListPage[Permission](**response).model_dump(), + ) + + async def get_permission(self, slug: str) -> Permission: + response = await self._http_client.request( + f"{AUTHORIZATION_PERMISSIONS_PATH}/{slug}", + method=REQUEST_METHOD_GET, + ) + + return Permission.model_validate(response) + + async def update_permission( + self, + slug: str, + *, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> Permission: + 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_PERMISSIONS_PATH}/{slug}", + method=REQUEST_METHOD_PATCH, + json=json, + ) + + return Permission.model_validate(response) + + async def delete_permission(self, slug: str) -> None: + await self._http_client.request( + f"{AUTHORIZATION_PERMISSIONS_PATH}/{slug}", + method=REQUEST_METHOD_DELETE, + ) + + # Organization Roles + + async def create_organization_role( + self, + organization_id: str, + *, + slug: str, + name: str, + description: Optional[str] = None, + ) -> OrganizationRole: + json: Dict[str, Any] = {"slug": slug, "name": name} + if description is not None: + json["description"] = description + + response = await self._http_client.request( + f"authorization/organizations/{organization_id}/roles", + method=REQUEST_METHOD_POST, + json=json, + ) + + return OrganizationRole.model_validate(response) + + async def list_organization_roles(self, organization_id: str) -> RoleList: + response = await self._http_client.request( + f"authorization/organizations/{organization_id}/roles", + method=REQUEST_METHOD_GET, + ) + + return RoleList.model_validate(response) + + async def get_organization_role(self, organization_id: str, slug: str) -> Role: + response = await self._http_client.request( + f"authorization/organizations/{organization_id}/roles/{slug}", + method=REQUEST_METHOD_GET, + ) + + return _role_adapter.validate_python(response) + + async def update_organization_role( + self, + organization_id: str, + slug: str, + *, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> OrganizationRole: + 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/organizations/{organization_id}/roles/{slug}", + method=REQUEST_METHOD_PATCH, + json=json, + ) + + return OrganizationRole.model_validate(response) + + async def set_organization_role_permissions( + self, + organization_id: str, + slug: str, + *, + permissions: Sequence[str], + ) -> OrganizationRole: + response = await self._http_client.request( + f"authorization/organizations/{organization_id}/roles/{slug}/permissions", + method=REQUEST_METHOD_PUT, + json={"permissions": list(permissions)}, + ) + + return OrganizationRole.model_validate(response) + + async def add_organization_role_permission( + self, + organization_id: str, + slug: str, + *, + permission_slug: str, + ) -> OrganizationRole: + response = await self._http_client.request( + f"authorization/organizations/{organization_id}/roles/{slug}/permissions", + method=REQUEST_METHOD_POST, json={"slug": permission_slug}, ) @@ -712,3 +1328,405 @@ async def add_environment_role_permission( ) return EnvironmentRole.model_validate(response) + + # Resources + + async def get_resource(self, resource_id: str) -> AuthorizationResource: + response = await self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_GET, + ) + + return AuthorizationResource.model_validate(response) + + async def create_resource( + self, + *, + external_id: str, + name: str, + description: Optional[str] = None, + resource_type_slug: str, + organization_id: str, + parent: Optional[ParentResourceIdentifier] = None, + ) -> AuthorizationResource: + json: Dict[str, Any] = { + "resource_type_slug": resource_type_slug, + "organization_id": organization_id, + "external_id": external_id, + "name": name, + } + if parent is not None: + json.update(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 AuthorizationResource.model_validate(response) + + async def update_resource( + self, + resource_id: str, + *, + name: Optional[str] = None, + description: Union[str, None, _Unset] = UNSET, + ) -> AuthorizationResource: + json: Dict[str, Any] = {} + if name is not None: + json["name"] = name + if not isinstance(description, _Unset): + json["description"] = description + + response = await self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_PATCH, + json=json, + exclude_none=False, + ) + + return AuthorizationResource.model_validate(response) + + async def delete_resource( + self, + resource_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> None: + params = ( + {"cascade_delete": str(cascade_delete).lower()} + if cascade_delete is not None + else None + ) + await self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}", + method=REQUEST_METHOD_DELETE, + params=params, + ) + + async def list_resources( + self, + *, + organization_id: Optional[str] = None, + resource_type_slug: Optional[str] = None, + parent_resource_id: Optional[str] = None, + parent_resource_type_slug: Optional[str] = None, + parent_external_id: Optional[str] = None, + search: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> AuthorizationResourcesList: + list_params: ResourceListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + } + if organization_id is not None: + list_params["organization_id"] = organization_id + if resource_type_slug is not None: + list_params["resource_type_slug"] = resource_type_slug + if parent_resource_id is not None: + list_params["parent_resource_id"] = parent_resource_id + if parent_resource_type_slug is not None: + list_params["parent_resource_type_slug"] = parent_resource_type_slug + if parent_external_id is not None: + list_params["parent_external_id"] = parent_external_id + if search is not None: + list_params["search"] = search + + response = await self._http_client.request( + AUTHORIZATION_RESOURCES_PATH, + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + AuthorizationResource, ResourceListFilters, ListMetadata + ]( + list_method=self.list_resources, + list_args=list_params, + **ListPage[AuthorizationResource](**response).model_dump(), + ) + + async def get_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + ) -> AuthorizationResource: + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}", + method=REQUEST_METHOD_GET, + ) + + return AuthorizationResource.model_validate(response) + + async def update_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + name: Optional[str] = None, + description: Union[str, None, _Unset] = UNSET, + ) -> AuthorizationResource: + json: Dict[str, Any] = {} + if name is not None: + json["name"] = name + if not isinstance(description, _Unset): + json["description"] = description + + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}", + method=REQUEST_METHOD_PATCH, + json=json, + exclude_none=False, + ) + + return AuthorizationResource.model_validate(response) + + async def delete_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> None: + path = f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}" + params = ( + {"cascade_delete": str(cascade_delete).lower()} + if cascade_delete is not None + else None + ) + await self._http_client.request( + path, + method=REQUEST_METHOD_DELETE, + params=params, + ) + + async def check( + self, + organization_membership_id: str, + *, + permission_slug: str, + resource: ResourceIdentifier, + ) -> AccessCheckResponse: + json: Dict[str, Any] = {"permission_slug": permission_slug} + json.update(resource) + + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/check", + method=REQUEST_METHOD_POST, + json=json, + ) + + return AccessCheckResponse.model_validate(response) + + # Role Assignments + + async def assign_role( + self, + organization_membership_id: str, + *, + role_slug: str, + resource_identifier: ResourceIdentifier, + ) -> RoleAssignment: + json: Dict[str, Any] = {"role_slug": role_slug} + json.update(resource_identifier) + + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/role_assignments", + method=REQUEST_METHOD_POST, + json=json, + ) + + return RoleAssignment.model_validate(response) + + async def remove_role( + self, + organization_membership_id: str, + *, + role_slug: str, + resource_identifier: ResourceIdentifier, + ) -> None: + json: Dict[str, Any] = {"role_slug": role_slug} + json.update(resource_identifier) + + await self._http_client.delete_with_body( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/role_assignments", + json=json, + ) + + async def remove_role_assignment( + self, + organization_membership_id: str, + role_assignment_id: str, + ) -> None: + await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/role_assignments/{role_assignment_id}", + method=REQUEST_METHOD_DELETE, + ) + + 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 = { + "organization_membership_id": organization_membership_id, + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + query_params: ListArgs = { + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/role_assignments", + method=REQUEST_METHOD_GET, + params=query_params, + ) + + return WorkOSListResource[ + RoleAssignment, RoleAssignmentListFilters, ListMetadata + ]( + list_method=self.list_role_assignments, + list_args=list_params, + **ListPage[RoleAssignment](**response).model_dump(), + ) + + async def list_resources_for_membership( + self, + organization_membership_id: str, + *, + permission_slug: str, + parent_resource: ParentResourceIdentifier, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> AuthorizationResourcesForMembershipList: + list_params: ResourcesForMembershipListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + "permission_slug": permission_slug, + } + + http_params: Dict[str, Any] = {**list_params} + http_params.update(parent_resource) + + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/resources", + method=REQUEST_METHOD_GET, + params=http_params, + ) + + return AuthorizationResourcesForMembershipList( + list_method=partial( + self.list_resources_for_membership, + organization_membership_id, + parent_resource=parent_resource, + ), + list_args=list_params, + **ListPage[AuthorizationResource](**response).model_dump(), + ) + + async def list_memberships_for_resource( + self, + resource_id: str, + *, + permission_slug: str, + assignment: Optional[Assignment] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> AuthorizationOrganizationMembershipList: + list_params: AuthorizationOrganizationMembershipListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + "permission_slug": permission_slug, + } + if assignment is not None: + list_params["assignment"] = assignment + + response = await self._http_client.request( + f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + AuthorizationOrganizationMembership, + AuthorizationOrganizationMembershipListFilters, + ListMetadata, + ]( + list_method=partial(self.list_memberships_for_resource, resource_id), + list_args=list_params, + **ListPage[AuthorizationOrganizationMembership](**response).model_dump(), + ) + + async def list_memberships_for_resource_by_external_id( + self, + organization_id: str, + resource_type_slug: str, + external_id: str, + *, + permission_slug: str, + assignment: Optional[Assignment] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> AuthorizationOrganizationMembershipList: + list_params: AuthorizationOrganizationMembershipListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + "permission_slug": permission_slug, + } + if assignment is not None: + list_params["assignment"] = assignment + + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type_slug}/{external_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + AuthorizationOrganizationMembership, + AuthorizationOrganizationMembershipListFilters, + ListMetadata, + ]( + list_method=partial( + self.list_memberships_for_resource_by_external_id, + organization_id, + resource_type_slug, + external_id, + ), + list_args=list_params, + **ListPage[AuthorizationOrganizationMembership](**response).model_dump(), + ) diff --git a/src/workos/fga.py b/src/workos/fga.py index 2c76a320..4a1b7332 100644 --- a/src/workos/fga.py +++ b/src/workos/fga.py @@ -53,6 +53,8 @@ class WarrantQueryListResource( warnings: Optional[Sequence[FGAWarning]] = None +# Deprecated: Use the Authorization module instead. +# See: workos.authorization class FGAModule(Protocol): def get_resource( self, *, resource_type: str, resource_id: str diff --git a/src/workos/types/authorization/__init__.py b/src/workos/types/authorization/__init__.py index 609a4f2d..a6c93040 100644 --- a/src/workos/types/authorization/__init__.py +++ b/src/workos/types/authorization/__init__.py @@ -1,14 +1,32 @@ +from workos.types.authorization.access_check_response import AccessCheckResponse from workos.types.authorization.environment_role import ( EnvironmentRole, EnvironmentRoleList, ) +from workos.types.authorization.organization_membership import ( + AuthorizationOrganizationMembership, +) from workos.types.authorization.organization_role import ( OrganizationRole, OrganizationRoleEvent, OrganizationRoleList, ) from workos.types.authorization.permission import Permission +from workos.types.authorization.parent_resource_identifier import ( + ParentResourceIdentifier, +) +from workos.types.authorization.authorization_resource import AuthorizationResource +from workos.types.authorization.resource_identifier import ( + ResourceIdentifier, + ResourceIdentifierByExternalId, + ResourceIdentifierById, +) from workos.types.authorization.role import ( Role, RoleList, ) +from workos.types.authorization.role_assignment import ( + RoleAssignment, + RoleAssignmentResource, + RoleAssignmentRole, +) diff --git a/src/workos/types/authorization/access_check_response.py b/src/workos/types/authorization/access_check_response.py new file mode 100644 index 00000000..2515b763 --- /dev/null +++ b/src/workos/types/authorization/access_check_response.py @@ -0,0 +1,5 @@ +from workos.types.workos_model import WorkOSModel + + +class AccessCheckResponse(WorkOSModel): + authorized: bool diff --git a/src/workos/types/authorization/assignment.py b/src/workos/types/authorization/assignment.py new file mode 100644 index 00000000..ea87fca9 --- /dev/null +++ b/src/workos/types/authorization/assignment.py @@ -0,0 +1,3 @@ +from typing import Literal + +Assignment = Literal["direct", "indirect"] diff --git a/src/workos/types/authorization/authorization_resource.py b/src/workos/types/authorization/authorization_resource.py new file mode 100644 index 00000000..ea722f9b --- /dev/null +++ b/src/workos/types/authorization/authorization_resource.py @@ -0,0 +1,18 @@ +from typing import Literal, Optional + +from workos.types.workos_model import WorkOSModel + + +class AuthorizationResource(WorkOSModel): + """Representation of an Authorization Resource.""" + + object: Literal["authorization_resource"] + id: str + external_id: str + name: str + description: Optional[str] = None + resource_type_slug: str + organization_id: str + parent_resource_id: Optional[str] = None + created_at: str + updated_at: str diff --git a/src/workos/types/authorization/organization_membership.py b/src/workos/types/authorization/organization_membership.py new file mode 100644 index 00000000..f24e9a27 --- /dev/null +++ b/src/workos/types/authorization/organization_membership.py @@ -0,0 +1,5 @@ +from workos.types.user_management.organization_membership import ( + BaseOrganizationMembership, +) + +AuthorizationOrganizationMembership = BaseOrganizationMembership diff --git a/src/workos/types/authorization/parent_resource_identifier.py b/src/workos/types/authorization/parent_resource_identifier.py new file mode 100644 index 00000000..c105f87b --- /dev/null +++ b/src/workos/types/authorization/parent_resource_identifier.py @@ -0,0 +1,15 @@ +from typing import Union + +from typing_extensions import TypedDict + + +class ParentResourceById(TypedDict): + parent_resource_id: str + + +class ParentResourceByExternalId(TypedDict): + parent_resource_external_id: str + parent_resource_type_slug: str + + +ParentResourceIdentifier = Union[ParentResourceById, ParentResourceByExternalId] diff --git a/src/workos/types/authorization/resource_identifier.py b/src/workos/types/authorization/resource_identifier.py new file mode 100644 index 00000000..081a175d --- /dev/null +++ b/src/workos/types/authorization/resource_identifier.py @@ -0,0 +1,15 @@ +from typing import Union + +from typing_extensions import TypedDict + + +class ResourceIdentifierById(TypedDict): + resource_id: str + + +class ResourceIdentifierByExternalId(TypedDict): + resource_external_id: str + resource_type_slug: str + + +ResourceIdentifier = Union[ResourceIdentifierById, ResourceIdentifierByExternalId] diff --git a/src/workos/types/authorization/role_assignment.py b/src/workos/types/authorization/role_assignment.py new file mode 100644 index 00000000..9ca59936 --- /dev/null +++ b/src/workos/types/authorization/role_assignment.py @@ -0,0 +1,22 @@ +from typing import Literal + +from workos.types.workos_model import WorkOSModel + + +class RoleAssignmentRole(WorkOSModel): + slug: str + + +class RoleAssignmentResource(WorkOSModel): + id: str + external_id: str + resource_type_slug: str + + +class RoleAssignment(WorkOSModel): + object: Literal["role_assignment"] + id: str + role: RoleAssignmentRole + resource: RoleAssignmentResource + created_at: str + updated_at: str diff --git a/src/workos/types/list_resource.py b/src/workos/types/list_resource.py index e8621ab9..e2f21841 100644 --- a/src/workos/types/list_resource.py +++ b/src/workos/types/list_resource.py @@ -19,7 +19,12 @@ from typing_extensions import Required, TypedDict from workos.types.api_keys import ApiKey from workos.types.audit_logs import AuditLogAction, AuditLogSchema +from workos.types.authorization.organization_membership import ( + AuthorizationOrganizationMembership, +) from workos.types.authorization.permission import Permission +from workos.types.authorization.authorization_resource import AuthorizationResource +from workos.types.authorization.role_assignment import RoleAssignment from workos.types.directory_sync import ( Directory, DirectoryGroup, @@ -28,9 +33,9 @@ from workos.types.events import Event from workos.types.feature_flags import FeatureFlag from workos.types.fga import ( - Warrant, - AuthorizationResource, + AuthorizationResource as FGAAuthorizationResource, AuthorizationResourceType, + Warrant, WarrantQueryResult, ) from workos.types.mfa import AuthenticationFactor @@ -60,6 +65,9 @@ OrganizationMembership, Permission, AuthorizationResource, + RoleAssignment, + AuthorizationOrganizationMembership, + FGAAuthorizationResource, AuthorizationResourceType, User, UserManagementSession, diff --git a/src/workos/types/user_management/organization_membership.py b/src/workos/types/user_management/organization_membership.py index 67ba1f21..be90124b 100644 --- a/src/workos/types/user_management/organization_membership.py +++ b/src/workos/types/user_management/organization_membership.py @@ -8,20 +8,23 @@ OrganizationMembershipStatus = Literal["active", "inactive", "pending"] +class BaseOrganizationMembership(WorkOSModel): + object: Literal["organization_membership"] + id: str + user_id: str + organization_id: str + status: LiteralOrUntyped[OrganizationMembershipStatus] + created_at: str + updated_at: str + + class OrganizationMembershipRole(TypedDict): slug: str -class OrganizationMembership(WorkOSModel): +class OrganizationMembership(BaseOrganizationMembership): """Representation of an WorkOS Organization Membership.""" - object: Literal["organization_membership"] - id: str - user_id: str - organization_id: str role: OrganizationMembershipRole roles: Optional[Sequence[OrganizationMembershipRole]] = None - status: LiteralOrUntyped[OrganizationMembershipStatus] custom_attributes: Mapping[str, Any] = {} - created_at: str - updated_at: str diff --git a/src/workos/utils/_base_http_client.py b/src/workos/utils/_base_http_client.py index 49dcbcf5..3bcddf32 100644 --- a/src/workos/utils/_base_http_client.py +++ b/src/workos/utils/_base_http_client.py @@ -123,6 +123,8 @@ def _prepare_request( json: JsonType = None, headers: HeadersType = None, exclude_default_auth_headers: bool = False, + force_include_body: bool = False, + exclude_none: bool = True, ) -> PreparedRequest: """Executes a request against the WorkOS API. @@ -133,8 +135,9 @@ def _prepare_request( method Optional[str]: One of the supported methods as defined by the REQUEST_METHOD_X constants params Optional[dict]: Query params or body payload to be added to the request headers Optional[dict]: Custom headers to be added to the request - token Optional[str]: Bearer token - + exclude_default_auth_headers (bool): If True, excludes default auth headers from the request + force_include_body (bool): If True, allows sending a body in a bodyless request (used for DELETE requests) + exclude_none (bool): If True (default), strips keys with None values from the JSON body so only defined fields are sent. Returns: dict: Response from WorkOS """ @@ -149,7 +152,7 @@ def _prepare_request( REQUEST_METHOD_GET, ] - if bodyless_http_method and json is not None: + if bodyless_http_method and json is not None and not force_include_body: raise ValueError(f"Cannot send a body with a {parsed_method} request") # Remove any parameters that are None @@ -157,11 +160,11 @@ def _prepare_request( params = {k: v for k, v in params.items() if v is not None} # Remove any body values that are None - if json is not None and isinstance(json, Mapping): + if exclude_none and json is not None and isinstance(json, Mapping): json = {k: v for k, v in json.items() if v is not None} # We'll spread these return values onto the HTTP client request method - if bodyless_http_method: + if bodyless_http_method and not force_include_body: return { "method": parsed_method, "url": url, diff --git a/src/workos/utils/http_client.py b/src/workos/utils/http_client.py index 203c7df0..5c7deac5 100644 --- a/src/workos/utils/http_client.py +++ b/src/workos/utils/http_client.py @@ -14,7 +14,7 @@ ParamsType, ResponseJson, ) -from workos.utils.request_helper import REQUEST_METHOD_GET +from workos.utils.request_helper import REQUEST_METHOD_DELETE, REQUEST_METHOD_GET class SyncHttpxClientWrapper(httpx.Client): @@ -88,6 +88,7 @@ def request( json: JsonType = None, headers: HeadersType = None, exclude_default_auth_headers: bool = False, + exclude_none: bool = True, ) -> ResponseJson: """Executes a request against the WorkOS API. @@ -98,6 +99,7 @@ def request( method (str): One of the supported methods as defined by the REQUEST_METHOD_X constants params (ParamsType): Query params to be added to the request json (JsonType): Body payload to be added to the request + exclude_none (bool): If True, removes None values from the JSON body Returns: ResponseJson: Response from WorkOS @@ -109,6 +111,28 @@ def request( json=json, headers=headers, exclude_default_auth_headers=exclude_default_auth_headers, + exclude_none=exclude_none, + ) + response = self._client.request(**prepared_request_parameters) + return self._handle_response(response) + + def delete_with_body( + self, + path: str, + json: JsonType = None, + params: ParamsType = None, + headers: HeadersType = None, + exclude_default_auth_headers: bool = False, + ) -> ResponseJson: + """Executes a DELETE request with a JSON body against the WorkOS API.""" + prepared_request_parameters = self._prepare_request( + path=path, + method=REQUEST_METHOD_DELETE, + json=json, + params=params, + headers=headers, + exclude_default_auth_headers=exclude_default_auth_headers, + force_include_body=True, ) response = self._client.request(**prepared_request_parameters) return self._handle_response(response) @@ -185,6 +209,7 @@ async def request( json: JsonType = None, headers: HeadersType = None, exclude_default_auth_headers: bool = False, + exclude_none: bool = True, ) -> ResponseJson: """Executes a request against the WorkOS API. @@ -195,6 +220,7 @@ async def request( method (str): One of the supported methods as defined by the REQUEST_METHOD_X constants params (ParamsType): Query params to be added to the request json (JsonType): Body payload to be added to the request + exclude_none (bool): If True, removes None values from the JSON body Returns: ResponseJson: Response from WorkOS @@ -206,6 +232,28 @@ async def request( json=json, headers=headers, exclude_default_auth_headers=exclude_default_auth_headers, + exclude_none=exclude_none, + ) + response = await self._client.request(**prepared_request_parameters) + return self._handle_response(response) + + async def delete_with_body( + self, + path: str, + json: JsonType = None, + params: ParamsType = None, + headers: HeadersType = None, + exclude_default_auth_headers: bool = False, + ) -> ResponseJson: + """Executes a DELETE request with a JSON body against the WorkOS API.""" + prepared_request_parameters = self._prepare_request( + path=path, + method=REQUEST_METHOD_DELETE, + json=json, + params=params, + headers=headers, + exclude_default_auth_headers=exclude_default_auth_headers, + force_include_body=True, ) response = await self._client.request(**prepared_request_parameters) return self._handle_response(response) diff --git a/tests/test_async_http_client.py b/tests/test_async_http_client.py index 633ed71a..b842c51f 100644 --- a/tests/test_async_http_client.py +++ b/tests/test_async_http_client.py @@ -313,6 +313,7 @@ async def test_request_removes_none_parameter_values( method="get", params={"organization_id": None, "test": "value"}, ) + assert request_kwargs["params"] == {"test": "value"} async def test_request_removes_none_json_values( @@ -325,4 +326,43 @@ async def test_request_removes_none_json_values( method="post", json={"organization_id": None, "test": "value"}, ) + + assert request_kwargs["url"].endswith("/test") assert request_kwargs["json"] == {"test": "value"} + + async def test_delete_with_body_sends_json( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request(self.http_client, {}, 200) + + await self.http_client.delete_with_body( + path="/test", + json={"obj": "json"}, + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["json"] == {"obj": "json"} + + async def test_delete_with_body_sends_params( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request(self.http_client, {}, 200) + + await self.http_client.delete_with_body( + path="/test", + json={"obj1": "json"}, + params={"obj2": "params"}, + ) + + assert request_kwargs["json"] == {"obj1": "json"} + assert request_kwargs["params"] == {"obj2": "params"} + + async def test_delete_without_body_raises_value_error(self): + with pytest.raises( + ValueError, match="Cannot send a body with a delete request" + ): + await self.http_client.request( + path="/test", + method="delete", + json={"should": "fail"}, + ) diff --git a/tests/test_authorization_check.py b/tests/test_authorization_check.py new file mode 100644 index 00000000..2cc19e6c --- /dev/null +++ b/tests/test_authorization_check.py @@ -0,0 +1,142 @@ +from typing import Union + +import pytest +from tests.utils.syncify import syncify +from workos.authorization import AsyncAuthorization, Authorization +from workos.types.authorization.resource_identifier import ( + ResourceIdentifierByExternalId, + ResourceIdentifierById, +) + +MOCK_ORG_MEMBERSHIP_ID = "org_membership_01ABC" +MOCK_PERMISSION_SLUG = "document:read" +MOCK_RESOURCE_ID = "res_01ABC" +MOCK_RESOURCE_TYPE = "document" +MOCK_EXTERNAL_ID = "ext_123" + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestAuthorizationCheck: + @pytest.fixture(autouse=True) + def setup(self, module_instance: Union[Authorization, AsyncAuthorization]): + self.http_client = module_instance._http_client + self.authorization = module_instance + + def test_check_authorized_by_resource_id( + self, capture_and_mock_http_client_request + ): + mock_response = {"authorized": True} + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_response, 200 + ) + + resource: ResourceIdentifierById = {"resource_id": MOCK_RESOURCE_ID} + response = syncify( + self.authorization.check( + MOCK_ORG_MEMBERSHIP_ID, + permission_slug=MOCK_PERMISSION_SLUG, + resource=resource, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith( + f"/authorization/organization_memberships/{MOCK_ORG_MEMBERSHIP_ID}/check" + ) + assert request_kwargs["json"] == { + "permission_slug": MOCK_PERMISSION_SLUG, + "resource_id": MOCK_RESOURCE_ID, + } + + assert response.authorized is True + + def test_check_authorized_by_external_id( + self, capture_and_mock_http_client_request + ): + mock_response = {"authorized": True} + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_response, 200 + ) + + resource: ResourceIdentifierByExternalId = { + "resource_external_id": MOCK_EXTERNAL_ID, + "resource_type_slug": MOCK_RESOURCE_TYPE, + } + response = syncify( + self.authorization.check( + MOCK_ORG_MEMBERSHIP_ID, + permission_slug=MOCK_PERMISSION_SLUG, + resource=resource, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith( + f"/authorization/organization_memberships/{MOCK_ORG_MEMBERSHIP_ID}/check" + ) + assert request_kwargs["json"] == { + "permission_slug": MOCK_PERMISSION_SLUG, + "resource_external_id": MOCK_EXTERNAL_ID, + "resource_type_slug": MOCK_RESOURCE_TYPE, + } + + assert response.authorized is True + + def test_check_not_authorized_by_resource_id( + self, capture_and_mock_http_client_request + ): + mock_response = {"authorized": False} + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_response, 200 + ) + + resource: ResourceIdentifierById = {"resource_id": MOCK_RESOURCE_ID} + response = syncify( + self.authorization.check( + MOCK_ORG_MEMBERSHIP_ID, + permission_slug=MOCK_PERMISSION_SLUG, + resource=resource, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith( + f"/authorization/organization_memberships/{MOCK_ORG_MEMBERSHIP_ID}/check" + ) + assert request_kwargs["json"] == { + "permission_slug": MOCK_PERMISSION_SLUG, + "resource_id": MOCK_RESOURCE_ID, + } + + assert response.authorized is False + + def test_check_not_authorized_by_external_id( + self, capture_and_mock_http_client_request + ): + mock_response = {"authorized": False} + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_response, 200 + ) + + resource: ResourceIdentifierByExternalId = { + "resource_external_id": MOCK_EXTERNAL_ID, + "resource_type_slug": MOCK_RESOURCE_TYPE, + } + response = syncify( + self.authorization.check( + MOCK_ORG_MEMBERSHIP_ID, + permission_slug=MOCK_PERMISSION_SLUG, + resource=resource, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith( + f"/authorization/organization_memberships/{MOCK_ORG_MEMBERSHIP_ID}/check" + ) + assert request_kwargs["json"] == { + "permission_slug": MOCK_PERMISSION_SLUG, + "resource_external_id": MOCK_EXTERNAL_ID, + "resource_type_slug": MOCK_RESOURCE_TYPE, + } + assert response.authorized is False diff --git a/tests/test_authorization_resource.py b/tests/test_authorization_resource.py new file mode 100644 index 00000000..b08c4e91 --- /dev/null +++ b/tests/test_authorization_resource.py @@ -0,0 +1,767 @@ +from typing import Union + +import pytest +from tests.utils.fixtures.mock_resource import MockAuthorizationResource +from tests.utils.fixtures.mock_resource_list import MockAuthorizationResourceList +from tests.utils.list_resource import list_response_of +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 MockAuthorizationResource().dict() + + @pytest.fixture + def mock_resources_list_two(self): + return MockAuthorizationResourceList().dict() + + @pytest.fixture + def mock_resources_empty_list(self): + return list_response_of(data=[]) + + # --- 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 + ) + + response = syncify(self.authorization.get_resource("res_01ABC")) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + + assert response.dict() == MockAuthorizationResource().dict() + + def test_get_resource_without_parent(self, capture_and_mock_http_client_request): + mock_resource = MockAuthorizationResource(parent_resource_id=None).dict() + capture_and_mock_http_client_request(self.http_client, mock_resource, 200) + + response = syncify(self.authorization.get_resource("res_01ABC")) + + assert ( + response.dict() == MockAuthorizationResource(parent_resource_id=None).dict() + ) + + def test_get_resource_without_description( + self, capture_and_mock_http_client_request + ): + mock_resource = MockAuthorizationResource(description=None).dict() + capture_and_mock_http_client_request(self.http_client, mock_resource, 200) + + response = syncify(self.authorization.get_resource("res_01ABC")) + + assert response.dict() == MockAuthorizationResource(description=None).dict() + + def test_get_resource_without_parent_and_description( + self, capture_and_mock_http_client_request + ): + mock_resource = MockAuthorizationResource( + parent_resource_id=None, description=None + ).dict() + capture_and_mock_http_client_request(self.http_client, mock_resource, 200) + + response = syncify(self.authorization.get_resource("res_01ABC")) + + assert ( + response.dict() + == MockAuthorizationResource( + parent_resource_id=None, description=None + ).dict() + ) + + # --- create_resource --- + + 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, 200 + ) + + response = syncify( + self.authorization.create_resource( + external_id="ext_123", + name="Test Resource", + description="A test resource", + resource_type_slug="document", + organization_id="org_01EHT88Z8J8795GZNQ4ZP1J81T", + parent={"parent_resource_id": "res_01XYZ"}, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith("/authorization/resources") + assert request_kwargs["json"] == { + "resource_type_slug": "document", + "organization_id": "org_01EHT88Z8J8795GZNQ4ZP1J81T", + "external_id": "ext_123", + "name": "Test Resource", + "description": "A test resource", + "parent_resource_id": "res_01XYZ", + } + assert "parent_resource_external_id" not in request_kwargs["json"] + assert "parent_resource_type_slug" not in request_kwargs["json"] + + assert response.dict() == MockAuthorizationResource().dict() + + def test_create_resource_with_parent_by_id_no_description( + self, capture_and_mock_http_client_request + ): + mock_resource = MockAuthorizationResource(description=None).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 200 + ) + + response = syncify( + self.authorization.create_resource( + external_id="ext_123", + name="Test Resource", + description=None, + resource_type_slug="document", + organization_id="org_01EHT88Z8J8795GZNQ4ZP1J81T", + parent={"parent_resource_id": "res_01XYZ"}, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith("/authorization/resources") + assert request_kwargs["json"] == { + "resource_type_slug": "document", + "organization_id": "org_01EHT88Z8J8795GZNQ4ZP1J81T", + "external_id": "ext_123", + "name": "Test Resource", + "parent_resource_id": "res_01XYZ", + } + assert "description" not in request_kwargs["json"] + assert "parent_resource_external_id" not in request_kwargs["json"] + assert "parent_resource_type_slug" not in request_kwargs["json"] + + assert response.dict() == MockAuthorizationResource(description=None).dict() + + 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, 200 + ) + + response = syncify( + self.authorization.create_resource( + external_id="ext_123", + name="Test Resource", + description="A test resource", + resource_type_slug="document", + organization_id="org_01EHT88Z8J8795GZNQ4ZP1J81T", + parent={ + "parent_resource_external_id": "parent_ext_456", + "parent_resource_type_slug": "folder", + }, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith("/authorization/resources") + assert request_kwargs["json"] == { + "resource_type_slug": "document", + "organization_id": "org_01EHT88Z8J8795GZNQ4ZP1J81T", + "external_id": "ext_123", + "name": "Test Resource", + "description": "A test resource", + "parent_resource_external_id": "parent_ext_456", + "parent_resource_type_slug": "folder", + } + + assert "parent_resource_id" not in request_kwargs["json"] + + assert response.dict() == MockAuthorizationResource().dict() + + def test_create_resource_with_parent_by_external_id_no_description( + self, capture_and_mock_http_client_request + ): + mock_resource = MockAuthorizationResource(description=None).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 200 + ) + + response = syncify( + self.authorization.create_resource( + external_id="ext_123", + name="Test Resource", + description=None, + resource_type_slug="document", + organization_id="org_01EHT88Z8J8795GZNQ4ZP1J81T", + parent={ + "parent_resource_external_id": "parent_ext_456", + "parent_resource_type_slug": "folder", + }, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["json"] == { + "resource_type_slug": "document", + "organization_id": "org_01EHT88Z8J8795GZNQ4ZP1J81T", + "external_id": "ext_123", + "name": "Test Resource", + "parent_resource_external_id": "parent_ext_456", + "parent_resource_type_slug": "folder", + } + + assert "description" not in request_kwargs["json"] + assert "parent_resource_id" not in request_kwargs["json"] + + assert response.dict() == MockAuthorizationResource(description=None).dict() + + def test_create_resource_without_parent(self, capture_and_mock_http_client_request): + mock_resource = MockAuthorizationResource(parent_resource_id=None).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 200 + ) + + response = syncify( + self.authorization.create_resource( + external_id="ext_123", + name="Test Resource", + description="A test resource", + resource_type_slug="document", + organization_id="org_01EHT88Z8J8795GZNQ4ZP1J81T", + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["json"] == { + "resource_type_slug": "document", + "organization_id": "org_01EHT88Z8J8795GZNQ4ZP1J81T", + "external_id": "ext_123", + "name": "Test Resource", + "description": "A test resource", + } + + assert ( + response.dict() == MockAuthorizationResource(parent_resource_id=None).dict() + ) + + # --- update_resource --- + def test_update_resource_name_only( + self, mock_resource, capture_and_mock_http_client_request + ): + updated_resource = MockAuthorizationResource( + name="New Name", + ).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, updated_resource, 200 + ) + + response = syncify( + self.authorization.update_resource("res_01ABC", name="New Name") + ) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + assert request_kwargs["json"] == {"name": "New Name"} + + assert ( + response.dict() + == MockAuthorizationResource( + name="New Name", + ).dict() + ) + + def test_update_resource_description_only( + self, mock_resource, capture_and_mock_http_client_request + ): + updated_resource = MockAuthorizationResource( + description="Updated description only", + ).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, updated_resource, 200 + ) + + response = syncify( + self.authorization.update_resource( + "res_01ABC", description="Updated description only" + ) + ) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + assert request_kwargs["json"] == { + "description": "Updated description only", + } + assert ( + response.dict() + == MockAuthorizationResource( + description="Updated description only", + ).dict() + ) + + def test_update_resource_remove_description( + self, mock_resource, capture_and_mock_http_client_request + ): + updated_resource = MockAuthorizationResource(description=None).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, updated_resource, 200 + ) + + response = syncify( + self.authorization.update_resource("res_01ABC", description=None) + ) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + assert request_kwargs["json"] == {"description": None} + assert ( + response.dict() + == MockAuthorizationResource( + description=None, + ).dict() + ) + + def test_update_resource_with_name_and_description( + self, capture_and_mock_http_client_request + ): + updated_resource = MockAuthorizationResource( + name="Updated Name", + description="Updated description", + ).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, updated_resource, 200 + ) + + response = syncify( + self.authorization.update_resource( + "res_01ABC", + name="Updated Name", + description="Updated description", + ) + ) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + assert request_kwargs["json"] == { + "name": "Updated Name", + "description": "Updated description", + } + assert ( + response.dict() + == MockAuthorizationResource( + name="Updated Name", + description="Updated description", + ).dict() + ) + + # --- 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=204, + ) + + response = syncify(self.authorization.delete_resource("res_01ABC")) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + assert request_kwargs.get("params") is None + assert response is None + + def test_delete_resource_with_cascade_true( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=204, + ) + + response = syncify( + self.authorization.delete_resource("res_01ABC", cascade_delete=True) + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + assert request_kwargs["params"] == {"cascade_delete": "true"} + assert response is None + + def test_delete_resource_with_cascade_false( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=204, + ) + + response = syncify( + self.authorization.delete_resource("res_01ABC", cascade_delete=False) + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith("/authorization/resources/res_01ABC") + assert request_kwargs["params"] == {"cascade_delete": "false"} + assert response is None + + # --- list_resources --- + def test_list_resources_returns_paginated_list( + self, + mock_resources_list_two, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + response = syncify(self.authorization.list_resources()) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith("/authorization/resources") + assert request_kwargs["params"] == {"limit": 10, "order": "desc"} + + assert response.object == "list" + assert len(response.data) == 2 + + assert response.data[0].object == "authorization_resource" + assert response.data[0].id == "authz_resource_01HXYZ123ABC456DEF789ABC" + assert response.data[0].external_id == "doc-12345678" + assert response.data[0].name == "Q5 Budget Report" + assert response.data[0].description == "Financial report for Q5 2025" + assert response.data[0].resource_type_slug == "document" + assert response.data[0].organization_id == "org_01HXYZ123ABC456DEF789ABC" + assert ( + response.data[0].parent_resource_id + == "authz_resource_01HXYZ123ABC456DEF789XYZ" + ) + assert response.data[0].created_at == "2024-01-15T09:30:00.000Z" + assert response.data[0].updated_at == "2024-01-15T09:30:00.000Z" + + assert response.data[1].object == "authorization_resource" + assert response.data[1].id == "authz_resource_01HXYZ123ABC456DEF789DEF" + assert response.data[1].external_id == "folder-123" + assert response.data[1].name == "Finance Folder" + assert response.data[1].description is None + assert response.data[1].resource_type_slug == "folder" + assert response.data[1].organization_id == "org_01HXYZ123ABC456DEF789ABC" + assert response.data[1].parent_resource_id is None + assert response.data[1].created_at == "2024-01-14T08:00:00.000Z" + assert response.data[1].updated_at == "2024-01-14T08:00:00.000Z" + + assert response.list_metadata.before is None + assert response.list_metadata.after == "authz_resource_01HXYZ123ABC456DEF789DEF" + + def test_list_resources_returns_empty_list( + self, + mock_resources_empty_list, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_empty_list, 200 + ) + + response = syncify(self.authorization.list_resources()) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith("/authorization/resources") + assert request_kwargs["params"] == {"limit": 10, "order": "desc"} + + assert len(response.data) == 0 + assert response.list_metadata.before is None + assert response.list_metadata.after is None + + def test_list_resources_request_with_no_parameters( + self, + mock_resources_list_two, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources()) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith("/authorization/resources") + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_organization_id( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(organization_id="org_123")) + + assert request_kwargs["params"]["organization_id"] == "org_123" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_resource_type_slug( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(resource_type_slug="document")) + + assert request_kwargs["params"]["resource_type_slug"] == "document" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert "organization_id" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_parent_resource_id( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(parent_resource_id="res_parent_123")) + + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_123" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_parent_resource_type_slug( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(parent_resource_type_slug="folder")) + + assert request_kwargs["params"]["parent_resource_type_slug"] == "folder" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_parent_external_id( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(parent_external_id="parent_ext_456")) + + assert request_kwargs["params"]["parent_external_id"] == "parent_ext_456" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_search( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(search="Budget")) + + assert request_kwargs["params"]["search"] == "Budget" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_limit( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(limit=25)) + + assert request_kwargs["params"]["limit"] == 25 + assert request_kwargs["params"]["order"] == "desc" + + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_before( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(before="cursor_before")) + + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_after( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(after="cursor_after")) + + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + + def test_list_resources_with_order_asc( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(order="asc")) + + assert request_kwargs["params"]["order"] == "asc" + assert request_kwargs["params"]["limit"] == 10 + + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_order_desc( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify(self.authorization.list_resources(order="desc")) + + assert request_kwargs["params"]["order"] == "desc" + assert request_kwargs["params"]["limit"] == 10 + + assert "organization_id" not in request_kwargs["params"] + assert "resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_id" not in request_kwargs["params"] + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_external_id" not in request_kwargs["params"] + assert "search" not in request_kwargs["params"] + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_resources_with_all_parameters( + self, mock_resources_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list_two, 200 + ) + + syncify( + self.authorization.list_resources( + organization_id="org_123", + resource_type_slug="document", + parent_resource_id="res_parent_123", + parent_resource_type_slug="folder", + parent_external_id="parent_ext_456", + search="Budget", + limit=5, + before="cursor_before", + after="cursor_after", + order="asc", + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith("/authorization/resources") + assert request_kwargs["params"]["organization_id"] == "org_123" + assert request_kwargs["params"]["resource_type_slug"] == "document" + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_123" + assert request_kwargs["params"]["parent_resource_type_slug"] == "folder" + assert request_kwargs["params"]["parent_external_id"] == "parent_ext_456" + assert request_kwargs["params"]["search"] == "Budget" + assert request_kwargs["params"]["limit"] == 5 + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["order"] == "asc" diff --git a/tests/test_authorization_resource_external_id.py b/tests/test_authorization_resource_external_id.py new file mode 100644 index 00000000..6cff6773 --- /dev/null +++ b/tests/test_authorization_resource_external_id.py @@ -0,0 +1,297 @@ +from typing import Union + +import pytest +from tests.utils.fixtures.mock_resource import MockAuthorizationResource +from tests.utils.syncify import syncify +from workos.authorization import AsyncAuthorization, Authorization + + +MOCK_ORG_ID = "org_01EHT88Z8J8795GZNQ4ZP1J81T" +MOCK_RESOURCE_TYPE = "document" +MOCK_EXTERNAL_ID = "ext_123" + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestAuthorizationResourceExternalId: + @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 MockAuthorizationResource().dict() + + # --- get_resource_by_external_id --- + def test_get_resource_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, 200 + ) + + response = syncify( + self.authorization.get_resource_by_external_id( + MOCK_ORG_ID, MOCK_RESOURCE_TYPE, MOCK_EXTERNAL_ID + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + + assert response.dict() == MockAuthorizationResource().dict() + + def test_get_resource_by_external_id_without_parent( + self, capture_and_mock_http_client_request + ): + mock_resource = MockAuthorizationResource(parent_resource_id=None).dict() + capture_and_mock_http_client_request(self.http_client, mock_resource, 200) + + response = syncify( + self.authorization.get_resource_by_external_id( + MOCK_ORG_ID, MOCK_RESOURCE_TYPE, MOCK_EXTERNAL_ID + ) + ) + + assert ( + response.dict() == MockAuthorizationResource(parent_resource_id=None).dict() + ) + + def test_get_resource_by_external_id_without_description( + self, capture_and_mock_http_client_request + ): + mock_resource = MockAuthorizationResource(description=None).dict() + capture_and_mock_http_client_request(self.http_client, mock_resource, 200) + + response = syncify( + self.authorization.get_resource_by_external_id( + MOCK_ORG_ID, MOCK_RESOURCE_TYPE, MOCK_EXTERNAL_ID + ) + ) + + assert response.dict() == MockAuthorizationResource(description=None).dict() + + def test_get_resource_by_external_id_without_parent_and_description( + self, capture_and_mock_http_client_request + ): + mock_resource = MockAuthorizationResource( + parent_resource_id=None, description=None + ).dict() + capture_and_mock_http_client_request(self.http_client, mock_resource, 200) + + response = syncify( + self.authorization.get_resource_by_external_id( + MOCK_ORG_ID, MOCK_RESOURCE_TYPE, MOCK_EXTERNAL_ID + ) + ) + + assert ( + response.dict() + == MockAuthorizationResource( + parent_resource_id=None, description=None + ).dict() + ) + + # --- update_resource_by_external_id --- + + def test_update_resource_by_external_id_name_only( + self, mock_resource, capture_and_mock_http_client_request + ): + updated_resource = MockAuthorizationResource(name="New Name").dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, updated_resource, 200 + ) + + response = syncify( + self.authorization.update_resource_by_external_id( + MOCK_ORG_ID, MOCK_RESOURCE_TYPE, MOCK_EXTERNAL_ID, name="New Name" + ) + ) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + assert request_kwargs["json"] == {"name": "New Name"} + + assert ( + response.dict() + == MockAuthorizationResource( + name="New Name", + ).dict() + ) + + def test_update_resource_by_external_id_description_only( + self, capture_and_mock_http_client_request + ): + updated_resource = MockAuthorizationResource( + description="Updated description only", + ).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, updated_resource, 200 + ) + + response = syncify( + self.authorization.update_resource_by_external_id( + MOCK_ORG_ID, + MOCK_RESOURCE_TYPE, + MOCK_EXTERNAL_ID, + description="Updated description only", + ) + ) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + assert request_kwargs["json"] == { + "description": "Updated description only", + } + + assert ( + response.dict() + == MockAuthorizationResource( + description="Updated description only", + ).dict() + ) + + def test_update_resource_by_external_id_name_and_description( + self, capture_and_mock_http_client_request + ): + updated_resource = MockAuthorizationResource( + name="Updated Name", + description="Updated description", + ).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, updated_resource, 200 + ) + + response = syncify( + self.authorization.update_resource_by_external_id( + MOCK_ORG_ID, + MOCK_RESOURCE_TYPE, + MOCK_EXTERNAL_ID, + name="Updated Name", + description="Updated description", + ) + ) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + assert request_kwargs["json"] == { + "name": "Updated Name", + "description": "Updated description", + } + + assert ( + response.dict() + == MockAuthorizationResource( + name="Updated Name", + description="Updated description", + ).dict() + ) + + def test_update_resource_by_external_id_remove_description( + self, capture_and_mock_http_client_request + ): + updated_resource = MockAuthorizationResource(description=None).dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, updated_resource, 200 + ) + + response = syncify( + self.authorization.update_resource_by_external_id( + MOCK_ORG_ID, + MOCK_RESOURCE_TYPE, + MOCK_EXTERNAL_ID, + description=None, + ) + ) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + assert request_kwargs["json"] == {"description": None} + + assert ( + response.dict() + == MockAuthorizationResource( + description=None, + ).dict() + ) + + # --- delete_resource_by_external_id --- + + def test_delete_resource_by_external_id_without_cascade( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=204, + ) + + response = syncify( + self.authorization.delete_resource_by_external_id( + MOCK_ORG_ID, MOCK_RESOURCE_TYPE, MOCK_EXTERNAL_ID + ) + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + assert request_kwargs.get("params") is None + assert response is None + + def test_delete_resource_by_external_id_with_cascade_true( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=204, + ) + + response = syncify( + self.authorization.delete_resource_by_external_id( + MOCK_ORG_ID, + MOCK_RESOURCE_TYPE, + MOCK_EXTERNAL_ID, + cascade_delete=True, + ) + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + assert request_kwargs["params"] == {"cascade_delete": "true"} + assert response is None + + def test_delete_resource_by_external_id_with_cascade_false( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=204, + ) + + response = syncify( + self.authorization.delete_resource_by_external_id( + MOCK_ORG_ID, + MOCK_RESOURCE_TYPE, + MOCK_EXTERNAL_ID, + cascade_delete=False, + ) + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + assert request_kwargs["params"] == {"cascade_delete": "false"} + assert response is None diff --git a/tests/test_authorization_resource_memberships.py b/tests/test_authorization_resource_memberships.py new file mode 100644 index 00000000..bcaf2107 --- /dev/null +++ b/tests/test_authorization_resource_memberships.py @@ -0,0 +1,904 @@ +from typing import Union + +import pytest +from tests.utils.fixtures.mock_organization_membership import ( + MockAuthorizationOrganizationMembershipList, +) +from tests.utils.fixtures.mock_resource_list import MockAuthorizationResourceList +from tests.utils.syncify import syncify +from workos.authorization import AsyncAuthorization, Authorization +from workos.types.authorization.parent_resource_identifier import ( + ParentResourceByExternalId, + ParentResourceById, +) + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestListResourcesForMembership: + @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_resources_list(self): + return MockAuthorizationResourceList().model_dump() + + def test_list_resources_for_membership_with_parent_by_id_returns_paginated_list( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceById(parent_resource_id="res_parent_123") + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + response = syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/resources" + ) + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_123" + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert response.object == "list" + assert len(response.data) == 2 + assert response.data[0].object == "authorization_resource" + assert response.data[0].id == "authz_resource_01HXYZ123ABC456DEF789ABC" + assert response.data[0].external_id == "doc-12345678" + assert response.data[0].name == "Q5 Budget Report" + assert response.data[0].description == "Financial report for Q5 2025" + assert response.data[0].resource_type_slug == "document" + assert response.data[0].organization_id == "org_01HXYZ123ABC456DEF789ABC" + assert ( + response.data[0].parent_resource_id + == "authz_resource_01HXYZ123ABC456DEF789XYZ" + ) + assert response.data[0].created_at == "2024-01-15T09:30:00.000Z" + assert response.data[0].updated_at == "2024-01-15T09:30:00.000Z" + assert response.data[1].object == "authorization_resource" + assert response.data[1].id == "authz_resource_01HXYZ123ABC456DEF789DEF" + assert response.data[1].external_id == "folder-123" + assert response.data[1].name == "Finance Folder" + assert response.data[1].description is None + assert response.data[1].resource_type_slug == "folder" + assert response.data[1].organization_id == "org_01HXYZ123ABC456DEF789ABC" + assert response.data[1].parent_resource_id is None + assert response.data[1].created_at == "2024-01-14T08:00:00.000Z" + assert response.data[1].updated_at == "2024-01-14T08:00:00.000Z" + assert response.list_metadata.before is None + assert response.list_metadata.after == "authz_resource_01HXYZ123ABC456DEF789DEF" + + def test_list_resources_for_membership_with_parent_by_id_with_limit( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceById(parent_resource_id="res_parent_123") + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + limit=25, + ) + ) + + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_123" + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["limit"] == 25 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_resources_for_membership_with_parent_by_id_with_before( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceById(parent_resource_id="res_parent_123") + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + before="cursor_before", + ) + ) + + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_123" + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_resources_for_membership_with_parent_by_id_with_after( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceById(parent_resource_id="res_parent_123") + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + after="cursor_after", + ) + ) + + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_123" + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_resources_for_membership_with_parent_by_id_with_order_asc( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceById(parent_resource_id="res_parent_123") + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + order="asc", + ) + ) + + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_123" + assert request_kwargs["params"]["order"] == "asc" + assert request_kwargs["params"]["limit"] == 10 + + def test_list_resources_for_membership_with_parent_by_id_with_order_desc( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceById(parent_resource_id="res_parent_123") + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + order="desc", + ) + ) + + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_123" + assert request_kwargs["params"]["order"] == "desc" + assert request_kwargs["params"]["limit"] == 10 + + def test_list_resources_for_membership_with_parent_by_id_with_all_parameters( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceById(parent_resource_id="res_parent_123") + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + response = syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + limit=5, + before="cursor_before", + after="cursor_after", + order="asc", + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/resources" + ) + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_123" + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["limit"] == 5 + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["order"] == "asc" + + assert response.object == "list" + assert len(response.data) == 2 + + # --- list_resources_for_membership with ParentResourceByExternalId --- + + def test_list_resources_for_membership_with_parent_by_external_id_returns_paginated_list( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceByExternalId( + parent_resource_external_id="parent_ext_456", + parent_resource_type_slug="folder", + ) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + response = syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/resources" + ) + assert ( + request_kwargs["params"]["parent_resource_external_id"] == "parent_ext_456" + ) + assert request_kwargs["params"]["parent_resource_type_slug"] == "folder" + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + assert "parent_resource_id" not in request_kwargs["params"] + + assert response.object == "list" + assert len(response.data) == 2 + assert response.data[0].object == "authorization_resource" + assert response.data[0].id == "authz_resource_01HXYZ123ABC456DEF789ABC" + assert response.data[0].external_id == "doc-12345678" + assert response.data[0].name == "Q5 Budget Report" + assert response.data[0].description == "Financial report for Q5 2025" + assert response.data[0].resource_type_slug == "document" + assert response.data[0].organization_id == "org_01HXYZ123ABC456DEF789ABC" + assert ( + response.data[0].parent_resource_id + == "authz_resource_01HXYZ123ABC456DEF789XYZ" + ) + assert response.data[0].created_at == "2024-01-15T09:30:00.000Z" + assert response.data[0].updated_at == "2024-01-15T09:30:00.000Z" + assert response.data[1].object == "authorization_resource" + assert response.data[1].id == "authz_resource_01HXYZ123ABC456DEF789DEF" + assert response.data[1].external_id == "folder-123" + assert response.data[1].name == "Finance Folder" + assert response.data[1].description is None + assert response.data[1].resource_type_slug == "folder" + assert response.data[1].organization_id == "org_01HXYZ123ABC456DEF789ABC" + assert response.data[1].parent_resource_id is None + assert response.data[1].created_at == "2024-01-14T08:00:00.000Z" + assert response.data[1].updated_at == "2024-01-14T08:00:00.000Z" + assert response.list_metadata.before is None + + def test_list_resources_for_membership_with_parent_by_external_id_with_limit( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceByExternalId( + parent_resource_external_id="parent_ext_456", + parent_resource_type_slug="folder", + ) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + limit=25, + ) + ) + + assert ( + request_kwargs["params"]["parent_resource_external_id"] == "parent_ext_456" + ) + assert request_kwargs["params"]["parent_resource_type_slug"] == "folder" + assert request_kwargs["params"]["limit"] == 25 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_resources_for_membership_with_parent_by_external_id_with_before( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceByExternalId( + parent_resource_external_id="parent_ext_456", + parent_resource_type_slug="folder", + ) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + before="cursor_before", + ) + ) + + assert ( + request_kwargs["params"]["parent_resource_external_id"] == "parent_ext_456" + ) + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_resources_for_membership_with_parent_by_external_id_with_after( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceByExternalId( + parent_resource_external_id="parent_ext_456", + parent_resource_type_slug="folder", + ) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + after="cursor_after", + ) + ) + + assert ( + request_kwargs["params"]["parent_resource_external_id"] == "parent_ext_456" + ) + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_resources_for_membership_with_parent_by_external_id_with_order_asc( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceByExternalId( + parent_resource_external_id="parent_ext_456", + parent_resource_type_slug="folder", + ) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + order="asc", + ) + ) + + assert ( + request_kwargs["params"]["parent_resource_external_id"] == "parent_ext_456" + ) + assert request_kwargs["params"]["order"] == "asc" + assert request_kwargs["params"]["limit"] == 10 + + def test_list_resources_for_membership_with_parent_by_external_id_with_order_desc( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceByExternalId( + parent_resource_external_id="parent_ext_456", + parent_resource_type_slug="folder", + ) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + order="desc", + ) + ) + + assert ( + request_kwargs["params"]["parent_resource_external_id"] == "parent_ext_456" + ) + assert request_kwargs["params"]["order"] == "desc" + assert request_kwargs["params"]["limit"] == 10 + + def test_list_resources_for_membership_with_parent_by_external_id_with_all_parameters( + self, mock_resources_list, capture_and_mock_http_client_request + ): + parent = ParentResourceByExternalId( + parent_resource_external_id="parent_ext_456", + parent_resource_type_slug="folder", + ) + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + response = syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="document:read", + parent_resource=parent, + limit=5, + before="cursor_before", + after="cursor_after", + order="asc", + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/resources" + ) + assert ( + request_kwargs["params"]["parent_resource_external_id"] == "parent_ext_456" + ) + assert request_kwargs["params"]["parent_resource_type_slug"] == "folder" + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["limit"] == 5 + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["order"] == "asc" + assert "parent_resource_id" not in request_kwargs["params"] + + assert response.object == "list" + assert len(response.data) == 2 + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestListMembershipsForResource: + @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_memberships_list_two(self): + return MockAuthorizationOrganizationMembershipList().model_dump() + + # --- list_memberships_for_resource (by resource_id) --- + + def test_list_memberships_for_resource_returns_paginated_list( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + response = syncify( + self.authorization.list_memberships_for_resource( + "authz_resource_01HXYZ123ABC456DEF789ABC", + permission_slug="document:read", + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/resources/authz_resource_01HXYZ123ABC456DEF789ABC/organization_memberships" + ) + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert response.object == "list" + assert len(response.data) == 2 + assert response.data[0].object == "organization_membership" + assert response.data[0].id == "om_01ABC" + assert response.data[0].user_id == "user_123" + assert response.data[0].organization_id == "org_456" + assert response.data[0].status == "active" + assert response.data[0].created_at == "2024-01-01T00:00:00Z" + assert response.data[0].updated_at == "2024-01-01T00:00:00Z" + assert response.data[1].object == "organization_membership" + assert response.data[1].id == "om_01DEF" + assert response.data[1].user_id == "user_789" + assert response.data[1].organization_id == "org_456" + assert response.data[1].status == "active" + assert response.data[1].created_at == "2024-01-02T00:00:00Z" + assert response.data[1].updated_at == "2024-01-02T00:00:00Z" + assert response.list_metadata.before is None + assert response.list_metadata.after == "om_01DEF" + + def test_list_memberships_for_resource_with_assignment_direct( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource( + "authz_resource_01HXYZ", + permission_slug="document:read", + assignment="direct", + ) + ) + + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["assignment"] == "direct" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_memberships_for_resource_with_assignment_indirect( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource( + "authz_resource_01HXYZ", + permission_slug="document:read", + assignment="indirect", + ) + ) + + assert request_kwargs["params"]["assignment"] == "indirect" + assert request_kwargs["params"]["permission_slug"] == "document:read" + + def test_list_memberships_for_resource_with_limit( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource( + "authz_resource_01HXYZ", + permission_slug="document:read", + limit=25, + ) + ) + + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["limit"] == 25 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_memberships_for_resource_with_before( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource( + "authz_resource_01HXYZ", + permission_slug="document:read", + before="cursor_before", + ) + ) + + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_memberships_for_resource_with_after( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource( + "authz_resource_01HXYZ", + permission_slug="document:read", + after="cursor_after", + ) + ) + + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_memberships_for_resource_with_order_asc( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource( + "authz_resource_01HXYZ", + permission_slug="document:read", + order="asc", + ) + ) + + assert request_kwargs["params"]["order"] == "asc" + assert request_kwargs["params"]["limit"] == 10 + + def test_list_memberships_for_resource_with_order_desc( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource( + "authz_resource_01HXYZ", + permission_slug="document:read", + order="desc", + ) + ) + + assert request_kwargs["params"]["order"] == "desc" + assert request_kwargs["params"]["limit"] == 10 + + def test_list_memberships_for_resource_with_all_parameters( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + response = syncify( + self.authorization.list_memberships_for_resource( + "authz_resource_01HXYZ", + permission_slug="document:read", + assignment="direct", + limit=5, + before="cursor_before", + after="cursor_after", + order="asc", + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/resources/authz_resource_01HXYZ/organization_memberships" + ) + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["assignment"] == "direct" + assert request_kwargs["params"]["limit"] == 5 + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["order"] == "asc" + + assert response.object == "list" + assert len(response.data) == 2 + + # --- list_memberships_for_resource_by_external_id --- + + def test_list_memberships_for_resource_by_external_id_returns_paginated_list( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + response = syncify( + self.authorization.list_memberships_for_resource_by_external_id( + organization_id="org_123", + resource_type_slug="document", + external_id="doc-ext-456", + permission_slug="document:read", + ) + ) + + assert request_kwargs["method"] == "get" + assert ( + "/authorization/organizations/org_123/resources/document/doc-ext-456/organization_memberships" + in request_kwargs["url"] + ) + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + assert response.object == "list" + assert len(response.data) == 2 + assert response.data[0].object == "organization_membership" + assert response.data[0].id == "om_01ABC" + assert response.data[0].user_id == "user_123" + assert response.data[0].organization_id == "org_456" + assert response.data[0].status == "active" + assert response.data[0].created_at == "2024-01-01T00:00:00Z" + assert response.data[0].updated_at == "2024-01-01T00:00:00Z" + assert response.data[1].object == "organization_membership" + assert response.data[1].id == "om_01DEF" + assert response.data[1].user_id == "user_789" + assert response.data[1].organization_id == "org_456" + assert response.data[1].status == "active" + assert response.data[1].created_at == "2024-01-02T00:00:00Z" + assert response.data[1].updated_at == "2024-01-02T00:00:00Z" + assert response.list_metadata.before is None + assert response.list_metadata.after == "om_01DEF" + + def test_list_memberships_for_resource_by_external_id_with_assignment_direct( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource_by_external_id( + organization_id="org_123", + resource_type_slug="document", + external_id="doc-ext-456", + permission_slug="document:read", + assignment="direct", + ) + ) + + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["assignment"] == "direct" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_memberships_for_resource_by_external_id_with_assignment_indirect( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource_by_external_id( + organization_id="org_123", + resource_type_slug="folder", + external_id="folder-ext-789", + permission_slug="document:read", + assignment="indirect", + ) + ) + + assert request_kwargs["params"]["assignment"] == "indirect" + assert ( + "/authorization/organizations/org_123/resources/folder/folder-ext-789/organization_memberships" + in request_kwargs["url"] + ) + + def test_list_memberships_for_resource_by_external_id_with_limit( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource_by_external_id( + organization_id="org_123", + resource_type_slug="document", + external_id="doc-ext-456", + permission_slug="document:read", + limit=25, + ) + ) + + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["limit"] == 25 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_memberships_for_resource_by_external_id_with_before( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource_by_external_id( + organization_id="org_123", + resource_type_slug="document", + external_id="doc-ext-456", + permission_slug="document:read", + before="cursor_before", + ) + ) + + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_memberships_for_resource_by_external_id_with_after( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource_by_external_id( + organization_id="org_123", + resource_type_slug="document", + external_id="doc-ext-456", + permission_slug="document:read", + after="cursor_after", + ) + ) + + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + + def test_list_memberships_for_resource_by_external_id_with_order_asc( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource_by_external_id( + organization_id="org_123", + resource_type_slug="document", + external_id="doc-ext-456", + permission_slug="document:read", + order="asc", + ) + ) + + assert request_kwargs["params"]["order"] == "asc" + assert request_kwargs["params"]["limit"] == 10 + + def test_list_memberships_for_resource_by_external_id_with_order_desc( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource_by_external_id( + organization_id="org_123", + resource_type_slug="document", + external_id="doc-ext-456", + permission_slug="document:read", + order="desc", + ) + ) + + assert request_kwargs["params"]["order"] == "desc" + assert request_kwargs["params"]["limit"] == 10 + + def test_list_memberships_for_resource_by_external_id_with_all_parameters( + self, mock_memberships_list_two, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list_two, 200 + ) + + response = syncify( + self.authorization.list_memberships_for_resource_by_external_id( + organization_id="org_123", + resource_type_slug="document", + external_id="doc-ext-456", + permission_slug="document:read", + assignment="direct", + limit=5, + before="cursor_before", + after="cursor_after", + order="asc", + ) + ) + + assert request_kwargs["method"] == "get" + assert ( + "/authorization/organizations/org_123/resources/document/doc-ext-456/organization_memberships" + in request_kwargs["url"] + ) + assert request_kwargs["params"]["permission_slug"] == "document:read" + assert request_kwargs["params"]["assignment"] == "direct" + assert request_kwargs["params"]["limit"] == 5 + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["order"] == "asc" + + assert response.object == "list" + assert len(response.data) == 2 diff --git a/tests/test_authorization_role_assignments.py b/tests/test_authorization_role_assignments.py new file mode 100644 index 00000000..ed040ece --- /dev/null +++ b/tests/test_authorization_role_assignments.py @@ -0,0 +1,358 @@ +from typing import Union + +import pytest +from tests.utils.fixtures.mock_role_assignment import ( + MockRoleAssignment, + MockRoleAssignmentsList, +) +from tests.utils.list_resource import list_response_of +from tests.utils.syncify import syncify +from workos.authorization import AsyncAuthorization, Authorization + + +@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 + + @pytest.fixture + def mock_role_assignments_list(self): + return MockRoleAssignmentsList().dict() + + @pytest.fixture + def mock_role_assignments_empty_list(self): + return list_response_of(data=[]) + + def test_assign_role_by_resource_id(self, capture_and_mock_http_client_request): + mock_role_assignment = MockRoleAssignment().dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignment, 201 + ) + + response = syncify( + self.authorization.assign_role( + "om_01ABC", + role_slug="admin", + resource_identifier={"resource_id": "res_01XYZ"}, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/role_assignments" + ) + assert request_kwargs["json"] == { + "role_slug": "admin", + "resource_id": "res_01XYZ", + } + assert "resource_external_id" not in request_kwargs["json"] + assert "resource_type_slug" not in request_kwargs["json"] + + assert response.dict() == mock_role_assignment + + def test_assign_role_by_external_id_and_resource_type_slug( + self, capture_and_mock_http_client_request + ): + mock_role_assignment = MockRoleAssignment().dict() + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignment, 201 + ) + + response = syncify( + self.authorization.assign_role( + "om_01ABC", + role_slug="editor", + resource_identifier={ + "resource_external_id": "ext_doc_456", + "resource_type_slug": "document", + }, + ) + ) + + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/role_assignments" + ) + assert request_kwargs["json"] == { + "role_slug": "editor", + "resource_external_id": "ext_doc_456", + "resource_type_slug": "document", + } + assert "resource_id" not in request_kwargs["json"] + + assert response.dict() == mock_role_assignment + + def test_remove_role_by_resource_id(self, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, status_code=204 + ) + + syncify( + self.authorization.remove_role( + "om_01ABC", + role_slug="admin", + resource_identifier={"resource_id": "res_01XYZ"}, + ) + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/role_assignments" + ) + assert request_kwargs["json"] == { + "role_slug": "admin", + "resource_id": "res_01XYZ", + } + assert "resource_external_id" not in request_kwargs["json"] + assert "resource_type_slug" not in request_kwargs["json"] + + def test_remove_role_by_external_id_and_resource_type_slug( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, status_code=204 + ) + + syncify( + self.authorization.remove_role( + "om_01ABC", + role_slug="editor", + resource_identifier={ + "resource_external_id": "ext_doc_456", + "resource_type_slug": "document", + }, + ) + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/role_assignments" + ) + assert request_kwargs["json"] == { + "role_slug": "editor", + "resource_external_id": "ext_doc_456", + "resource_type_slug": "document", + } + assert "resource_id" not in request_kwargs["json"] + + 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 + ) + + syncify( + self.authorization.remove_role_assignment( + "om_01ABC", + role_assignment_id="ra_01XYZ", + ) + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/role_assignments/ra_01XYZ" + ) + + def test_list_role_assignments_returns_paginated_list( + self, + mock_role_assignments_list, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignments_list, 200 + ) + + response = syncify( + self.authorization.list_role_assignments( + organization_membership_id="om_01ABC" + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/role_assignments" + ) + assert request_kwargs["params"] == {"limit": 10, "order": "desc"} + + assert response.object == "list" + assert len(response.data) == 2 + + assert response.data[0].object == "role_assignment" + assert response.data[0].id == "ra_01ABC" + assert response.data[0].role.slug == "admin" + assert response.data[0].resource.id == "res_01ABC" + assert response.data[0].resource.external_id == "ext_123" + assert response.data[0].resource.resource_type_slug == "document" + assert response.data[0].created_at == "2024-01-15T09:30:00.000Z" + assert response.data[0].updated_at == "2024-01-15T09:30:00.000Z" + + assert response.data[1].object == "role_assignment" + assert response.data[1].id == "ra_01DEF" + assert response.data[1].role.slug == "editor" + assert response.data[1].resource.id == "res_01XYZ" + assert response.data[1].resource.external_id == "ext_456" + assert response.data[1].resource.resource_type_slug == "folder" + assert response.data[1].created_at == "2024-01-14T08:00:00.000Z" + assert response.data[1].updated_at == "2024-01-14T08:00:00.000Z" + + assert response.list_metadata.before is None + assert response.list_metadata.after == "ra_01DEF" + + def test_list_role_assignments_returns_empty_list( + self, + mock_role_assignments_empty_list, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignments_empty_list, 200 + ) + + response = syncify( + self.authorization.list_role_assignments( + organization_membership_id="om_01ABC" + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/role_assignments" + ) + assert request_kwargs["params"] == {"limit": 10, "order": "desc"} + + assert len(response.data) == 0 + assert response.list_metadata.before is None + assert response.list_metadata.after is None + + def test_list_role_assignments_with_limit( + self, + mock_role_assignments_list, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignments_list, 200 + ) + + syncify( + self.authorization.list_role_assignments( + organization_membership_id="om_01ABC", + limit=25, + ) + ) + + assert request_kwargs["params"]["limit"] == 25 + assert request_kwargs["params"]["order"] == "desc" + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_role_assignments_with_before( + self, + mock_role_assignments_list, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignments_list, 200 + ) + + syncify( + self.authorization.list_role_assignments( + organization_membership_id="om_01ABC", + before="cursor_before", + ) + ) + + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + assert "after" not in request_kwargs["params"] + + def test_list_role_assignments_with_after( + self, + mock_role_assignments_list, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignments_list, 200 + ) + + syncify( + self.authorization.list_role_assignments( + organization_membership_id="om_01ABC", + after="cursor_after", + ) + ) + + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["order"] == "desc" + assert "before" not in request_kwargs["params"] + + def test_list_role_assignments_with_order_desc( + self, + mock_role_assignments_list, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignments_list, 200 + ) + + syncify( + self.authorization.list_role_assignments( + organization_membership_id="om_01ABC", + order="desc", + ) + ) + + assert request_kwargs["params"]["order"] == "desc" + assert request_kwargs["params"]["limit"] == 10 + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_role_assignments_with_order_asc( + self, + mock_role_assignments_list, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignments_list, 200 + ) + + syncify( + self.authorization.list_role_assignments( + organization_membership_id="om_01ABC", + order="asc", + ) + ) + + assert request_kwargs["params"]["order"] == "asc" + assert request_kwargs["params"]["limit"] == 10 + assert "before" not in request_kwargs["params"] + assert "after" not in request_kwargs["params"] + + def test_list_role_assignments_with_all_parameters( + self, + mock_role_assignments_list, + capture_and_mock_http_client_request, + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_role_assignments_list, 200 + ) + + syncify( + self.authorization.list_role_assignments( + organization_membership_id="om_01ABC", + limit=5, + before="cursor_before", + after="cursor_after", + order="asc", + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/role_assignments" + ) + assert request_kwargs["params"]["limit"] == 5 + assert request_kwargs["params"]["before"] == "cursor_before" + assert request_kwargs["params"]["after"] == "cursor_after" + assert request_kwargs["params"]["order"] == "asc" diff --git a/tests/test_sync_http_client.py b/tests/test_sync_http_client.py index edbba0b4..9023ce15 100644 --- a/tests/test_sync_http_client.py +++ b/tests/test_sync_http_client.py @@ -372,3 +372,36 @@ def test_request_removes_none_json_values( json={"organization_id": None, "test": "value"}, ) assert request_kwargs["json"] == {"test": "value"} + + def test_delete_with_body_sends_json(self, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request(self.http_client, {}, 200) + + self.http_client.delete_with_body( + path="/test", + json={"obj": "json"}, + ) + + assert request_kwargs["method"] == "delete" + assert request_kwargs["json"] == {"obj": "json"} + + def test_delete_with_body_sends_params(self, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request(self.http_client, {}, 200) + + self.http_client.delete_with_body( + path="/test", + json={"obj1": "json"}, + params={"obj2": "params"}, + ) + + assert request_kwargs["json"] == {"obj1": "json"} + assert request_kwargs["params"] == {"obj2": "params"} + + def test_delete_without_body_raises_value_error(self): + with pytest.raises( + ValueError, match="Cannot send a body with a delete request" + ): + self.http_client.request( + path="/test", + method="delete", + json={"should": "fail"}, + ) diff --git a/tests/utils/fixtures/mock_organization_membership.py b/tests/utils/fixtures/mock_organization_membership.py index b363b48b..9314fd14 100644 --- a/tests/utils/fixtures/mock_organization_membership.py +++ b/tests/utils/fixtures/mock_organization_membership.py @@ -1,8 +1,50 @@ import datetime +from typing import Optional, Sequence +from workos.types.authorization.organization_membership import ( + AuthorizationOrganizationMembership, +) +from workos.types.list_resource import ListMetadata, ListPage from workos.types.user_management import OrganizationMembership +class MockAuthorizationOrganizationMembershipList( + ListPage[AuthorizationOrganizationMembership] +): + def __init__( + self, + data: Optional[Sequence[AuthorizationOrganizationMembership]] = None, + before: Optional[str] = None, + after: Optional[str] = "om_01DEF", + ): + if data is None: + data = [ + AuthorizationOrganizationMembership( + object="organization_membership", + id="om_01ABC", + user_id="user_123", + organization_id="org_456", + status="active", + created_at="2024-01-01T00:00:00Z", + updated_at="2024-01-01T00:00:00Z", + ), + AuthorizationOrganizationMembership( + object="organization_membership", + id="om_01DEF", + user_id="user_789", + organization_id="org_456", + status="active", + created_at="2024-01-02T00:00:00Z", + updated_at="2024-01-02T00:00:00Z", + ), + ] + super().__init__( + object="list", + data=data, + list_metadata=ListMetadata(before=before, after=after), + ) + + class MockOrganizationMembership(OrganizationMembership): def __init__(self, id): now = datetime.datetime.now().isoformat() diff --git a/tests/utils/fixtures/mock_resource.py b/tests/utils/fixtures/mock_resource.py new file mode 100644 index 00000000..957a01c5 --- /dev/null +++ b/tests/utils/fixtures/mock_resource.py @@ -0,0 +1,30 @@ +from typing import Optional + +from workos.types.authorization.authorization_resource import AuthorizationResource + + +class MockAuthorizationResource(AuthorizationResource): + def __init__( + self, + id: str = "res_01ABC", + external_id: str = "ext_123", + name: str = "Test Resource", + description: Optional[str] = "A test resource for unit tests", + resource_type_slug: str = "document", + organization_id: str = "org_01EHT88Z8J8795GZNQ4ZP1J81T", + parent_resource_id: Optional[str] = "res_01XYZ", + created_at: str = "2024-01-15T12:00:00.000Z", + updated_at: str = "2024-01-15T12:00:00.000Z", + ): + super().__init__( + object="authorization_resource", + id=id, + external_id=external_id, + name=name, + description=description, + resource_type_slug=resource_type_slug, + organization_id=organization_id, + parent_resource_id=parent_resource_id, + created_at=created_at, + updated_at=updated_at, + ) diff --git a/tests/utils/fixtures/mock_resource_list.py b/tests/utils/fixtures/mock_resource_list.py new file mode 100644 index 00000000..a21c5c72 --- /dev/null +++ b/tests/utils/fixtures/mock_resource_list.py @@ -0,0 +1,45 @@ +from typing import Optional, Sequence + +from workos.types.authorization.authorization_resource import AuthorizationResource +from workos.types.list_resource import ListMetadata, ListPage + + +class MockAuthorizationResourceList(ListPage[AuthorizationResource]): + def __init__( + self, + data: Optional[Sequence[AuthorizationResource]] = None, + before: Optional[str] = None, + after: Optional[str] = "authz_resource_01HXYZ123ABC456DEF789DEF", + ): + if data is None: + data = [ + AuthorizationResource( + object="authorization_resource", + id="authz_resource_01HXYZ123ABC456DEF789ABC", + external_id="doc-12345678", + name="Q5 Budget Report", + description="Financial report for Q5 2025", + resource_type_slug="document", + organization_id="org_01HXYZ123ABC456DEF789ABC", + parent_resource_id="authz_resource_01HXYZ123ABC456DEF789XYZ", + created_at="2024-01-15T09:30:00.000Z", + updated_at="2024-01-15T09:30:00.000Z", + ), + AuthorizationResource( + object="authorization_resource", + id="authz_resource_01HXYZ123ABC456DEF789DEF", + external_id="folder-123", + name="Finance Folder", + description=None, + resource_type_slug="folder", + organization_id="org_01HXYZ123ABC456DEF789ABC", + parent_resource_id=None, + created_at="2024-01-14T08:00:00.000Z", + updated_at="2024-01-14T08:00:00.000Z", + ), + ] + super().__init__( + object="list", + data=data, + list_metadata=ListMetadata(before=before, after=after), + ) diff --git a/tests/utils/fixtures/mock_role_assignment.py b/tests/utils/fixtures/mock_role_assignment.py new file mode 100644 index 00000000..e9c789f0 --- /dev/null +++ b/tests/utils/fixtures/mock_role_assignment.py @@ -0,0 +1,68 @@ +from typing import Optional, Sequence + +from workos.types.authorization.role_assignment import ( + RoleAssignment, + RoleAssignmentResource, + RoleAssignmentRole, +) +from workos.types.list_resource import ListMetadata, ListPage + + +class MockRoleAssignment(RoleAssignment): + def __init__( + self, + id: str = "ra_01ABC", + role_slug: str = "admin", + resource_id: str = "res_01ABC", + resource_external_id: str = "ext_123", + resource_type_slug: str = "document", + created_at: str = "2024-01-01T00:00:00Z", + updated_at: str = "2024-01-01T00:00:00Z", + ): + super().__init__( + object="role_assignment", + id=id, + role=RoleAssignmentRole(slug=role_slug), + resource=RoleAssignmentResource( + id=resource_id, + external_id=resource_external_id, + resource_type_slug=resource_type_slug, + ), + created_at=created_at, + updated_at=updated_at, + ) + + +class MockRoleAssignmentsList(ListPage[RoleAssignment]): + def __init__( + self, + data: Optional[Sequence[RoleAssignment]] = None, + before: Optional[str] = None, + after: Optional[str] = "ra_01DEF", + ): + if data is None: + data = [ + MockRoleAssignment( + id="ra_01ABC", + role_slug="admin", + resource_id="res_01ABC", + resource_external_id="ext_123", + resource_type_slug="document", + created_at="2024-01-15T09:30:00.000Z", + updated_at="2024-01-15T09:30:00.000Z", + ), + MockRoleAssignment( + id="ra_01DEF", + role_slug="editor", + resource_id="res_01XYZ", + resource_external_id="ext_456", + resource_type_slug="folder", + created_at="2024-01-14T08:00:00.000Z", + updated_at="2024-01-14T08:00:00.000Z", + ), + ] + super().__init__( + object="list", + data=data, + list_metadata=ListMetadata(before=before, after=after), + )