Merge branch 'main' into feature/session-helpers

Author: Paul Asjes

tests/test_sso.py

@@ -38,6 +38,16 @@ def mock_magic_link_profile(self):
     def mock_connection(self):
         return MockConnection("conn_01E4ZCR3C56J083X43JQXF3JK5").dict()
 
+    @pytest.fixture
+    def mock_connection_updated(self):
+        connection = MockConnection("conn_01FHT48Z8J8295GZNQ4ZP1J81T").dict()
+
+        connection["options"] = {
+            "signing_cert": "signing_cert",
+        }
+
+        return connection
+
     @pytest.fixture
     def mock_connections(self):
         connection_list = [MockConnection(id=str(i)).dict() for i in range(10)]
@@ -339,6 +349,33 @@ def test_list_connections_with_connection_type(
             "order": "desc",
         }
 
+    def test_update_connection(
+        self, mock_connection_updated, capture_and_mock_http_client_request
+    ):
+        request_kwargs = capture_and_mock_http_client_request(
+            self.http_client, mock_connection_updated, 200
+        )
+
+        updated_connection = syncify(
+            self.sso.update_connection(
+                connection_id="conn_01EHT88Z8J8795GZNQ4ZP1J81T",
+                saml_options_signing_key="signing_key",
+                saml_options_signing_cert="signing_cert",
+            )
+        )
+
+        assert request_kwargs["url"].endswith(
+            "/connections/conn_01EHT88Z8J8795GZNQ4ZP1J81T"
+        )
+
+        assert request_kwargs["method"] == "put"
+        assert request_kwargs["json"] == {
+            "options": {"signing_key": "signing_key", "signing_cert": "signing_cert"}
+        }
+        assert updated_connection.id == "conn_01FHT48Z8J8295GZNQ4ZP1J81T"
+        assert updated_connection.name == "Foo Corporation"
+        assert updated_connection.options.signing_cert == "signing_cert"
+
     def test_delete_connection(self, capture_and_mock_http_client_request):
         request_kwargs = capture_and_mock_http_client_request(
             self.http_client,

tests/test_user_management.py

@@ -215,6 +215,26 @@ def test_authorization_url_has_expected_query_params_with_provider(self):
             "response_type": RESPONSE_TYPE_CODE,
         }
 
+    def test_authorization_url_has_expected_query_params_with_prompt(self):
+        provider = "GoogleOAuth"
+        redirect_uri = "https://localhost/auth/callback"
+        prompt = "consent"
+        authorization_url = self.user_management.get_authorization_url(
+            provider=provider,
+            redirect_uri=redirect_uri,
+            prompt=prompt,
+        )
+
+        parsed_url = urlparse(authorization_url)
+        assert parsed_url.path == "/user_management/authorize"
+        assert dict(parse_qsl(str(parsed_url.query))) == {
+            "client_id": self.http_client.client_id,
+            "redirect_uri": redirect_uri,
+            "response_type": RESPONSE_TYPE_CODE,
+            "provider": provider,
+            "prompt": prompt,
+        }
+
     def test_authorization_url_has_expected_query_params_with_domain_hint(self):
         connection_id = "connection_123"
         redirect_uri = "https://localhost/auth/callback"

tests/utils/fixtures/mock_connection.py

@@ -1,5 +1,9 @@
 import datetime
-from workos.types.sso import ConnectionDomain, ConnectionWithDomains
+from workos.types.sso import (
+    ConnectionDomain,
+    ConnectionWithDomains,
+    SamlConnectionOptions,
+)
 
 
 class MockConnection(ConnectionWithDomains):
@@ -14,6 +18,7 @@ def __init__(self, id):
             state="active",
             created_at=now,
             updated_at=now,
