diff --git a/src/workos/_base_client.py b/src/workos/_base_client.py index 24d2f5e8..d883aec2 100644 --- a/src/workos/_base_client.py +++ b/src/workos/_base_client.py @@ -6,6 +6,7 @@ from workos.api_keys import ApiKeysModule from workos.audit_logs import AuditLogsModule from workos.authorization import AuthorizationModule +from workos.connect import ConnectModule from workos.directory_sync import DirectorySyncModule from workos.events import EventsModule from workos.fga import FGAModule @@ -79,6 +80,10 @@ def api_keys(self) -> ApiKeysModule: ... @abstractmethod def authorization(self) -> AuthorizationModule: ... + @property + @abstractmethod + def connect(self) -> ConnectModule: ... + @property @abstractmethod def audit_logs(self) -> AuditLogsModule: ... diff --git a/src/workos/async_client.py b/src/workos/async_client.py index c050865f..d2b3921c 100644 --- a/src/workos/async_client.py +++ b/src/workos/async_client.py @@ -4,6 +4,7 @@ from workos.api_keys import AsyncApiKeys from workos.audit_logs import AsyncAuditLogs from workos.authorization import AsyncAuthorization +from workos.connect import AsyncConnect from workos.directory_sync import AsyncDirectorySync from workos.events import AsyncEvents from workos.fga import FGAModule @@ -50,6 +51,12 @@ def __init__( timeout=self.request_timeout, ) + @property + def connect(self) -> AsyncConnect: + if not getattr(self, "_connect", None): + self._connect = AsyncConnect(self._http_client) + return self._connect + @property def api_keys(self) -> AsyncApiKeys: if not getattr(self, "_api_keys", None): diff --git a/src/workos/client.py b/src/workos/client.py index 68b9a287..7e942924 100644 --- a/src/workos/client.py +++ b/src/workos/client.py @@ -4,6 +4,7 @@ from workos.api_keys import ApiKeys from workos.audit_logs import AuditLogs from workos.authorization import Authorization +from workos.connect import Connect from workos.directory_sync import DirectorySync from workos.fga import FGA from workos.organizations import Organizations @@ -50,6 +51,12 @@ def __init__( timeout=self.request_timeout, ) + @property + def connect(self) -> Connect: + if not getattr(self, "_connect", None): + self._connect = Connect(self._http_client) + return self._connect + @property def api_keys(self) -> ApiKeys: if not getattr(self, "_api_keys", None): diff --git a/src/workos/connect.py b/src/workos/connect.py new file mode 100644 index 00000000..19db5c51 --- /dev/null +++ b/src/workos/connect.py @@ -0,0 +1,479 @@ +from functools import partial +from typing import Optional, Protocol, Sequence + +from workos.types.connect import ClientSecret, ConnectApplication +from workos.types.connect.connect_application import ApplicationType +from workos.types.connect.list_filters import ( + ClientSecretListFilters, + ConnectApplicationListFilters, +) +from workos.types.list_resource import ListMetadata, ListPage, WorkOSListResource +from workos.typing.sync_or_async import SyncOrAsync +from workos.utils.http_client import AsyncHTTPClient, SyncHTTPClient +from workos.utils.pagination_order import PaginationOrder +from workos.utils.request_helper import ( + DEFAULT_LIST_RESPONSE_LIMIT, + REQUEST_METHOD_DELETE, + REQUEST_METHOD_GET, + REQUEST_METHOD_POST, + REQUEST_METHOD_PUT, +) + +CONNECT_APPLICATIONS_PATH = "connect/applications" +CONNECT_CLIENT_SECRETS_PATH = "connect/client_secrets" + +ConnectApplicationsListResource = WorkOSListResource[ + ConnectApplication, ConnectApplicationListFilters, ListMetadata +] + +ClientSecretsListResource = WorkOSListResource[ + ClientSecret, ClientSecretListFilters, ListMetadata +] + + +class ConnectModule(Protocol): + """Offers methods through the WorkOS Connect service.""" + + def list_applications( + self, + *, + organization_id: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> SyncOrAsync[ConnectApplicationsListResource]: + """Retrieve a list of connect applications. + + Kwargs: + organization_id (str): Filter by organization ID. (Optional) + limit (int): Maximum number of records to return. (Optional) + before (str): Pagination cursor to receive records before a provided ID. (Optional) + after (str): Pagination cursor to receive records after a provided ID. (Optional) + order (Literal["asc","desc"]): Sort records in either ascending or descending order. (Optional) + + Returns: + ConnectApplicationsListResource: Applications list response from WorkOS. + """ + ... + + def get_application(self, application_id: str) -> SyncOrAsync[ConnectApplication]: + """Gets details for a single connect application. + + Args: + application_id (str): Application ID or client ID. + + Returns: + ConnectApplication: Application response from WorkOS. + """ + ... + + def create_application( + self, + *, + name: str, + application_type: ApplicationType, + is_first_party: bool, + description: Optional[str] = None, + scopes: Optional[Sequence[str]] = None, + redirect_uris: Optional[Sequence[str]] = None, + uses_pkce: Optional[bool] = None, + organization_id: Optional[str] = None, + ) -> SyncOrAsync[ConnectApplication]: + """Create a connect application. + + Kwargs: + name (str): Application name. + application_type (ApplicationType): "oauth" or "m2m". + is_first_party (bool): Whether this is a first-party application. + description (str): Application description. (Optional) + scopes (Sequence[str]): Permission slugs. (Optional) + redirect_uris (Sequence[str]): OAuth redirect URIs. (Optional) + uses_pkce (bool): PKCE support (OAuth only). (Optional) + organization_id (str): Organization ID. Required for M2M and third-party OAuth. (Optional) + + Returns: + ConnectApplication: Created application response from WorkOS. + """ + ... + + def update_application( + self, + *, + application_id: str, + name: Optional[str] = None, + description: Optional[str] = None, + scopes: Optional[Sequence[str]] = None, + redirect_uris: Optional[Sequence[str]] = None, + ) -> SyncOrAsync[ConnectApplication]: + """Update a connect application. + + Kwargs: + application_id (str): Application ID or client ID. + name (str): Updated application name. (Optional) + description (str): Updated description. Pass None to clear. (Optional) + scopes (Sequence[str]): Updated permission slugs. (Optional) + redirect_uris (Sequence[str]): Updated OAuth redirect URIs. (Optional) + + Returns: + ConnectApplication: Updated application response from WorkOS. + """ + ... + + def delete_application(self, application_id: str) -> SyncOrAsync[None]: + """Delete a connect application. + + Args: + application_id (str): Application ID or client ID. + + Returns: + None + """ + ... + + def create_client_secret(self, application_id: str) -> SyncOrAsync[ClientSecret]: + """Create a client secret for a connect application. + + Args: + application_id (str): Application ID or client ID. + + Returns: + ClientSecret: Created client secret response from WorkOS. + """ + ... + + def list_client_secrets( + self, + application_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> SyncOrAsync[ClientSecretsListResource]: + """List client secrets for a connect application. + + Args: + application_id (str): Application ID or client ID. + + Kwargs: + limit (int): Maximum number of records to return. (Optional) + before (str): Pagination cursor to receive records before a provided ID. (Optional) + after (str): Pagination cursor to receive records after a provided ID. (Optional) + order (Literal["asc","desc"]): Sort records in either ascending or descending order. (Optional) + + Returns: + ClientSecretsListResource: Client secrets list response from WorkOS. + """ + ... + + def delete_client_secret(self, client_secret_id: str) -> SyncOrAsync[None]: + """Delete a client secret. + + Args: + client_secret_id (str): Client secret ID. + + Returns: + None + """ + ... + + +class Connect(ConnectModule): + _http_client: SyncHTTPClient + + def __init__(self, http_client: SyncHTTPClient): + self._http_client = http_client + + def list_applications( + self, + *, + organization_id: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> ConnectApplicationsListResource: + list_params: ConnectApplicationListFilters = { + "organization_id": organization_id, + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + response = self._http_client.request( + CONNECT_APPLICATIONS_PATH, + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + ConnectApplication, ConnectApplicationListFilters, ListMetadata + ]( + list_method=self.list_applications, + list_args=list_params, + **ListPage[ConnectApplication](**response).model_dump(), + ) + + def get_application(self, application_id: str) -> ConnectApplication: + response = self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}", + method=REQUEST_METHOD_GET, + ) + + return ConnectApplication.model_validate(response) + + def create_application( + self, + *, + name: str, + application_type: ApplicationType, + is_first_party: bool, + description: Optional[str] = None, + scopes: Optional[Sequence[str]] = None, + redirect_uris: Optional[Sequence[str]] = None, + uses_pkce: Optional[bool] = None, + organization_id: Optional[str] = None, + ) -> ConnectApplication: + json = { + "name": name, + "application_type": application_type, + "is_first_party": is_first_party, + "description": description, + "scopes": scopes, + "redirect_uris": redirect_uris, + "uses_pkce": uses_pkce, + "organization_id": organization_id, + } + + response = self._http_client.request( + CONNECT_APPLICATIONS_PATH, + method=REQUEST_METHOD_POST, + json=json, + ) + + return ConnectApplication.model_validate(response) + + def update_application( + self, + *, + application_id: str, + name: Optional[str] = None, + description: Optional[str] = None, + scopes: Optional[Sequence[str]] = None, + redirect_uris: Optional[Sequence[str]] = None, + ) -> ConnectApplication: + json = { + "name": name, + "description": description, + "scopes": scopes, + "redirect_uris": redirect_uris, + } + + response = self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}", + method=REQUEST_METHOD_PUT, + json=json, + ) + + return ConnectApplication.model_validate(response) + + def delete_application(self, application_id: str) -> None: + self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}", + method=REQUEST_METHOD_DELETE, + ) + + def create_client_secret(self, application_id: str) -> ClientSecret: + response = self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}/client_secrets", + method=REQUEST_METHOD_POST, + json={}, + ) + + return ClientSecret.model_validate(response) + + def list_client_secrets( + self, + application_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> ClientSecretsListResource: + list_params: ClientSecretListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + response = self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}/client_secrets", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ClientSecret, ClientSecretListFilters, ListMetadata]( + list_method=partial(self.list_client_secrets, application_id), + list_args=list_params, + **ListPage[ClientSecret](**response).model_dump(), + ) + + def delete_client_secret(self, client_secret_id: str) -> None: + self._http_client.request( + f"{CONNECT_CLIENT_SECRETS_PATH}/{client_secret_id}", + method=REQUEST_METHOD_DELETE, + ) + + +class AsyncConnect(ConnectModule): + _http_client: AsyncHTTPClient + + def __init__(self, http_client: AsyncHTTPClient): + self._http_client = http_client + + async def list_applications( + self, + *, + organization_id: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> ConnectApplicationsListResource: + list_params: ConnectApplicationListFilters = { + "organization_id": organization_id, + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + response = await self._http_client.request( + CONNECT_APPLICATIONS_PATH, + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + ConnectApplication, ConnectApplicationListFilters, ListMetadata + ]( + list_method=self.list_applications, + list_args=list_params, + **ListPage[ConnectApplication](**response).model_dump(), + ) + + async def get_application(self, application_id: str) -> ConnectApplication: + response = await self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}", + method=REQUEST_METHOD_GET, + ) + + return ConnectApplication.model_validate(response) + + async def create_application( + self, + *, + name: str, + application_type: ApplicationType, + is_first_party: bool, + description: Optional[str] = None, + scopes: Optional[Sequence[str]] = None, + redirect_uris: Optional[Sequence[str]] = None, + uses_pkce: Optional[bool] = None, + organization_id: Optional[str] = None, + ) -> ConnectApplication: + json = { + "name": name, + "application_type": application_type, + "is_first_party": is_first_party, + "description": description, + "scopes": scopes, + "redirect_uris": redirect_uris, + "uses_pkce": uses_pkce, + "organization_id": organization_id, + } + + response = await self._http_client.request( + CONNECT_APPLICATIONS_PATH, + method=REQUEST_METHOD_POST, + json=json, + ) + + return ConnectApplication.model_validate(response) + + async def update_application( + self, + *, + application_id: str, + name: Optional[str] = None, + description: Optional[str] = None, + scopes: Optional[Sequence[str]] = None, + redirect_uris: Optional[Sequence[str]] = None, + ) -> ConnectApplication: + json = { + "name": name, + "description": description, + "scopes": scopes, + "redirect_uris": redirect_uris, + } + + response = await self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}", + method=REQUEST_METHOD_PUT, + json=json, + ) + + return ConnectApplication.model_validate(response) + + async def delete_application(self, application_id: str) -> None: + await self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}", + method=REQUEST_METHOD_DELETE, + ) + + async def create_client_secret(self, application_id: str) -> ClientSecret: + response = await self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}/client_secrets", + method=REQUEST_METHOD_POST, + json={}, + ) + + return ClientSecret.model_validate(response) + + async def list_client_secrets( + self, + application_id: str, + *, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> ClientSecretsListResource: + list_params: ClientSecretListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + } + + response = await self._http_client.request( + f"{CONNECT_APPLICATIONS_PATH}/{application_id}/client_secrets", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ClientSecret, ClientSecretListFilters, ListMetadata]( + list_method=partial(self.list_client_secrets, application_id), + list_args=list_params, + **ListPage[ClientSecret](**response).model_dump(), + ) + + async def delete_client_secret(self, client_secret_id: str) -> None: + await self._http_client.request( + f"{CONNECT_CLIENT_SECRETS_PATH}/{client_secret_id}", + method=REQUEST_METHOD_DELETE, + ) diff --git a/src/workos/types/connect/__init__.py b/src/workos/types/connect/__init__.py new file mode 100644 index 00000000..87fce74b --- /dev/null +++ b/src/workos/types/connect/__init__.py @@ -0,0 +1,2 @@ +from .connect_application import * +from .client_secret import * diff --git a/src/workos/types/connect/client_secret.py b/src/workos/types/connect/client_secret.py new file mode 100644 index 00000000..338c1b62 --- /dev/null +++ b/src/workos/types/connect/client_secret.py @@ -0,0 +1,13 @@ +from typing import Literal, Optional + +from workos.types.workos_model import WorkOSModel + + +class ClientSecret(WorkOSModel): + object: Literal["connect_application_secret"] + id: str + secret: Optional[str] = None + secret_hint: str + last_used_at: Optional[str] = None + created_at: str + updated_at: str diff --git a/src/workos/types/connect/connect_application.py b/src/workos/types/connect/connect_application.py new file mode 100644 index 00000000..4f4a5c83 --- /dev/null +++ b/src/workos/types/connect/connect_application.py @@ -0,0 +1,29 @@ +from typing import Literal, Optional, Sequence + +from workos.types.workos_model import WorkOSModel +from workos.typing.literals import LiteralOrUntyped + + +ApplicationType = Literal["oauth", "m2m"] + + +class RedirectUri(WorkOSModel): + uri: str + default: Optional[bool] = None + + +class ConnectApplication(WorkOSModel): + object: Literal["connect_application"] + id: str + client_id: str + name: str + description: Optional[str] = None + application_type: LiteralOrUntyped[ApplicationType] + organization_id: Optional[str] = None + scopes: Sequence[str] = [] + created_at: str + updated_at: str + redirect_uris: Optional[Sequence[RedirectUri]] = None + uses_pkce: Optional[bool] = None + is_first_party: Optional[bool] = None + was_dynamically_registered: Optional[bool] = None diff --git a/src/workos/types/connect/list_filters.py b/src/workos/types/connect/list_filters.py new file mode 100644 index 00000000..ff93ca1a --- /dev/null +++ b/src/workos/types/connect/list_filters.py @@ -0,0 +1,11 @@ +from typing import Optional + +from workos.types.list_resource import ListArgs + + +class ConnectApplicationListFilters(ListArgs, total=False): + organization_id: Optional[str] + + +class ClientSecretListFilters(ListArgs, total=False): + pass diff --git a/src/workos/types/list_resource.py b/src/workos/types/list_resource.py index e2f21841..c1eb22f3 100644 --- a/src/workos/types/list_resource.py +++ b/src/workos/types/list_resource.py @@ -25,6 +25,7 @@ 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.connect import ClientSecret, ConnectApplication from workos.types.directory_sync import ( Directory, DirectoryGroup, @@ -54,6 +55,8 @@ AuditLogAction, AuditLogSchema, AuthenticationFactor, + ClientSecret, + ConnectApplication, ConnectionWithDomains, Directory, DirectoryGroup, diff --git a/tests/test_connect.py b/tests/test_connect.py new file mode 100644 index 00000000..8ae56dc7 --- /dev/null +++ b/tests/test_connect.py @@ -0,0 +1,297 @@ +from typing import Union +import pytest +from tests.types.test_auto_pagination_function import TestAutoPaginationFunction +from tests.utils.fixtures.mock_client_secret import MockClientSecret +from tests.utils.fixtures.mock_connect_application import MockConnectApplication +from tests.utils.list_resource import list_response_of +from tests.utils.syncify import syncify +from workos.connect import AsyncConnect, Connect + + +@pytest.mark.sync_and_async(Connect, AsyncConnect) +class TestConnect: + @pytest.fixture(autouse=True) + def setup(self, module_instance: Union[Connect, AsyncConnect]): + self.http_client = module_instance._http_client + self.connect = module_instance + + @pytest.fixture + def mock_application(self): + return MockConnectApplication("app_01ABC").dict() + + @pytest.fixture + def mock_oauth_application(self): + return MockConnectApplication("app_01ABC", application_type="oauth").dict() + + @pytest.fixture + def mock_applications(self): + application_list = [MockConnectApplication(id=str(i)).dict() for i in range(10)] + return { + "data": application_list, + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + + @pytest.fixture + def mock_applications_multiple_data_pages(self): + applications_list = [ + MockConnectApplication(id=f"app_{i + 1}").dict() for i in range(40) + ] + return list_response_of(data=applications_list) + + @pytest.fixture + def mock_client_secret(self): + return MockClientSecret("cs_01ABC", include_secret=True).dict() + + @pytest.fixture + def mock_client_secrets(self): + secret_list = [MockClientSecret(id=f"cs_{i}").dict() for i in range(10)] + return { + "data": secret_list, + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + + @pytest.fixture + def mock_client_secrets_multiple_data_pages(self): + secrets_list = [MockClientSecret(id=f"cs_{i + 1}").dict() for i in range(40)] + return list_response_of(data=secrets_list) + + # --- Application Tests --- + + def test_list_applications( + self, mock_applications, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_applications, 200 + ) + + response = syncify(self.connect.list_applications()) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith("/connect/applications") + assert list(map(lambda x: x.dict(), response.data)) == mock_applications["data"] + + def test_list_applications_with_organization_id( + self, mock_applications, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_applications, 200 + ) + + syncify(self.connect.list_applications(organization_id="org_01ABC")) + + assert request_kwargs["method"] == "get" + assert request_kwargs["params"]["organization_id"] == "org_01ABC" + + def test_get_application( + self, mock_application, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_application, 200 + ) + + application = syncify(self.connect.get_application(application_id="app_01ABC")) + + assert application.dict() == mock_application + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith("/connect/applications/app_01ABC") + + def test_get_oauth_application( + self, mock_oauth_application, capture_and_mock_http_client_request + ): + capture_and_mock_http_client_request( + self.http_client, mock_oauth_application, 200 + ) + + application = syncify(self.connect.get_application(application_id="app_01ABC")) + + assert application.dict() == mock_oauth_application + assert application.application_type == "oauth" + assert application.redirect_uris is not None + assert application.uses_pkce is True + assert application.is_first_party is True + + def test_create_m2m_application( + self, mock_application, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_application, 201 + ) + + application = syncify( + self.connect.create_application( + name="Test Application", + application_type="m2m", + is_first_party=True, + organization_id="org_01ABC", + scopes=["read", "write"], + ) + ) + + assert application.id == "app_01ABC" + assert application.name == "Test Application" + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith("/connect/applications") + assert request_kwargs["json"]["name"] == "Test Application" + assert request_kwargs["json"]["application_type"] == "m2m" + assert request_kwargs["json"]["organization_id"] == "org_01ABC" + + def test_create_oauth_application( + self, mock_oauth_application, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_oauth_application, 201 + ) + + application = syncify( + self.connect.create_application( + name="Test Application", + application_type="oauth", + is_first_party=True, + redirect_uris=["https://example.com/callback"], + uses_pkce=True, + ) + ) + + assert application.application_type == "oauth" + assert request_kwargs["method"] == "post" + assert request_kwargs["json"]["application_type"] == "oauth" + assert request_kwargs["json"]["redirect_uris"] == [ + "https://example.com/callback" + ] + + def test_update_application( + self, mock_application, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_application, 200 + ) + + syncify( + self.connect.update_application( + application_id="app_01ABC", + name="Updated Name", + scopes=["read"], + ) + ) + + assert request_kwargs["method"] == "put" + assert request_kwargs["url"].endswith("/connect/applications/app_01ABC") + assert request_kwargs["json"]["name"] == "Updated Name" + assert request_kwargs["json"]["scopes"] == ["read"] + + def test_delete_application(self, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + None, + 202, + headers={"content-type": "text/plain; charset=utf-8"}, + ) + + response = syncify(self.connect.delete_application(application_id="app_01ABC")) + + assert request_kwargs["url"].endswith("/connect/applications/app_01ABC") + assert request_kwargs["method"] == "delete" + assert response is None + + def test_list_applications_auto_pagination_for_single_page( + self, + mock_applications, + test_auto_pagination: TestAutoPaginationFunction, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.connect.list_applications, + expected_all_page_data=mock_applications["data"], + ) + + def test_list_applications_auto_pagination_for_multiple_pages( + self, + mock_applications_multiple_data_pages, + test_auto_pagination: TestAutoPaginationFunction, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.connect.list_applications, + expected_all_page_data=mock_applications_multiple_data_pages["data"], + ) + + # --- Client Secret Tests --- + + def test_create_client_secret( + self, mock_client_secret, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_client_secret, 201 + ) + + secret = syncify(self.connect.create_client_secret(application_id="app_01ABC")) + + assert secret.id == "cs_01ABC" + assert secret.secret == "sk_test_secret_value_123" + assert secret.secret_hint == "...abcd" + assert request_kwargs["method"] == "post" + assert request_kwargs["url"].endswith( + "/connect/applications/app_01ABC/client_secrets" + ) + assert request_kwargs["json"] == {} + + def test_list_client_secrets( + self, mock_client_secrets, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_client_secrets, 200 + ) + + response = syncify(self.connect.list_client_secrets(application_id="app_01ABC")) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/connect/applications/app_01ABC/client_secrets" + ) + assert ( + list(map(lambda x: x.dict(), response.data)) == mock_client_secrets["data"] + ) + + def test_delete_client_secret(self, capture_and_mock_http_client_request): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + None, + 202, + headers={"content-type": "text/plain; charset=utf-8"}, + ) + + response = syncify( + self.connect.delete_client_secret(client_secret_id="cs_01ABC") + ) + + assert request_kwargs["url"].endswith("/connect/client_secrets/cs_01ABC") + assert request_kwargs["method"] == "delete" + assert response is None + + def test_list_client_secrets_auto_pagination_for_single_page( + self, + mock_client_secrets, + test_auto_pagination: TestAutoPaginationFunction, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.connect.list_client_secrets, + expected_all_page_data=mock_client_secrets["data"], + list_function_params={"application_id": "app_01ABC"}, + url_path_keys=["application_id"], + ) + + def test_list_client_secrets_auto_pagination_for_multiple_pages( + self, + mock_client_secrets_multiple_data_pages, + test_auto_pagination: TestAutoPaginationFunction, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.connect.list_client_secrets, + expected_all_page_data=mock_client_secrets_multiple_data_pages["data"], + list_function_params={"application_id": "app_01ABC"}, + url_path_keys=["application_id"], + ) diff --git a/tests/utils/fixtures/mock_client_secret.py b/tests/utils/fixtures/mock_client_secret.py new file mode 100644 index 00000000..70bec678 --- /dev/null +++ b/tests/utils/fixtures/mock_client_secret.py @@ -0,0 +1,19 @@ +import datetime + +from workos.types.connect import ClientSecret + + +class MockClientSecret(ClientSecret): + def __init__(self, id: str, include_secret: bool = False): + now = datetime.datetime.now().isoformat() + kwargs = { + "object": "connect_application_secret", + "id": id, + "secret_hint": "...abcd", + "last_used_at": None, + "created_at": now, + "updated_at": now, + } + if include_secret: + kwargs["secret"] = "sk_test_secret_value_123" + super().__init__(**kwargs) diff --git a/tests/utils/fixtures/mock_connect_application.py b/tests/utils/fixtures/mock_connect_application.py new file mode 100644 index 00000000..f4c12305 --- /dev/null +++ b/tests/utils/fixtures/mock_connect_application.py @@ -0,0 +1,29 @@ +import datetime + +from workos.types.connect import ConnectApplication +from workos.types.connect.connect_application import ApplicationType + + +class MockConnectApplication(ConnectApplication): + def __init__(self, id: str, application_type: ApplicationType = "m2m"): + now = datetime.datetime.now().isoformat() + kwargs = { + "object": "connect_application", + "id": id, + "client_id": f"client_{id}", + "name": "Test Application", + "application_type": application_type, + "scopes": ["read", "write"], + "created_at": now, + "updated_at": now, + } + if application_type == "m2m": + kwargs["organization_id"] = "org_01ABC" + elif application_type == "oauth": + kwargs["redirect_uris"] = [ + {"uri": "https://example.com/callback", "default": True} + ] + kwargs["uses_pkce"] = True + kwargs["is_first_party"] = True + kwargs["was_dynamically_registered"] = False + super().__init__(**kwargs)