add tests

Author: THardy98

packages/test/src/activities/heartbeat-cancellation-details.ts

@@ -0,0 +1,24 @@
+import { ActivityCancellationDetails } from '@temporalio/common';
+import * as activity from '@temporalio/activity';
+
+export async function heartbeatCancellationDetailsActivity(
+  catchErr: boolean
+): Promise<ActivityCancellationDetails | undefined> {
+  // Exit early if we've already run this activity.
+  if (activity.activityInfo().heartbeatDetails === 'finally-complete') {
+    return activity.cancellationDetails();
+  }
+  // eslint-disable-next-line no-constant-condition
+  while (true) {
+    try {
+      activity.heartbeat();
+      await activity.sleep(100);
+    } catch (err) {
+      if (err instanceof activity.CancelledFailure && catchErr) {
+        return activity.cancellationDetails();
+      }
+      activity.heartbeat('finally-complete');
+      throw err;
+    }
+  }
+}

packages/test/src/helpers-integration.ts

@@ -30,14 +30,15 @@ import * as workflow from '@temporalio/workflow';
 import { temporal } from '@temporalio/proto';
 import { defineSearchAttributeKey, SearchAttributeType } from '@temporalio/common/lib/search-attributes';
 import { ConnectionInjectorInterceptor } from './activities/interceptors';
-import { Worker, TestWorkflowEnvironment, test as anyTest, bundlerOptions } from './helpers';
+import { Worker, TestWorkflowEnvironment, test as anyTest, bundlerOptions, waitUntil } from './helpers';
 
 export interface Context {
   env: TestWorkflowEnvironment;
   workflowBundle: WorkflowBundle;
 }
 
 const defaultDynamicConfigOptions = [
+  'frontend.activityAPIsEnabled=true',
   'frontend.enableExecuteMultiOperation=true',
   'frontend.workerVersioningDataAPIs=true',
   'frontend.workerVersioningWorkflowAPIs=true',
@@ -284,6 +285,48 @@ export function configurableHelpers<T>(
   };
 }
 
+export async function assertPendingActivityExistsEventually(
+  handle: WorkflowHandle<workflow.Workflow>,
+  activityId: string,
+  timeoutMs: number
+): Promise<temporal.api.workflow.v1.IPendingActivityInfo> {
+  let activityInfo: temporal.api.workflow.v1.IPendingActivityInfo | undefined;
+  try {
+    await waitUntil(async () => {
+      const desc = await handle.describe();
+      activityInfo = desc.raw.pendingActivities?.find((pa) => pa.activityId === activityId);
+      return activityInfo !== undefined;
+    }, timeoutMs);
+  } catch {
+    throw new Error(`Unable to find pending activity for activity ${activityId}`);
+  }
+  return activityInfo as temporal.api.workflow.v1.IPendingActivityInfo;
+}
+
+export async function setActivityPauseState(handle: WorkflowHandle, activityId: string, pause: boolean): Promise<void> {
+  const desc = await handle.describe();
+  const req = {
+    namespace: handle.client.options.namespace,
+    execution: {
+      workflowId: desc.raw.workflowExecutionInfo?.execution?.workflowId,
+      runId: desc.raw.workflowExecutionInfo?.execution?.runId,
+    },
+    id: activityId,
+  };
+  if (pause) {
+    await handle.client.workflowService.pauseActivity(req);
+  } else {
+    await handle.client.workflowService.unpauseActivity(req);
+  }
+  await waitUntil(async () => {
+    const info = await assertPendingActivityExistsEventually(handle, activityId, 10000);
+    if (pause) {
+      return info.paused ?? false;
+    }
+    return !info.paused;
+  }, 10000);
+}
+
 export function helpers(t: ExecutionContext<Context>, testEnv: TestWorkflowEnvironment = t.context.env): Helpers {
   return configurableHelpers(t, t.context.workflowBundle, testEnv);
 }

packages/test/src/test-integration-workflows.ts

@@ -2,7 +2,7 @@ import { setTimeout as setTimeoutPromise } from 'timers/promises';
 import { randomUUID } from 'crypto';
 import { ExecutionContext } from 'ava';
 import { firstValueFrom, Subject } from 'rxjs';
-import { WorkflowFailedError } from '@temporalio/client';
+import { WorkflowFailedError, WorkflowHandle } from '@temporalio/client';
 import * as activity from '@temporalio/activity';
 import { msToNumber, tsToMs } from '@temporalio/common/lib/time';
 import { TestWorkflowEnvironment } from '@temporalio/testing';
