diff --git a/.github/scripts/cloud_namespace.py b/.github/scripts/cloud_namespace.py index c13d1e65e..04c064117 100644 --- a/.github/scripts/cloud_namespace.py +++ b/.github/scripts/cloud_namespace.py @@ -9,8 +9,10 @@ from temporalio.api.cloud.cloudservice.v1 import ( CreateNamespaceRequest, DeleteNamespaceRequest, + DeleteNexusEndpointRequest, GetAsyncOperationRequest, GetNamespaceRequest, + GetNexusEndpointsRequest, ) from temporalio.api.cloud.namespace.v1 import MtlsAuthSpec, NamespaceSpec from temporalio.api.cloud.operation.v1 import AsyncOperation @@ -51,8 +53,10 @@ async def wait_for_operation( async def create() -> None: client = await cloud_client() - namespace_name = "sdk-python-ci-{}-{}".format( - os.environ["GITHUB_RUN_ID"], os.environ["GITHUB_RUN_ATTEMPT"] + namespace_name = "sdk-python-ci-{}-{}{}".format( + os.environ["GITHUB_RUN_ID"], + os.environ["GITHUB_RUN_ATTEMPT"], + os.environ.get("TEMPORAL_CLOUD_NAMESPACE_SUFFIX", ""), ) result = await client.cloud_service.create_namespace( CreateNamespaceRequest( @@ -80,6 +84,27 @@ async def delete(namespace: str) -> None: existing = await client.cloud_service.get_namespace( GetNamespaceRequest(namespace=namespace) ) + endpoints = [] + page_token = "" + while True: + response = await client.cloud_service.get_nexus_endpoints( + GetNexusEndpointsRequest( + target_namespace_id=existing.namespace.namespace, + page_token=page_token, + ) + ) + endpoints.extend(response.endpoints) + if not response.next_page_token: + break + page_token = response.next_page_token + for endpoint in endpoints: + result = await client.cloud_service.delete_nexus_endpoint( + DeleteNexusEndpointRequest( + endpoint_id=endpoint.id, + resource_version=endpoint.resource_version, + ) + ) + await wait_for_operation(client, result.async_operation) result = await client.cloud_service.delete_namespace( DeleteNamespaceRequest( namespace=namespace, diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b1098dcdc..c632ab98b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -254,8 +254,9 @@ jobs: env: TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }} TEMPORAL_CLIENT_CLOUD_API_VERSION: v0.19.1 + TEMPORAL_CLOUD_NAMESPACE_SUFFIX: -general - run: mkdir junit-xml - - run: poe test -s --workflow-environment envconfig --junit-xml=junit-xml/cloud.xml + - run: poe test -s --workflow-environment envconfig --ignore=tests/nexus --junit-xml=junit-xml/cloud.xml timeout-minutes: 15 env: TEMPORAL_ADDRESS: ${{ steps.create-cloud-namespace.outputs.namespace }}.tmprl.cloud:7233 @@ -278,6 +279,80 @@ jobs: path: junit-xml retention-days: 14 + # Nexus endpoint provisioning is slow on Cloud, so run these tests separately. + cloud-nexus-test: + if: ${{ github.event.pull_request.head.repo.full_name == '' || github.event.pull_request.head.repo.full_name == 'temporalio/sdk-python' }} + timeout-minutes: 60 + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + submodules: recursive + - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.14" + - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + with: + workspaces: temporalio/bridge -> target + key: ${{ env.pythonLocation }} + - uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3 + with: + version: "23.x" + repo-token: ${{ secrets.GITHUB_TOKEN }} + - uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8 + - run: uv tool install poethepoet + - run: uv sync --all-extras + - run: poe build-develop + - name: Generate Cloud test certificates + run: | + cert_dir="$RUNNER_TEMP/cloud-test-certs" + mkdir "$cert_dir" + openssl req -x509 -newkey rsa:2048 -nodes -days 1 \ + -keyout "$cert_dir/ca.key" -out "$cert_dir/ca.pem" \ + -subj '/CN=Temporal Python SDK Cloud CI CA' + openssl req -newkey rsa:2048 -nodes \ + -keyout "$cert_dir/client.key" -out "$cert_dir/client.csr" \ + -subj '/CN=Temporal Python SDK Cloud CI' + openssl x509 -req -days 1 -in "$cert_dir/client.csr" \ + -CA "$cert_dir/ca.pem" -CAkey "$cert_dir/ca.key" -CAcreateserial \ + -out "$cert_dir/client.pem" -extfile <(printf 'extendedKeyUsage=clientAuth') + { + echo "TEMPORAL_CLOUD_CLIENT_CA_PATH=$cert_dir/ca.pem" + echo "TEMPORAL_TLS_CLIENT_CERT_PATH=$cert_dir/client.pem" + echo "TEMPORAL_TLS_CLIENT_KEY_PATH=$cert_dir/client.key" + } >> "$GITHUB_ENV" + - name: Create Cloud namespace + id: create-cloud-namespace + run: uv run python .github/scripts/cloud_namespace.py create + env: + TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }} + TEMPORAL_CLIENT_CLOUD_API_VERSION: v0.19.1 + TEMPORAL_CLOUD_NAMESPACE_SUFFIX: -nexus + - run: mkdir junit-xml + - run: poe test -n 16 -s --workflow-environment envconfig tests/nexus --junit-xml=junit-xml/cloud-nexus.xml + timeout-minutes: 45 + env: + TEMPORAL_ADDRESS: ${{ steps.create-cloud-namespace.outputs.namespace }}.tmprl.cloud:7233 + TEMPORAL_NAMESPACE: ${{ steps.create-cloud-namespace.outputs.namespace }} + TEMPORAL_IS_CLOUD_TESTS: true + TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }} + TEMPORAL_CLIENT_CLOUD_API_VERSION: v0.19.1 + TEMPORAL_CLIENT_CLOUD_NAMESPACE: ${{ steps.create-cloud-namespace.outputs.namespace }} + - name: Delete Cloud namespace + if: ${{ always() && steps.create-cloud-namespace.outputs.namespace != '' }} + run: uv run python .github/scripts/cloud_namespace.py delete "${{ steps.create-cloud-namespace.outputs.namespace }}" + env: + TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }} + TEMPORAL_CLIENT_CLOUD_API_VERSION: v0.19.1 + - name: Upload junit-xml artifacts + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + if: always() + with: + name: junit-xml--${{github.run_id}}--${{github.run_attempt}}--cloud-nexus + path: junit-xml + retention-days: 14 + # Runs the sdk features repo tests with this repo's current SDK code features-tests: uses: temporalio/features/.github/workflows/python.yaml@main diff --git a/tests/nexus/conftest.py b/tests/nexus/conftest.py new file mode 100644 index 000000000..8f45bef3f --- /dev/null +++ b/tests/nexus/conftest.py @@ -0,0 +1,281 @@ +import asyncio +import logging +import os +import time +import uuid +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from datetime import timedelta + +import nexusrpc +import pytest +import pytest_asyncio +from nexusrpc.handler import StartOperationContext, service_handler, sync_operation + +from temporalio.api.cloud.cloudservice.v1 import ( + CreateNexusEndpointRequest, + DeleteNexusEndpointRequest, + GetAsyncOperationRequest, + GetNamespaceRequest, + GetNexusEndpointRequest, +) +from temporalio.api.cloud.nexus.v1 import ( + AllowedCloudNamespacePolicySpec, + Endpoint, + EndpointPolicySpec, + EndpointSpec, + EndpointTargetSpec, + WorkerTargetSpec, +) +from temporalio.api.cloud.operation.v1 import AsyncOperation +from temporalio.api.cloud.resource.v1 import ResourceState +from temporalio.client import CloudOperationsClient, NexusOperationFailureError +from temporalio.service import RPCError, RPCStatusCode +from temporalio.testing import WorkflowEnvironment +from temporalio.worker import Worker +from tests.helpers.nexus import make_nexus_endpoint_name + +logger = logging.getLogger(__name__) + + +@dataclass +class _CloudNexusEndpointClient: + client: CloudOperationsClient + namespace_id: str + + async def wait_for_operation(self, operation: AsyncOperation) -> None: + deadline = time.monotonic() + 10 * 60 + while True: + operation = ( + await self.client.cloud_service.get_async_operation( + GetAsyncOperationRequest(async_operation_id=operation.id) + ) + ).async_operation + if operation.state == AsyncOperation.STATE_FULFILLED: + return + if operation.state in { + AsyncOperation.STATE_FAILED, + AsyncOperation.STATE_CANCELLED, + AsyncOperation.STATE_REJECTED, + }: + raise RuntimeError( + "Cloud operation " + f"{operation.id} " + f"{AsyncOperation.State.Name(operation.state).lower()}: " + f"{operation.failure_reason}" + ) + if time.monotonic() >= deadline: + raise TimeoutError( + f"Timed out waiting for Cloud operation {operation.id}" + ) + delay = max( + operation.check_duration.seconds + + operation.check_duration.nanos / 1_000_000_000, + 1, + ) + await asyncio.sleep(min(delay, deadline - time.monotonic())) + + async def wait_for_endpoint(self, endpoint_id: str) -> Endpoint: + deadline = time.monotonic() + 10 * 60 + while True: + endpoint = ( + await self.client.cloud_service.get_nexus_endpoint( + GetNexusEndpointRequest(endpoint_id=endpoint_id) + ) + ).endpoint + if endpoint.state == ResourceState.RESOURCE_STATE_ACTIVE: + return endpoint + if endpoint.state in { + ResourceState.RESOURCE_STATE_ACTIVATION_FAILED, + ResourceState.RESOURCE_STATE_UPDATE_FAILED, + ResourceState.RESOURCE_STATE_DELETE_FAILED, + ResourceState.RESOURCE_STATE_SUSPENDED, + ResourceState.RESOURCE_STATE_EXPIRED, + }: + raise RuntimeError( + "Cloud Nexus endpoint " + f"{endpoint_id} " + f"{ResourceState.Name(endpoint.state).lower()}" + ) + if time.monotonic() >= deadline: + raise TimeoutError( + f"Timed out waiting for Cloud Nexus endpoint {endpoint_id}" + ) + await asyncio.sleep(1) + + +@dataclass(frozen=True) +class NexusEndpoint: + name: str + task_queue: str + + +@nexusrpc.service +class _EndpointReadinessService: + ready: nexusrpc.Operation[None, None] + + +@service_handler(service=_EndpointReadinessService) +class _EndpointReadinessHandler: + @sync_operation + async def ready(self, _ctx: StartOperationContext, _input: None) -> None: + return None + + +@pytest_asyncio.fixture(scope="session") # type: ignore[reportUntypedFunctionDecorator] +async def cloud_nexus_endpoint_client() -> AsyncGenerator[ + _CloudNexusEndpointClient | None, None +]: + if "TEMPORAL_IS_CLOUD_TESTS" not in os.environ: + yield None + return + + client = await CloudOperationsClient.connect( + api_key=os.environ["TEMPORAL_CLIENT_CLOUD_API_KEY"], + version=os.environ["TEMPORAL_CLIENT_CLOUD_API_VERSION"], + ) + namespace = await client.cloud_service.get_namespace( + GetNamespaceRequest(namespace=os.environ["TEMPORAL_NAMESPACE"]) + ) + yield _CloudNexusEndpointClient(client, namespace.namespace.namespace) + + +@pytest_asyncio.fixture(autouse=True) # type: ignore[reportUntypedFunctionDecorator] +async def cloud_nexus_endpoints( + cloud_nexus_endpoint_client: _CloudNexusEndpointClient | None, + env: WorkflowEnvironment, + monkeypatch: pytest.MonkeyPatch, +) -> AsyncGenerator[None, None]: + if cloud_nexus_endpoint_client is None: + yield + return + + endpoints: list[Endpoint] = [] + + async def create_nexus_endpoint(endpoint_name: str, task_queue: str) -> Endpoint: + logger.info( + "Creating Cloud Nexus endpoint %s for task queue %s", + endpoint_name, + task_queue, + ) + response = await cloud_nexus_endpoint_client.client.cloud_service.create_nexus_endpoint( + CreateNexusEndpointRequest( + spec=EndpointSpec( + name=endpoint_name, + target_spec=EndpointTargetSpec( + worker_target_spec=WorkerTargetSpec( + namespace_id=cloud_nexus_endpoint_client.namespace_id, + task_queue=task_queue, + ) + ), + policy_specs=[ + EndpointPolicySpec( + allowed_cloud_namespace_policy_spec=AllowedCloudNamespacePolicySpec( + namespace_id=cloud_nexus_endpoint_client.namespace_id + ) + ) + ], + ) + ) + ) + await cloud_nexus_endpoint_client.wait_for_operation(response.async_operation) + endpoint = ( + await cloud_nexus_endpoint_client.client.cloud_service.get_nexus_endpoint( + GetNexusEndpointRequest(endpoint_id=response.endpoint_id) + ) + ).endpoint + endpoints.append(endpoint) + endpoint = await cloud_nexus_endpoint_client.wait_for_endpoint( + response.endpoint_id + ) + endpoints[-1] = endpoint + logger.info( + "Cloud Nexus endpoint %s (%s) is active", endpoint_name, endpoint.id + ) + return endpoint + + monkeypatch.setattr(env, "create_nexus_endpoint", create_nexus_endpoint) + try: + yield + finally: + for endpoint in reversed(endpoints): + logger.info( + "Deleting Cloud Nexus endpoint %s (%s)", endpoint.spec.name, endpoint.id + ) + response = await cloud_nexus_endpoint_client.client.cloud_service.delete_nexus_endpoint( + DeleteNexusEndpointRequest( + endpoint_id=endpoint.id, + resource_version=endpoint.resource_version, + ) + ) + await cloud_nexus_endpoint_client.wait_for_operation( + response.async_operation + ) + + +@pytest_asyncio.fixture # type: ignore[reportUntypedFunctionDecorator] +async def nexus_endpoint( + cloud_nexus_endpoint_client: _CloudNexusEndpointClient | None, + env: WorkflowEnvironment, +) -> NexusEndpoint: + """Create and, on Cloud, route-check a Nexus endpoint before a test worker.""" + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with time-skipping server") + + task_queue = str(uuid.uuid4()) + endpoint = NexusEndpoint( + name=make_nexus_endpoint_name(task_queue), task_queue=task_queue + ) + await env.create_nexus_endpoint(endpoint.name, endpoint.task_queue) + + if cloud_nexus_endpoint_client is None: + return endpoint + + deadline = time.monotonic() + 10 * 60 + nexus_client = env.client.create_nexus_client( + _EndpointReadinessService, endpoint.name + ) + attempt = 0 + async with Worker( + env.client, + task_queue=endpoint.task_queue, + nexus_service_handlers=[_EndpointReadinessHandler()], + ): + while True: + attempt += 1 + try: + operation = await nexus_client.start_operation( + _EndpointReadinessService.ready, + None, + id=f"cloud-nexus-readiness-{uuid.uuid4()}", + schedule_to_close_timeout=timedelta(seconds=10), + ) + await asyncio.wait_for(operation.result(), timeout=15) + break + except RPCError as err: + retryable = ( + err.status == RPCStatusCode.NOT_FOUND + and str(err) == "endpoint not found" + ) + except NexusOperationFailureError as err: + retryable = str(err.cause) in { + "endpoint not registered", + "nexus endpoint not found", + } + except TimeoutError: + retryable = True + if not retryable: + raise + if time.monotonic() >= deadline: + raise TimeoutError( + f"Timed out waiting for Cloud Nexus endpoint {endpoint.name} " + "to route operations" + ) + logger.info( + "Cloud Nexus endpoint %s did not route readiness operation on " + "attempt %d; retrying", + endpoint.name, + attempt, + ) + await asyncio.sleep(1) + return endpoint diff --git a/tests/nexus/test_dynamic_creation_of_user_handler_classes.py b/tests/nexus/test_dynamic_creation_of_user_handler_classes.py index 214e02ab9..17de5407d 100644 --- a/tests/nexus/test_dynamic_creation_of_user_handler_classes.py +++ b/tests/nexus/test_dynamic_creation_of_user_handler_classes.py @@ -9,10 +9,7 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker from tests.helpers.nexus import make_nexus_endpoint_name - -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server +from tests.nexus.conftest import NexusEndpoint @workflow.defn @@ -56,11 +53,12 @@ async def run(self, input: int, task_queue: str) -> int: async def test_run_nexus_service_from_programmatically_created_service_handler( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue service_handler = nexusrpc.handler._core.ServiceHandler( service=nexusrpc.ServiceDefinition( @@ -79,7 +77,6 @@ async def test_run_nexus_service_from_programmatically_created_service_handler( }, ) - await env.create_nexus_endpoint(make_nexus_endpoint_name(task_queue), task_queue) async with Worker( client, task_queue=task_queue, diff --git a/tests/nexus/test_nexus_client_updates.py b/tests/nexus/test_nexus_client_updates.py index 97dd251da..302ede54e 100644 --- a/tests/nexus/test_nexus_client_updates.py +++ b/tests/nexus/test_nexus_client_updates.py @@ -3,7 +3,6 @@ import uuid import nexusrpc -import pytest from nexusrpc.handler import StartOperationContext, service_handler, sync_operation import temporalio.nexus @@ -12,10 +11,6 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server - @nexusrpc.service class ClientTestService: diff --git a/tests/nexus/test_nexus_worker_shutdown.py b/tests/nexus/test_nexus_worker_shutdown.py index 2a94027d5..bd9063237 100644 --- a/tests/nexus/test_nexus_worker_shutdown.py +++ b/tests/nexus/test_nexus_worker_shutdown.py @@ -23,10 +23,6 @@ make_nexus_endpoint_name, ) -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server - @nexusrpc.service class ShutdownTestService: diff --git a/tests/nexus/test_signal_link_propagation_e2e.py b/tests/nexus/test_signal_link_propagation_e2e.py index 9e51d4b93..bf062594e 100644 --- a/tests/nexus/test_signal_link_propagation_e2e.py +++ b/tests/nexus/test_signal_link_propagation_e2e.py @@ -55,10 +55,7 @@ make_nexus_endpoint_name, workflow_event_link_event_type, ) - -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server +from tests.nexus.conftest import NexusEndpoint EventType = temporalio.api.enums.v1.EventType @@ -270,12 +267,12 @@ def _assert_backlink( async def test_sync_signal_operation_links( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ) -> None: if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) - await env.create_nexus_endpoint(make_nexus_endpoint_name(task_queue), task_queue) + task_queue = nexus_endpoint.task_queue callee_id = f"callee-{uuid.uuid4()}" caller_id = f"caller-{uuid.uuid4()}" @@ -319,12 +316,12 @@ async def test_sync_signal_operation_links( async def test_async_signal_operation_links( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ) -> None: if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) - await env.create_nexus_endpoint(make_nexus_endpoint_name(task_queue), task_queue) + task_queue = nexus_endpoint.task_queue callee_id = f"async-callee-{uuid.uuid4()}" caller_id = f"async-caller-{uuid.uuid4()}" @@ -404,12 +401,12 @@ def _assert_standalone_forward_link( async def test_standalone_sync_signal_operation_links( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ) -> None: if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) - await env.create_nexus_endpoint(make_nexus_endpoint_name(task_queue), task_queue) + task_queue = nexus_endpoint.task_queue callee_id = f"standalone-callee-{uuid.uuid4()}" operation_id = f"standalone-op-{uuid.uuid4()}" @@ -443,12 +440,12 @@ async def test_standalone_sync_signal_operation_links( async def test_standalone_async_signal_operation_links( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ) -> None: if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) - await env.create_nexus_endpoint(make_nexus_endpoint_name(task_queue), task_queue) + task_queue = nexus_endpoint.task_queue callee_id = f"standalone-async-callee-{uuid.uuid4()}" operation_id = f"standalone-async-op-{uuid.uuid4()}" @@ -501,12 +498,12 @@ async def _callee_result() -> str: async def test_start_from_handler_attaches_on_conflict_options( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ) -> None: if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) - await env.create_nexus_endpoint(make_nexus_endpoint_name(task_queue), task_queue) + task_queue = nexus_endpoint.task_queue callee_id = f"conflict-callee-{uuid.uuid4()}" operation_id = f"conflict-op-{uuid.uuid4()}" diff --git a/tests/nexus/test_standalone_operations.py b/tests/nexus/test_standalone_operations.py index 26a8316b4..b470f734c 100644 --- a/tests/nexus/test_standalone_operations.py +++ b/tests/nexus/test_standalone_operations.py @@ -63,19 +63,14 @@ expected_nexus_operation_link, expected_workflow_event_link, links_from_workflow_execution_started_event, - make_nexus_endpoint_name, ) +from tests.nexus.conftest import NexusEndpoint # --------------------------------------------------------------------------- # Data types # --------------------------------------------------------------------------- -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server - - @dataclass class EchoInput: value: str @@ -199,7 +194,7 @@ async def raise_err( async def test_start_sync_operation_and_get_result( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): """Start a sync nexus operation, call handle.result(), verify return value.""" if env.supports_time_skipping: @@ -207,8 +202,8 @@ async def test_start_sync_operation_and_get_result( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -216,8 +211,6 @@ async def test_start_sync_operation_and_get_result( nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -239,7 +232,7 @@ async def test_start_sync_operation_and_get_result( async def test_start_async_operation_and_poll_result( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): """Start a workflow_run operation, poll result, verify.""" if env.supports_time_skipping: @@ -247,8 +240,8 @@ async def test_start_async_operation_and_poll_result( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -256,8 +249,6 @@ async def test_start_async_operation_and_poll_result( nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -273,7 +264,7 @@ async def test_start_async_operation_and_poll_result( async def test_started_workflow_has_link_to_standalone_nexus_operation( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): """Start a workflow_run operation and verify its workflow links back to the Nexus op.""" if env.supports_time_skipping: @@ -281,8 +272,8 @@ async def test_started_workflow_has_link_to_standalone_nexus_operation( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name service_handler = StandaloneTestServiceHandler() async with Worker( @@ -291,8 +282,6 @@ async def test_started_workflow_has_link_to_standalone_nexus_operation( nexus_service_handlers=[service_handler], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -329,15 +318,17 @@ async def test_started_workflow_has_link_to_standalone_nexus_operation( assert result.value == input_value -async def test_execute_operation(client: Client, env: WorkflowEnvironment): +async def test_execute_operation( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Use execute_operation convenience method, verify it returns result directly.""" if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -345,8 +336,6 @@ async def test_execute_operation(client: Client, env: WorkflowEnvironment): nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -363,7 +352,7 @@ async def test_execute_operation(client: Client, env: WorkflowEnvironment): async def test_execute_operation_named_service( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): """Verify that the name on the service decorator is respected by the standalone nexus client""" if env.supports_time_skipping: @@ -371,8 +360,8 @@ async def test_execute_operation_named_service( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -381,8 +370,6 @@ async def test_execute_operation_named_service( nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - # Create client using the service that is uses the name "StandaloneTestService" nexus_client = client.create_nexus_client( service=NamedService, endpoint=endpoint_name @@ -399,15 +386,17 @@ async def test_execute_operation_named_service( assert result.value == "execute" -async def test_errors(client: Client, env: WorkflowEnvironment): +async def test_errors( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Execute operations that raise errors""" if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -415,8 +404,6 @@ async def test_errors(client: Client, env: WorkflowEnvironment): nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -459,15 +446,17 @@ async def test_errors(client: Client, env: WorkflowEnvironment): assert isinstance(err.value.__cause__.__cause__, ApplicationError) -async def test_describe_operation(client: Client, env: WorkflowEnvironment): +async def test_describe_operation( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Start op, get result first, then describe, verify fields populated.""" if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -475,8 +464,6 @@ async def test_describe_operation(client: Client, env: WorkflowEnvironment): nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -506,7 +493,9 @@ async def test_describe_operation(client: Client, env: WorkflowEnvironment): assert summary == StandaloneTestService.echo_async.name -async def test_cancel_operation(client: Client, env: WorkflowEnvironment): +async def test_cancel_operation( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Start blocking async op, cancel it, verify awaiting result raises NexusOperationFailureError from a CancelledError. """ @@ -515,8 +504,8 @@ async def test_cancel_operation(client: Client, env: WorkflowEnvironment): "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -524,8 +513,6 @@ async def test_cancel_operation(client: Client, env: WorkflowEnvironment): nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -548,7 +535,9 @@ async def test_cancel_operation(client: Client, env: WorkflowEnvironment): assert isinstance(err.value.__cause__, CancelledError) -async def test_terminate_operation(client: Client, env: WorkflowEnvironment): +async def test_terminate_operation( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Start blocking async op, terminate it, verify awaiting the result raises NexusOperationFailureError from a TerminatedError. """ @@ -557,8 +546,8 @@ async def test_terminate_operation(client: Client, env: WorkflowEnvironment): "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -566,8 +555,6 @@ async def test_terminate_operation(client: Client, env: WorkflowEnvironment): nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -590,15 +577,17 @@ async def test_terminate_operation(client: Client, env: WorkflowEnvironment): assert isinstance(err.value.__cause__, TerminatedError) -async def test_list_operations(client: Client, env: WorkflowEnvironment): +async def test_list_operations( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Start multiple ops, list them, verify iteration yields correct results.""" if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -606,8 +595,6 @@ async def test_list_operations(client: Client, env: WorkflowEnvironment): nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -638,15 +625,17 @@ async def check_ids() -> None: await assert_eventually(check_ids) -async def test_count_operations(client: Client, env: WorkflowEnvironment): +async def test_count_operations( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Start ops, count, verify count.""" if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -654,8 +643,6 @@ async def test_count_operations(client: Client, env: WorkflowEnvironment): nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -681,15 +668,17 @@ async def check_count() -> None: await assert_eventually(check_count) -async def test_get_nexus_operation_handle(client: Client, env: WorkflowEnvironment): +async def test_get_nexus_operation_handle( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Start op, get result, then get handle by ID and get result again.""" if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -697,8 +686,6 @@ async def test_get_nexus_operation_handle(client: Client, env: WorkflowEnvironme nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -727,7 +714,7 @@ async def test_get_nexus_operation_handle(client: Client, env: WorkflowEnvironme async def test_id_conflict_policy_use_existing( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): """Start op, re-start with USE_EXISTING, verify same op/run ID and expected result""" if env.supports_time_skipping: @@ -735,8 +722,8 @@ async def test_id_conflict_policy_use_existing( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name service_handler = StandaloneTestServiceHandler() @@ -746,8 +733,6 @@ async def test_id_conflict_policy_use_existing( nexus_service_handlers=[service_handler], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -793,15 +778,17 @@ async def test_id_conflict_policy_use_existing( assert first_result.value == second_result.value -async def test_id_conflict_policy_fail(client: Client, env: WorkflowEnvironment): +async def test_id_conflict_policy_fail( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Start op, re-start with FAIL, verify raises NexusOperationAlreadyStartedError.""" if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, @@ -809,8 +796,6 @@ async def test_id_conflict_policy_fail(client: Client, env: WorkflowEnvironment) nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) @@ -907,15 +892,17 @@ def intercept_client(self, next: OutboundInterceptor) -> OutboundInterceptor: return _RecordingOutboundInterceptor(next, self) -async def test_interceptor_receives_inputs(client: Client, env: WorkflowEnvironment): +async def test_interceptor_receives_inputs( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """Custom OutboundInterceptor records calls, verify correct input types.""" if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name interceptor = _RecordingInterceptor() intercepted_client = Client( @@ -930,8 +917,6 @@ async def test_interceptor_receives_inputs(client: Client, env: WorkflowEnvironm nexus_service_handlers=[StandaloneTestServiceHandler()], workflows=[EchoHandlerWorkflow, BlockingHandlerWorkflow], ): - await env.create_nexus_endpoint(endpoint_name, task_queue) - nexus_client = intercepted_client.create_nexus_client( service=StandaloneTestService, endpoint=endpoint_name ) diff --git a/tests/nexus/test_temporal_extstore.py b/tests/nexus/test_temporal_extstore.py index f23a69259..dadf73d00 100644 --- a/tests/nexus/test_temporal_extstore.py +++ b/tests/nexus/test_temporal_extstore.py @@ -39,12 +39,9 @@ from temporalio.types import MethodAsyncSingleParam from temporalio.worker import UnsandboxedWorkflowRunner, Worker from tests.helpers.nexus import make_nexus_endpoint_name +from tests.nexus.conftest import NexusEndpoint from tests.test_extstore import InMemoryTestDriver -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server - PAYLOAD_SIZE = 4096 PAYLOAD_SIZE_THRESHOLD = 1024 _STORE_FAILURE_MESSAGE = "external storage store failed" @@ -154,9 +151,9 @@ async def _run_caller( env: WorkflowEnvironment, driver: InMemoryTestDriver, workflow_run: MethodAsyncSingleParam[Any, str, int], + task_queue: str, ) -> int: client = _client_with_extstore(env, driver) - task_queue = str(uuid.uuid4()) async with Worker( client, task_queue=task_queue, @@ -164,9 +161,6 @@ async def _run_caller( nexus_service_handlers=[ExtStoreNexusServiceHandler()], workflow_runner=UnsandboxedWorkflowRunner(), ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(task_queue), task_queue - ) return await client.execute_workflow( workflow_run, task_queue, @@ -185,13 +179,17 @@ def _cause_chain(err: BaseException) -> list[BaseException]: return chain -async def test_nexus_operation_input_offloaded_and_retrieved(env: WorkflowEnvironment): +async def test_nexus_operation_input_offloaded_and_retrieved( + env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): """The offloaded operation input is retrieved before the handler runs.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with the Java test server") driver = InMemoryTestDriver() - result = await _run_caller(env, driver, SizeOpCallerWorkflow.run) + result = await _run_caller( + env, driver, SizeOpCallerWorkflow.run, nexus_endpoint.task_queue + ) assert result == PAYLOAD_SIZE assert driver._store_calls >= 1 @@ -200,13 +198,16 @@ async def test_nexus_operation_input_offloaded_and_retrieved(env: WorkflowEnviro async def test_nexus_operation_sync_result_offloaded_and_retrieved( env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): """A large synchronous result is offloaded and retrieved by the caller.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with the Java test server") driver = InMemoryTestDriver() - result = await _run_caller(env, driver, BigResultOpCallerWorkflow.run) + result = await _run_caller( + env, driver, BigResultOpCallerWorkflow.run, nexus_endpoint.task_queue + ) assert result == PAYLOAD_SIZE assert driver._store_calls >= 1 @@ -215,13 +216,16 @@ async def test_nexus_operation_sync_result_offloaded_and_retrieved( async def test_nexus_operation_transient_retrieve_failure_recovers( env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): """A transient retrieve failure fails the task retryably; it then recovers.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with the Java test server") driver = TransientFailureDriver(fail_first_retrieve=True) - result = await _run_caller(env, driver, SizeOpCallerWorkflow.run) + result = await _run_caller( + env, driver, SizeOpCallerWorkflow.run, nexus_endpoint.task_queue + ) assert result == PAYLOAD_SIZE assert driver.retrieve_attempts >= 2 @@ -229,13 +233,16 @@ async def test_nexus_operation_transient_retrieve_failure_recovers( async def test_nexus_operation_transient_store_failure_recovers( env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): """A transient store failure fails the task retryably; it then recovers.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with the Java test server") driver = TransientFailureDriver(fail_first_store=True) - result = await _run_caller(env, driver, BigResultOpCallerWorkflow.run) + result = await _run_caller( + env, driver, BigResultOpCallerWorkflow.run, nexus_endpoint.task_queue + ) assert result == PAYLOAD_SIZE assert driver.store_attempts >= 2 @@ -254,6 +261,7 @@ async def store( async def test_nexus_operation_store_failure_fails_operation( env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): """A non-retryable store failure fails the operation and surfaces the driver error to the caller (deterministically, with no retries).""" @@ -262,7 +270,9 @@ async def test_nexus_operation_store_failure_fails_operation( driver = PermanentFailStoreDriver() with pytest.raises(WorkflowFailureError) as exc_info: - await _run_caller(env, driver, BigResultOpCallerWorkflow.run) + await _run_caller( + env, driver, BigResultOpCallerWorkflow.run, nexus_endpoint.task_queue + ) causes = _cause_chain(exc_info.value) assert [type(c) for c in causes] == [ diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index 85948deb3..42602180f 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -39,10 +39,7 @@ expected_nexus_operation_link, make_nexus_endpoint_name, ) - -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server +from tests.nexus.conftest import NexusEndpoint @dataclass @@ -430,11 +427,9 @@ async def run(self, input: Input) -> str: async def test_temporal_operation_start_workflow( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue async with Worker( env.client, task_queue=task_queue, @@ -461,15 +456,13 @@ async def test_temporal_operation_start_workflow( async def test_temporal_operation_update_workflow( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ) -> None: if ( env.supports_time_skipping ): # time skipping server uses different dynamic configs pytest.skip("Update workflow tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue async with Worker( env.client, task_queue=task_queue, @@ -844,11 +837,9 @@ async def wait_operation_started(self): async def test_temporal_operation_cancel_workflow( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue async with Worker( env.client, task_queue=task_queue, @@ -879,16 +870,15 @@ async def test_temporal_operation_cancel_workflow( async def test_customized_temporal_operation_cancel_workflow( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name service_handler = TestServiceHandler() async with Worker( @@ -953,11 +943,9 @@ async def run(self, input: Input) -> str: async def test_temporal_operation_double_start_raises_handler_err( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue async with Worker( env.client, task_queue=task_queue, @@ -982,11 +970,9 @@ async def test_temporal_operation_double_start_raises_handler_err( async def test_temporal_operation_concurrent_start_raises_handler_err( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue async with Worker( env.client, task_queue=task_queue, @@ -1004,12 +990,10 @@ async def test_temporal_operation_concurrent_start_raises_handler_err( async def test_temporal_operation_failed_start_allows_retry( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) + task_queue = nexus_endpoint.task_queue conflict_id = f"failed-start-rollback-{uuid.uuid4()}" - await env.create_nexus_endpoint(endpoint_name, task_queue) async with Worker( env.client, task_queue=task_queue, @@ -1039,16 +1023,15 @@ async def test_temporal_operation_failed_start_allows_retry( async def test_temporal_operation_mixed_start_raises_handler_err( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( env.client, task_queue=task_queue, @@ -1083,10 +1066,10 @@ async def run(self, input: Input) -> str: return await client.execute_operation(TestService.sync_result, input) -async def test_temporal_operation_sync_result(client: Client, env: WorkflowEnvironment): - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) +async def test_temporal_operation_sync_result( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): + task_queue = nexus_endpoint.task_queue async with Worker( env.client, task_queue=task_queue, @@ -1113,16 +1096,15 @@ async def test_temporal_operation_sync_result(client: Client, env: WorkflowEnvir async def test_temporal_operation_start_activity( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( env.client, task_queue=task_queue, @@ -1140,16 +1122,15 @@ async def test_temporal_operation_start_activity( async def test_temporal_operation_backing_activity_does_not_duplicate_links( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name activity_id = f"link-activity-{uuid.uuid4()}" @service_handler @@ -1201,16 +1182,15 @@ async def echo_activity( async def test_temporal_operation_start_activity_raises_error( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( env.client, task_queue=task_queue, @@ -1240,16 +1220,15 @@ async def test_temporal_operation_start_activity_raises_error( async def test_temporal_operation_cancel_activity( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( env.client, task_queue=task_queue, @@ -1277,16 +1256,15 @@ async def check_cancelled(): async def test_customized_temporal_operation_cancel_activity( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name service_handler = TestServiceHandler() async with Worker( @@ -1318,16 +1296,15 @@ async def check_cancelled(): async def test_temporal_operation_double_start_activity_raises_handler_err( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( env.client, task_queue=task_queue, @@ -1487,11 +1464,11 @@ async def run( ], ) async def test_temporal_operation_overloads( - client: Client, env: WorkflowEnvironment, op: str + client: Client, + op: str, + nexus_endpoint: NexusEndpoint, ): - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue async with Worker( client, task_queue=task_queue, @@ -1516,11 +1493,9 @@ async def test_temporal_operation_overloads( async def test_temporal_operation_includes_token_in_callback( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue async with Worker( env.client, task_queue=task_queue, @@ -1599,15 +1574,14 @@ async def do_update(self, value: str) -> str: async def test_temporal_operation_includes_activity_token_in_callback( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip( "Standalone Nexus Operation tests don't work with time-skipping server" ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name @service_handler class ActivityTokenHandler: diff --git a/tests/nexus/test_use_existing_conflict_policy.py b/tests/nexus/test_use_existing_conflict_policy.py index 8ffa9f8f8..82f2b2829 100644 --- a/tests/nexus/test_use_existing_conflict_policy.py +++ b/tests/nexus/test_use_existing_conflict_policy.py @@ -4,19 +4,14 @@ import uuid from dataclasses import dataclass -import pytest from nexusrpc.handler import service_handler from temporalio import nexus, workflow from temporalio.client import Client from temporalio.common import WorkflowIDConflictPolicy -from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker from tests.helpers.nexus import make_nexus_endpoint_name - -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server +from tests.nexus.conftest import NexusEndpoint @dataclass @@ -93,9 +88,9 @@ async def nexus_operations_have_started(self) -> None: async def test_multiple_operation_invocations_can_connect_to_same_handler_workflow( - client: Client, env: WorkflowEnvironment + client: Client, nexus_endpoint: NexusEndpoint ): - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue workflow_id = str(uuid.uuid4()) async with Worker( @@ -104,9 +99,6 @@ async def test_multiple_operation_invocations_can_connect_to_same_handler_workfl workflows=[CallerWorkflow, HandlerWorkflow], task_queue=task_queue, ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(task_queue), task_queue - ) caller_handle = await client.start_workflow( CallerWorkflow.run, args=[ diff --git a/tests/nexus/test_workflow_caller.py b/tests/nexus/test_workflow_caller.py index 0c47f17c1..6f6fb2197 100644 --- a/tests/nexus/test_workflow_caller.py +++ b/tests/nexus/test_workflow_caller.py @@ -71,6 +71,7 @@ links_from_workflow_execution_started_event, make_nexus_endpoint_name, ) +from tests.nexus.conftest import NexusEndpoint # TODO(nexus-preview): test worker shutdown, wait_all_completed, drain etc @@ -90,11 +91,6 @@ class OpDefinitionType(IntEnum): LONGHAND = 1 -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server - - @dataclass class SyncResponse: op_definition_type: OpDefinitionType @@ -668,8 +664,8 @@ async def run(self, input: WorkflowRunHeaderTestCallerWfInput) -> HeaderTestOutp # -async def test_sync_operation_happy_path(client: Client, env: WorkflowEnvironment): - task_queue = str(uuid.uuid4()) +async def test_sync_operation_happy_path(client: Client, nexus_endpoint: NexusEndpoint): + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[ServiceImpl()], @@ -677,8 +673,6 @@ async def test_sync_operation_happy_path(client: Client, env: WorkflowEnvironmen task_queue=task_queue, workflow_failure_exception_types=[Exception], ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) wf_output = await client.execute_workflow( CallerWorkflow.run, args=[ @@ -727,16 +721,17 @@ async def run(self, task_queue: str) -> dict[str, str]: return await nexus_client.execute_operation(NexusInfoService.get_info, None) -async def test_nexus_info_includes_namespace(client: Client, env: WorkflowEnvironment): - task_queue = str(uuid.uuid4()) +async def test_nexus_info_includes_namespace( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): + task_queue = nexus_endpoint.task_queue + endpoint_name = nexus_endpoint.name async with Worker( client, nexus_service_handlers=[NexusInfoService()], workflows=[NexusInfoCallerWorkflow], task_queue=task_queue, ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) result = await client.execute_workflow( NexusInfoCallerWorkflow.run, task_queue, @@ -751,9 +746,9 @@ async def test_nexus_info_includes_namespace(client: Client, env: WorkflowEnviro async def test_workflow_run_operation_happy_path( - client: Client, env: WorkflowEnvironment + client: Client, nexus_endpoint: NexusEndpoint ): - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[ServiceImpl()], @@ -761,8 +756,6 @@ async def test_workflow_run_operation_happy_path( task_queue=task_queue, workflow_failure_exception_types=[Exception], ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) wf_output = await client.execute_workflow( CallerWorkflow.run, args=[ @@ -904,12 +897,13 @@ async def start_nexus_operation( async def test_start_operation_headers( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): """Test headers from workflow and interceptors are propagated to start operation handler.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue inbound_interceptor = HeaderModifyingNexusInterceptor() async with Worker( @@ -919,9 +913,6 @@ async def test_start_operation_headers( task_queue=task_queue, interceptors=[HeaderAddingOutboundInterceptor(), inbound_interceptor], ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) - workflow_headers = {"x-custom-from-workflow": "workflow-value"} result = await client.execute_workflow( HeaderTestCallerWorkflow.run, @@ -950,10 +941,10 @@ async def test_start_operation_headers( async def test_workflow_run_operation_headers( client: Client, - env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): """Test that headers are propagated to @workflow_run_operation handlers.""" - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue test_headers = {"x-custom-workflow-run": "workflow-run-value"} async with Worker( @@ -962,9 +953,6 @@ async def test_workflow_run_operation_headers( workflows=[WorkflowRunHeaderTestCallerWorkflow, HeaderEchoWorkflow], task_queue=task_queue, ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) - result = await client.execute_workflow( WorkflowRunHeaderTestCallerWorkflow.run, WorkflowRunHeaderTestCallerWfInput( @@ -981,12 +969,13 @@ async def test_workflow_run_operation_headers( async def test_cancel_operation_headers( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): """Test headers from workflow and interceptor are propagated to cancel operation handler.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue workflow_id = str(uuid.uuid4()) inbound_interceptor = HeaderModifyingNexusInterceptor() service_handler = HeaderTestServiceImpl() @@ -998,9 +987,6 @@ async def test_cancel_operation_headers( task_queue=task_queue, interceptors=[inbound_interceptor], ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) - workflow_headers = {"x-custom-cancel": "cancel-value"} await client.execute_workflow( CancelHeaderTestCallerWorkflow.run, @@ -1041,11 +1027,12 @@ async def test_sync_response( request_cancel: bool, op_definition_type: OpDefinitionType, caller_reference: CallerReference, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[ServiceImpl()], @@ -1053,8 +1040,6 @@ async def test_sync_response( task_queue=task_queue, workflow_failure_exception_types=[Exception], ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) caller_wf_handle = await client.start_workflow( CallerWorkflow.run, args=[ @@ -1115,11 +1100,12 @@ async def test_async_response( request_cancel: bool, op_definition_type: OpDefinitionType, caller_reference: CallerReference, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[ServiceImpl()], @@ -1128,7 +1114,6 @@ async def test_async_response( workflow_failure_exception_types=[Exception], ): caller_wf_handle, handler_wf_handle = await _start_wf_and_nexus_op( - env, client, task_queue, exception_in_operation_start, @@ -1204,7 +1189,6 @@ async def test_async_response( async def _start_wf_and_nexus_op( - env: WorkflowEnvironment, client: Client, task_queue: str, exception_in_operation_start: bool, @@ -1218,8 +1202,6 @@ async def _start_wf_and_nexus_op( """ Start the caller workflow and wait until the Nexus operation has started. """ - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) operation_workflow_id = str(uuid.uuid4()) # Start the caller workflow and wait until it confirms the Nexus operation has started. @@ -1279,11 +1261,12 @@ async def test_untyped_caller( op_definition_type: OpDefinitionType, caller_reference: CallerReference, response_type: ResponseType, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, workflows=[UntypedCallerWorkflow, HandlerWorkflow], @@ -1304,8 +1287,6 @@ async def test_untyped_caller( op_definition_type=op_definition_type, exception_in_operation_start=exception_in_operation_start, ) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) caller_wf_handle = await client.start_workflow( UntypedCallerWorkflow.run, args=[ @@ -1435,7 +1416,7 @@ async def run( async def test_service_interface_and_implementation_names( - client: Client, env: WorkflowEnvironment + client: Client, nexus_endpoint: NexusEndpoint ): # Note that: # - The caller can specify the service & operation via a reference to either the @@ -1450,7 +1431,7 @@ async def test_service_interface_and_implementation_names( # # This test checks that the request is routed to the expected service under a variety # of scenarios related to the above considerations. - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[ @@ -1463,8 +1444,6 @@ async def test_service_interface_and_implementation_names( task_queue=task_queue, workflow_failure_exception_types=[Exception], ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) assert await client.execute_workflow( ServiceInterfaceAndImplCallerWorkflow.run, args=(CallerReference.INTERFACE, NameOverride.YES, task_queue), @@ -1564,11 +1543,12 @@ async def run(self, _input: str, task_queue: str) -> str: async def test_workflow_run_operation_can_execute_workflow_before_starting_backing_workflow( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, workflows=[ @@ -1580,8 +1560,6 @@ async def test_workflow_run_operation_can_execute_workflow_before_starting_backi ], task_queue=task_queue, ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) result = await client.execute_workflow( WorkflowCallingNexusOperationThatExecutesWorkflowBeforeStartingBackingWorkflow.run, args=("result-1", task_queue), @@ -1619,9 +1597,9 @@ async def run(self, input: str, task_queue: str) -> str: async def test_nexus_operation_summary( client: Client, - env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): - task_queue = f"task-queue-{uuid.uuid4()}" + task_queue = nexus_endpoint.task_queue async with Worker( client, workflows=[ExecuteNexusOperationWithSummaryWorkflow], @@ -1630,8 +1608,6 @@ async def test_nexus_operation_summary( ], task_queue=task_queue, ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) wf_id = f"wf-{uuid.uuid4()}" handle = await client.start_workflow( ExecuteNexusOperationWithSummaryWorkflow.run, @@ -1901,9 +1877,11 @@ async def run(self, op: str, input: OverloadTestValue) -> OverloadTestValue: ], ) async def test_workflow_run_operation_overloads( - client: Client, env: WorkflowEnvironment, op: str + client: Client, + op: str, + nexus_endpoint: NexusEndpoint, ): - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, task_queue=task_queue, @@ -1914,8 +1892,6 @@ async def test_workflow_run_operation_overloads( ], nexus_service_handlers=[OverloadTestServiceHandler()], ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) res = await client.execute_workflow( OverloadTestCallerWorkflow.run, args=[op, OverloadTestValue(value=2)], @@ -1970,10 +1946,10 @@ async def run(self, task_queue: str) -> None: ) -async def test_workflow_caller_custom_metrics(client: Client, env: WorkflowEnvironment): - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) +async def test_workflow_caller_custom_metrics( + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint +): + task_queue = nexus_endpoint.task_queue # Create new runtime with Prom server prom_addr = f"127.0.0.1:{find_free_port()}" @@ -2043,7 +2019,7 @@ async def test_workflow_caller_custom_metrics(client: Client, env: WorkflowEnvir async def test_workflow_caller_buffered_metrics( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): # Create runtime with metric buffer buffer = MetricBuffer(10000) @@ -2058,9 +2034,7 @@ async def test_workflow_caller_buffered_metrics( client = await env.connect_client( runtime=runtime, ) - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue async with new_worker( client, CustomMetricsWorkflow, @@ -2219,13 +2193,17 @@ def non_async_cancel_op(self) -> OperationHandler[None, str]: @pytest.mark.parametrize("use_async_cancel", [True, False]) async def test_task_executor_operation_cancel_method( - self, client: Client, env: WorkflowEnvironment, use_async_cancel: bool + self, + client: Client, + env: WorkflowEnvironment, + use_async_cancel: bool, + nexus_endpoint: NexusEndpoint, ): """Test that both async and non-async cancel methods work for TaskExecutor-based operations.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, task_queue=task_queue, @@ -2237,9 +2215,6 @@ async def test_task_executor_operation_cancel_method( ], nexus_task_executor=concurrent.futures.ThreadPoolExecutor(), ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) - caller_wf_handle = await client.start_workflow( CancelTestCallerWorkflow.run, args=[use_async_cancel, task_queue], @@ -2268,12 +2243,13 @@ async def test_task_executor_operation_cancel_method( async def test_request_deadline_is_accessible_in_operation( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): """Test that request_deadline is accessible in StartOperationContext.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue service_handler = RequestDeadlineServiceImpl() async with Worker( @@ -2282,9 +2258,6 @@ async def test_request_deadline_is_accessible_in_operation( workflows=[CancelDeadlineCallerWorkflow], task_queue=task_queue, ): - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) - await client.execute_workflow( CancelDeadlineCallerWorkflow.run, task_queue, diff --git a/tests/nexus/test_workflow_caller_cancellation_types.py b/tests/nexus/test_workflow_caller_cancellation_types.py index eca269984..7a777a684 100644 --- a/tests/nexus/test_workflow_caller_cancellation_types.py +++ b/tests/nexus/test_workflow_caller_cancellation_types.py @@ -24,10 +24,7 @@ from temporalio.worker import Worker from tests.helpers import LogCapturer, assert_event_subsequence, assert_eventually from tests.helpers.nexus import make_nexus_endpoint_name - -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server +from tests.nexus.conftest import NexusEndpoint @dataclass @@ -264,6 +261,7 @@ async def run(self, input: Input) -> CancellationResult: async def test_cancellation_type( env: WorkflowEnvironment, cancellation_type_name: str, + nexus_endpoint: NexusEndpoint, ): cancellation_type = workflow.NexusOperationCancellationType[cancellation_type_name] global test_context @@ -273,6 +271,7 @@ async def test_cancellation_type( ) client = env.client + task_queue = nexus_endpoint.task_queue log_capturer = LogCapturer() with log_capturer.logs_captured( @@ -280,14 +279,10 @@ async def test_cancellation_type( ): async with Worker( client, - task_queue=str(uuid.uuid4()), + task_queue=task_queue, workflows=[CallerWorkflow, HandlerWorkflow], nexus_service_handlers=[ServiceHandler()], ) as worker: - await env.create_nexus_endpoint( - make_nexus_endpoint_name(worker.task_queue), worker.task_queue - ) - # Start the caller workflow, wait for the nexus op to have started and retrieve the nexus op # token with_start_workflow = WithStartWorkflowOperation( diff --git a/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py b/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py index 44d91a7d1..5c42f6202 100644 --- a/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py +++ b/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py @@ -25,15 +25,12 @@ from temporalio.worker import Worker from tests.helpers import assert_event_subsequence, assert_eventually from tests.helpers.nexus import make_nexus_endpoint_name +from tests.nexus.conftest import NexusEndpoint from tests.nexus.test_workflow_caller_cancellation_types import ( get_event_time, has_event, ) -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server - @dataclass class TestContext: @@ -228,6 +225,7 @@ async def run(self, input: Input) -> CancellationResult: async def test_cancellation_type( env: WorkflowEnvironment, cancellation_type_name: str, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") @@ -237,17 +235,14 @@ async def test_cancellation_type( test_context = TestContext(cancellation_type=cancellation_type) client = env.client + task_queue = nexus_endpoint.task_queue async with Worker( client, - task_queue=str(uuid.uuid4()), + task_queue=task_queue, workflows=[CallerWorkflow, HandlerWorkflow], nexus_service_handlers=[ServiceHandler()], ) as worker: - await env.create_nexus_endpoint( - make_nexus_endpoint_name(worker.task_queue), worker.task_queue - ) - # Start the caller workflow, wait for the nexus op to have started and retrieve the nexus op # token with_start_workflow = WithStartWorkflowOperation( diff --git a/tests/nexus/test_workflow_caller_error_chains.py b/tests/nexus/test_workflow_caller_error_chains.py index 1012d8a94..cf3da7685 100644 --- a/tests/nexus/test_workflow_caller_error_chains.py +++ b/tests/nexus/test_workflow_caller_error_chains.py @@ -24,10 +24,7 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker from tests.helpers.nexus import make_nexus_endpoint_name - -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server +from tests.nexus.conftest import NexusEndpoint @dataclass @@ -627,21 +624,21 @@ async def run(self, input: ErrorTestInput) -> None: ids=lambda tc: tc.name, ) async def test_errors_raised_by_nexus_operation( - client: Client, env: WorkflowEnvironment, test_case: ErrorTestCase + client: Client, + env: WorkflowEnvironment, + test_case: ErrorTestCase, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[ErrorTestService()], workflows=[ErrorTestCallerWorkflow], task_queue=task_queue, ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(task_queue), task_queue - ) await client.execute_workflow( ErrorTestCallerWorkflow.run, ErrorTestInput( diff --git a/tests/nexus/test_workflow_caller_errors.py b/tests/nexus/test_workflow_caller_errors.py index 0f1b6a789..97556b09d 100644 --- a/tests/nexus/test_workflow_caller_errors.py +++ b/tests/nexus/test_workflow_caller_errors.py @@ -41,10 +41,7 @@ from temporalio.worker import Worker from tests.helpers import LogCapturer, assert_eq_eventually from tests.helpers.nexus import make_nexus_endpoint_name - -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server +from tests.nexus.conftest import NexusEndpoint operation_invocation_counts = Counter[str]() @@ -173,7 +170,10 @@ async def run(self, input: RPCErrorInput) -> None: ], ) async def test_nexus_operation_is_retried( - client: Client, env: WorkflowEnvironment, operation_name: str + client: Client, + env: WorkflowEnvironment, + operation_name: str, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") @@ -181,7 +181,7 @@ async def test_nexus_operation_is_retried( input = ErrorTestInput( service_name="ErrorTestService", operation_name=operation_name, - task_queue=str(uuid.uuid4()), + task_queue=nexus_endpoint.task_queue, id=str(uuid.uuid4()), ) async with Worker( @@ -191,9 +191,6 @@ async def test_nexus_operation_is_retried( workflows=[CallerWorkflow], task_queue=input.task_queue, ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(input.task_queue), input.task_queue - ) asyncio.create_task( client.execute_workflow( CallerWorkflow.run, @@ -235,6 +232,7 @@ async def test_nexus_operation_fails_without_retry_as_handler_error( operation_name: str, handler_error_type: nexusrpc.HandlerErrorType, handler_error_message: str, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") @@ -246,7 +244,7 @@ async def test_nexus_operation_fails_without_retry_as_handler_error( else "NonExistentService" ), operation_name=operation_name, - task_queue=str(uuid.uuid4()), + task_queue=nexus_endpoint.task_queue, id=str(uuid.uuid4()), ) async with Worker( @@ -256,9 +254,6 @@ async def test_nexus_operation_fails_without_retry_as_handler_error( workflows=[CallerWorkflow], task_queue=input.task_queue, ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(input.task_queue), input.task_queue - ) try: await client.execute_workflow( CallerWorkflow.run, @@ -317,12 +312,12 @@ async def run(self, operation: str) -> None: async def test_error_raised_by_timeout_of_nexus_start_operation( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[StartTimeoutTestService()], @@ -330,9 +325,6 @@ async def test_error_raised_by_timeout_of_nexus_start_operation( task_queue=task_queue, nexus_task_executor=concurrent.futures.ThreadPoolExecutor(), ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(task_queue), task_queue - ) try: await client.execute_workflow( StartTimeoutTestCallerWorkflow.run, @@ -399,12 +391,12 @@ async def run(self) -> None: async def test_error_raised_by_schedule_to_start_timeout_of_nexus_operation( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[ScheduleToStartTimeoutTestService()], @@ -412,9 +404,6 @@ async def test_error_raised_by_schedule_to_start_timeout_of_nexus_operation( task_queue=task_queue, nexus_task_executor=concurrent.futures.ThreadPoolExecutor(), ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(task_queue), task_queue - ) try: await client.execute_workflow( ScheduleToStartTimeoutTestCallerWorkflow.run, @@ -475,12 +464,12 @@ async def run(self) -> None: async def test_error_raised_by_start_to_close_timeout_of_nexus_operation( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[StartToCloseTimeoutTestService()], @@ -488,9 +477,6 @@ async def test_error_raised_by_start_to_close_timeout_of_nexus_operation( task_queue=task_queue, nexus_task_executor=concurrent.futures.ThreadPoolExecutor(), ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(task_queue), task_queue - ) try: await client.execute_workflow( StartToCloseTimeoutTestCallerWorkflow.run, @@ -556,12 +542,12 @@ async def run(self) -> None: async def test_error_raised_by_timeout_of_nexus_cancel_operation( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue async with Worker( client, nexus_service_handlers=[CancellationTimeoutTestService()], @@ -569,9 +555,6 @@ async def test_error_raised_by_timeout_of_nexus_cancel_operation( task_queue=task_queue, ): with LogCapturer().logs_captured(logger) as capturer: - await env.create_nexus_endpoint( - make_nexus_endpoint_name(task_queue), task_queue - ) try: await client.execute_workflow( CancellationTimeoutTestCallerWorkflow.run, @@ -609,13 +592,14 @@ async def test_rpc_error_fails_without_retry( env: WorkflowEnvironment, status_code: RPCStatusCode, expected_handler_error_type: nexusrpc.HandlerErrorType, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") input = RPCErrorInput( status_code_value=status_code.value, - task_queue=str(uuid.uuid4()), + task_queue=nexus_endpoint.task_queue, id=str(uuid.uuid4()), ) async with Worker( @@ -625,9 +609,6 @@ async def test_rpc_error_fails_without_retry( workflows=[RPCErrorCallerWorkflow], task_queue=input.task_queue, ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(input.task_queue), input.task_queue - ) try: await client.execute_workflow( RPCErrorCallerWorkflow.run, @@ -668,13 +649,14 @@ async def test_rpc_error_is_retried( client: Client, env: WorkflowEnvironment, status_code: RPCStatusCode, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") input = RPCErrorInput( status_code_value=status_code.value, - task_queue=str(uuid.uuid4()), + task_queue=nexus_endpoint.task_queue, id=str(uuid.uuid4()), ) async with Worker( @@ -684,10 +666,6 @@ async def test_rpc_error_is_retried( workflows=[RPCErrorCallerWorkflow], task_queue=input.task_queue, ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(input.task_queue), input.task_queue - ) - handle = await client.start_workflow( RPCErrorCallerWorkflow.run, input, @@ -743,12 +721,12 @@ def from_payloads( async def test_nexus_operation_retried_on_codec_decode_failure( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue codec = FailOnFirstDecodeCodec() handler_client = Client( client.service_client, @@ -774,9 +752,6 @@ async def test_nexus_operation_retried_on_codec_decode_failure( task_queue=task_queue, ), ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(input.task_queue), input.task_queue - ) await client.execute_workflow( CallerWorkflow.run, input, @@ -787,12 +762,12 @@ async def test_nexus_operation_retried_on_codec_decode_failure( async def test_nexus_operation_fails_without_retry_on_converter_failure( - client: Client, env: WorkflowEnvironment + client: Client, env: WorkflowEnvironment, nexus_endpoint: NexusEndpoint ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) + task_queue = nexus_endpoint.task_queue handler_client = Client( client.service_client, namespace=client.namespace, @@ -819,9 +794,6 @@ async def test_nexus_operation_fails_without_retry_on_converter_failure( task_queue=task_queue, ), ): - await env.create_nexus_endpoint( - make_nexus_endpoint_name(input.task_queue), input.task_queue - ) try: await client.execute_workflow( CallerWorkflow.run, diff --git a/tests/nexus/test_workflow_run_operation.py b/tests/nexus/test_workflow_run_operation.py index 7135fde71..faa78faca 100644 --- a/tests/nexus/test_workflow_run_operation.py +++ b/tests/nexus/test_workflow_run_operation.py @@ -22,10 +22,7 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker from tests.helpers.nexus import make_nexus_endpoint_name - -# Cloud CI's namespace credentials cannot manage Nexus endpoints. -# See https://github.com/temporalio/sdk-python/issues/1704. -pytestmark = pytest.mark.requires_local_server +from tests.nexus.conftest import NexusEndpoint @dataclass @@ -138,12 +135,12 @@ async def test_workflow_run_operation( client: Client, env: WorkflowEnvironment, service_handler_cls: type[Any], + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) - await env.create_nexus_endpoint(make_nexus_endpoint_name(task_queue), task_queue) + task_queue = nexus_endpoint.task_queue assert (service_defn := nexusrpc.get_service_definition(service_handler_cls)) async with Worker( client, @@ -164,14 +161,13 @@ async def test_workflow_run_operation( async def test_request_deadline_is_accessible_in_workflow_run_operation( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): """Test that request_deadline is accessible in WorkflowRunOperationContext.""" if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) - endpoint_name = make_nexus_endpoint_name(task_queue) - await env.create_nexus_endpoint(endpoint_name, task_queue) + task_queue = nexus_endpoint.task_queue service_handler = RequestDeadlineHandler() async with Worker( env.client, @@ -198,12 +194,12 @@ async def test_request_deadline_is_accessible_in_workflow_run_operation( async def test_workflow_run_operation_includes_token_in_callback( client: Client, env: WorkflowEnvironment, + nexus_endpoint: NexusEndpoint, ): if env.supports_time_skipping: pytest.skip("Nexus tests don't work with time-skipping server") - task_queue = str(uuid.uuid4()) - await env.create_nexus_endpoint(make_nexus_endpoint_name(task_queue), task_queue) + task_queue = nexus_endpoint.task_queue async with Worker( client, task_queue=task_queue,