diff --git a/docs/channels.rst b/docs/channels.rst index 7a5cbbf4..0d45ebcc 100644 --- a/docs/channels.rst +++ b/docs/channels.rst @@ -6,4 +6,5 @@ Channels :name: channels Quickstart + Configuration Reference diff --git a/docs/channels_configuration.rst b/docs/channels_configuration.rst new file mode 100644 index 00000000..ddeab0ce --- /dev/null +++ b/docs/channels_configuration.rst @@ -0,0 +1,123 @@ +====================== +Channels Configuration +====================== + +This document describes the environment variables and configurations used for establishing different gRPC channel connections to Camunda (Zeebe) instances, either with or without authentication. + +Environment Variables +--------------------- + +The following environment variables are used to configure channels. The variables are grouped according to their relevance and usage context in each type of channel. + +These variables are only considered if a corresponding argument was not passed (Unset) during initialization of a channel. + +Common Variables +---------------- + +This variables is used across all types of channels: + +**ZEEBE_ADDRESS** + :Description: + The default address of the Zeebe Gateway. + + :Usage: + Used in both secure and insecure channel configurations. + :func:`pyzeebe.create_insecure_channel` + :func:`pyzeebe.create_secure_channel` + + :Default: + ``"localhost:26500"`` + +Common OAuth2 Variables +----------------------- + +These variables are specifically for connecting to generic OAuth2 or Camunda Cloud instances. + +**CAMUNDA_CLIENT_ID** / **ZEEBE_CLIENT_ID** + :Description: + The client ID required for OAuth2 client credential authentication. + + :Usage: + Required for OAuth2 and Camunda Cloud channels. + :func:`pyzeebe.create_oauth2_client_credentials_channel` + :func:`pyzeebe.create_camunda_cloud_channel` + +**CAMUNDA_CLIENT_SECRET** / **ZEEBE_CLIENT_SECRET** + :Description: + The client secret for the OAuth2 client. + + :Usage: + Required for OAuth2 and Camunda Cloud channels. + :func:`pyzeebe.create_oauth2_client_credentials_channel` + :func:`pyzeebe.create_camunda_cloud_channel` + +OAuth2 Variables (Self-Managed) +------------------------------- + +These variables are primarily used for OAuth2 authentication in self-managed Camunda 8 instances. + +**CAMUNDA_OAUTH_URL** / **ZEEBE_AUTHORIZATION_SERVER_URL** + :Description: + Specifies the URL of the authorization server issuing access tokens to the client. + + :Usage: + Required if channel initialization argument was not specified. + :func:`pyzeebe.create_oauth2_client_credentials_channel` + +**CAMUNDA_TOKEN_AUDIENCE** / **ZEEBE_TOKEN_AUDIENCE** + :Description: + Specifies the audience for the OAuth2 token. + + :Usage: + Used when creating OAuth2 or Camunda Cloud channels. + :func:`pyzeebe.create_oauth2_client_credentials_channel` + + :Default: + ``None`` if not provided. + +Camunda Cloud Variables (SaaS) +------------------------------ + +These variables are specifically for connecting to Camunda Cloud instances. + +**CAMUNDA_OAUTH_URL** / **ZEEBE_AUTHORIZATION_SERVER_URL** + :Description: + Specifies the URL of the authorization server issuing access tokens to the client. + + :Usage: + Used in the OAuth2 and Camunda Cloud channel configurations. + :func:`pyzeebe.create_camunda_cloud_channel` + + :Default: + ``"https://login.cloud.camunda.io/oauth/token"`` if not specified. + +**CAMUNDA_CLUSTER_ID** + :Description: + The unique identifier for the Camunda Cloud cluster to connect to. + + :Usage: + Required if channel initialization argument was not specified. + :func:`pyzeebe.create_camunda_cloud_channel` + +**CAMUNDA_CLUSTER_REGION** + :Description: + The region where the Camunda Cloud cluster is hosted. + + :Usage: + Required for Camunda Cloud channels. + :func:`pyzeebe.create_camunda_cloud_channel` + + :Default: + ``"bru-2"`` if not provided. + +**CAMUNDA_TOKEN_AUDIENCE** / **ZEEBE_TOKEN_AUDIENCE** + :Description: + Specifies the audience for the OAuth2 token. + + :Usage: + Used when creating OAuth2 or Camunda Cloud channels. + :func:`pyzeebe.create_camunda_cloud_channel` + + :Default: + ``"zeebe.camunda.io"`` if not provided. + diff --git a/docs/channels_quickstart.rst b/docs/channels_quickstart.rst index e725c85c..95f22e89 100644 --- a/docs/channels_quickstart.rst +++ b/docs/channels_quickstart.rst @@ -19,10 +19,6 @@ This Channel can be configured with the parameters `channel_credentials` and `ch You can override the default `channel_options` by passing e.g. `channel_options = (("grpc.keepalive_time_ms", 60_000),)` - for a keepalive time of 60 seconds. - -Pyzeebe provides a couple standard ways to achieve this: - - Insecure -------- @@ -151,6 +147,12 @@ Example: This method use the :py:class:`.Oauth2ClientCredentialsMetadataPlugin` under the hood. +Configuration +------------- + +It is possible to omit any arguments to the channel initialization functions and instead provide environment variables. +See :doc:`Channels Configuration ` for additional details. + Custom Oauth2 Authorization Flow --------------------------------- diff --git a/docs/channels_reference.rst b/docs/channels_reference.rst index f9d26799..61e14dc3 100644 --- a/docs/channels_reference.rst +++ b/docs/channels_reference.rst @@ -26,3 +26,23 @@ Credentials :members: :special-members: :private-members: + + +Utilities (Environment) +----------------------- + +.. autofunction:: pyzeebe.channel.utils.get_zeebe_address + +.. autofunction:: pyzeebe.channel.utils.get_camunda_oauth_url + +.. autofunction:: pyzeebe.channel.utils.get_camunda_client_id + +.. autofunction:: pyzeebe.channel.utils.get_camunda_client_secret + +.. autofunction:: pyzeebe.channel.utils.get_camunda_cluster_id + +.. autofunction:: pyzeebe.channel.utils.get_camunda_cluster_region + +.. autofunction:: pyzeebe.channel.utils.get_camunda_token_audience + +.. autofunction:: pyzeebe.channel.utils.get_camunda_address diff --git a/pyzeebe/channel/insecure_channel.py b/pyzeebe/channel/insecure_channel.py index 846cff72..39e86360 100644 --- a/pyzeebe/channel/insecure_channel.py +++ b/pyzeebe/channel/insecure_channel.py @@ -3,12 +3,12 @@ import grpc from pyzeebe.channel.channel_options import get_channel_options -from pyzeebe.channel.utils import create_address -from pyzeebe.types import ChannelArgumentType +from pyzeebe.channel.utils import get_zeebe_address +from pyzeebe.types import ChannelArgumentType, Unset def create_insecure_channel( - grpc_address: str | None = None, channel_options: ChannelArgumentType | None = None + grpc_address: str = Unset, channel_options: ChannelArgumentType | None = None ) -> grpc.aio.Channel: """ Create an insecure channel @@ -22,5 +22,7 @@ def create_insecure_channel( Returns: grpc.aio.Channel: A GRPC Channel connected to the Zeebe gateway. """ - grpc_address = create_address(grpc_address=grpc_address) + if grpc_address is Unset: + grpc_address = get_zeebe_address() + return grpc.aio.insecure_channel(target=grpc_address, options=get_channel_options(channel_options)) diff --git a/pyzeebe/channel/oauth_channel.py b/pyzeebe/channel/oauth_channel.py index d54746a4..a4382cb4 100644 --- a/pyzeebe/channel/oauth_channel.py +++ b/pyzeebe/channel/oauth_channel.py @@ -5,52 +5,89 @@ import grpc from pyzeebe.channel.channel_options import get_channel_options +from pyzeebe.channel.utils import ( + get_camunda_address, + get_camunda_client_id, + get_camunda_client_secret, + get_camunda_cluster_id, + get_camunda_cluster_region, + get_camunda_oauth_url, + get_camunda_token_audience, + get_zeebe_address, +) from pyzeebe.credentials.oauth import Oauth2ClientCredentialsMetadataPlugin -from pyzeebe.types import ChannelArgumentType +from pyzeebe.types import ChannelArgumentType, Unset def create_oauth2_client_credentials_channel( - grpc_address: str, - client_id: str, - client_secret: str, - authorization_server: str, - scope: str | None = None, - audience: str | None = None, - channel_credentials: grpc.ChannelCredentials | None = None, + grpc_address: str = Unset, + client_id: str = Unset, + client_secret: str = Unset, + authorization_server: str = Unset, + scope: str | None = Unset, + audience: str | None = Unset, + channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(), channel_options: ChannelArgumentType | None = None, leeway: int = 60, expire_in: int | None = None, ) -> grpc.aio.Channel: """Create a gRPC channel for connecting to Camunda 8 (Self-Managed) with OAuth2ClientCredentials. - - https://oauth.net/2/grant-types/client-credentials/ - - https://datatracker.ietf.org/doc/html/rfc6749#section-11.2.2 + https://oauth.net/2/grant-types/client-credentials/ + https://datatracker.ietf.org/doc/html/rfc6749#section-11.2.2 Args: - grpc_address (str): Zeebe Gateway Address. - - client_id (str): The client id. - client_secret (str): The client secret. - authorization_server (str): The authorization server issuing access tokens + grpc_address (str, optional): Zeebe Gateway Address. + Defaults to value from ZEEBE_ADDRESS environment variable + or "{CAMUNDA_CLUSTER_ID}.{CAMUNDA_CLUSTER_REGION}.zeebe.camunda.io:443" + or "localhost:26500". + client_id (str, optional): The client id. + Defaults to value from CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable + client_secret (str, optional): The client secret. + Defaults to value from CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable + authorization_server (str, optional): The authorization server issuing access tokens to the client after successfully authenticating the client. - scope (Optional[str]): The scope of the access request. Defaults to None. - audience (Optional[str]): The audience for authentication. Defaults to None. + Defaults to value from CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL environment variable + scope (str | None, optional): The scope of the access request. + audience (str | None, optional): The audience for authentication. + Defaults to value from CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE environment variable channel_credentials (grpc.ChannelCredentials): The gRPC channel credentials. Defaults to grpc.ssl_channel_credentials(). - channel_options (Optional[ChannelArgumentType]): Additional options for the gRPC channel. + channel_options (ChannelArgumentType | None): Additional options for the gRPC channel. Defaults to None. See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments leeway (int): The number of seconds to consider the token as expired before the actual expiration time. Defaults to 60. - expire_in (Optional[int]): The number of seconds the token is valid for. Defaults to None. + expire_in (int | None): The number of seconds the token is valid for. Defaults to None. Should only be used if the token does not contain an "expires_in" attribute. Returns: grpc.aio.Channel: A gRPC channel connected to the Zeebe Gateway. + + Raises: + InvalidOAuthCredentialsError: One of the provided camunda credentials is not correct """ + if grpc_address is Unset: + grpc_address = get_zeebe_address() + + if client_id is Unset: + client_id = get_camunda_client_id() + + if client_secret is Unset: + client_secret = get_camunda_client_secret() + + if authorization_server is Unset: + authorization_server = get_camunda_oauth_url() + + if scope is Unset: + scope = None + + if audience is Unset: + audience = get_camunda_token_audience() + oauth2_client_credentials = Oauth2ClientCredentialsMetadataPlugin( client_id=client_id, client_secret=client_secret, @@ -67,21 +104,23 @@ def create_oauth2_client_credentials_channel( ) channel: grpc.aio.Channel = grpc.aio.secure_channel( - target=grpc_address, credentials=composite_credentials, options=get_channel_options(channel_options) + target=grpc_address, + credentials=composite_credentials, + options=get_channel_options(channel_options), ) return channel def create_camunda_cloud_channel( - client_id: str, - client_secret: str, - cluster_id: str, - region: str = "bru-2", - authorization_server: str = "https://login.cloud.camunda.io/oauth/token", - scope: str | None = None, - audience: str | None = "zeebe.camunda.io", - channel_credentials: grpc.ChannelCredentials | None = None, + client_id: str = Unset, + client_secret: str = Unset, + cluster_id: str = Unset, + region: str = Unset, + authorization_server: str = Unset, + scope: str | None = Unset, + audience: str | None = Unset, + channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(), channel_options: ChannelArgumentType | None = None, leeway: int = 60, expire_in: int | None = None, @@ -89,32 +128,60 @@ def create_camunda_cloud_channel( """Create a gRPC channel for connecting to Camunda 8 Cloud (SaaS). Args: - client_id (str): The client id. - 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". - authorization_server (Optional[str]): The authorization server issuing access tokens + client_id (str, optional): The client id. + Defaults to value from CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable. + client_secret (str, optional): The client secret. + Defaults to value from CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable. + cluster_id (str, optional): The ID of the cluster to connect to. + Defaults to value from CAMUNDA_CLUSTER_ID environment variable. + region (str, optional): The region of the cluster. + Defaults to value from CAMUNDA_CLUSTER_REGION environment variable or 'bru-2'. + authorization_server (str, optional): 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". + Defaults to value from CAMUNDA_OAUTH_URL + or ZEEBE_AUTHORIZATION_SERVER_URL environment variable + or "https://login.cloud.camunda.io/oauth/token". + scope (str | None, optional): The scope of the access request. + audience (str | None, optional): The audience for authentication. + Defaults to value from CAMUNDA_TOKEN_AUDIENCE + or ZEEBE_TOKEN_AUDIENCE environment variable + or "zeebe.camunda.io". channel_credentials (grpc.ChannelCredentials): The gRPC channel credentials. Defaults to grpc.ssl_channel_credentials(). - channel_options (Optional[ChannelArgumentType]): Additional options for the gRPC channel. + channel_options (ChannelArgumentType | None): Additional options for the gRPC channel. Defaults to None. See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments leeway (int): The number of seconds to consider the token as expired before the actual expiration time. Defaults to 60. - expire_in (Optional[int]): The number of seconds the token is valid for. Defaults to None. + expire_in (int | None): The number of seconds the token is valid for. Defaults to None. Should only be used if the token does not contain an "expires_in" attribute. Returns: grpc.aio.Channel: The gRPC channel for connecting to Camunda Cloud. """ - grpc_address = f"{cluster_id}.{region}.zeebe.camunda.io:443" + if client_id is Unset: + client_id = get_camunda_client_id() + + if client_secret is Unset: + client_secret = get_camunda_client_secret() + + if cluster_id is Unset: + cluster_id = get_camunda_cluster_id() + + if region is Unset: + region = get_camunda_cluster_region("bru-2") + + if authorization_server is Unset: + authorization_server = get_camunda_oauth_url("https://login.cloud.camunda.io/oauth/token") + + if scope is Unset: + scope = None + + if audience is Unset: + audience = get_camunda_token_audience("zeebe.camunda.io") oauth2_client_credentials = Oauth2ClientCredentialsMetadataPlugin( client_id=client_id, @@ -141,6 +208,8 @@ def create_camunda_cloud_channel( channel_credentials or grpc.ssl_channel_credentials(), call_credentials ) + grpc_address = get_camunda_address(cluster_id=cluster_id, cluster_region=region) + channel: grpc.aio.Channel = grpc.aio.secure_channel( target=grpc_address, credentials=composite_credentials, options=get_channel_options(channel_options) ) diff --git a/pyzeebe/channel/secure_channel.py b/pyzeebe/channel/secure_channel.py index e28b4f91..5ffb53da 100644 --- a/pyzeebe/channel/secure_channel.py +++ b/pyzeebe/channel/secure_channel.py @@ -3,12 +3,12 @@ import grpc from pyzeebe.channel.channel_options import get_channel_options -from pyzeebe.channel.utils import create_address -from pyzeebe.types import ChannelArgumentType +from pyzeebe.channel.utils import get_zeebe_address +from pyzeebe.types import ChannelArgumentType, Unset def create_secure_channel( - grpc_address: str | None = None, + grpc_address: str = Unset, channel_options: ChannelArgumentType | None = None, channel_credentials: grpc.ChannelCredentials | None = None, ) -> grpc.aio.Channel: @@ -26,7 +26,10 @@ def create_secure_channel( Returns: grpc.aio.Channel: A GRPC Channel connected to the Zeebe gateway. """ - grpc_address = create_address(grpc_address=grpc_address) + + if grpc_address is Unset: + grpc_address = get_zeebe_address() + credentials = channel_credentials or grpc.ssl_channel_credentials() return grpc.aio.secure_channel( target=grpc_address, credentials=credentials, options=get_channel_options(channel_options) diff --git a/pyzeebe/channel/utils.py b/pyzeebe/channel/utils.py index 7b81909d..57fbe015 100644 --- a/pyzeebe/channel/utils.py +++ b/pyzeebe/channel/utils.py @@ -5,9 +5,151 @@ DEFAULT_ZEEBE_ADDRESS = "localhost:26500" -def create_address( - grpc_address: str | None = None, -) -> str: - if grpc_address: - return grpc_address - return os.getenv("ZEEBE_ADDRESS", DEFAULT_ZEEBE_ADDRESS) +def get_zeebe_address(default: str | None = None) -> str: + """ + Get the Zeebe Gateway Address. + + Args: + default (str, optional): Default value to be used if no other value was discovered. + + Returns: + str: ZEEBE_ADDRESS environment variable or provided default or "localhost:26500" + """ + return os.getenv("ZEEBE_ADDRESS") or default or DEFAULT_ZEEBE_ADDRESS + + +def get_camunda_oauth_url(default: str | None = None) -> str: + """ + Get the Camunda OAuth URL or Zeebe Authorization Server URL. + + Args: + default (str, optional): Default value to be used if no other value was discovered. + + Returns: + str: CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL environment variable or provided default + + Raises: + EnvironmentError: If neither CAMUNDA_OAUTH_URL nor ZEEBE_AUTHORIZATION_SERVER_URL is provided. + """ + r = os.getenv("CAMUNDA_OAUTH_URL") or os.getenv("ZEEBE_AUTHORIZATION_SERVER_URL") or default + + if r is None: + raise EnvironmentError("No CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL provided!") + + return r + + +def get_camunda_client_id() -> str: + """ + Get the Camunda Client ID. + + Returns: + str: CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable + + Raises: + EnvironmentError: If neither CAMUNDA_CLIENT_ID nor ZEEBE_CLIENT_ID is provided. + """ + r = os.getenv("CAMUNDA_CLIENT_ID") or os.getenv("ZEEBE_CLIENT_ID") + + if r is None: + raise EnvironmentError("No CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID provided!") + + return r + + +def get_camunda_client_secret() -> str: + """ + Get the Camunda Client Secret. + + Returns: + str: CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable + + Raises: + EnvironmentError: If neither CAMUNDA_CLIENT_SECRET nor ZEEBE_CLIENT_SECRET is provided. + """ + r = os.getenv("CAMUNDA_CLIENT_SECRET") or os.getenv("ZEEBE_CLIENT_SECRET") + + if r is None: + raise EnvironmentError("No CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET provided!") + + return r + + +def get_camunda_cluster_id() -> str: + """ + Get the Camunda Cluster ID. + + Returns: + str: CAMUNDA_CLUSTER_ID environment variable + + Raises: + EnvironmentError: If CAMUNDA_CLUSTER_ID is not provided. + """ + r = os.getenv("CAMUNDA_CLUSTER_ID") + + if r is None: + raise EnvironmentError("No CAMUNDA_CLUSTER_ID provided!") + + return r + + +def get_camunda_cluster_region(default: str | None = None) -> str: + """ + Get the Camunda Cluster Region. + + Args: + default (str, optional): Default value to be used if no other value was discovered. + + Returns: + str: CAMUNDA_CLUSTER_REGION environment variable or provided default + + Raises: + EnvironmentError: If CAMUNDA_CLUSTER_REGION is not provided. + """ + r = os.getenv("CAMUNDA_CLUSTER_REGION") or default + + if r is None: + raise EnvironmentError("No CAMUNDA_CLUSTER_REGION provided!") + + return r + + +def get_camunda_token_audience(default: str | None = None) -> str: + """ + Get the Camunda Token Audience. + + Args: + default (str, optional): Default value to be used if no other value was discovered. + + Returns: + str: CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE environment variable or provided default + + Raises: + EnvironmentError: If neither CAMUNDA_TOKEN_AUDIENCE nor ZEEBE_TOKEN_AUDIENCE is provided. + """ + r = os.getenv("CAMUNDA_TOKEN_AUDIENCE") or os.getenv("ZEEBE_TOKEN_AUDIENCE") or default + + if r is None: + raise EnvironmentError("No CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE provided!") + + return r + + +def get_camunda_address(cluster_id: str | None = None, cluster_region: str | None = None) -> str: + """ + Get the Camunda Cloud gRPC server address. + + Args: + cluster_id (str, optional): The Camunda cluster ID provided as parameter. + cluster_region (str, optional): The Camunda cluster region provided as parameter. + + Returns: + str: The Camunda Cloud gRPC server address. + + Raises: + EnvironmentError: If either cluster_id or cluster_region is not provided. + """ + if (cluster_id is None) or (cluster_region is None): + raise EnvironmentError("The cluster_id and cluster_region must be provided!") + + return f"{cluster_id}.{cluster_region}.zeebe.camunda.io:443" diff --git a/pyzeebe/types.py b/pyzeebe/types.py index 6191be46..696fab30 100644 --- a/pyzeebe/types.py +++ b/pyzeebe/types.py @@ -5,5 +5,6 @@ Headers: TypeAlias = Mapping[str, Any] Variables: TypeAlias = Mapping[str, Any] +Unset = str("UNSET") ChannelArgumentType: TypeAlias = Sequence[tuple[str, Any]] diff --git a/tests/unit/channel/insecure_channel_test.py b/tests/unit/channel/insecure_channel_test.py index e76fe6ae..35a07d83 100644 --- a/tests/unit/channel/insecure_channel_test.py +++ b/tests/unit/channel/insecure_channel_test.py @@ -1,3 +1,4 @@ +import os from unittest.mock import Mock, patch import grpc @@ -5,7 +6,7 @@ from pyzeebe import create_insecure_channel from pyzeebe.channel.channel_options import get_channel_options -from pyzeebe.channel.utils import create_address +from pyzeebe.channel.utils import get_zeebe_address class TestCreateInsecureChannel: @@ -29,4 +30,14 @@ def test_uses_default_address(self, insecure_channel_mock: Mock): create_insecure_channel() insecure_channel_call = insecure_channel_mock.mock_calls[0] - assert insecure_channel_call.kwargs["target"] == create_address() + assert insecure_channel_call.kwargs["target"] == get_zeebe_address() + + @patch.dict( + os.environ, + {"ZEEBE_ADDRESS": "ZEEBE_ADDRESS"}, + ) + def test_uses_zeebe_address_environment_variable(self, insecure_channel_mock: Mock): + create_insecure_channel() + + insecure_channel_call = insecure_channel_mock.mock_calls[0] + assert insecure_channel_call.kwargs["target"] == "ZEEBE_ADDRESS" diff --git a/tests/unit/channel/oauth_channel_test.py b/tests/unit/channel/oauth_channel_test.py index 19e122c3..cc6a6b9d 100644 --- a/tests/unit/channel/oauth_channel_test.py +++ b/tests/unit/channel/oauth_channel_test.py @@ -1,3 +1,4 @@ +import os from unittest import mock import grpc @@ -42,6 +43,24 @@ def test_create_oauth2_client_credentials_channel( assert isinstance(channel, grpc.aio.Channel) +@mock.patch.dict( + os.environ, + { + "ZEEBE_ADDRESS": "ZEEBE_ADDRESS", + "CAMUNDA_CLIENT_ID": "CAMUNDA_CLIENT_ID", + "CAMUNDA_CLIENT_SECRET": "CAMUNDA_CLIENT_SECRET", + "CAMUNDA_OAUTH_URL": "CAMUNDA_OAUTH_URL", + "CAMUNDA_TOKEN_AUDIENCE": "CAMUNDA_TOKEN_AUDIENCE", + }, +) +def test_create_oauth2_client_credentials_channel_using_environment_variables( + mock_oauth2metadataplugin, +): + channel = create_oauth2_client_credentials_channel() + + assert isinstance(channel, grpc.aio.Channel) + + def test_create_camunda_cloud_channel( mock_oauth2metadataplugin, ): @@ -68,3 +87,20 @@ def test_create_camunda_cloud_channel( ) assert isinstance(channel, grpc.aio.Channel) + + +@mock.patch.dict( + os.environ, + { + "CAMUNDA_CLUSTER_ID": "CAMUNDA_CLUSTER_ID", + "CAMUNDA_CLUSTER_REGION": "CAMUNDA_CLUSTER_REGION", + "CAMUNDA_CLIENT_ID": "CAMUNDA_CLIENT_ID", + "CAMUNDA_CLIENT_SECRET": "CAMUNDA_CLIENT_SECRET", + }, +) +def test_create_camunda_cloud_channel_using_environment_variables( + mock_oauth2metadataplugin, +): + channel = create_camunda_cloud_channel() + + assert isinstance(channel, grpc.aio.Channel) diff --git a/tests/unit/channel/secure_channel_test.py b/tests/unit/channel/secure_channel_test.py index 181f5f80..d9bf419c 100644 --- a/tests/unit/channel/secure_channel_test.py +++ b/tests/unit/channel/secure_channel_test.py @@ -1,3 +1,4 @@ +import os from unittest.mock import Mock, patch import grpc @@ -5,7 +6,7 @@ from pyzeebe import create_secure_channel from pyzeebe.channel.channel_options import get_channel_options -from pyzeebe.channel.utils import create_address +from pyzeebe.channel.utils import get_zeebe_address class TestCreateSecureChannel: @@ -35,4 +36,14 @@ def test_uses_default_address(self, secure_channel_mock: Mock): create_secure_channel() secure_channel_call = secure_channel_mock.mock_calls[0] - assert secure_channel_call.kwargs["target"] == create_address() + assert secure_channel_call.kwargs["target"] == get_zeebe_address() + + @patch.dict( + os.environ, + {"ZEEBE_ADDRESS": "ZEEBE_ADDRESS"}, + ) + def test_uses_zeebe_address_environment_variable(self, secure_channel_mock: Mock): + create_secure_channel() + + secure_channel_call = secure_channel_mock.mock_calls[0] + assert secure_channel_call.kwargs["target"] == "ZEEBE_ADDRESS" diff --git a/tests/unit/channel/utils_test.py b/tests/unit/channel/utils_test.py index f4d0a9ef..0a6537d3 100644 --- a/tests/unit/channel/utils_test.py +++ b/tests/unit/channel/utils_test.py @@ -1,25 +1,190 @@ import os +from unittest.mock import patch from uuid import uuid4 -from pyzeebe.channel.utils import DEFAULT_ZEEBE_ADDRESS, create_address +import pytest +from pyzeebe.channel.utils import ( + DEFAULT_ZEEBE_ADDRESS, + get_camunda_address, + get_camunda_client_id, + get_camunda_client_secret, + get_camunda_cluster_id, + get_camunda_cluster_region, + get_camunda_oauth_url, + get_camunda_token_audience, + get_zeebe_address, +) -class TestCreateAddress: + +class TestGetZeebeAddress: def test_returns_passed_address(self): address = str(uuid4()) - assert address == create_address(address) + assert address == get_zeebe_address(address) def test_returns_default_address(self): - address = create_address() + address = get_zeebe_address() assert address == DEFAULT_ZEEBE_ADDRESS + @patch.dict( + os.environ, + {"ZEEBE_ADDRESS": "ZEEBE_ADDRESS"}, + ) def test_returns_env_var_if_provided(self): - zeebe_address = str(uuid4()) - os.environ["ZEEBE_ADDRESS"] = zeebe_address + address = get_zeebe_address("zeebe_address") + + assert address == "ZEEBE_ADDRESS" + + +class TestGetCamundaOauthUrl: + @patch.dict( + os.environ, + {"CAMUNDA_OAUTH_URL": "CAMUNDA_OAUTH_URL", "ZEEBE_AUTHORIZATION_SERVER_URL": "ZEEBE_AUTHORIZATION_SERVER_URL"}, + ) + def test_camunda_oauth_url_has_highest_priority(self): + result = get_camunda_oauth_url("oauth_url") + + assert result == "CAMUNDA_OAUTH_URL" + + @patch.dict(os.environ, {"ZEEBE_AUTHORIZATION_SERVER_URL": "ZEEBE_AUTHORIZATION_SERVER_URL"}) + def test_zeebe_authorization_server_url_has_second_highest_priority(self): + result = get_camunda_oauth_url("oauth_url") + + assert result == "ZEEBE_AUTHORIZATION_SERVER_URL" + + def test_param_has_lowest_priority(self): + result = get_camunda_oauth_url("oauth_url") + + assert result == "oauth_url" + + @patch.dict(os.environ, {}) + def test_none_has_fourth_highest_priority(self): + with pytest.raises(EnvironmentError): + get_camunda_oauth_url() + + +class TestGetCamundaClientId: + @patch.dict(os.environ, {"CAMUNDA_CLIENT_ID": "CAMUNDA_CLIENT_ID", "ZEEBE_CLIENT_ID": "ZEEBE_CLIENT_ID"}) + def test_is_calculated_from_camunda_environment_variable_as_highest_priority(self): + result = get_camunda_client_id() + + assert result == "CAMUNDA_CLIENT_ID" + + @patch.dict(os.environ, {"ZEEBE_CLIENT_ID": "ZEEBE_CLIENT_ID"}) + def test_is_calculated_from_zeebe_environment_variable_as_second_priority(self): + result = get_camunda_client_id() + + assert result == "ZEEBE_CLIENT_ID" + + @patch.dict(os.environ, {}) + def test_throw_exception_if_not_configured(self): + with pytest.raises(EnvironmentError): + get_camunda_client_id() + + +class TestGetCamundaClientSecret: + @patch.dict( + os.environ, {"CAMUNDA_CLIENT_SECRET": "CAMUNDA_CLIENT_SECRET", "ZEEBE_CLIENT_SECRET": "ZEEBE_CLIENT_SECRET"} + ) + def test_is_calculated_from_camunda_environment_variable_as_highest_priority(self): + result = get_camunda_client_secret() + + assert result == "CAMUNDA_CLIENT_SECRET" + + @patch.dict(os.environ, {"ZEEBE_CLIENT_SECRET": "ZEEBE_CLIENT_SECRET"}) + def test_is_calculated_from_zeebe_environment_variable_as_second_priority(self): + result = get_camunda_client_secret() + + assert result == "ZEEBE_CLIENT_SECRET" + + @patch.dict(os.environ, {}) + def test_throw_exception_if_not_configured(self): + with pytest.raises(EnvironmentError): + get_camunda_client_secret() + + +class TestGetCamundaCloudClusterId: + @patch.dict(os.environ, {"CAMUNDA_CLUSTER_ID": "CAMUNDA_CLUSTER_ID"}) + def test_is_calculated_from_camunda_environment_variable_as_highest_priority(self): + result = get_camunda_cluster_id() + + assert result == "CAMUNDA_CLUSTER_ID" + + def test_environment_error(self): + with pytest.raises(EnvironmentError): + get_camunda_cluster_id() + + +class TestGetCamundaCloudClusterRegion: + @patch.dict(os.environ, {"CAMUNDA_CLUSTER_REGION": "CAMUNDA_CLUSTER_REGION"}) + def test_is_calculated_from_camunda_environment_variable_as_highest_priority(self): + result = get_camunda_cluster_region("cluster_region_param") + + assert result == "CAMUNDA_CLUSTER_REGION" + + def test_is_calculated_from_default_value_parameter_as_second_priority(self): + result = get_camunda_cluster_region("cluster_region_param") + + assert result == "cluster_region_param" + + @patch.dict(os.environ, {}) + def test_raises_environment_error_if_no_default_value_as_fallback_provided(self): + with pytest.raises(EnvironmentError): + get_camunda_cluster_region() + + +class TestGetCamundaTokenAudience: + @patch.dict( + os.environ, + { + "CAMUNDA_TOKEN_AUDIENCE": "CAMUNDA_TOKEN_AUDIENCE", + "ZEEBE_TOKEN_AUDIENCE": "ZEEBE_TOKEN_AUDIENCE", + }, + ) + def test_is_calculated_from_camunda_token_audience_as_highest_priority(self): + result = get_camunda_token_audience("token_audience_param") + + assert result == "CAMUNDA_TOKEN_AUDIENCE" + + @patch.dict( + os.environ, + { + "ZEEBE_TOKEN_AUDIENCE": "ZEEBE_TOKEN_AUDIENCE", + }, + ) + def test_is_calculated_from_zeebe_token_audience_as_second_highest_priority(self): + result = get_camunda_token_audience("token_audience_param") + + assert result == "ZEEBE_TOKEN_AUDIENCE" + + def test_is_calculated_from_camunda_token_audience_as_third_highest_priority(self): + result = get_camunda_token_audience("token_audience_param") + + assert result == "token_audience_param" + + @patch.dict(os.environ, {}) + def test_raises_environment_error_as_fourth_highest_priority(self): + with pytest.raises(EnvironmentError): + get_camunda_token_audience() + + +class TestGetCamundaAddress: + def test_is_calculated_from_parameters_as_highest_priority(self): + result = get_camunda_address("cluster_id_param", "camunda_region_param") + + assert result == f"cluster_id_param.camunda_region_param.zeebe.camunda.io:443" + + def test_raises_error_if_cluster_id_is_none(self): + with pytest.raises(EnvironmentError): + get_camunda_address(None, "camunda_region_param") - address = create_address() + def test_raises_error_if_cluster_region_is_none(self): + with pytest.raises(EnvironmentError): + get_camunda_address("cluster_id_param", None) - assert address == zeebe_address + def test_raises_error_if_all_args_are_none(self): + with pytest.raises(EnvironmentError): + get_camunda_address(None, None)