@@ -11,6 +11,7 @@ import * as workflow from '@temporalio/workflow';
 import { defineQuery, defineSignal } from '@temporalio/workflow';
 import { SdkFlags } from '@temporalio/workflow/lib/flags';
 import {
+  ActivityCancellationDetails,
   ActivityCancellationType,
   ApplicationFailure,
   defineSearchAttributeKey,
@@ -22,9 +23,17 @@ import {
 import { signalSchedulingWorkflow } from './activities/helpers';
 import { activityStartedSignal } from './workflows/definitions';
 import * as workflows from './workflows';
-import { Context, createLocalTestEnvironment, helpers, makeTestFunction } from './helpers-integration';
+import {
+  assertPendingActivityExistsEventually,
+  Context,
+  createLocalTestEnvironment,
+  helpers,
+  makeTestFunction,
+  setActivityPauseState,
+} from './helpers-integration';
 import { overrideSdkInternalFlag } from './mock-internal-flags';
 import { asSdkLoggerSink, loadHistory, RUN_TIME_SKIPPING_TESTS, waitUntil } from './helpers';
+import { heartbeatCancellationDetailsActivity } from './activities/heartbeat-cancellation-details';
 
 const test = makeTestFunction({
   workflowsPath: __filename,
@@ -1414,3 +1423,118 @@ test('Workflow can return root workflow', async (t) => {
     t.deepEqual(result, 'empty test-root-workflow-length');
   });
 });
+
+export async function heartbeatPauseWorkflow(
+  activityId: string,
+  catchErr: boolean,
+  maximumAttempts: number
+): Promise<Array<ActivityCancellationDetails | undefined>> {
+  const { heartbeatCancellationDetailsActivity } = workflow.proxyActivities({
+    startToCloseTimeout: '5s',
+    activityId,
+    retry: {
+      maximumAttempts,
+    },
+  });
+  const { heartbeatCancellationDetailsActivity2 } = workflow.proxyActivities({
+    startToCloseTimeout: '5s',
+    activityId: `${activityId}-2`,
+    retry: {
+      maximumAttempts,
+    },
+  });
+  const results = [];
+  results.push(
+    await heartbeatCancellationDetailsActivity(catchErr),
+    await heartbeatCancellationDetailsActivity2(catchErr)
+  );
+  return results;
+}
+
+test('Activity pause returns expected cancellation details', async (t) => {
+  const { createWorker, startWorkflow } = helpers(t);
+  const worker = await createWorker({
+    activities: {
+      heartbeatCancellationDetailsActivity,
+      heartbeatCancellationDetailsActivity2: heartbeatCancellationDetailsActivity,
+    },
+    maxHeartbeatThrottleInterval: '200ms',
+    defaultHeartbeatThrottleInterval: '200ms',
+  });
+
+  await worker.runUntil(async () => {
+    const testActivityId = randomUUID();
+    const handle = await startWorkflow(heartbeatPauseWorkflow, { args: [testActivityId, true, 1] });
+
+    const activityInfo = await assertPendingActivityExistsEventually(handle, testActivityId, 5000);
+    t.true(activityInfo.paused === false);
+    await setActivityPauseState(handle, testActivityId, true);
+    const activityInfo2 = await assertPendingActivityExistsEventually(handle, `${testActivityId}-2`, 5000);
+    t.true(activityInfo2.paused === false);
+    await setActivityPauseState(handle, `${testActivityId}-2`, true);
+    const result = await handle.result();
+    t.deepEqual(result[0], {
+      cancelRequested: false,
+      notFound: false,
+      paused: true,
+      timedOut: false,
+      workerShutdown: false,
+      reset: false,
+    });
+    t.deepEqual(result[1], {
+      cancelRequested: false,
+      notFound: false,
+      paused: true,
+      timedOut: false,
+      workerShutdown: false,
+      reset: false,
+    });
+  });
+});
+
+test('Activity can pause and unpause', async (t) => {
+  const { createWorker, startWorkflow } = helpers(t);
+  async function checkHeartbeatDetailsExist(handle: WorkflowHandle, activityId: string) {
+    const activityInfo = await assertPendingActivityExistsEventually(handle, activityId, 3000);
+    if (activityInfo.heartbeatDetails?.payloads) {
+      for (const payload of activityInfo.heartbeatDetails?.payloads || []) {
+        if (payload.data && payload.data?.length > 0) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
+
+  const worker = await createWorker({
+    activities: {
+      heartbeatCancellationDetailsActivity,
+      heartbeatCancellationDetailsActivity2: heartbeatCancellationDetailsActivity,
+    },
+    maxHeartbeatThrottleInterval: '200ms',
+    defaultHeartbeatThrottleInterval: '200ms',
+  });
+
+  await worker.runUntil(async () => {
+    const testActivityId = randomUUID();
+    const handle = await startWorkflow(heartbeatPauseWorkflow, { args: [testActivityId, false, 2] });
+    const activityInfo = await assertPendingActivityExistsEventually(handle, testActivityId, 5000);
+    t.true(activityInfo.paused === false);
+    await setActivityPauseState(handle, testActivityId, true);
+    await waitUntil(async () => {
+      return await checkHeartbeatDetailsExist(handle, testActivityId);
+    }, 3000);
+    await setActivityPauseState(handle, testActivityId, false);
+    const activityInfo2 = await assertPendingActivityExistsEventually(handle, `${testActivityId}-2`, 5000);
+    t.true(activityInfo2.paused === false);
+    await setActivityPauseState(handle, `${testActivityId}-2`, true);
+    await waitUntil(async () => {
+      return await checkHeartbeatDetailsExist(handle, `${testActivityId}-2`);
+    }, 3000);
+    await setActivityPauseState(handle, `${testActivityId}-2`, false);
+    const result = await handle.result();
+    // Undefined values are converted to null by data converter.
+    t.true(result[0] === null);
+    t.true(result[1] === null);
+  });
+});

packages/testing/src/mocking-activity-environment.ts

@@ -30,7 +30,7 @@ export interface MockActivityEnvironmentOptions {
  * will immediately be in a cancelled state.
  */
 export class MockActivityEnvironment extends events.EventEmitter {
-  public cancel: (reason?: any) => void = () => undefined;
+  public cancel: (reason?: any, details?: any) => void = () => undefined;
   public readonly context: activity.Context;
   private readonly activity: Activity;