fix: create_camunda_cloud_channel

pyzeebe/channel/camunda_cloud_channel.py

@@ -56,10 +56,10 @@ def _create_camunda_cloud_credentials(
 ) -> grpc.ChannelCredentials:
     try:
         access_token = _get_access_token(
-            "https://login.cloud.camunda.io/oauth/token",
-            client_id,
-            client_secret,
-            f"{cluster_id}.{region}.zeebe.camunda.io",
+            url="https://login.cloud.camunda.io/oauth/token",
+            client_id=client_id,
+            client_secret=client_secret,
+            audience="zeebe.camunda.io",
         )
         return _create_oauth_credentials(access_token)
     except InvalidOAuthCredentialsError as oauth_error:

pyzeebe/channel/oauth_channel.py

@@ -81,9 +81,9 @@ def create_camunda_cloud_channel(
     client_secret: str,
     cluster_id: str,
     region: str = "bru-2",
-    scope: str = "Zeebe",
     authorization_server: str = "https://login.cloud.camunda.io/oauth/token",
-    audience: str = "zeebe.camunda.io",
+    scope: str | None = None,
+    audience: str | None = "zeebe.camunda.io",
     channel_credentials: grpc.ChannelCredentials | None = None,
     channel_options: ChannelArgumentType | None = None,
     leeway: int = 60,
@@ -96,10 +96,10 @@ def create_camunda_cloud_channel(
         client_secret (str): The client secret.
         cluster_id (str): The ID of the cluster to connect to.
         region (Optional[str]): The region of the cluster. Defaults to "bru-2".
-        scope (Optional[str]): The scope of the access request. Defaults to "Zeebe".
         authorization_server (Optional[str]): The authorization server issuing access tokens
             to the client after successfully authenticating the client.
             Defaults to "https://login.cloud.camunda.io/oauth/token".
+        scope (Optional[str]): The scope of the access request. Can be set to CAMUNDA_CLUSTER_ID. Defaults to None.
         audience (Optional[str]): The audience for authentication. Defaults to "zeebe.camunda.io".
 
         channel_credentials (grpc.ChannelCredentials): The gRPC channel credentials.

tests/unit/channel/camunda_cloud_channel_test.py

@@ -89,9 +89,7 @@ def test_gets_access_token_using_correct_parameters(
         cluster_id: str,
         region: str,
     ):
-        expected_request_body = (
-            f"client_id={client_id}&client_secret={client_secret}&audience={cluster_id}.{region}.zeebe.camunda.io"
-        )
+        expected_request_body = f"client_id={client_id}&client_secret={client_secret}&audience=zeebe.camunda.io"
 
         create_camunda_cloud_channel(client_id, client_secret, cluster_id, region)
 

tests/unit/channel/oauth_channel_test.py

@@ -23,7 +23,21 @@ def test_create_oauth2_client_credentials_channel(
     client_id = "client_id"
     client_secret = "client_secret"
     authorization_server = "https://authorization.server"
-    channel = create_oauth2_client_credentials_channel(grpc_address, client_id, client_secret, authorization_server)
+    scope = "scope"
+    audience = "audience"
+
+    channel = create_oauth2_client_credentials_channel(
+        grpc_address=grpc_address,
+        client_id=client_id,
+        client_secret=client_secret,
+        authorization_server=authorization_server,
+        scope=scope,
+        audience=audience,
+        channel_credentials=None,
+        channel_options=None,
+        leeway=60,
+        expire_in=None,
+    )
 
     assert isinstance(channel, grpc.aio.Channel)
 
@@ -35,12 +49,22 @@ def test_create_camunda_cloud_channel(
     client_secret = "client_secret"
     cluster_id = "cluster_id"
     region = "bru-2"
-    scope = "Zeebe"
     authorization_server = "https://login.cloud.camunda.io/oauth/token"
+    scope = None
     audience = "zeebe.camunda.io"
 
     channel = create_camunda_cloud_channel(
-        client_id, client_secret, cluster_id, region, scope, authorization_server, audience
+        client_id=client_id,
+        client_secret=client_secret,
+        cluster_id=cluster_id,
+        region=region,
+        authorization_server=authorization_server,
+        scope=scope,
+        audience=audience,
+        channel_credentials=None,
+        channel_options=None,
+        leeway=60,
+        expire_in=None,
     )
 
     assert isinstance(channel, grpc.aio.Channel)