new test helper

Author: yuandrew

tests/common/mod.rs

@@ -39,8 +39,8 @@ use temporal_sdk::{
     },
 };
 use temporal_sdk_core::{
-    ClientOptions, ClientOptionsBuilder, CoreRuntime, RuntimeOptionsBuilder, WorkerConfigBuilder,
-    init_replay_worker, init_worker,
+    ClientOptions, ClientOptionsBuilder, CoreRuntime, RuntimeOptions, RuntimeOptionsBuilder,
+    WorkerConfigBuilder, init_replay_worker, init_worker,
     replay::{HistoryForReplay, ReplayWorkerInput},
     telemetry::{build_otlp_metric_exporter, start_prometheus_metric_exporter},
 };
@@ -811,6 +811,13 @@ pub(crate) fn get_integ_telem_options() -> TelemetryOptions {
     .unwrap()
 }
 
+pub(crate) fn get_integ_runtime_options(telemopts: TelemetryOptions) -> RuntimeOptions {
+    RuntimeOptionsBuilder::default()
+        .telemetry_options(telemopts)
+        .build()
+        .unwrap()
+}
+
 #[async_trait::async_trait(?Send)]
 pub(crate) trait WorkflowHandleExt {
     async fn fetch_history_and_replay(

tests/global_metric_tests.rs

@@ -2,11 +2,12 @@
 #[allow(dead_code)]
 mod common;
 
+use crate::common::get_integ_runtime_options;
 use common::CoreWfStarter;
 use parking_lot::Mutex;
 use std::{sync::Arc, time::Duration};
 use temporal_sdk_core::{
-    CoreRuntime, RuntimeOptionsBuilder,
+    CoreRuntime,
     telemetry::{build_otlp_metric_exporter, construct_filter_string, telemetry_init_global},
 };
 use temporal_sdk_core_api::telemetry::{
@@ -79,12 +80,8 @@ async fn otel_errors_logged_as_errors() {
         })
         .build()
         .unwrap();
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
 
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let mut starter = CoreWfStarter::new_with_runtime("otel_errors_logged_as_errors", rt);
     let _worker = starter.get_worker().await;
 

tests/heavy_tests.rs

@@ -2,6 +2,7 @@
 #[allow(dead_code)]
 mod common;
 
+use crate::common::get_integ_runtime_options;
 use common::{
     CoreWfStarter, init_integ_telem, prom_metrics, rand_6_chars, workflows::la_problem_workflow,
 };
@@ -19,9 +20,7 @@ use std::{
 };
 use temporal_client::{GetWorkflowResultOpts, WfClientExt, WorkflowClientTrait, WorkflowOptions};
 use temporal_sdk::{ActContext, ActivityOptions, WfContext, WorkflowResult};
-use temporal_sdk_core::{
-    CoreRuntime, ResourceBasedTuner, ResourceSlotOptions, RuntimeOptionsBuilder,
-};
+use temporal_sdk_core::{CoreRuntime, ResourceBasedTuner, ResourceSlotOptions};
 use temporal_sdk_core_api::worker::PollerBehavior;
 use temporal_sdk_core_protos::{
     coresdk::{AsJsonPayloadExt, workflow_commands::ActivityCancellationType},
@@ -196,11 +195,7 @@ async fn workflow_load() {
     // cause us to encounter the tracing span drop bug
     telemopts.logging = None;
     init_integ_telem();
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let mut starter = CoreWfStarter::new_with_runtime("workflow_load", rt);
     starter
         .worker_config

tests/integ_tests/metrics_tests.rs

@@ -1,3 +1,4 @@
+use crate::common::get_integ_runtime_options;
 use crate::{
     common::{
         ANY_PORT, CoreWfStarter, NAMESPACE, OTEL_URL_ENV_VAR, PROMETHEUS_QUERY_API,
@@ -22,8 +23,7 @@ use temporal_sdk::{
     NexusOperationOptions, WfContext,
 };
 use temporal_sdk_core::{
-    CoreRuntime, FixedSizeSlotSupplier, RuntimeOptionsBuilder, TokioRuntimeBuilder, TunerBuilder,
-    init_worker,
+    CoreRuntime, FixedSizeSlotSupplier, TokioRuntimeBuilder, TunerBuilder, init_worker,
     telemetry::{WORKFLOW_TASK_EXECUTION_LATENCY_HISTOGRAM_NAME, build_otlp_metric_exporter},
 };
 use temporal_sdk_core_api::{
@@ -98,11 +98,7 @@ async fn prometheus_metrics_exported(
         });
     }
     let (telemopts, addr, _aborter) = prom_metrics(Some(opts_builder.build().unwrap()));
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let opts = get_integ_server_options();
     let mut raw_client = opts
         .connect_no_namespace(rt.telemetry().get_temporal_metric_meter())
@@ -153,11 +149,7 @@ async fn prometheus_metrics_exported(
 async fn one_slot_worker_reports_available_slot() {
     let (telemopts, addr, _aborter) = prom_metrics(None);
     let tq = "one_slot_worker_tq";
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
 
     let worker_cfg = WorkerConfigBuilder::default()
         .namespace(NAMESPACE)
@@ -410,11 +402,7 @@ async fn query_of_closed_workflow_doesnt_tick_terminal_metric(
     completion: workflow_command::Variant,
 ) {
     let (telemopts, addr, _aborter) = prom_metrics(None);
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let mut starter =
         CoreWfStarter::new_with_runtime("query_of_closed_workflow_doesnt_tick_terminal_metric", rt);
     // Disable cache to ensure replay happens completely
@@ -536,11 +524,11 @@ async fn query_of_closed_workflow_doesnt_tick_terminal_metric(
 
 #[test]
 fn runtime_new() {
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(get_integ_telem_options())
-        .build()
-        .unwrap();
-    let mut rt = CoreRuntime::new(runtimeopts, TokioRuntimeBuilder::default()).unwrap();
+    let mut rt = CoreRuntime::new(
+        get_integ_runtime_options(get_integ_telem_options()),
+        TokioRuntimeBuilder::default(),
+    )
+    .unwrap();
     let handle = rt.tokio_handle();
     let _rt = handle.enter();
     let (telemopts, addr, _aborter) = prom_metrics(None);
@@ -586,11 +574,7 @@ async fn latency_metrics(
             .build()
             .unwrap(),
     ));
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let mut starter = CoreWfStarter::new_with_runtime("latency_metrics", rt);
     let worker = starter.get_worker().await;
     starter.start_wf().await;
@@ -644,11 +628,7 @@ async fn latency_metrics(
 #[tokio::test]
 async fn request_fail_codes() {
     let (telemopts, addr, _aborter) = prom_metrics(None);
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let opts = get_integ_server_options();
     let mut client = opts
         .connect(NAMESPACE, rt.telemetry().get_temporal_metric_meter())
@@ -691,11 +671,8 @@ async fn request_fail_codes_otel() {
     let mut telemopts = TelemetryOptionsBuilder::default();
     let exporter = Arc::new(exporter);
     telemopts.metrics(exporter as Arc<dyn CoreMeter>);
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts.build().unwrap())
-        .build()
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts.build().unwrap()))
         .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
     let opts = get_integ_server_options();
     let mut client = opts
         .connect(NAMESPACE, rt.telemetry().get_temporal_metric_meter())
@@ -745,11 +722,7 @@ async fn docker_metrics_with_prometheus(
         .metric_prefix(test_uid.clone())
         .build()
         .unwrap();
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let test_name = "docker_metrics_with_prometheus";
     let mut starter = CoreWfStarter::new_with_runtime(test_name, rt);
     let worker = starter.get_worker().await;
@@ -803,11 +776,7 @@ async fn docker_metrics_with_prometheus(
 #[tokio::test]
 async fn activity_metrics() {
     let (telemopts, addr, _aborter) = prom_metrics(None);
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let wf_name = "activity_metrics";
     let mut starter = CoreWfStarter::new_with_runtime(wf_name, rt);
     starter
@@ -941,11 +910,7 @@ async fn activity_metrics() {
 #[tokio::test]
 async fn nexus_metrics() {
     let (telemopts, addr, _aborter) = prom_metrics(None);
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let wf_name = "nexus_metrics";
     let mut starter = CoreWfStarter::new_with_runtime(wf_name, rt);
     starter.worker_config.no_remote_activities(true);
@@ -1122,11 +1087,7 @@ async fn nexus_metrics() {
 #[tokio::test]
 async fn evict_on_complete_does_not_count_as_forced_eviction() {
     let (telemopts, addr, _aborter) = prom_metrics(None);
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let wf_name = "evict_on_complete_does_not_count_as_forced_eviction";
     let mut starter = CoreWfStarter::new_with_runtime(wf_name, rt);
     starter.worker_config.no_remote_activities(true);
@@ -1209,11 +1170,7 @@ where
 #[tokio::test]
 async fn metrics_available_from_custom_slot_supplier() {
     let (telemopts, addr, _aborter) = prom_metrics(None);
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let mut starter =
         CoreWfStarter::new_with_runtime("metrics_available_from_custom_slot_supplier", rt);
     starter.worker_config.no_remote_activities(true);

tests/integ_tests/workflow_tests.rs

@@ -18,6 +18,7 @@ mod stickyness;
 mod timers;
 mod upsert_search_attrs;
 
+use crate::common::get_integ_runtime_options;
 use crate::{
     common::{
         CoreWfStarter, history_from_proto_binary, init_core_and_create_wf,
@@ -35,7 +36,7 @@ use temporal_sdk::{
     ActivityOptions, LocalActivityOptions, TimerOptions, WfContext, interceptors::WorkerInterceptor,
 };
 use temporal_sdk_core::{
-    CoreRuntime, RuntimeOptionsBuilder,
+    CoreRuntime,
     replay::HistoryForReplay,
     test_help::{MockPollCfg, WorkerTestHelpers, drain_pollers_and_shutdown},
 };
@@ -67,7 +68,6 @@ use temporal_sdk_core_protos::{
     test_utils::schedule_activity_cmd,
 };
 use tokio::{join, sync::Notify, time::sleep};
-
 // TODO: We should get expected histories for these tests and confirm that the history at the end
 //  matches.
 
@@ -764,11 +764,7 @@ async fn nondeterminism_errors_fail_workflow_when_configured_to(
     #[values(true, false)] whole_worker: bool,
 ) {
     let (telemopts, addr, _aborter) = prom_metrics(None);
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let wf_name = "nondeterminism_errors_fail_workflow_when_configured_to";
     let mut starter = CoreWfStarter::new_with_runtime(wf_name, rt);
     starter.worker_config.no_remote_activities(true);

tests/main.rs

@@ -27,11 +27,12 @@ mod integ_tests {
     mod workflow_tests;
 
     use crate::common::{
-        CoreWfStarter, get_integ_server_options, get_integ_telem_options, rand_6_chars,
+        CoreWfStarter, get_integ_runtime_options, get_integ_server_options,
+        get_integ_telem_options, rand_6_chars,
     };
     use std::time::Duration;
     use temporal_client::{NamespacedClient, WorkflowService};
-    use temporal_sdk_core::{CoreRuntime, RuntimeOptionsBuilder, init_worker};
+    use temporal_sdk_core::{CoreRuntime, init_worker};
     use temporal_sdk_core_api::worker::WorkerConfigBuilder;
     use temporal_sdk_core_protos::temporal::api::{
         nexus::v1::{EndpointSpec, EndpointTarget, endpoint_target},
@@ -44,11 +45,9 @@ mod integ_tests {
     #[ignore] // Really a compile time check more than anything
     async fn lang_bridge_example() {
         let opts = get_integ_server_options();
-        let runtimeopts = RuntimeOptionsBuilder::default()
-            .telemetry_options(get_integ_telem_options())
-            .build()
-            .unwrap();
-        let runtime = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+        let runtime =
+            CoreRuntime::new_assume_tokio(get_integ_runtime_options(get_integ_telem_options()))
+                .unwrap();
         let mut retrying_client = opts
             .connect_no_namespace(runtime.telemetry().get_temporal_metric_meter())
             .await

tests/manual_tests.rs

@@ -5,6 +5,7 @@
 #[allow(dead_code)]
 mod common;
 
+use crate::common::get_integ_runtime_options;
 use common::{CoreWfStarter, prom_metrics, rand_6_chars};
 use futures_util::{
     StreamExt,
@@ -20,7 +21,7 @@ use std::{
 };
 use temporal_client::{GetWorkflowResultOpts, WfClientExt, WorkflowClientTrait, WorkflowOptions};
 use temporal_sdk::{ActContext, ActivityOptions, WfContext};
-use temporal_sdk_core::{CoreRuntime, RuntimeOptionsBuilder};
+use temporal_sdk_core::CoreRuntime;
 use temporal_sdk_core_api::{telemetry::PrometheusExporterOptionsBuilder, worker::PollerBehavior};
 use temporal_sdk_core_protos::coresdk::AsJsonPayloadExt;
 use tracing::info;
@@ -41,11 +42,7 @@ async fn poller_load_spiky() {
         } else {
             prom_metrics(None)
         };
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let mut starter = CoreWfStarter::new_with_runtime("poller_load", rt);
     starter
         .worker_config
@@ -204,11 +201,7 @@ async fn poller_load_sustained() {
         } else {
             prom_metrics(None)
         };
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let mut starter = CoreWfStarter::new_with_runtime("poller_load", rt);
     starter
         .worker_config
@@ -299,11 +292,7 @@ async fn poller_load_spike_then_sustained() {
         } else {
             prom_metrics(None)
         };
-    let runtimeopts = RuntimeOptionsBuilder::default()
-        .telemetry_options(telemopts)
-        .build()
-        .unwrap();
-    let rt = CoreRuntime::new_assume_tokio(runtimeopts).unwrap();
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
     let mut starter = CoreWfStarter::new_with_runtime("poller_load", rt);
     starter
         .worker_config