test: run integration suite against Cloud

.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,19 @@ 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 --workflow-environment envconfig --junit-xml=junit-xml/cloud.xml
+        timeout-minutes: 15
         env:
+          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()

temporalio/testing/_workflow.py

@@ -373,6 +373,27 @@ 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,
+            "runtime": config.runtime or temporalio.runtime.Runtime.default(),
+        }
+        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

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,43 @@ 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:
+    return env_type == "envconfig"
+
+
+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 +134,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",

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(
@@ -112,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
@@ -138,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]
 
@@ -162,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)
@@ -185,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]
 
@@ -214,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
@@ -244,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]
 
@@ -337,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(
@@ -359,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"
 
 
@@ -386,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.
@@ -469,9 +484,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(

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,

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:

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,

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
 ):

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(

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,

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:
@@ -2722,8 +2723,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,
     )
 

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:

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:
@@ -48,9 +51,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,
     )

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:

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
 
 

tests/nexus/test_standalone_operations.py

@@ -71,6 +71,9 @@
 # ---------------------------------------------------------------------------
 
 
+pytestmark = pytest.mark.requires_local_server
+
+
 @dataclass
 class EchoInput:
     value: str