From 62fa78b33bd03b88fde8a14885fe16e39cfe6e2a Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Tue, 28 Jul 2026 12:11:40 -0700 Subject: [PATCH 01/10] test: run integration suite against Cloud --- .github/workflows/ci.yml | 18 ++-- temporalio/testing/_workflow.py | 20 +++++ tests/conftest.py | 44 +++++++++- .../aws/s3driver/test_s3driver_worker.py | 8 +- .../workflow_streams/test_workflow_streams.py | 3 +- tests/nexus/test_nexus_client_updates.py | 4 +- tests/nexus/test_workflow_caller.py | 8 +- tests/test_client.py | 23 ++--- tests/test_cloud.py | 67 +-------------- tests/test_envconfig.py | 83 +++++++++++++++++++ tests/test_plugins.py | 1 + tests/test_runtime.py | 29 +++---- tests/worker/test_extstore.py | 49 +++-------- tests/worker/test_payload_size_limits.py | 14 +--- tests/worker/test_worker.py | 16 ++-- tests/worker/test_workflow.py | 26 +++--- 16 files changed, 215 insertions(+), 198 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 86dbe6253..aa1d7b8ed 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -204,10 +204,10 @@ jobs: path: junit-xml retention-days: 14 - # Run tests against Temporal Cloud (skipped on forks) + # Run the test suite against Temporal Cloud (skipped on forks) cloud-test: if: ${{ github.event.pull_request.head.repo.full_name == '' || github.event.pull_request.head.repo.full_name == 'temporalio/sdk-python' }} - timeout-minutes: 15 + timeout-minutes: 30 runs-on: ubuntu-latest steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 @@ -230,16 +230,20 @@ jobs: - run: uv tool install poethepoet - run: uv sync --all-extras - run: poe build-develop - - run: poe test -s tests/test_cloud.py --junit-xml=junit-xml/cloud.xml - timeout-minutes: 10 + - run: mkdir junit-xml + - run: poe test -s --junit-xml=junit-xml/cloud.xml + timeout-minutes: 15 env: + TEMPORAL_TEST_ENV_CONFIG_SERVER: true + TEMPORAL_ADDRESS: sdk-ci.a2dd6.tmprl.cloud:7233 + TEMPORAL_NAMESPACE: sdk-ci.a2dd6 + TEMPORAL_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }} + TEMPORAL_TLS_CLIENT_CERT_DATA: ${{ secrets.TEMPORAL_CLIENT_CERT }} + TEMPORAL_TLS_CLIENT_KEY_DATA: ${{ secrets.TEMPORAL_CLIENT_KEY }} TEMPORAL_IS_CLOUD_TESTS: true TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }} TEMPORAL_CLIENT_CLOUD_API_VERSION: 2024-05-13-00 TEMPORAL_CLIENT_CLOUD_NAMESPACE: sdk-ci.a2dd6 - TEMPORAL_CLIENT_CLOUD_TARGET: sdk-ci.a2dd6.tmprl.cloud:7233 - TEMPORAL_CLIENT_CERT: ${{ secrets.TEMPORAL_CLIENT_CERT }} - TEMPORAL_CLIENT_KEY: ${{ secrets.TEMPORAL_CLIENT_KEY }} - name: "Upload junit-xml artifacts" uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 if: always() diff --git a/temporalio/testing/_workflow.py b/temporalio/testing/_workflow.py index 979222dea..3c829a132 100644 --- a/temporalio/testing/_workflow.py +++ b/temporalio/testing/_workflow.py @@ -373,6 +373,26 @@ def client(self) -> temporalio.client.Client: """Client to this environment.""" return self._client + async def connect_client(self, **kwargs: Any) -> temporalio.client.Client: + """Create another client connected to this environment. + + Namespace and connection credentials from this environment's client are + used by default. + Keyword arguments are forwarded to :py:meth:`temporalio.client.Client.connect` + and override those defaults. + """ + config = self.client.service_client.config + connect_kwargs: dict[str, Any] = { + "namespace": self.client.namespace, + "api_key": config.api_key, + "tls": config.tls, + "rpc_metadata": config.rpc_metadata, + } + connect_kwargs.update(kwargs) + return await temporalio.client.Client.connect( + config.target_host, **connect_kwargs + ) + async def shutdown(self) -> None: """Shut down this environment.""" pass diff --git a/tests/conftest.py b/tests/conftest.py index bfe005ede..909b9bf1e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,6 +10,7 @@ from opentelemetry.util._once import Once from temporalio.client import Client +from temporalio.envconfig import ClientConfigProfile from temporalio.testing import WorkflowEnvironment from temporalio.worker import SharedStateManager from tests.helpers.worker import ExternalPythonWorker, ExternalWorker @@ -58,10 +59,47 @@ def pytest_addoption(parser): # type: ignore[reportMissingParameterType] "-E", "--workflow-environment", default="local", - help="Which workflow environment to use ('local', 'time-skipping', or ip:port for existing server)", + help="Which workflow environment to use ('local', 'time-skipping', 'envconfig', or ip:port for existing server)", ) +def _uses_envconfig_server(env_type: str) -> bool: + if env_type == "envconfig": + return True + return env_type == "local" and os.getenv( + "TEMPORAL_TEST_ENV_CONFIG_SERVER", "" + ).lower() not in ("", "0", "false", "no", "off") + + +def pytest_configure(config: pytest.Config) -> None: + config.addinivalue_line( + "markers", + "requires_local_server: test requires local-server-only behavior and cannot run against an envconfig server", + ) + + +def pytest_collection_modifyitems( + config: pytest.Config, items: list[pytest.Item] +) -> None: + if not _uses_envconfig_server(config.getoption("--workflow-environment")): + return + skip_local_only = pytest.mark.skip( + reason="requires a local Temporal server, not the configured envconfig server" + ) + for item in items: + if item.get_closest_marker("requires_local_server"): + item.add_marker(skip_local_only) + + +async def _create_env_from_envconfig() -> WorkflowEnvironment: + config = ClientConfigProfile.load().to_client_connect_config() + if not config.get("target_host"): + raise ValueError( + "An envconfig workflow environment requires TEMPORAL_ADDRESS or an envconfig profile with an address" + ) + return WorkflowEnvironment.from_client(await Client.connect(**config)) + + @pytest.fixture(scope="session") def event_loop(): loop = asyncio.get_event_loop_policy().new_event_loop() # type: ignore[reportDeprecated] @@ -100,7 +138,9 @@ def env_type(request: pytest.FixtureRequest) -> str: @pytest_asyncio.fixture(scope="session") # type: ignore[reportUntypedFunctionDecorator] async def env(env_type: str) -> AsyncGenerator[WorkflowEnvironment, None]: - if env_type == "local": + if _uses_envconfig_server(env_type): + env = await _create_env_from_envconfig() + elif env_type == "local": env = await WorkflowEnvironment.start_local( dev_server_extra_args=[ "--dynamic-config-value", diff --git a/tests/contrib/aws/s3driver/test_s3driver_worker.py b/tests/contrib/aws/s3driver/test_s3driver_worker.py index 61729535f..12b813adf 100644 --- a/tests/contrib/aws/s3driver/test_s3driver_worker.py +++ b/tests/contrib/aws/s3driver/test_s3driver_worker.py @@ -62,9 +62,7 @@ async def tmprl_client( ) -> AsyncIterator[Client]: """Temporal client wired with ExternalStorage backed by the moto S3 server.""" driver = S3StorageDriver(client=new_aioboto3_client(aioboto3_client), bucket=BUCKET) - yield await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + yield await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( @@ -469,9 +467,7 @@ async def test_s3_store_failure_surfaces_in_workflow_history( aws_secret_access_key="testing", ) as client: driver = S3StorageDriver(client=new_aioboto3_client(client), bucket=bad_bucket) - bad_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + bad_client = await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( diff --git a/tests/contrib/workflow_streams/test_workflow_streams.py b/tests/contrib/workflow_streams/test_workflow_streams.py index 12026ff1a..851fdabf1 100644 --- a/tests/contrib/workflow_streams/test_workflow_streams.py +++ b/tests/contrib/workflow_streams/test_workflow_streams.py @@ -2722,8 +2722,7 @@ async def test_cross_namespace_nexus_stream( ) ) - handler_client = await Client.connect( - client.service_client.config.target_host, + handler_client = await env.connect_client( namespace=handler_ns, ) diff --git a/tests/nexus/test_nexus_client_updates.py b/tests/nexus/test_nexus_client_updates.py index f63d5482c..302ede54e 100644 --- a/tests/nexus/test_nexus_client_updates.py +++ b/tests/nexus/test_nexus_client_updates.py @@ -48,9 +48,7 @@ async def test_nexus_client_updates_when_worker_client_changes( """Test that Nexus operations get the updated client when worker.client is changed.""" # Create a second client (simulating a new client after cert rotation) # Must use the same runtime - client2 = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client2 = await env.connect_client( data_converter=env.client.data_converter, runtime=env.client.service_client.config.runtime, ) diff --git a/tests/nexus/test_workflow_caller.py b/tests/nexus/test_workflow_caller.py index 89ce2719a..328637525 100644 --- a/tests/nexus/test_workflow_caller.py +++ b/tests/nexus/test_workflow_caller.py @@ -1979,9 +1979,7 @@ async def test_workflow_caller_custom_metrics(client: Client, env: WorkflowEnvir ) # New client with the runtime - client = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client = await env.connect_client( runtime=runtime, ) @@ -2052,9 +2050,7 @@ async def test_workflow_caller_buffered_metrics( assert not buffer.retrieve_updates() # Create a new client on the runtime and execute the custom metric workflow - client = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client = await env.connect_client( runtime=runtime, ) task_queue = str(uuid.uuid4()) diff --git a/tests/test_client.py b/tests/test_client.py index d611eda3a..e7fd7172a 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -556,7 +556,7 @@ async def test_interceptor(client: Client, worker: ExternalWorker): assert interceptor.traces[4][1].id == handle.id -async def test_lazy_client(client: Client, env: WorkflowEnvironment): +async def test_lazy_client(env: WorkflowEnvironment): # TODO(cretz): Fix if env.supports_time_skipping: pytest.skip( @@ -564,23 +564,17 @@ async def test_lazy_client(client: Client, env: WorkflowEnvironment): ) # Create another client that is lazy. This test just makes sure the # functionality continues to work. - lazy_client = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, - lazy=True, - ) + lazy_client = await env.connect_client(lazy=True) assert not lazy_client.service_client.worker_service_client._bridge_client await lazy_client.workflow_service.get_system_info(GetSystemInfoRequest()) assert lazy_client.service_client.worker_service_client._bridge_client -async def test_client_connected_skips_connect_lock(client: Client): +async def test_client_connected_skips_connect_lock(env: WorkflowEnvironment): # Once connected, RPCs must not touch the lazy-connect lock. Acquiring it # per-RPC put an event-loop-bound primitive on the hot path, pinning a # connected client to the loop it connected on. - other = await Client.connect( - client.service_client.config.target_host, namespace=client.namespace - ) + other = await env.connect_client() svc = other.service_client.worker_service_client assert svc._bridge_client @@ -599,18 +593,13 @@ async def acquire(self) -> Literal[True]: assert counting.acquire_count == 0 -def test_client_reuse_across_event_loops(client: Client): +def test_client_reuse_across_event_loops(env: WorkflowEnvironment): # A connected client must not be pinned to the loop (or thread) it # connected on. This mirrors the long-lived-loop reuse pattern used by # gevent/gunicorn and synchronous services. - target_host = client.service_client.config.target_host - namespace = client.namespace - connect_loop = asyncio.new_event_loop() try: - reused_client = connect_loop.run_until_complete( - Client.connect(target_host, namespace=namespace) - ) + reused_client = connect_loop.run_until_complete(env.connect_client()) finally: connect_loop.close() diff --git a/tests/test_cloud.py b/tests/test_cloud.py index b701bdf94..d7fefb4da 100644 --- a/tests/test_cloud.py +++ b/tests/test_cloud.py @@ -1,18 +1,11 @@ -"""Tests that run against Temporal Cloud.""" +"""Tests that run against the Temporal Cloud Operations API.""" -import multiprocessing import os -from collections.abc import AsyncGenerator, Iterator import pytest -import pytest_asyncio from temporalio.api.cloud.cloudservice.v1 import GetNamespaceRequest -from temporalio.client import Client, CloudOperationsClient -from temporalio.service import TLSConfig -from temporalio.testing import WorkflowEnvironment -from temporalio.worker import SharedStateManager -from tests.helpers.worker import ExternalPythonWorker, ExternalWorker +from temporalio.client import CloudOperationsClient # Skip entire module unless explicitly enabled pytestmark = pytest.mark.skipif( @@ -21,54 +14,6 @@ ) -@pytest_asyncio.fixture(scope="module") # type: ignore[reportUntypedFunctionDecorator] -async def env() -> AsyncGenerator[WorkflowEnvironment, None]: - tls_config: bool | TLSConfig = True - client_cert = os.environ.get("TEMPORAL_CLIENT_CERT") - client_key = os.environ.get("TEMPORAL_CLIENT_KEY") - if client_cert and client_key: - tls_config = TLSConfig( - client_cert=client_cert.encode(), - client_private_key=client_key.encode(), - ) - client = await Client.connect( - os.environ["TEMPORAL_CLIENT_CLOUD_TARGET"], - namespace=os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"], - api_key=os.environ.get("TEMPORAL_CLIENT_CLOUD_API_KEY"), - tls=tls_config, - ) - env = WorkflowEnvironment.from_client(client) - yield env - await env.shutdown() - - -@pytest_asyncio.fixture # type: ignore[reportUntypedFunctionDecorator] -async def client(env: WorkflowEnvironment) -> Client: - return env.client - - -@pytest_asyncio.fixture(scope="module") # type: ignore[reportUntypedFunctionDecorator] -async def worker( - env: WorkflowEnvironment, -) -> AsyncGenerator[ExternalWorker, None]: - w = ExternalPythonWorker(env) - yield w - await w.close() - - -@pytest.fixture(scope="module") -def shared_state_manager() -> Iterator[SharedStateManager]: - mp_mgr = multiprocessing.Manager() - mgr = SharedStateManager.create_from_multiprocessing(mp_mgr) - try: - yield mgr - finally: - mp_mgr.shutdown() - - -# --- Cloud-specific tests --- - - async def test_cloud_client_simple(): client = await CloudOperationsClient.connect( api_key=os.environ["TEMPORAL_CLIENT_CLOUD_API_KEY"], @@ -78,11 +23,3 @@ async def test_cloud_client_simple(): GetNamespaceRequest(namespace=os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"]) ) assert os.environ["TEMPORAL_CLIENT_CLOUD_NAMESPACE"] == result.namespace.namespace - - -# --- Delegated tests --- -# Import test functions to re-run them against cloud fixtures. - -from tests.worker.test_activity import ( # noqa: E402 - test_activity_info, # pyright: ignore[reportUnusedImport] # noqa: F401 -) diff --git a/tests/test_envconfig.py b/tests/test_envconfig.py index c1a7e32ab..0a6afd94b 100644 --- a/tests/test_envconfig.py +++ b/tests/test_envconfig.py @@ -1,12 +1,17 @@ import os import textwrap from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest +import temporalio.service from temporalio.client import Client from temporalio.envconfig import ClientConfig, ClientConfigProfile, ClientConfigTLS from temporalio.service import TLSConfig +from temporalio.testing import WorkflowEnvironment +from tests import conftest # A base TOML config with a default and a custom profile TOML_CONFIG_BASE = textwrap.dedent( @@ -147,6 +152,84 @@ def test_load_profile_from_data_env_overrides(): assert config.get("target_host") == "env-address" +@pytest.mark.parametrize( + ("env_type", "flag", "expected"), + [ + ("envconfig", None, True), + ("local", "true", True), + ("local", "false", False), + ("time-skipping", "true", False), + ], +) +def test_envconfig_server_selection( + monkeypatch: pytest.MonkeyPatch, + env_type: str, + flag: str | None, + expected: bool, +): + if flag is not None: + monkeypatch.setenv("TEMPORAL_TEST_ENV_CONFIG_SERVER", flag) + assert conftest._uses_envconfig_server(env_type) is expected + + +async def test_envconfig_workflow_environment_uses_client_connect_config( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv("TEMPORAL_ADDRESS", "env-address") + monkeypatch.setenv("TEMPORAL_NAMESPACE", "env-namespace") + monkeypatch.setenv("TEMPORAL_API_KEY", "env-api-key") + monkeypatch.setenv("TEMPORAL_TLS", "false") + monkeypatch.setenv("TEMPORAL_GRPC_META_TEST_HEADER", "env-value") + client = object() + environment = object() + connect = AsyncMock(return_value=client) + monkeypatch.setattr(Client, "connect", connect) + monkeypatch.setattr( + WorkflowEnvironment, + "from_client", + lambda actual_client: environment, + ) + + assert await conftest._create_env_from_envconfig() is environment + connect.assert_awaited_once_with( + target_host="env-address", + namespace="env-namespace", + api_key="env-api-key", + tls=False, + rpc_metadata={"test-header": "env-value"}, + ) + + +async def test_workflow_environment_connect_client_inherits_connection_options( + monkeypatch: pytest.MonkeyPatch, +): + source_client = SimpleNamespace( + namespace="env-namespace", + service_client=SimpleNamespace( + config=temporalio.service.ConnectConfig( + target_host="env-address", + api_key="env-api-key", + tls=False, + rpc_metadata={"test-header": "env-value"}, + ) + ), + ) + environment = WorkflowEnvironment(source_client) # type: ignore[arg-type] + connect = AsyncMock(return_value=object()) + monkeypatch.setattr(Client, "connect", connect) + + await environment.connect_client(lazy=True) + + connect.assert_awaited_once_with( + "env-address", + namespace="env-namespace", + api_key="env-api-key", + tls=False, + rpc_metadata={"test-header": "env-value"}, + lazy=True, + ) + + def test_load_profile_env_overrides(base_config_file: Path): """Test that environment variables correctly override file settings.""" env = { diff --git a/tests/test_plugins.py b/tests/test_plugins.py index e8823af27..9414e8df0 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -60,6 +60,7 @@ async def connect_service_client( return await next(config) +@pytest.mark.requires_local_server async def test_client_plugin(client: Client, env: WorkflowEnvironment): if env.supports_time_skipping: pytest.skip("Client connect is only designed for local") diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 55501883d..6416ce930 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -21,6 +21,7 @@ TelemetryFilter, _RuntimeRef, ) +from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker from tests.helpers import ( LogHandler, @@ -37,22 +38,18 @@ async def run(self, name: str) -> str: return f"Hello, {name}!" -async def test_different_runtimes(client: Client): +async def test_different_runtimes(env: WorkflowEnvironment): # Create two workers in separate runtimes and run workflows on them. # Confirm they each have different Prometheus addresses. prom_addr1 = f"127.0.0.1:{find_free_port()}" - client1 = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client1 = await env.connect_client( runtime=Runtime( telemetry=TelemetryConfig(metrics=PrometheusConfig(bind_address=prom_addr1)) ), ) prom_addr2 = f"127.0.0.1:{find_free_port()}" - client2 = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client2 = await env.connect_client( runtime=Runtime( telemetry=TelemetryConfig(metrics=PrometheusConfig(bind_address=prom_addr2)) ), @@ -151,16 +148,14 @@ async def run(self) -> None: raise RuntimeError("Intentional error") -async def test_runtime_task_fail_log_forwarding(client: Client): +async def test_runtime_task_fail_log_forwarding(env: WorkflowEnvironment): # Client with lo capturing runtime log_queue: queue.Queue[logging.LogRecord] = queue.Queue() log_queue_list = cast(list[logging.LogRecord], log_queue.queue) handler = logging.handlers.QueueHandler(log_queue) logger = logging.getLogger(f"log-{uuid.uuid4()}") logger.setLevel(logging.WARN) - client = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client = await env.connect_client( runtime=Runtime( telemetry=TelemetryConfig( logging=LoggingConfig( @@ -199,7 +194,7 @@ async def has_log() -> bool: assert record.temporal_log.fields["run_id"] == handle.result_run_id # type: ignore -async def test_prometheus_histogram_bucket_overrides(client: Client): +async def test_prometheus_histogram_bucket_overrides(env: WorkflowEnvironment): # Set up a Prometheus configuration with custom histogram bucket overrides prom_addr = f"127.0.0.1:{find_free_port()}" special_value = float(1234.5678) @@ -229,9 +224,7 @@ async def test_prometheus_histogram_bucket_overrides(client: Client): custom_histogram.record(600) # Create client with overrides - client_with_overrides = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client_with_overrides = await env.connect_client( runtime=runtime, ) @@ -270,7 +263,7 @@ async def check_metrics() -> None: await assert_eventually(check_metrics) -async def test_opentelemetry_histogram_bucket_overrides(client: Client): +async def test_opentelemetry_histogram_bucket_overrides(env: WorkflowEnvironment): # Set up an OpenTelemetry configuration with custom histogram bucket overrides import threading from http.server import BaseHTTPRequestHandler, HTTPServer @@ -336,9 +329,7 @@ def do_POST(self): # Run a workflow so built-in histograms (e.g. temporal_long_request_latency) # are recorded and exported. - client_with_overrides = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client_with_overrides = await env.connect_client( runtime=runtime, ) task_queue = f"task-queue-{uuid.uuid4()}" diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index 998a2bd27..2f8fde5fe 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -181,9 +181,7 @@ async def test_extstore_activity_input_no_retrieve( WorkflowFailureError wrapping an ActivityError.""" driver = BadTestDriver(no_retrieve=True) - client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client = await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( @@ -225,9 +223,7 @@ async def test_extstore_activity_result_no_store( terminates with a WorkflowFailureError wrapping an ActivityError.""" driver = BadTestDriver(no_store=True) - client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client = await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( @@ -274,9 +270,7 @@ async def test_extstore_worker_missing_driver( """ driver = InMemoryTestDriver() - far_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + far_client = await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( @@ -286,10 +280,7 @@ async def test_extstore_worker_missing_driver( ), ) - worker_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, - ) + worker_client = await env.connect_client() async with new_worker( worker_client, ExtStoreWorkflow, activities=[ext_store_activity] @@ -315,9 +306,7 @@ async def test_extstore_payload_not_found_fails_workflow( """When a non-retryable ApplicationError is raised while retrieving workflow input, the workflow must fail terminally (not retry as a task failure). """ - client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client = await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( @@ -363,9 +352,7 @@ async def _run_extstore_workflow_and_fetch_history( activity_output_size: int = 10, ) -> WorkflowHandle: """Helper: run ExtStoreWorkflow with the given driver and return its history handle.""" - extstore_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + extstore_client = await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( @@ -515,9 +502,7 @@ async def test_extstore_chained_activities( """ driver = InMemoryTestDriver() - client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client = await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( @@ -567,9 +552,7 @@ def __init__(self, driver_name: str): driver2 = InMemoryTestDriver(driver_name="driver2") driver3 = DifferentTestDriver(driver_name="driver3") - client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client = await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( @@ -685,9 +668,7 @@ async def test_tmprl1104_with_extstore_download(env: WorkflowEnvironment) -> Non payload_size_threshold=512, ), ) - client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client = await env.connect_client( data_converter=data_converter, ) @@ -737,9 +718,7 @@ async def test_tmprl1104_with_extstore_upload(env: WorkflowEnvironment) -> None: payload_size_threshold=512, ), ) - client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client = await env.connect_client( data_converter=data_converter, ) @@ -791,9 +770,7 @@ async def test_tmprl1104_with_extstore_download_and_upload( payload_size_threshold=512, ), ) - client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client = await env.connect_client( data_converter=data_converter, ) @@ -914,9 +891,7 @@ async def _make_tracking_client( env: WorkflowEnvironment, ) -> tuple[Client, ContextTrackingStorageDriver]: driver = ContextTrackingStorageDriver() - client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + client = await env.connect_client( data_converter=dataclasses.replace( temporalio.converter.default(), external_storage=ExternalStorage( diff --git a/tests/worker/test_payload_size_limits.py b/tests/worker/test_payload_size_limits.py index b527f429c..18cc7e75a 100644 --- a/tests/worker/test_payload_size_limits.py +++ b/tests/worker/test_payload_size_limits.py @@ -7,7 +7,7 @@ import temporalio.api.enums.v1 from temporalio import activity, workflow -from temporalio.client import Client, PayloadLimitsConfig, WorkflowFailureError +from temporalio.client import PayloadLimitsConfig, WorkflowFailureError from temporalio.exceptions import ( TerminatedError, TimeoutError, @@ -91,9 +91,7 @@ async def test_oversized_payload_fails_task_with_error_log(env: WorkflowEnvironm dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION, ) as env: worker_logger = logging.getLogger(f"log-{uuid.uuid4()}") - worker_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + worker_client = await env.connect_client( runtime=_forwarding_runtime(worker_logger), ) @@ -179,9 +177,7 @@ async def test_disable_payload_error_limit_sends_to_server(env: WorkflowEnvironm async def test_payload_size_warning_forwarded(env: WorkflowEnvironment): """The connection's warn threshold produces a forwarded [TMPRL1103] warning for over-threshold payloads.""" worker_logger = logging.getLogger(f"log-{uuid.uuid4()}") - worker_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + worker_client = await env.connect_client( runtime=_forwarding_runtime(worker_logger), payload_limits=PayloadLimitsConfig(payloads_warn_size=1024), ) @@ -219,9 +215,7 @@ async def test_memo_size_warning_forwarded(env: WorkflowEnvironment): """The connection's memo warn threshold produces a forwarded [TMPRL1103] warning for an over-threshold memo.""" worker_logger = logging.getLogger(f"log-{uuid.uuid4()}") - worker_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, + worker_client = await env.connect_client( runtime=_forwarding_runtime(worker_logger), payload_limits=PayloadLimitsConfig(memo_warn_size=1024), ) diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 3a4e9165b..6f2667a80 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -1206,7 +1206,9 @@ async def test_workflows_can_use_versioning_override( ) -async def test_can_run_autoscaling_polling_worker(client: Client): +async def test_can_run_autoscaling_polling_worker( + client: Client, env: WorkflowEnvironment +): # Create new runtime with Prom server prom_addr = f"127.0.0.1:{find_free_port()}" runtime = Runtime( @@ -1214,9 +1216,7 @@ async def test_can_run_autoscaling_polling_worker(client: Client): metrics=PrometheusConfig(bind_address=prom_addr), ) ) - client = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client = await env.connect_client( runtime=runtime, ) @@ -1467,13 +1467,13 @@ def test_fork_use_worker( self.run(mp_fork_ctx) -async def test_activity_client_updates_when_worker_client_changes(client: Client): +async def test_activity_client_updates_when_worker_client_changes( + client: Client, env: WorkflowEnvironment +): """Test that activities get the updated client when worker.client is changed.""" # Create a second client (simulating a new client after cert rotation) # Must use the same runtime - client2 = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client2 = await env.connect_client( data_converter=client.data_converter, runtime=client.service_client.config.runtime, ) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index dbd85ef20..b78e1e8c0 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -4613,7 +4613,7 @@ async def run(self) -> None: ) -async def test_workflow_custom_metrics(client: Client): +async def test_workflow_custom_metrics(client: Client, env: WorkflowEnvironment): # Run worker with default runtime which is noop meter just to confirm it # doesn't fail async with new_worker( @@ -4639,9 +4639,7 @@ async def test_workflow_custom_metrics(client: Client): assert str(err.value).startswith("Invalid value type for key") # New client with the runtime - client = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client = await env.connect_client( runtime=runtime, ) @@ -4718,7 +4716,7 @@ async def test_workflow_custom_metrics(client: Client): ) -async def test_workflow_buffered_metrics(client: Client): +async def test_workflow_buffered_metrics(client: Client, env: WorkflowEnvironment): # Create runtime with metric buffer buffer = MetricBuffer(10000) runtime = Runtime( @@ -4779,9 +4777,7 @@ async def test_workflow_buffered_metrics(client: Client): assert runtime_updates2[1].value == 400 # Create a new client on the runtime and execute the custom metric workflow - client = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + client = await env.connect_client( runtime=runtime, ) async with new_worker( @@ -4842,12 +4838,10 @@ async def test_workflow_buffered_metrics(client: Client): ) -async def test_workflow_metrics_other_types(client: Client): +async def test_workflow_metrics_other_types(env: WorkflowEnvironment): async def do_stuff(buffer: MetricBuffer) -> None: runtime = Runtime(telemetry=TelemetryConfig(metrics=buffer)) - new_client = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + new_client = await env.connect_client( runtime=runtime, ) async with new_worker(new_client, HelloWorkflow) as worker: @@ -5922,11 +5916,11 @@ async def any_task_completed(handle: WorkflowHandle) -> bool: await handle2.terminate() -async def test_workflow_replace_worker_client_diff_runtimes_fail(client: Client): +async def test_workflow_replace_worker_client_diff_runtimes_fail( + client: Client, env: WorkflowEnvironment +): other_runtime = Runtime(telemetry=TelemetryConfig()) - other_client = await Client.connect( - client.service_client.config.target_host, - namespace=client.namespace, + other_client = await env.connect_client( runtime=other_runtime, ) async with new_worker(client, HelloWorkflow) as worker: From 60e49caf6f02ef17c83727006fd3fd476bdb9d04 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Tue, 28 Jul 2026 13:23:23 -0700 Subject: [PATCH 02/10] test: scope Cloud integration coverage --- .github/workflows/ci.yml | 3 +- tests/conftest.py | 6 +- .../aws/s3driver/test_s3driver_worker.py | 35 ++++++++--- tests/contrib/langsmith/test_integration.py | 3 + tests/contrib/langsmith/test_plugin.py | 1 + .../langsmith/test_tracing_env_override.py | 1 + tests/contrib/openai_agents/test_openai.py | 1 + .../opentelemetry/test_opentelemetry.py | 1 + .../test_opentelemetry_plugin.py | 1 + .../workflow_streams/test_workflow_streams.py | 1 + ...ynamic_creation_of_user_handler_classes.py | 2 + tests/nexus/test_nexus_client_updates.py | 3 + tests/nexus/test_nexus_worker_shutdown.py | 2 + .../nexus/test_signal_link_propagation_e2e.py | 2 + tests/nexus/test_standalone_operations.py | 3 + tests/nexus/test_temporal_operation.py | 2 + .../test_use_existing_conflict_policy.py | 3 + tests/nexus/test_workflow_caller.py | 3 + ...test_workflow_caller_cancellation_types.py | 2 + ...llation_types_when_cancel_handler_fails.py | 2 + .../test_workflow_caller_error_chains.py | 2 + tests/nexus/test_workflow_caller_errors.py | 2 + tests/nexus/test_workflow_run_operation.py | 2 + tests/test_activity.py | 2 + tests/test_client.py | 63 ++++++------------- tests/test_envconfig.py | 15 ++--- tests/test_serialization_context.py | 35 ++++++----- tests/test_service.py | 1 + tests/testing/test_workflow.py | 1 + tests/worker/test_interceptor.py | 1 + tests/worker/test_replayer.py | 8 +++ tests/worker/test_worker.py | 1 + tests/worker/test_workflow.py | 5 ++ 33 files changed, 130 insertions(+), 85 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index aa1d7b8ed..e97b81c80 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -231,10 +231,9 @@ jobs: - run: uv sync --all-extras - run: poe build-develop - run: mkdir junit-xml - - run: poe test -s --junit-xml=junit-xml/cloud.xml + - run: poe test -s --workflow-environment envconfig --junit-xml=junit-xml/cloud.xml timeout-minutes: 15 env: - TEMPORAL_TEST_ENV_CONFIG_SERVER: true TEMPORAL_ADDRESS: sdk-ci.a2dd6.tmprl.cloud:7233 TEMPORAL_NAMESPACE: sdk-ci.a2dd6 TEMPORAL_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }} diff --git a/tests/conftest.py b/tests/conftest.py index 909b9bf1e..9c57bc0d1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -64,11 +64,7 @@ def pytest_addoption(parser): # type: ignore[reportMissingParameterType] def _uses_envconfig_server(env_type: str) -> bool: - if env_type == "envconfig": - return True - return env_type == "local" and os.getenv( - "TEMPORAL_TEST_ENV_CONFIG_SERVER", "" - ).lower() not in ("", "0", "false", "no", "off") + return env_type == "envconfig" def pytest_configure(config: pytest.Config) -> None: diff --git a/tests/contrib/aws/s3driver/test_s3driver_worker.py b/tests/contrib/aws/s3driver/test_s3driver_worker.py index 12b813adf..fcf17fc16 100644 --- a/tests/contrib/aws/s3driver/test_s3driver_worker.py +++ b/tests/contrib/aws/s3driver/test_s3driver_worker.py @@ -110,7 +110,8 @@ async def test_s3_driver_workflow_input_key( # worker stores activity input with ri=run_id — same bytes, two S3 objects. assert len(keys) == 2 assert all( - f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k for k in keys + f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k + for k in keys ) # Client-side store: ri=null because run ID is not yet known. assert sum(1 for k in keys if "/ri/null/" in k) == 1 @@ -136,7 +137,10 @@ async def test_s3_driver_workflow_output_key( keys = await _list_keys(aioboto3_client) # Activity result and workflow result dedup to same key assert len(keys) == 1 - assert f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in keys[0] + assert ( + f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" + in keys[0] + ) # Run ID is known for both activity completion and workflow completion assert "/ri/null/" not in keys[0] @@ -160,7 +164,8 @@ async def test_s3_driver_workflow_activity_input_key( assert len(keys) == 2 # Both keys are under the workflow wi/ri prefix, not the activity. assert all( - f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k for k in keys + f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k + for k in keys ) # Activity input is keyed under the scheduling workflow, not the activity. assert all("/ai/" not in k for k in keys) @@ -183,7 +188,10 @@ async def test_s3_driver_workflow_activity_output_key( keys = await _list_keys(aioboto3_client) # Activity result and workflow result are both LARGE so they deduplicate to one object. assert len(keys) == 1 - assert f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in keys[0] + assert ( + f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" + in keys[0] + ) # ri=run_id for both stores (run ID is known by the time the activity completes). assert "/ri/null/" not in keys[0] @@ -212,7 +220,8 @@ async def test_s3_driver_standalone_activity_input_key( assert len(keys) == 2 # Both keyed under the activity, not a workflow. assert all( - f"/ns/default/at/large_io_activity/ai/{activity_id}/ri/" in k for k in keys + f"/ns/{tmprl_client.namespace}/at/large_io_activity/ai/{activity_id}/ri/" in k + for k in keys ) assert all("/wt/" not in k for k in keys) # Client-side store does not have run ID information @@ -242,7 +251,10 @@ async def test_s3_driver_standalone_activity_output_key( keys = await _list_keys(aioboto3_client) # Only the output is large; keyed under the activity. assert len(keys) == 1 - assert f"/ns/default/at/large_output_activity/ai/{activity_id}/ri/" in keys[0] + assert ( + f"/ns/{tmprl_client.namespace}/at/large_output_activity/ai/{activity_id}/ri/" + in keys[0] + ) assert "/ri/null/" not in keys[0] assert "/wt/" not in keys[0] @@ -335,7 +347,10 @@ async def test_s3_driver_child_workflow_input_key( # Child input is the only large payload — stored under the child's wi/ri. assert len(keys) == 1 # Keyed under the child: child input is stored in the child's context. - assert f"/ns/default/wt/ChildWorkflow/wi/{child_workflow_id}/ri/" in keys[0] + assert ( + f"/ns/{tmprl_client.namespace}/wt/ChildWorkflow/wi/{child_workflow_id}/ri/" + in keys[0] + ) async def test_s3_driver_identified_casing( @@ -357,7 +372,8 @@ async def test_s3_driver_identified_casing( assert len(keys) == 2 # Workflow ID is percent-encoded but casing is preserved verbatim. assert all( - f"/ns/default/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k for k in keys + f"/ns/{tmprl_client.namespace}/wt/LargeIOWorkflow/wi/{workflow_id}/ri/" in k + for k in keys ), "Workflow ID should preserve original case in the key" @@ -384,7 +400,8 @@ async def test_s3_driver_content_dedup( assert len(keys) == 2 # Both are under the same workflow wi/ri prefix despite crossing activity boundaries. assert all( - f"/ns/default/wt/DocumentIngestionWorkflow/wi/{workflow_id}/ri/" in k + f"/ns/{tmprl_client.namespace}/wt/DocumentIngestionWorkflow/wi/{workflow_id}/ri/" + in k for k in keys ) # The two keys differ by content hash only. diff --git a/tests/contrib/langsmith/test_integration.py b/tests/contrib/langsmith/test_integration.py index a89d1ea4a..426b9a3af 100644 --- a/tests/contrib/langsmith/test_integration.py +++ b/tests/contrib/langsmith/test_integration.py @@ -572,6 +572,7 @@ async def test_benign_error_not_marked( class TestComprehensiveTracing: + @pytest.mark.requires_local_server async def test_comprehensive_with_temporal_runs( self, client: Client, env: WorkflowEnvironment ) -> None: @@ -765,6 +766,7 @@ async def user_pipeline() -> WorkflowHandle[Any, Any]: " HandleUpdate:my_unvalidated_update", ] + @pytest.mark.requires_local_server async def test_comprehensive_without_temporal_runs( self, client: Client, env: WorkflowEnvironment ) -> None: @@ -1138,6 +1140,7 @@ async def run(self) -> str: class TestNexusInboundTracing: """Verifies nexus handlers receive tracing_context for @traceable collection.""" + @pytest.mark.requires_local_server async def test_nexus_direct_traceable_without_temporal_runs( self, client: Client, diff --git a/tests/contrib/langsmith/test_plugin.py b/tests/contrib/langsmith/test_plugin.py index 17c21cb7c..0dad5566f 100644 --- a/tests/contrib/langsmith/test_plugin.py +++ b/tests/contrib/langsmith/test_plugin.py @@ -53,6 +53,7 @@ def test_construction_stores_all_config(self) -> None: class TestPluginIntegration: """End-to-end test using LangSmithPlugin as a Temporal client plugin.""" + @pytest.mark.requires_local_server async def test_comprehensive_plugin_trace_hierarchy( self, client: Client, env: WorkflowEnvironment ) -> None: diff --git a/tests/contrib/langsmith/test_tracing_env_override.py b/tests/contrib/langsmith/test_tracing_env_override.py index 9d871e1d3..c8aaa2fb0 100644 --- a/tests/contrib/langsmith/test_tracing_env_override.py +++ b/tests/contrib/langsmith/test_tracing_env_override.py @@ -162,6 +162,7 @@ async def test_no_runs_when_langchain_tracing_v2_disabled( f"{[r.name for r in collector.runs]}" ) + @pytest.mark.requires_local_server async def test_no_runs_when_tracing_disabled_for_nexus_start( self, client: Client, diff --git a/tests/contrib/openai_agents/test_openai.py b/tests/contrib/openai_agents/test_openai.py index 96cc25133..23c8939a5 100644 --- a/tests/contrib/openai_agents/test_openai.py +++ b/tests/contrib/openai_agents/test_openai.py @@ -471,6 +471,7 @@ async def test_tool_failure_workflow(client: Client): @pytest.mark.parametrize("use_local_model", [True, False]) +@pytest.mark.requires_local_server async def test_nexus_tool_workflow( client: Client, env: WorkflowEnvironment, use_local_model: bool ): diff --git a/tests/contrib/opentelemetry/test_opentelemetry.py b/tests/contrib/opentelemetry/test_opentelemetry.py index 71e2fa41d..1bab931ac 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry.py +++ b/tests/contrib/opentelemetry/test_opentelemetry.py @@ -457,6 +457,7 @@ async def test_opentelemetry_tracing_update_with_start( ] +@pytest.mark.requires_local_server async def test_opentelemetry_tracing_nexus(client: Client, env: WorkflowEnvironment): if env.supports_time_skipping: pytest.skip( diff --git a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py index 3fd50e89b..337270d0c 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py +++ b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py @@ -233,6 +233,7 @@ def validate_update_status(self, status: str) -> None: raise ValueError("Status cannot be empty") +@pytest.mark.requires_local_server async def test_opentelemetry_comprehensive_tracing( client: Client, env: WorkflowEnvironment, diff --git a/tests/contrib/workflow_streams/test_workflow_streams.py b/tests/contrib/workflow_streams/test_workflow_streams.py index 851fdabf1..e7cedd038 100644 --- a/tests/contrib/workflow_streams/test_workflow_streams.py +++ b/tests/contrib/workflow_streams/test_workflow_streams.py @@ -2699,6 +2699,7 @@ async def test_subscribe_iterates_through_more_ready(client: Client) -> None: @pytest.mark.asyncio +@pytest.mark.requires_local_server async def test_cross_namespace_nexus_stream( client: Client, env: WorkflowEnvironment ) -> None: 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 f7306a46b..22c1a818b 100644 --- a/tests/nexus/test_dynamic_creation_of_user_handler_classes.py +++ b/tests/nexus/test_dynamic_creation_of_user_handler_classes.py @@ -10,6 +10,8 @@ from temporalio.worker import Worker from tests.helpers.nexus import make_nexus_endpoint_name +pytestmark = pytest.mark.requires_local_server + @workflow.defn class MyWorkflow: diff --git a/tests/nexus/test_nexus_client_updates.py b/tests/nexus/test_nexus_client_updates.py index 302ede54e..c99822770 100644 --- a/tests/nexus/test_nexus_client_updates.py +++ b/tests/nexus/test_nexus_client_updates.py @@ -3,6 +3,7 @@ import uuid import nexusrpc +import pytest from nexusrpc.handler import StartOperationContext, service_handler, sync_operation import temporalio.nexus @@ -11,6 +12,8 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker +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 bd9063237..45dfcb8ad 100644 --- a/tests/nexus/test_nexus_worker_shutdown.py +++ b/tests/nexus/test_nexus_worker_shutdown.py @@ -23,6 +23,8 @@ make_nexus_endpoint_name, ) +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 e489ad8a7..e7906f7a5 100644 --- a/tests/nexus/test_signal_link_propagation_e2e.py +++ b/tests/nexus/test_signal_link_propagation_e2e.py @@ -56,6 +56,8 @@ workflow_event_link_event_type, ) +pytestmark = pytest.mark.requires_local_server + EventType = temporalio.api.enums.v1.EventType diff --git a/tests/nexus/test_standalone_operations.py b/tests/nexus/test_standalone_operations.py index 8193ba7ba..e0cda9475 100644 --- a/tests/nexus/test_standalone_operations.py +++ b/tests/nexus/test_standalone_operations.py @@ -71,6 +71,9 @@ # --------------------------------------------------------------------------- +pytestmark = pytest.mark.requires_local_server + + @dataclass class EchoInput: value: str diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index b6b1a92c7..0d1b7d63e 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -20,6 +20,8 @@ from tests.helpers import EventType, assert_event_subsequence, assert_eventually from tests.helpers.nexus import make_nexus_endpoint_name +pytestmark = pytest.mark.requires_local_server + @dataclass class Input: diff --git a/tests/nexus/test_use_existing_conflict_policy.py b/tests/nexus/test_use_existing_conflict_policy.py index e7cee9e1c..94a3821ba 100644 --- a/tests/nexus/test_use_existing_conflict_policy.py +++ b/tests/nexus/test_use_existing_conflict_policy.py @@ -4,6 +4,7 @@ import uuid from dataclasses import dataclass +import pytest from nexusrpc.handler import service_handler from temporalio import nexus, workflow @@ -13,6 +14,8 @@ from temporalio.worker import Worker from tests.helpers.nexus import make_nexus_endpoint_name +pytestmark = pytest.mark.requires_local_server + @dataclass class OpInput: diff --git a/tests/nexus/test_workflow_caller.py b/tests/nexus/test_workflow_caller.py index 328637525..f776e6e18 100644 --- a/tests/nexus/test_workflow_caller.py +++ b/tests/nexus/test_workflow_caller.py @@ -90,6 +90,9 @@ class OpDefinitionType(IntEnum): LONGHAND = 1 +pytestmark = pytest.mark.requires_local_server + + @dataclass class SyncResponse: op_definition_type: OpDefinitionType diff --git a/tests/nexus/test_workflow_caller_cancellation_types.py b/tests/nexus/test_workflow_caller_cancellation_types.py index bf33983a5..fa39009a5 100644 --- a/tests/nexus/test_workflow_caller_cancellation_types.py +++ b/tests/nexus/test_workflow_caller_cancellation_types.py @@ -25,6 +25,8 @@ from tests.helpers import LogCapturer, assert_event_subsequence, assert_eventually from tests.helpers.nexus import make_nexus_endpoint_name +pytestmark = pytest.mark.requires_local_server + @dataclass class TestContext: 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 4cdeeeb15..5a0970c95 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 @@ -30,6 +30,8 @@ has_event, ) +pytestmark = pytest.mark.requires_local_server + @dataclass class TestContext: diff --git a/tests/nexus/test_workflow_caller_error_chains.py b/tests/nexus/test_workflow_caller_error_chains.py index 9ff84f405..18868288a 100644 --- a/tests/nexus/test_workflow_caller_error_chains.py +++ b/tests/nexus/test_workflow_caller_error_chains.py @@ -25,6 +25,8 @@ from temporalio.worker import Worker from tests.helpers.nexus import make_nexus_endpoint_name +pytestmark = pytest.mark.requires_local_server + @dataclass class ExpectedError: diff --git a/tests/nexus/test_workflow_caller_errors.py b/tests/nexus/test_workflow_caller_errors.py index 9246b9fd7..eb85155fe 100644 --- a/tests/nexus/test_workflow_caller_errors.py +++ b/tests/nexus/test_workflow_caller_errors.py @@ -42,6 +42,8 @@ from tests.helpers import LogCapturer, assert_eq_eventually from tests.helpers.nexus import make_nexus_endpoint_name +pytestmark = pytest.mark.requires_local_server + operation_invocation_counts = Counter[str]() logger = getLogger(__name__) diff --git a/tests/nexus/test_workflow_run_operation.py b/tests/nexus/test_workflow_run_operation.py index 851f408ec..51032a23a 100644 --- a/tests/nexus/test_workflow_run_operation.py +++ b/tests/nexus/test_workflow_run_operation.py @@ -23,6 +23,8 @@ from temporalio.worker import Worker from tests.helpers.nexus import make_nexus_endpoint_name +pytestmark = pytest.mark.requires_local_server + @dataclass class Input: diff --git a/tests/test_activity.py b/tests/test_activity.py index 6efa0d644..1d9ecd7ab 100644 --- a/tests/test_activity.py +++ b/tests/test_activity.py @@ -202,6 +202,7 @@ async def count_activities(self, input: CountActivitiesInput): return await super().count_activities(input) +@pytest.mark.requires_local_server async def test_start_activity_calls_interceptor( client: Client, env: WorkflowEnvironment ): @@ -461,6 +462,7 @@ async def test_get_result(client: Client, env: WorkflowEnvironment): assert await result_via_execute_activity == 2 +@pytest.mark.requires_local_server async def test_start_activity_start_delay(client: Client, env: WorkflowEnvironment): if env.supports_time_skipping: pytest.skip( diff --git a/tests/test_client.py b/tests/test_client.py index e7fd7172a..c545769fb 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -657,21 +657,24 @@ async def test_list_workflows_and_fetch_history( ) expected_id_and_input.append((workflow_id, f'"user{i}"')) - # List them and get their history - actual_id_and_input = sorted( - [ - ( - hist.workflow_id, - hist.events[0] - .workflow_execution_started_event_attributes.input.payloads[0] - .data.decode(), - ) - async for hist in client.list_workflows( - f"WorkflowId = '{workflow_id}'" - ).map_histories() - ] - ) - assert actual_id_and_input == expected_id_and_input + # Visibility is eventually consistent, so wait for all runs before fetching + # their histories. + async def list_id_and_input() -> list[tuple[str, str]]: + return sorted( + [ + ( + hist.workflow_id, + hist.events[0] + .workflow_execution_started_event_attributes.input.payloads[0] + .data.decode(), + ) + async for hist in client.list_workflows( + f"WorkflowId = '{workflow_id}'" + ).map_histories() + ] + ) + + await assert_eq_eventually(expected_id_and_input, list_id_and_input) # Verify listing can limit results limited = [ @@ -833,8 +836,6 @@ async def test_schedule_basics( pytest.skip("Java test server doesn't support schedules") elif os.getenv("TEMPORAL_TEST_PROTO3"): pytest.skip("Older proto library cannot compare repeated fields") - await assert_no_schedules(client) - # Create a schedule with a lot of stuff schedule = Schedule( action=ScheduleActionStartWorkflow( @@ -1072,10 +1073,9 @@ async def list_ids() -> list[str]: assert list_descs[0].id in [f"{handle.id}-3", f"{handle.id}-4"] assert list_descs[1].id in [f"{handle.id}-3", f"{handle.id}-4"] - # Delete all of the schedules - for id in await list_ids(): + # Delete the schedules created by this test. + for id in expected_ids: await client.get_schedule_handle(id).delete() - await assert_no_schedules(client) async def test_schedule_calendar_spec_defaults( @@ -1083,8 +1083,6 @@ async def test_schedule_calendar_spec_defaults( ): if env.supports_time_skipping: pytest.skip("Java test server doesn't support schedules") - await assert_no_schedules(client) - handle = await client.create_schedule( f"schedule-{uuid.uuid4()}", Schedule( @@ -1113,7 +1111,6 @@ async def test_schedule_calendar_spec_defaults( assert time == desc.info.next_action_times[i - 1] + timedelta(days=1) await handle.delete() - await assert_no_schedules(client) async def test_schedule_trigger_immediately( @@ -1121,8 +1118,6 @@ async def test_schedule_trigger_immediately( ): if env.supports_time_skipping: pytest.skip("Java test server doesn't support schedules") - await assert_no_schedules(client) - # Create paused schedule that triggers immediately handle = await client.create_schedule( f"schedule-{uuid.uuid4()}", @@ -1154,7 +1149,6 @@ async def test_schedule_trigger_immediately( ) await handle.delete() - await assert_no_schedules(client) async def test_schedule_backfill( @@ -1162,8 +1156,6 @@ async def test_schedule_backfill( ): if env.supports_time_skipping: pytest.skip("Java test server doesn't support schedules") - await assert_no_schedules(client) - begin = datetime(year=2020, month=1, day=20, hour=5) # Create paused schedule that runs every minute and has two backfills @@ -1214,7 +1206,6 @@ async def test_schedule_backfill( ) finally: await handle.delete() - await assert_no_schedules(client) async def test_schedule_create_limited_actions_validation( @@ -1245,8 +1236,6 @@ async def test_schedule_workflow_search_attribute_update( ): if env.supports_time_skipping: pytest.skip("Java test server doesn't support schedules") - await assert_no_schedules(client) - # Put search attribute on server text_attr_key = SearchAttributeKey.for_text("python-test-schedule-text") untyped_keyword_key = SearchAttributeKey.for_keyword("python-test-schedule-keyword") @@ -1339,7 +1328,6 @@ def update_schedule_typed_attrs( assert desc.typed_search_attributes[text_attr_key] == "some-schedule-attr1" await handle.delete() - await assert_no_schedules(client) @pytest.mark.parametrize( @@ -1356,8 +1344,6 @@ async def test_schedule_search_attribute_update( ): if env.supports_time_skipping: pytest.skip("Java test server doesn't support schedules") - await assert_no_schedules(client) - # Put search attributes on server key_1 = SearchAttributeKey.for_text("python-test-schedule-sa-update-key-1") key_2 = SearchAttributeKey.for_keyword("python-test-schedule-sa-update-key-2") @@ -1475,15 +1461,6 @@ async def expectation() -> bool: raise ValueError(f"Invalid test case: {test_case}") await handle.delete() - await assert_no_schedules(client) - - -async def assert_no_schedules(client: Client) -> None: - # Listing appears eventually consistent - async def schedule_count() -> int: - return len([d async for d in await client.list_schedules()]) - - await assert_eq_eventually(0, schedule_count) async def test_build_id_interactions(client: Client, env: WorkflowEnvironment): diff --git a/tests/test_envconfig.py b/tests/test_envconfig.py index 0a6afd94b..9a6afc95e 100644 --- a/tests/test_envconfig.py +++ b/tests/test_envconfig.py @@ -13,6 +13,8 @@ from temporalio.testing import WorkflowEnvironment from tests import conftest +pytestmark = pytest.mark.requires_local_server + # A base TOML config with a default and a custom profile TOML_CONFIG_BASE = textwrap.dedent( """ @@ -153,22 +155,17 @@ def test_load_profile_from_data_env_overrides(): @pytest.mark.parametrize( - ("env_type", "flag", "expected"), + ("env_type", "expected"), [ - ("envconfig", None, True), - ("local", "true", True), - ("local", "false", False), - ("time-skipping", "true", False), + ("envconfig", True), + ("local", False), + ("time-skipping", False), ], ) def test_envconfig_server_selection( - monkeypatch: pytest.MonkeyPatch, env_type: str, - flag: str | None, expected: bool, ): - if flag is not None: - monkeypatch.setenv("TEMPORAL_TEST_ENV_CONFIG_SERVER", flag) assert conftest._uses_envconfig_server(env_type) is expected diff --git a/tests/test_serialization_context.py b/tests/test_serialization_context.py index 0fde2aa96..8d65d5f1f 100644 --- a/tests/test_serialization_context.py +++ b/tests/test_serialization_context.py @@ -217,19 +217,19 @@ async def test_payload_conversion_calls_follow_expected_sequence_and_contexts( workflow_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, ) ) child_workflow_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=f"{workflow_id}_child", ) ) activity_context = dataclasses.asdict( ActivitySerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, workflow_type=PayloadConversionWorkflow.__name__, activity_type=passthrough_activity.__name__, @@ -363,14 +363,14 @@ async def test_heartbeat_details_payload_conversion(client: Client): workflow_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, ) ) activity_context = dataclasses.asdict( ActivitySerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, workflow_type=HeartbeatDetailsSerializationContextTestWorkflow.__name__, activity_type=activity_with_heartbeat_details.__name__, @@ -455,13 +455,13 @@ async def test_local_activity_payload_conversion(client: Client): workflow_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, ) ) local_activity_context = dataclasses.asdict( ActivitySerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, workflow_type=LocalActivityWorkflow.__name__, activity_type=local_activity.__name__, @@ -572,11 +572,11 @@ async def test_async_activity_completion_payload_conversion( workflow_runner=UnsandboxedWorkflowRunner(), # so that we can use isinstance ): workflow_context = WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, ) activity_context = ActivitySerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, workflow_type=AsyncActivityCompletionSerializationContextTestWorkflow.__name__, activity_type=async_activity.__name__, @@ -649,7 +649,7 @@ def my_method(self) -> None: def test_subclassed_async_activity_handle(client: Client): activity_context = ActivitySerializationContext( - namespace="default", + namespace=client.namespace, workflow_id="workflow-id", workflow_type="workflow-type", activity_type="activity-type", @@ -742,7 +742,7 @@ async def test_signal_payload_conversion( workflow_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, ) ) @@ -811,7 +811,7 @@ async def test_query_payload_conversion( workflow_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, ) ) @@ -909,7 +909,7 @@ async def test_update_payload_conversion( workflow_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, ) ) @@ -1016,13 +1016,13 @@ async def test_external_workflow_signal_and_cancel_payload_conversion( signaler_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=signaler_workflow_id, ) ) target_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=target_workflow_id, ) ) @@ -1157,13 +1157,13 @@ async def test_failure_converter_with_context(client: Client): workflow_context = dataclasses.asdict( WorkflowSerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, ) ) activity_context = dataclasses.asdict( ActivitySerializationContext( - namespace="default", + namespace=client.namespace, workflow_id=workflow_id, workflow_type=FailureConverterTestWorkflow.__name__, activity_type=failing_activity.__name__, @@ -1731,6 +1731,7 @@ async def run(self, _data: str) -> None: ) +@pytest.mark.requires_local_server async def test_nexus_payload_codec_operations_lack_context( env: WorkflowEnvironment, ): diff --git a/tests/test_service.py b/tests/test_service.py index 0cf06fae0..954c87092 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -149,6 +149,7 @@ async def test_check_health(client: Client): assert err.value.status == temporalio.service.RPCStatusCode.NOT_FOUND +@pytest.mark.requires_local_server async def test_grpc_status(client: Client, env: WorkflowEnvironment): if env.supports_time_skipping: pytest.skip( diff --git a/tests/testing/test_workflow.py b/tests/testing/test_workflow.py index d5f2aae5c..b47d0ac05 100644 --- a/tests/testing/test_workflow.py +++ b/tests/testing/test_workflow.py @@ -252,6 +252,7 @@ def assert_proper_error(err: BaseException | None) -> None: assert_proper_error(err.value.cause) +@pytest.mark.requires_local_server async def test_search_attributes_on_dev_server( client: Client, env: WorkflowEnvironment ): diff --git a/tests/worker/test_interceptor.py b/tests/worker/test_interceptor.py index 431f8280d..4a1da399d 100644 --- a/tests/worker/test_interceptor.py +++ b/tests/worker/test_interceptor.py @@ -269,6 +269,7 @@ def update_validated_validator(self, param: str) -> None: raise ApplicationError("Invalid update") +@pytest.mark.requires_local_server async def test_worker_interceptor(client: Client, env: WorkflowEnvironment): # TODO(cretz): Fix if env.supports_time_skipping: diff --git a/tests/worker/test_replayer.py b/tests/worker/test_replayer.py index 22d771556..4f0a63fcf 100644 --- a/tests/worker/test_replayer.py +++ b/tests/worker/test_replayer.py @@ -311,6 +311,14 @@ async def test_replayer_multiple_from_client( ) await handle.result() + async def visible_run_ids() -> set[str]: + return { + workflow.run_id + async for workflow in client.list_workflows(f"WorkflowId = '{workflow_id}'") + } + + await assert_eq_eventually(set(expected_runs_and_non_det), visible_run_ids) + # Run replayer with list iterator mapped to histories and collect results async with Replayer(workflows=[SayHelloWorkflow]).workflow_replay_iterator( client.list_workflows(f"WorkflowId = '{workflow_id}'").map_histories() diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index 6f2667a80..b4f03d2cc 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -399,6 +399,7 @@ def my_signal(self, value: str) -> None: workflow.logger.info(f"Signal: {value}") +@pytest.mark.requires_local_server async def test_custom_slot_supplier(client: Client, env: WorkflowEnvironment): if env.supports_time_skipping: pytest.skip("Nexus tests don't work under Java test server") diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index b78e1e8c0..8073d4865 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -306,6 +306,7 @@ def get_history_info(self) -> HistoryInfo: ) +@pytest.mark.requires_local_server async def test_workflow_history_info( client: Client, env: WorkflowEnvironment, continue_as_new_suggest_history_count: int ): @@ -2016,6 +2017,7 @@ def do_search_attribute_update_typed(self) -> None: ) +@pytest.mark.requires_local_server async def test_workflow_search_attributes(client: Client, env_type: str): if env_type != "local": pytest.skip("Only testing search attributes on local which disables cache") @@ -2201,6 +2203,7 @@ async def run(self) -> None: # All we need to do is complete +@pytest.mark.requires_local_server async def test_workflow_no_initial_search_attributes(client: Client, env_type: str): if env_type != "local": pytest.skip("Only testing search attributes on local which disables cache") @@ -5861,6 +5864,7 @@ async def run(self) -> None: await asyncio.sleep(0.1) +@pytest.mark.requires_local_server async def test_workflow_replace_worker_client(client: Client, env: WorkflowEnvironment): if env.supports_time_skipping: pytest.skip("Only testing against two real servers") @@ -9050,6 +9054,7 @@ async def run(self, name: str) -> str: return f"Hello from child, {name}" +@pytest.mark.requires_local_server async def test_search_attribute_codec(client: Client, env_type: str): if env_type != "local": pytest.skip("Only testing search attributes on local which disables cache") From 6e74cf0916ce9b60cf3a546961f5bc965be2c140 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Tue, 28 Jul 2026 13:57:42 -0700 Subject: [PATCH 03/10] test: fix remaining Cloud test failures --- temporalio/testing/_workflow.py | 1 + tests/test_client.py | 9 +++++++-- tests/test_envconfig.py | 4 ++++ tests/worker/test_update_with_start.py | 10 ++++++---- tests/worker/test_worker.py | 1 - 5 files changed, 18 insertions(+), 7 deletions(-) diff --git a/temporalio/testing/_workflow.py b/temporalio/testing/_workflow.py index 3c829a132..3ab07e5aa 100644 --- a/temporalio/testing/_workflow.py +++ b/temporalio/testing/_workflow.py @@ -387,6 +387,7 @@ async def connect_client(self, **kwargs: Any) -> temporalio.client.Client: "api_key": config.api_key, "tls": config.tls, "rpc_metadata": config.rpc_metadata, + "runtime": config.runtime or temporalio.runtime.Runtime.default(), } connect_kwargs.update(kwargs) return await temporalio.client.Client.connect( diff --git a/tests/test_client.py b/tests/test_client.py index c545769fb..7ef0d6cad 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -829,6 +829,7 @@ def test_history_from_json(): ) +@pytest.mark.requires_local_server async def test_schedule_basics( client: Client, worker: ExternalWorker, env: WorkflowEnvironment ): @@ -1231,6 +1232,7 @@ async def test_schedule_create_limited_actions_validation( assert "are remaining actions set" in str(err.value) +@pytest.mark.requires_local_server async def test_schedule_workflow_search_attribute_update( client: Client, env: WorkflowEnvironment ): @@ -1339,6 +1341,7 @@ def update_schedule_typed_attrs( "partial-new-values-overwrites-and-drops", ], ) +@pytest.mark.requires_local_server async def test_schedule_search_attribute_update( client: Client, env: WorkflowEnvironment, test_case: str ): @@ -1556,12 +1559,14 @@ async def get_schedule_result() -> tuple[int, str | None]: result = await workflow_handle.result() return length, result - assert await get_schedule_result() == (1, "My First Result") + expected_first_result: tuple[int, str | None] = (1, "My First Result") + await assert_eq_eventually(expected_first_result, get_schedule_result) await handle.trigger() - assert await get_schedule_result() == ( + expected_second_result: tuple[int, str | None] = ( 2, "From last completion: My First Result", ) + await assert_eq_eventually(expected_second_result, get_schedule_result) await handle.delete() diff --git a/tests/test_envconfig.py b/tests/test_envconfig.py index 9a6afc95e..0b44db731 100644 --- a/tests/test_envconfig.py +++ b/tests/test_envconfig.py @@ -6,6 +6,7 @@ import pytest +import temporalio.runtime import temporalio.service from temporalio.client import Client from temporalio.envconfig import ClientConfig, ClientConfigProfile, ClientConfigTLS @@ -200,6 +201,7 @@ async def test_envconfig_workflow_environment_uses_client_connect_config( async def test_workflow_environment_connect_client_inherits_connection_options( monkeypatch: pytest.MonkeyPatch, ): + runtime = temporalio.runtime.Runtime.default() source_client = SimpleNamespace( namespace="env-namespace", service_client=SimpleNamespace( @@ -208,6 +210,7 @@ async def test_workflow_environment_connect_client_inherits_connection_options( api_key="env-api-key", tls=False, rpc_metadata={"test-header": "env-value"}, + runtime=runtime, ) ), ) @@ -223,6 +226,7 @@ async def test_workflow_environment_connect_client_inherits_connection_options( api_key="env-api-key", tls=False, rpc_metadata={"test-header": "env-value"}, + runtime=runtime, lazy=True, ) diff --git a/tests/worker/test_update_with_start.py b/tests/worker/test_update_with_start.py index 2ceb5e91b..0b3368725 100644 --- a/tests/worker/test_update_with_start.py +++ b/tests/worker/test_update_with_start.py @@ -192,6 +192,7 @@ async def _do_test( id_conflict_policy: WorkflowIDConflictPolicy, expect_error_when_workflow_exists: ExpectErrorWhenWorkflowExists, ): + workflow_id = f"{workflow_id}-{uuid.uuid4()}" await self._do_execute_update_test( client, workflow_id + "-execute-update", @@ -336,6 +337,7 @@ async def test_update_with_start_sets_first_execution_run_id( WorkflowForUpdateWithStartTest, activities=[activity_called_by_update], ) as worker: + workflow_id_prefix = f"wid-{uuid.uuid4()}" def make_start_op(workflow_id: str): return WithStartWorkflowOperation( @@ -348,7 +350,7 @@ def make_start_op(workflow_id: str): # conflict policy is FAIL # First UWS succeeds and sets the first execution run ID - start_op_1 = make_start_op("wid-1") + start_op_1 = make_start_op(f"{workflow_id_prefix}-1") update_handle_1 = await client.start_update_with_start_workflow( WorkflowForUpdateWithStartTest.my_non_blocking_update, "1", @@ -360,7 +362,7 @@ def make_start_op(workflow_id: str): # Second UWS start fails because the workflow already exists # first execution run ID is not set on the second UWS handle - start_op_2 = make_start_op("wid-1") + start_op_2 = make_start_op(f"{workflow_id_prefix}-1") for aw in [ client.start_update_with_start_workflow( @@ -375,7 +377,7 @@ def make_start_op(workflow_id: str): await aw # Third UWS start succeeds, but the update fails after acceptance - start_op_3 = make_start_op("wid-2") + start_op_3 = make_start_op(f"{workflow_id_prefix}-2") update_handle_3 = await client.start_update_with_start_workflow( WorkflowForUpdateWithStartTest.my_non_blocking_update, "fail-after-acceptance", @@ -394,7 +396,7 @@ def make_start_op(workflow_id: str): assert await wf_handle_3.result() == "workflow-result-0" # Fourth UWS is same as third, but we use execute_update instead of start_update. - start_op_4 = make_start_op("wid-3") + start_op_4 = make_start_op(f"{workflow_id_prefix}-3") with pytest.raises(WorkflowUpdateFailedError): await client.execute_update_with_start_workflow( WorkflowForUpdateWithStartTest.my_non_blocking_update, diff --git a/tests/worker/test_worker.py b/tests/worker/test_worker.py index b4f03d2cc..b6ce37c8b 100644 --- a/tests/worker/test_worker.py +++ b/tests/worker/test_worker.py @@ -1476,7 +1476,6 @@ async def test_activity_client_updates_when_worker_client_changes( # Must use the same runtime client2 = await env.connect_client( data_converter=client.data_converter, - runtime=client.service_client.config.runtime, ) captured_clients: list[Client] = [] From 1faa9afb6e92f285b10e001bc9d2944b18b9f946 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Tue, 28 Jul 2026 14:18:51 -0700 Subject: [PATCH 04/10] test: reenable Cloud start-delay coverage --- tests/test_activity.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/test_activity.py b/tests/test_activity.py index 1d9ecd7ab..6efa0d644 100644 --- a/tests/test_activity.py +++ b/tests/test_activity.py @@ -202,7 +202,6 @@ async def count_activities(self, input: CountActivitiesInput): return await super().count_activities(input) -@pytest.mark.requires_local_server async def test_start_activity_calls_interceptor( client: Client, env: WorkflowEnvironment ): @@ -462,7 +461,6 @@ async def test_get_result(client: Client, env: WorkflowEnvironment): assert await result_via_execute_activity == 2 -@pytest.mark.requires_local_server async def test_start_activity_start_delay(client: Client, env: WorkflowEnvironment): if env.supports_time_skipping: pytest.skip( From 386f944041b43ffd591c22275b11cb95891beac3 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Tue, 28 Jul 2026 14:34:48 -0700 Subject: [PATCH 05/10] test: scope replayer history test to local server --- tests/worker/test_replayer.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/worker/test_replayer.py b/tests/worker/test_replayer.py index 4f0a63fcf..01d20de6c 100644 --- a/tests/worker/test_replayer.py +++ b/tests/worker/test_replayer.py @@ -283,6 +283,7 @@ async def test_replayer_workflow_not_registered(client: Client) -> None: assert "SayHelloWorkflow is not registered" in str(err.value) +@pytest.mark.requires_local_server async def test_replayer_multiple_from_client( client: Client, env: WorkflowEnvironment ) -> None: From fa5d89ea964395fa8391abcba800a3f055530067 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Tue, 28 Jul 2026 14:35:22 -0700 Subject: [PATCH 06/10] test: wait for schedule action cleanup --- tests/test_client.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/test_client.py b/tests/test_client.py index 7ef0d6cad..e96c06299 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1561,6 +1561,11 @@ async def get_schedule_result() -> tuple[int, str | None]: expected_first_result: tuple[int, str | None] = (1, "My First Result") await assert_eq_eventually(expected_first_result, get_schedule_result) + + async def no_running_actions() -> bool: + return not (await handle.describe()).info.running_actions + + await assert_eq_eventually(True, no_running_actions) await handle.trigger() expected_second_result: tuple[int, str | None] = ( 2, From 1b1c18ce6b05a61cfd34ebe98035fa701c11b21c Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Tue, 28 Jul 2026 15:42:16 -0700 Subject: [PATCH 07/10] test: synchronize unfinished handler termination --- tests/worker/test_workflow.py | 99 ++++++++++++++++++++++------------- 1 file changed, 63 insertions(+), 36 deletions(-) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 8073d4865..05d341680 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -6275,6 +6275,9 @@ def _unfinished_handler_warning_cls(self) -> type: class UnfinishedHandlersOnWorkflowTerminationWorkflow: def __init__(self) -> None: self.handlers_may_finish = False + self.handler_started = False + self.ready_for_termination = False + self.termination_requested = False @workflow.run async def run( @@ -6291,18 +6294,23 @@ async def run( "-wait-all-handlers-finish-", "-no-wait-all-handlers-finish-" ], ) -> NoReturn: + if workflow_termination_type == "-fail-post-continue-as-new-run-": + raise ApplicationError("Deliberately failing post-ContinueAsNew run") + if handler_registration == "-late-registered-": if handler_dynamism == "-dynamic-": async def my_late_registered_dynamic_update( _name: str, _args: Sequence[RawValue] ) -> str: + self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) return "my-late-registered-dynamic-update-result" async def my_late_registered_dynamic_signal( _name: str, _args: Sequence[RawValue] ) -> None: + self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) workflow.set_dynamic_update_handler(my_late_registered_dynamic_update) @@ -6310,10 +6318,12 @@ async def my_late_registered_dynamic_signal( else: async def my_late_registered_update() -> str: + self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) return "my-late-registered-update-result" async def my_late_registered_signal() -> None: + self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) workflow.set_update_handler( @@ -6323,16 +6333,17 @@ async def my_late_registered_signal() -> None: "my_late_registered_signal", my_late_registered_signal ) + await workflow.wait_condition(lambda: self.handler_started) if handler_waiting == "-wait-all-handlers-finish-": self.handlers_may_finish = True await workflow.wait_condition(workflow.all_handlers_finished) + self.ready_for_termination = True + await workflow.wait_condition(lambda: self.termination_requested) if workflow_termination_type == "-failure-": raise ApplicationError( "Deliberately failing workflow with an unfinished handler" ) - elif workflow_termination_type == "-fail-post-continue-as-new-run-": - raise ApplicationError("Deliberately failing post-ContinueAsNew run") - elif workflow_termination_type == "-continue-as-new-": + if workflow_termination_type == "-continue-as-new-": # Fail next run so that test terminates workflow.continue_as_new( args=[ @@ -6348,22 +6359,34 @@ async def my_late_registered_signal() -> None: @workflow.update async def my_update(self) -> str: + self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) return "update-result" @workflow.signal async def my_signal(self) -> None: + self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) @workflow.update(dynamic=True) async def my_dynamic_update(self, _name: str, _args: Sequence[RawValue]) -> str: + self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) return "my-dynamic-update-result" @workflow.signal(dynamic=True) async def my_dynamic_signal(self, _name: str, _args: Sequence[RawValue]) -> None: + self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) + @workflow.signal + def request_termination(self) -> None: + self.termination_requested = True + + @workflow.query + def is_ready_for_termination(self) -> bool: + return self.ready_for_termination + @pytest.mark.parametrize("handler_type", ["-signal-", "-update-"]) @pytest.mark.parametrize( @@ -6430,9 +6453,6 @@ async def _run_workflow_and_get_warning(self) -> bool: update_id = "update-id" task_queue = "tq" - # We require a startWorkflow, an update, and maybe a cancellation request, to be delivered - # in the same WFT. To do this we start the worker after they've all been accepted by the - # server. handle = await self.client.start_workflow( UnfinishedHandlersOnWorkflowTerminationWorkflow.run, args=[ @@ -6444,36 +6464,6 @@ async def _run_workflow_and_get_warning(self) -> bool: id=workflow_id, task_queue=task_queue, ) - if self.workflow_termination_type == "-cancellation-": - await handle.cancel() - - if self.handler_type == "-update-": - update_method = ( - "__does_not_exist__" - if self.handler_dynamism == "-dynamic-" - else "my_late_registered_update" - if self.handler_registration == "-late-registered-" - else UnfinishedHandlersOnWorkflowTerminationWorkflow.my_update - ) - update_task = asyncio.create_task( - handle.execute_update( - update_method, # type: ignore - id=update_id, - ) - ) - await assert_eq_eventually( - True, - lambda: workflow_update_exists(self.client, workflow_id, update_id), - ) - else: - signal_method = ( - "__does_not_exist__" - if self.handler_dynamism == "-dynamic-" - else "my_late_registered_signal" - if self.handler_registration == "-late-registered-" - else UnfinishedHandlersOnWorkflowTerminationWorkflow.my_signal - ) - await handle.signal(signal_method) # type: ignore async with new_worker( self.client, @@ -6481,6 +6471,43 @@ async def _run_workflow_and_get_warning(self) -> bool: task_queue=task_queue, ): with pytest.WarningsRecorder() as warnings: + if self.handler_type == "-update-": + update_method = ( + "__does_not_exist__" + if self.handler_dynamism == "-dynamic-" + else "my_late_registered_update" + if self.handler_registration == "-late-registered-" + else UnfinishedHandlersOnWorkflowTerminationWorkflow.my_update + ) + update_task = asyncio.create_task( + handle.execute_update( + update_method, # type: ignore + id=update_id, + ) + ) + else: + signal_method = ( + "__does_not_exist__" + if self.handler_dynamism == "-dynamic-" + else "my_late_registered_signal" + if self.handler_registration == "-late-registered-" + else UnfinishedHandlersOnWorkflowTerminationWorkflow.my_signal + ) + await handle.signal(signal_method) # type: ignore + + async def is_ready_for_termination() -> bool: + return await handle.query( + UnfinishedHandlersOnWorkflowTerminationWorkflow.is_ready_for_termination + ) + + await assert_eq_eventually(True, is_ready_for_termination) + if self.workflow_termination_type == "-cancellation-": + await handle.cancel() + else: + await handle.signal( + UnfinishedHandlersOnWorkflowTerminationWorkflow.request_termination + ) + if self.handler_type == "-update-": assert update_task # type: ignore[reportUnboundVariable] if self.handler_waiting == "-wait-all-handlers-finish-": From 81294a3eb01ef6b4713928eeecbc59ec15dee8e7 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Tue, 28 Jul 2026 16:01:34 -0700 Subject: [PATCH 08/10] test: buffer schedule completion result action --- tests/test_client.py | 36 ++++++++++++++++++++++++++---------- 1 file changed, 26 insertions(+), 10 deletions(-) diff --git a/tests/test_client.py b/tests/test_client.py index e96c06299..b895c8243 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1516,13 +1516,20 @@ async def test_build_id_interactions(client: Client, env: WorkflowEnvironment): @workflow.defn class LastCompletionResultWorkflow: + def __init__(self) -> None: + self.complete = False + @workflow.run async def run(self) -> str: last_result = workflow.get_last_completion_result(type_hint=str) if last_result is not None: return "From last completion: " + last_result - else: - return "My First Result" + await workflow.wait_condition(lambda: self.complete) + return "My First Result" + + @workflow.signal + def allow_completion(self) -> None: + self.complete = True async def test_schedule_last_completion_result( @@ -1541,10 +1548,27 @@ async def test_schedule_last_completion_result( task_queue=worker.task_queue, ), spec=ScheduleSpec(), + policy=SchedulePolicy(overlap=ScheduleOverlapPolicy.BUFFER_ONE), ), ) await handle.trigger() + async def has_running_action() -> bool: + return bool((await handle.describe()).info.running_actions) + + await assert_eq_eventually(True, has_running_action) + first_action = cast( + ScheduleActionExecutionStartWorkflow, + (await handle.describe()).info.running_actions[0], + ) + # Buffer the second action so the scheduler starts it only after recording + # the first run's completion result. + await handle.trigger() + await client.get_workflow_handle( + first_action.workflow_id, + run_id=first_action.first_execution_run_id, + ).signal(LastCompletionResultWorkflow.allow_completion) + async def get_schedule_result() -> tuple[int, str | None]: desc = await handle.describe() length = len(desc.info.recent_actions) @@ -1559,14 +1583,6 @@ async def get_schedule_result() -> tuple[int, str | None]: result = await workflow_handle.result() return length, result - expected_first_result: tuple[int, str | None] = (1, "My First Result") - await assert_eq_eventually(expected_first_result, get_schedule_result) - - async def no_running_actions() -> bool: - return not (await handle.describe()).info.running_actions - - await assert_eq_eventually(True, no_running_actions) - await handle.trigger() expected_second_result: tuple[int, str | None] = ( 2, "From last completion: My First Result", From a4a3c48dd548ecadd91119b902164ed72d007219 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 29 Jul 2026 08:34:18 -0700 Subject: [PATCH 09/10] test: scope schedule completion result test to local server --- tests/test_client.py | 34 ++++++++-------------------------- 1 file changed, 8 insertions(+), 26 deletions(-) diff --git a/tests/test_client.py b/tests/test_client.py index b895c8243..15324cf78 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1516,22 +1516,18 @@ async def test_build_id_interactions(client: Client, env: WorkflowEnvironment): @workflow.defn class LastCompletionResultWorkflow: - def __init__(self) -> None: - self.complete = False - @workflow.run async def run(self) -> str: last_result = workflow.get_last_completion_result(type_hint=str) if last_result is not None: return "From last completion: " + last_result - await workflow.wait_condition(lambda: self.complete) - return "My First Result" - - @workflow.signal - def allow_completion(self) -> None: - self.complete = True + else: + return "My First Result" +# Cloud does not reliably provide a prior completion result to manually +# triggered schedule actions. +@pytest.mark.requires_local_server async def test_schedule_last_completion_result( client: Client, env: WorkflowEnvironment ): @@ -1548,27 +1544,10 @@ async def test_schedule_last_completion_result( task_queue=worker.task_queue, ), spec=ScheduleSpec(), - policy=SchedulePolicy(overlap=ScheduleOverlapPolicy.BUFFER_ONE), ), ) await handle.trigger() - async def has_running_action() -> bool: - return bool((await handle.describe()).info.running_actions) - - await assert_eq_eventually(True, has_running_action) - first_action = cast( - ScheduleActionExecutionStartWorkflow, - (await handle.describe()).info.running_actions[0], - ) - # Buffer the second action so the scheduler starts it only after recording - # the first run's completion result. - await handle.trigger() - await client.get_workflow_handle( - first_action.workflow_id, - run_id=first_action.first_execution_run_id, - ).signal(LastCompletionResultWorkflow.allow_completion) - async def get_schedule_result() -> tuple[int, str | None]: desc = await handle.describe() length = len(desc.info.recent_actions) @@ -1583,6 +1562,9 @@ async def get_schedule_result() -> tuple[int, str | None]: result = await workflow_handle.result() return length, result + expected_first_result: tuple[int, str | None] = (1, "My First Result") + await assert_eq_eventually(expected_first_result, get_schedule_result) + await handle.trigger() expected_second_result: tuple[int, str | None] = ( 2, "From last completion: My First Result", From cff87a10549af4c8ab56dcca60f633b006f1205a Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 29 Jul 2026 11:08:17 -0700 Subject: [PATCH 10/10] test: scope unfinished handler test to local server --- tests/worker/test_workflow.py | 102 +++++++++++++--------------------- 1 file changed, 39 insertions(+), 63 deletions(-) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 705d2999d..5bb2b13e1 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -6619,9 +6619,6 @@ def _unfinished_handler_warning_cls(self) -> type: class UnfinishedHandlersOnWorkflowTerminationWorkflow: def __init__(self) -> None: self.handlers_may_finish = False - self.handler_started = False - self.ready_for_termination = False - self.termination_requested = False @workflow.run async def run( @@ -6638,23 +6635,18 @@ async def run( "-wait-all-handlers-finish-", "-no-wait-all-handlers-finish-" ], ) -> NoReturn: - if workflow_termination_type == "-fail-post-continue-as-new-run-": - raise ApplicationError("Deliberately failing post-ContinueAsNew run") - if handler_registration == "-late-registered-": if handler_dynamism == "-dynamic-": async def my_late_registered_dynamic_update( _name: str, _args: Sequence[RawValue] ) -> str: - self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) return "my-late-registered-dynamic-update-result" async def my_late_registered_dynamic_signal( _name: str, _args: Sequence[RawValue] ) -> None: - self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) workflow.set_dynamic_update_handler(my_late_registered_dynamic_update) @@ -6662,12 +6654,10 @@ async def my_late_registered_dynamic_signal( else: async def my_late_registered_update() -> str: - self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) return "my-late-registered-update-result" async def my_late_registered_signal() -> None: - self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) workflow.set_update_handler( @@ -6677,17 +6667,16 @@ async def my_late_registered_signal() -> None: "my_late_registered_signal", my_late_registered_signal ) - await workflow.wait_condition(lambda: self.handler_started) if handler_waiting == "-wait-all-handlers-finish-": self.handlers_may_finish = True await workflow.wait_condition(workflow.all_handlers_finished) - self.ready_for_termination = True - await workflow.wait_condition(lambda: self.termination_requested) if workflow_termination_type == "-failure-": raise ApplicationError( "Deliberately failing workflow with an unfinished handler" ) - if workflow_termination_type == "-continue-as-new-": + elif workflow_termination_type == "-fail-post-continue-as-new-run-": + raise ApplicationError("Deliberately failing post-ContinueAsNew run") + elif workflow_termination_type == "-continue-as-new-": # Fail next run so that test terminates workflow.continue_as_new( args=[ @@ -6703,35 +6692,26 @@ async def my_late_registered_signal() -> None: @workflow.update async def my_update(self) -> str: - self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) return "update-result" @workflow.signal async def my_signal(self) -> None: - self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) @workflow.update(dynamic=True) async def my_dynamic_update(self, _name: str, _args: Sequence[RawValue]) -> str: - self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) return "my-dynamic-update-result" @workflow.signal(dynamic=True) async def my_dynamic_signal(self, _name: str, _args: Sequence[RawValue]) -> None: - self.handler_started = True await workflow.wait_condition(lambda: self.handlers_may_finish) - @workflow.signal - def request_termination(self) -> None: - self.termination_requested = True - - @workflow.query - def is_ready_for_termination(self) -> bool: - return self.ready_for_termination - +# Cloud cancels in-flight dynamic handlers before SDK teardown can emit the +# unfinished-handler warning this test asserts. +@pytest.mark.requires_local_server @pytest.mark.parametrize("handler_type", ["-signal-", "-update-"]) @pytest.mark.parametrize( "handler_registration", ["-late-registered-", "-not-late-registered-"] @@ -6797,6 +6777,9 @@ async def _run_workflow_and_get_warning(self) -> bool: update_id = "update-id" task_queue = "tq" + # We require a startWorkflow, an update, and maybe a cancellation request, to be delivered + # in the same WFT. To do this we start the worker after they've all been accepted by the + # server. handle = await self.client.start_workflow( UnfinishedHandlersOnWorkflowTerminationWorkflow.run, args=[ @@ -6808,6 +6791,36 @@ async def _run_workflow_and_get_warning(self) -> bool: id=workflow_id, task_queue=task_queue, ) + if self.workflow_termination_type == "-cancellation-": + await handle.cancel() + + if self.handler_type == "-update-": + update_method = ( + "__does_not_exist__" + if self.handler_dynamism == "-dynamic-" + else "my_late_registered_update" + if self.handler_registration == "-late-registered-" + else UnfinishedHandlersOnWorkflowTerminationWorkflow.my_update + ) + update_task = asyncio.create_task( + handle.execute_update( + update_method, # type: ignore + id=update_id, + ) + ) + await assert_eq_eventually( + True, + lambda: workflow_update_exists(self.client, workflow_id, update_id), + ) + else: + signal_method = ( + "__does_not_exist__" + if self.handler_dynamism == "-dynamic-" + else "my_late_registered_signal" + if self.handler_registration == "-late-registered-" + else UnfinishedHandlersOnWorkflowTerminationWorkflow.my_signal + ) + await handle.signal(signal_method) # type: ignore async with new_worker( self.client, @@ -6815,43 +6828,6 @@ async def _run_workflow_and_get_warning(self) -> bool: task_queue=task_queue, ): with pytest.WarningsRecorder() as warnings: - if self.handler_type == "-update-": - update_method = ( - "__does_not_exist__" - if self.handler_dynamism == "-dynamic-" - else "my_late_registered_update" - if self.handler_registration == "-late-registered-" - else UnfinishedHandlersOnWorkflowTerminationWorkflow.my_update - ) - update_task = asyncio.create_task( - handle.execute_update( - update_method, # type: ignore - id=update_id, - ) - ) - else: - signal_method = ( - "__does_not_exist__" - if self.handler_dynamism == "-dynamic-" - else "my_late_registered_signal" - if self.handler_registration == "-late-registered-" - else UnfinishedHandlersOnWorkflowTerminationWorkflow.my_signal - ) - await handle.signal(signal_method) # type: ignore - - async def is_ready_for_termination() -> bool: - return await handle.query( - UnfinishedHandlersOnWorkflowTerminationWorkflow.is_ready_for_termination - ) - - await assert_eq_eventually(True, is_ready_for_termination) - if self.workflow_termination_type == "-cancellation-": - await handle.cancel() - else: - await handle.signal( - UnfinishedHandlersOnWorkflowTerminationWorkflow.request_termination - ) - if self.handler_type == "-update-": assert update_task # type: ignore[reportUnboundVariable] if self.handler_waiting == "-wait-all-handlers-finish-":