+            options=SamlConnectionOptions(signing_cert="signing_cert"),
             domains=[
                 ConnectionDomain(
                     id="connection_domain_abc123",

workos/__about__.py

@@ -12,7 +12,7 @@
 
 __package_url__ = "https://github.com/workos-inc/workos-python"
 
-__version__ = "5.7.0"
+__version__ = "5.9.0"
 
 __author__ = "WorkOS"
 

workos/sso.py

@@ -14,6 +14,7 @@
     REQUEST_METHOD_POST,
     QueryParameters,
     RequestHelper,
+    REQUEST_METHOD_PUT,
 )
 from workos.types.list_resource import (
     ListArgs,
@@ -167,11 +168,29 @@ def list_connections(
         """
         ...
 
+    def update_connection(
+        self,
+        *,
+        connection_id: str,
+        saml_options_signing_key: Optional[str] = None,
+        saml_options_signing_cert: Optional[str] = None,
+    ) -> SyncOrAsync[ConnectionWithDomains]:
+        """Updates a single connection
+
+        Args:
+            connection_id (str): Connection unique identifier
+            saml_options_signing_key (str): Signing key for the connection (Optional)
+            saml_options_signing_cert (str): Signing certificate for the connection (Optional)
+        Returns:
+            None
+        """
+        ...
+
     def delete_connection(self, connection_id: str) -> SyncOrAsync[None]:
         """Deletes a single Connection
 
         Args:
-            connection (str): Connection unique identifier
+            connection_id (str): Connection unique identifier
 
         Returns:
             None
@@ -255,6 +274,28 @@ def list_connections(
             **ListPage[ConnectionWithDomains](**response).model_dump(),
         )
 
+    def update_connection(
+        self,
+        *,
+        connection_id: str,
+        saml_options_signing_key: Optional[str] = None,
+        saml_options_signing_cert: Optional[str] = None,
+    ) -> ConnectionWithDomains:
+        json = {
+            "options": {
+                "signing_key": saml_options_signing_key,
+                "signing_cert": saml_options_signing_cert,
+            }
+        }
+
+        response = self._http_client.request(
+            f"connections/{connection_id}",
+            method=REQUEST_METHOD_PUT,
+            json=json,
+        )
+
+        return ConnectionWithDomains.model_validate(response)
+
     def delete_connection(self, connection_id: str) -> None:
         self._http_client.request(
             f"connections/{connection_id}", method=REQUEST_METHOD_DELETE
@@ -335,6 +376,28 @@ async def list_connections(
             **ListPage[ConnectionWithDomains](**response).model_dump(),
         )
 
+    async def update_connection(
+        self,
+        *,
+        connection_id: str,
+        saml_options_signing_key: Optional[str] = None,
+        saml_options_signing_cert: Optional[str] = None,
+    ) -> ConnectionWithDomains:
+        json = {
+            "options": {
+                "signing_key": saml_options_signing_key,
+                "signing_cert": saml_options_signing_cert,
+            }
+        }
+
+        response = await self._http_client.request(
+            f"connections/{connection_id}",
+            method=REQUEST_METHOD_PUT,
+            json=json,
+        )
+
+        return ConnectionWithDomains.model_validate(response)
+
     async def delete_connection(self, connection_id: str) -> None:
         await self._http_client.request(
             f"connections/{connection_id}",

workos/types/sso/connection.py

@@ -1,4 +1,4 @@
-from typing import Literal, Sequence
+from typing import Literal, Sequence, Optional
 from workos.types.sso.connection_domain import ConnectionDomain
 from workos.types.workos_model import WorkOSModel
 from workos.typing.literals import LiteralOrUntyped
@@ -45,6 +45,12 @@
 ]
 
 
+class SamlConnectionOptions(WorkOSModel):
+    """Representation of options payload of a Connection Response."""
+
+    signing_cert: Optional[str]
+
+
 class Connection(WorkOSModel):
     object: Literal["connection"]
     id: str
@@ -54,6 +60,7 @@ class Connection(WorkOSModel):
     state: LiteralOrUntyped[ConnectionState]
     created_at: str
     updated_at: str
+    options: Optional[SamlConnectionOptions] = None
 
 
 class ConnectionWithDomains(Connection):

workos/user_management.py

@@ -341,6 +341,7 @@ def get_authorization_url(
         connection_id: Optional[str] = None,
         organization_id: Optional[str] = None,
         code_challenge: Optional[str] = None,
+        prompt: Optional[str] = None,
     ) -> str:
         """Generate an OAuth 2.0 authorization URL.
 
@@ -365,6 +366,9 @@ def get_authorization_url(
             state (str): An encoded string passed to WorkOS that'd be preserved through the authentication workflow, passed
                 back as a query parameter. (Optional)
             code_challenge (str): Code challenge is derived from the code verifier used for the PKCE flow. (Optional)
+            prompt (str): Used to specify whether the upstream provider should prompt the user for credentials or other
+                consent. Valid values depend on the provider. Currently only applies to provider values of 'GoogleOAuth',
+                'MicrosoftOAuth', or 'GitHubOAuth'. (Optional)
 
         Returns:
             str: URL to redirect a User to to begin the OAuth workflow with WorkOS
@@ -395,6 +399,8 @@ def get_authorization_url(
         if code_challenge:
             params["code_challenge"] = code_challenge
             params["code_challenge_method"] = "S256"
+        if prompt is not None:
+            params["prompt"] = prompt
 
         return RequestHelper.build_url_with_query_params(
             base_url=self._client_configuration.base_url,