From 5a695b50327ff962257e94e2991d9eb13982efa2 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 09:59:35 -0700 Subject: [PATCH 1/6] feat(durable-functions): add first-class testing API Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- packages/azure-functions-durable/CHANGELOG.md | 5 +- packages/azure-functions-durable/README.md | 111 ++++- packages/azure-functions-durable/package.json | 5 + .../src/testing/index.ts | 440 ++++++++++++++++++ .../test/unit/compat-exports.spec.ts | 12 + .../test/unit/testing.spec.ts | 246 ++++++++++ 6 files changed, 813 insertions(+), 6 deletions(-) create mode 100644 packages/azure-functions-durable/src/testing/index.ts create mode 100644 packages/azure-functions-durable/test/unit/testing.spec.ts diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index f2fc94e0..e8bc5e4c 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -2,8 +2,11 @@ ### New -### Fixes +- Added a first-class `durable-functions/testing` entry point with one-shot activity, + orchestration, and entity helpers plus an interactive in-memory orchestration harness for events, + termination, suspension, resumption, and real-time durable timers. +### Fixes ## v4.0.0-beta.1 (2026-07-31) diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index 801cf015..f0bf77ee 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -96,12 +96,13 @@ changed: **not** available — existing `response.getHeader(...)` calls **fail at runtime** and must be rewritten to index `response.headers[...]` by lower-cased key (response header names are lower-cased by `fetch`). -- **Some v3 top-level exports were removed** — `DummyOrchestrationContext` / `DummyEntityContext` - (testing utilities) and the entity-lock types above. `TaskFailedError` - is re-exported from the core SDK (aggregate failures surface as JS-native `AggregateError`); use the - core `TestOrchestrationWorker` / `TestOrchestrationClient` for orchestration unit tests. +- **The v3 dummy contexts were replaced by `durable-functions/testing`.** The new helpers run + orchestrators through the real in-memory replay engine and run entity batches directly, without a + Functions host or imports from `@microsoft/durabletask-js`. The entity-lock types above remain + removed. `TaskFailedError` is re-exported from the core SDK (aggregate failures surface as + JS-native `AggregateError`). - **A plain non-generator classic orchestrator is no longer supported.** A classic v3 orchestrator - written as a *synchronous, single-argument, non-generator* function `(context) => context.df.*` + written as a _synchronous, single-argument, non-generator_ function `(context) => context.df.*` (one that never `yield`s) is now treated as a **core-native** orchestrator and receives the core `OrchestrationContext`, which has no `.df`. This resolves [#321](https://github.com/microsoft/durabletask-js/issues/321), where a core-native @@ -149,6 +150,106 @@ app.http("startHello", { }); ``` +## Testing + +Import the first-class test helpers from `durable-functions/testing`. They create the compatibility +wrappers and core in-memory components internally, deserialize outputs, and clean up workers after +one-shot runs. + +### Activities and one-shot orchestrations + +```typescript +import type { OrchestrationContext } from "durable-functions"; +import { runActivity, runOrchestrator } from "durable-functions/testing"; + +const helloOrchestrator = function* (context: OrchestrationContext) { + const name = context.df.getInput(); + return yield context.df.callActivity("sayHello", name); +}; + +const activityOutput = await runActivity( + (name: string, context) => `${context.functionName}: Hello, ${name}!`, + "World", + { functionName: "sayHello" }, +); + +const orchestrationResult = await runOrchestrator(helloOrchestrator, { + input: "World", + activities: { + sayHello: (name: unknown) => `Hello, ${String(name)}!`, + }, +}); + +expect(activityOutput).toBe("sayHello: Hello, World!"); +expect(orchestrationResult.status).toBe("Completed"); +expect(orchestrationResult.output).toBe("Hello, World!"); +``` + +Failed orchestrations return `status: "Failed"` with plain `failure` details instead of requiring +manual parsing of core state. + +### Interactive orchestration tests + +Use a harness when a test needs to raise events, terminate, suspend, or resume an instance: + +```typescript +import type { OrchestrationContext } from "durable-functions"; +import { createOrchestrationHarness } from "durable-functions/testing"; + +const approvalOrchestrator = function* (context: OrchestrationContext) { + const approved = yield context.df.waitForExternalEvent("approved"); + return { approved }; +}; + +const harness = createOrchestrationHarness(); +harness.registerOrchestrator("approval", approvalOrchestrator); + +try { + const run = await harness.start("approval", { input: { orderId: "42" } }); + await run.waitForStart(); + await run.raiseEvent("approved", true); + + const result = await run.waitForCompletion(); + expect(result.output).toEqual({ approved: true }); +} finally { + await harness.dispose(); +} +``` + +Durable timers are supported with **real wall-clock delays**. The current core in-memory backend has +no virtual clock or timer-advance API, so use short timer delays in tests. The harness does not add a +synthetic fast-forward operation. + +### Entities + +`runEntity` executes a classic or core-native entity batch directly through the existing entity +executor, preserving per-operation rollback semantics: + +```typescript +import type { EntityHandler } from "durable-functions"; +import { runEntity } from "durable-functions/testing"; + +const counterEntity: EntityHandler = (context) => { + const state = context.df.getState(() => 0) ?? 0; + if (context.df.operationName === "add") { + context.df.setState(state + (context.df.getInput() ?? 0)); + } else if (context.df.operationName === "get") { + context.df.return(state); + } +}; + +const result = await runEntity(counterEntity, { + initialState: 0, + operations: [{ name: "add", input: 5 }, { name: "get" }], +}); + +expect(result.state).toBe(5); +expect(result.results).toEqual([undefined, 5]); +``` + +This is a direct, host-free entity unit seam. End-to-end entity messaging through the +orchestration harness is not supported by the current core in-memory backend. + ### Client (starter) functions `app.client.*` is sugar for the client-starter above — it adds the `durableClient` input binding and diff --git a/packages/azure-functions-durable/package.json b/packages/azure-functions-durable/package.json index 2fd4a593..9ae3e354 100644 --- a/packages/azure-functions-durable/package.json +++ b/packages/azure-functions-durable/package.json @@ -12,6 +12,11 @@ "types": "./dist/index.d.ts", "require": "./dist/index.js", "import": "./dist/index.js" + }, + "./testing": { + "types": "./dist/testing/index.d.ts", + "require": "./dist/testing/index.js", + "import": "./dist/testing/index.js" } }, "files": [ diff --git a/packages/azure-functions-durable/src/testing/index.ts b/packages/azure-functions-durable/src/testing/index.ts new file mode 100644 index 00000000..519f9be3 --- /dev/null +++ b/packages/azure-functions-durable/src/testing/index.ts @@ -0,0 +1,440 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { InvocationContext } from "@azure/functions"; +import { + EntityInstanceId, + InMemoryOrchestrationBackend, + OrchestrationState, + OrchestrationStatus, + TaskEntityShim, + TestOrchestrationClient, + TestOrchestrationWorker, +} from "@microsoft/durabletask-js"; +import type { EntityHandler as RegisteredEntityHandler, OrchestrationHandler } from "../app"; +import { wrapEntity } from "../entity-context"; +import { wrapOrchestrator } from "../orchestration-context"; + +const DEFAULT_TIMEOUT_MS = 10_000; +const DEFAULT_ORCHESTRATOR_NAME = "orchestrator"; +const DEFAULT_ENTITY_NAME = "entity"; +const DEFAULT_ENTITY_KEY = "test"; + +/** An Azure Functions activity handler used by the testing helpers. */ +export type TestActivityHandler = ( + input: TInput, + context: InvocationContext, +) => TOutput | Promise; + +/** Options for running an activity directly. */ +export interface ActivityTestOptions { + /** Function name exposed on the generated invocation context. */ + functionName?: string; + /** Invocation context to pass instead of creating one. */ + context?: InvocationContext; +} + +/** Runs one activity invocation without starting a Functions host. */ +export async function runActivity( + handler: TestActivityHandler, + input: TInput, + options: ActivityTestOptions = {}, +): Promise { + const context = options.context ?? new InvocationContext({ functionName: options.functionName ?? "activity" }); + return await handler(input, context); +} + +/** Plain failure details returned for a failed orchestration. */ +export interface OrchestrationTestFailure { + errorType: string; + message: string; + stackTrace?: string; +} + +/** Runtime states exposed by the Functions testing API. */ +export type OrchestrationTestStatus = "Pending" | "Running" | "Completed" | "Failed" | "Terminated" | "Suspended"; + +/** A deserialized snapshot of an orchestration run. */ +export interface OrchestrationTestResult { + instanceId: string; + status: OrchestrationTestStatus; + output?: TOutput; + customStatus?: unknown; + failure?: OrchestrationTestFailure; +} + +/** Options shared by one-shot and harness orchestration starts. */ +export interface OrchestrationStartOptions { + input?: TInput; + instanceId?: string; + startAt?: Date; +} + +/** Options for a one-shot orchestrator test. */ +export interface OrchestratorTestOptions extends OrchestrationStartOptions { + activities?: Readonly>>; + timeoutMs?: number; +} + +/** Options for an interactive orchestration harness. */ +export interface OrchestrationHarnessOptions { + timeoutMs?: number; +} + +/** A running orchestration controlled through an in-memory harness. */ +export interface OrchestrationRun { + readonly instanceId: string; + readonly status: OrchestrationTestStatus; + readonly output: TOutput | undefined; + readonly customStatus: unknown; + readonly failure: OrchestrationTestFailure | undefined; + waitForStart(timeoutMs?: number): Promise>; + waitForCompletion(timeoutMs?: number): Promise>; + raiseEvent(name: string, data?: TData): Promise; + terminate(output?: TOutputData): Promise; + suspend(): Promise; + resume(): Promise; + refresh(): Promise>; +} + +/** In-memory harness for interactive orchestration tests. */ +export interface OrchestrationHarness { + registerOrchestrator(name: string, handler: OrchestrationHandler): OrchestrationHarness; + registerActivity( + name: string, + handler: TestActivityHandler, + ): OrchestrationHarness; + start( + name: string, + options?: OrchestrationStartOptions, + ): Promise>; + dispose(): Promise; +} + +/** + * Creates a host-free orchestration harness backed by the real replay executor. + * + * @remarks Durable timers use real wall-clock time. The current core in-memory backend does not + * expose a virtual clock or timer-advance API, so tests should schedule short timer delays. + */ +export function createOrchestrationHarness(options: OrchestrationHarnessOptions = {}): OrchestrationHarness { + return new InMemoryOrchestrationHarness(options.timeoutMs ?? DEFAULT_TIMEOUT_MS); +} + +/** + * Runs one orchestrator to a terminal state and always releases its worker. + * + * @remarks Durable timers use real wall-clock time. The current core in-memory backend does not + * expose a virtual clock or timer-advance API, so tests should schedule short timer delays. + */ +export async function runOrchestrator( + handler: OrchestrationHandler, + options: OrchestratorTestOptions = {}, +): Promise> { + const harness = createOrchestrationHarness({ timeoutMs: options.timeoutMs }); + harness.registerOrchestrator(DEFAULT_ORCHESTRATOR_NAME, handler); + for (const [name, activity] of Object.entries(options.activities ?? {})) { + harness.registerActivity(name, activity); + } + + try { + const run = await harness.start(DEFAULT_ORCHESTRATOR_NAME, options); + return await run.waitForCompletion(); + } finally { + await harness.dispose(); + } +} + +/** One operation in a standalone entity batch. */ +export interface EntityTestOperation { + name: string; + input?: TInput; +} + +/** Options for a standalone entity batch. */ +export interface EntityTestOptions { + initialState?: TState; + operations: ReadonlyArray; + entityName?: string; + entityKey?: string; +} + +/** Deserialized state and operation return values from an entity batch. */ +export interface EntityTestResult { + state: TState | undefined; + results: Array; +} + +/** Error thrown when an entity operation in a test batch fails. */ +export class EntityOperationError extends Error { + readonly operationName: string; + readonly operationIndex: number; + readonly errorType: string; + readonly stackTrace?: string; + + constructor(operationName: string, operationIndex: number, failure: OrchestrationTestFailure) { + super(`Entity operation '${operationName}' at index ${operationIndex} failed: ${failure.message}`); + this.name = "EntityOperationError"; + this.operationName = operationName; + this.operationIndex = operationIndex; + this.errorType = failure.errorType; + this.stackTrace = failure.stackTrace; + } +} + +/** + * Runs an entity batch directly through the core entity executor. + * + * @remarks This is a standalone unit seam. It does not enqueue entity messages in the orchestration + * harness; end-to-end in-memory entity routing is not available in the current core test backend. + */ +export async function runEntity( + handler: RegisteredEntityHandler, + options: EntityTestOptions, +): Promise> { + const entityName = options.entityName ?? DEFAULT_ENTITY_NAME; + const entityKey = options.entityKey ?? DEFAULT_ENTITY_KEY; + const entityId = new EntityInstanceId(entityName, entityKey); + const executor = new TaskEntityShim(wrapEntity(handler)(), entityId); + const request = createEntityBatchRequest(options.initialState, options.operations); + const response = await executor.executeAsync(request); + + const results = response.getResultsList().map((operationResult, index) => { + const failureDetails = operationResult.getFailure()?.getFailuredetails(); + if (failureDetails) { + throw new EntityOperationError(options.operations[index].name, index, { + errorType: failureDetails.getErrortype(), + message: failureDetails.getErrormessage(), + stackTrace: failureDetails.getStacktrace()?.getValue(), + }); + } + return deserialize(operationResult.getSuccess()?.getResult()?.getValue()); + }); + + return { + state: deserialize(response.getEntitystate()?.getValue()), + results, + }; +} + +class InMemoryOrchestrationHarness implements OrchestrationHarness { + private readonly backend = new InMemoryOrchestrationBackend(); + private readonly worker = new TestOrchestrationWorker(this.backend); + private readonly client = new TestOrchestrationClient(this.backend); + private started = false; + private disposed = false; + + constructor(private readonly timeoutMs: number) {} + + registerOrchestrator(name: string, handler: OrchestrationHandler): OrchestrationHarness { + this.ensureCanRegister(); + this.worker.addNamedOrchestrator(name, wrapOrchestrator(handler)); + return this; + } + + registerActivity(name: string, handler: TestActivityHandler): OrchestrationHarness { + this.ensureCanRegister(); + this.worker.addNamedActivity(name, async (_context, input) => { + return await runActivity(handler, input as TInput, { functionName: name }); + }); + return this; + } + + async start( + name: string, + options: OrchestrationStartOptions = {}, + ): Promise> { + this.ensureNotDisposed(); + if (!this.started) { + await this.worker.start(); + this.started = true; + } + + const instanceId = await this.client.scheduleNewOrchestration( + name, + options.input, + options.instanceId, + options.startAt, + ); + return new InMemoryOrchestrationRun(instanceId, this.client, this.timeoutMs); + } + + async dispose(): Promise { + if (this.disposed) { + return; + } + this.disposed = true; + try { + if (this.started) { + await this.worker.stop(); + } + } finally { + this.backend.reset(); + await this.client.stop(); + } + } + + private ensureCanRegister(): void { + this.ensureNotDisposed(); + if (this.started) { + throw new Error("Orchestrators and activities must be registered before the first run starts."); + } + } + + private ensureNotDisposed(): void { + if (this.disposed) { + throw new Error("The orchestration harness has been disposed."); + } + } +} + +class InMemoryOrchestrationRun implements OrchestrationRun { + private currentStatus: OrchestrationTestStatus = "Pending"; + private currentOutput: TOutput | undefined; + private currentCustomStatus: unknown; + private currentFailure: OrchestrationTestFailure | undefined; + + constructor( + readonly instanceId: string, + private readonly client: TestOrchestrationClient, + private readonly timeoutMs: number, + ) {} + + get status(): OrchestrationTestStatus { + return this.currentStatus; + } + + get output(): TOutput | undefined { + return this.currentOutput; + } + + get customStatus(): unknown { + return this.currentCustomStatus; + } + + get failure(): OrchestrationTestFailure | undefined { + return this.currentFailure; + } + + async waitForStart(timeoutMs: number = this.timeoutMs): Promise> { + const state = await this.client.waitForOrchestrationStart(this.instanceId, true, timeoutMs / 1000); + return this.applyState(requireState(state, this.instanceId)); + } + + async waitForCompletion(timeoutMs: number = this.timeoutMs): Promise> { + const state = await this.client.waitForOrchestrationCompletion(this.instanceId, true, timeoutMs / 1000); + return this.applyState(requireState(state, this.instanceId)); + } + + async raiseEvent(name: string, data?: TData): Promise { + await this.client.raiseOrchestrationEvent(this.instanceId, name, data); + } + + async terminate(output?: TOutputData): Promise { + await this.client.terminateOrchestration(this.instanceId, output); + } + + async suspend(): Promise { + await this.client.suspendOrchestration(this.instanceId); + await this.refresh(); + } + + async resume(): Promise { + await this.client.resumeOrchestration(this.instanceId); + await this.refresh(); + } + + async refresh(): Promise> { + const state = await this.client.getOrchestrationState(this.instanceId, true); + return this.applyState(requireState(state, this.instanceId)); + } + + private applyState(state: OrchestrationState): OrchestrationTestResult { + this.currentStatus = toTestStatus(state.runtimeStatus); + this.currentOutput = deserialize(state.serializedOutput); + this.currentCustomStatus = deserialize(state.serializedCustomStatus); + this.currentFailure = state.failureDetails + ? { + errorType: state.failureDetails.errorType, + message: state.failureDetails.message, + stackTrace: state.failureDetails.stackTrace, + } + : undefined; + + return { + instanceId: this.instanceId, + status: this.currentStatus, + output: this.currentOutput, + customStatus: this.currentCustomStatus, + failure: this.currentFailure, + }; + } +} + +function createEntityBatchRequest( + initialState: TState | undefined, + operations: ReadonlyArray, +): Parameters[0] { + // TaskEntityShim reads only these protobuf accessors. Keeping this adapter local avoids exposing + // generated protocol types through the durable-functions testing API. + const request = { + getEntitystate: () => serializedValue(initialState), + getOperationsList: () => + operations.map((operation) => ({ + getOperation: () => operation.name, + getInput: () => serializedValue(operation.input), + })), + }; + return request as Parameters[0]; +} + +function serializedValue(value: unknown): { getValue(): string } | undefined { + if (value === undefined) { + return undefined; + } + const serialized = JSON.stringify(value); + if (serialized === undefined) { + throw new TypeError("Test values must be JSON-serializable."); + } + return { getValue: () => serialized }; +} + +function deserialize(value: string | undefined): T | undefined { + return value === undefined ? undefined : (JSON.parse(value) as T); +} + +function requireState(state: OrchestrationState | undefined, instanceId: string): OrchestrationState { + if (!state) { + throw new Error(`Orchestration '${instanceId}' was not found.`); + } + return state; +} + +function toTestStatus(status: OrchestrationStatus): OrchestrationTestStatus { + switch (status) { + case OrchestrationStatus.PENDING: + return "Pending"; + case OrchestrationStatus.RUNNING: + return "Running"; + case OrchestrationStatus.COMPLETED: + return "Completed"; + case OrchestrationStatus.FAILED: + return "Failed"; + case OrchestrationStatus.TERMINATED: + return "Terminated"; + case OrchestrationStatus.SUSPENDED: + return "Suspended"; + case OrchestrationStatus.CONTINUED_AS_NEW: + return "Running"; + default: + throw new Error(`Unexpected orchestration status value: ${String(status)}.`); + } +} + +/** Namespace-style access matching the rest of the Durable Functions API. */ +export const test = { + createOrchestrationHarness, + runActivity, + runEntity, + runOrchestrator, +}; diff --git a/packages/azure-functions-durable/test/unit/compat-exports.spec.ts b/packages/azure-functions-durable/test/unit/compat-exports.spec.ts index 039667e0..1db3f4dd 100644 --- a/packages/azure-functions-durable/test/unit/compat-exports.spec.ts +++ b/packages/azure-functions-durable/test/unit/compat-exports.spec.ts @@ -10,6 +10,7 @@ import type { OrchestrationHandler, } from "../../src"; import { TaskFailedError } from "../../src"; +import packageJson from "../../package.json"; describe("v3 compatibility type aliases", () => { it("exposes ActivityHandler / OrchestrationHandler / OrchestrationContext", () => { @@ -22,6 +23,17 @@ describe("v3 compatibility type aliases", () => { expect(typeof orchestrator).toBe("function"); }); + describe("package exports", () => { + it("publishes the testing subpath independently from the runtime entry point", () => { + const exports = packageJson.exports as Record; + expect(exports["./testing"]).toEqual({ + types: "./dist/testing/index.d.ts", + require: "./dist/testing/index.js", + import: "./dist/testing/index.js", + }); + }); + }); + it("exposes generic EntityContext / EntityHandler and DurableClient", () => { // Compile-guard: the generic aliases must accept a type argument (the legacy v3 surface uses // e.g. EntityHandler), even though our underlying types are non-generic. diff --git a/packages/azure-functions-durable/test/unit/testing.spec.ts b/packages/azure-functions-durable/test/unit/testing.spec.ts new file mode 100644 index 00000000..6556077c --- /dev/null +++ b/packages/azure-functions-durable/test/unit/testing.spec.ts @@ -0,0 +1,246 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { InvocationContext } from "@azure/functions"; +import { TaskEntity } from "@microsoft/durabletask-js"; +import { + EntityOperationError, + createOrchestrationHarness, + runActivity, + runEntity, + runOrchestrator, + test, +} from "../../src/testing"; +import type { EntityHandler, OrchestrationContext, OrchestrationHandler } from "../../src"; + +describe("durable-functions/testing", () => { + it("runs an activity with a Functions invocation context", async () => { + const result = await runActivity( + (input: string, context: InvocationContext) => `${context.functionName}:${input}`, + "World", + { functionName: "sayHello" }, + ); + + expect(result).toBe("sayHello:World"); + }); + + it("runs a classic orchestrator with inline Functions-style activities", async () => { + const orchestrator: OrchestrationHandler = function* ( + context: OrchestrationContext, + ): Generator { + const name = context.df.getInput(); + return yield context.df.callActivity("sayHello", name); + }; + + const result = await runOrchestrator(orchestrator, { + input: "World", + activities: { + sayHello: (input: unknown) => `Hello, ${String(input)}!`, + }, + instanceId: "one-shot", + }); + + expect(result).toMatchObject({ + instanceId: "one-shot", + status: "Completed", + output: "Hello, World!", + }); + }); + + it("returns deserialized custom status and failure details", async () => { + const orchestrator: OrchestrationHandler = function* ( + context: OrchestrationContext, + ): Generator { + context.df.setCustomStatus({ phase: "starting" }); + yield context.df.callActivity("fail"); + }; + + const result = await runOrchestrator(orchestrator, { + activities: { + fail: () => { + throw new TypeError("activity failed"); + }, + }, + }); + + expect(result.status).toBe("Failed"); + expect(result.customStatus).toEqual({ phase: "starting" }); + expect(result.failure).toMatchObject({ + errorType: expect.any(String), + message: expect.stringContaining("activity failed"), + }); + }); + + it("stops one-shot resources when waiting times out", async () => { + const orchestrator: OrchestrationHandler = function* ( + context: OrchestrationContext, + ): Generator { + yield context.df.waitForExternalEvent("never"); + }; + + await expect(runOrchestrator(orchestrator, { timeoutMs: 10 })).rejects.toThrow("Timeout waiting for orchestration"); + + await expect(runOrchestrator(async () => "still runs", { timeoutMs: 1000 })).resolves.toMatchObject({ + status: "Completed", + output: "still runs", + }); + }); + + it("drives external events through an orchestration harness", async () => { + const harness = createOrchestrationHarness(); + harness.registerOrchestrator("approval", function* (context: OrchestrationContext): Generator< + unknown, + { approved: boolean }, + boolean + > { + const approved = yield context.df.waitForExternalEvent("approved"); + return { approved }; + }); + + try { + const run = await harness.start<{ approved: boolean }>("approval", { + instanceId: "approval-1", + }); + await run.waitForStart(); + + expect(run.status).toBe("Running"); + + await run.raiseEvent("approved", true); + const result = await run.waitForCompletion(); + + expect(result.output).toEqual({ approved: true }); + expect(run.output).toEqual({ approved: true }); + } finally { + await harness.dispose(); + } + }); + + it("supports real-time durable timers", async () => { + const harness = createOrchestrationHarness({ timeoutMs: 1000 }); + harness.registerOrchestrator("timer", function* (context: OrchestrationContext): Generator< + unknown, + string, + unknown + > { + yield context.df.createTimer(new Date(context.df.currentUtcDateTime.getTime() + 10)); + return "timer fired"; + }); + + try { + const run = await harness.start("timer"); + await expect(run.waitForCompletion()).resolves.toMatchObject({ + status: "Completed", + output: "timer fired", + }); + } finally { + await harness.dispose(); + } + }); + + it("terminates, suspends, and resumes orchestration runs", async () => { + const harness = createOrchestrationHarness(); + harness.registerOrchestrator("interactive", function* (context: OrchestrationContext): Generator< + unknown, + string, + unknown + > { + yield context.df.waitForExternalEvent("finish"); + return "completed"; + }); + + try { + const suspended = await harness.start("interactive", { instanceId: "suspended" }); + await suspended.waitForStart(); + await suspended.suspend(); + expect(suspended.status).toBe("Suspended"); + await suspended.resume(); + expect(suspended.status).toBe("Running"); + + const terminated = await harness.start("interactive", { instanceId: "terminated" }); + await terminated.waitForStart(); + await terminated.terminate({ reason: "test" }); + const result = await terminated.waitForCompletion(); + + expect(result).toMatchObject({ + status: "Terminated", + output: { reason: "test" }, + }); + } finally { + await harness.dispose(); + } + }); + + it("runs a classic entity batch and deserializes state and operation results", async () => { + const entity: EntityHandler = (context) => { + const current = context.df.getState(() => 0) ?? 0; + switch (context.df.operationName) { + case "add": + context.df.setState(current + (context.df.getInput() ?? 0)); + break; + case "get": + context.df.return(current); + break; + } + }; + + const result = await runEntity(entity, { + initialState: 2, + entityName: "Counter", + entityKey: "Key", + operations: [{ name: "add", input: 3 }, { name: "get" }], + }); + + expect(result).toEqual({ + state: 5, + results: [undefined, 5], + }); + }); + + it("runs a core-native entity factory without exposing the core executor", async () => { + class Counter extends TaskEntity { + protected initializeState(): number { + return 0; + } + + add(value: number): number { + this.state += value; + return this.state; + } + } + + const result = await runEntity(() => new Counter(), { + operations: [{ name: "add", input: 4 }], + }); + + expect(result).toEqual({ state: 4, results: [4] }); + }); + + it("surfaces entity operation failures without committing their state", async () => { + const entity: EntityHandler = (context) => { + context.df.setState(99); + throw new RangeError("invalid operation"); + }; + + await expect( + runEntity(entity, { + initialState: 1, + operations: [{ name: "break" }], + }), + ).rejects.toMatchObject({ + name: "EntityOperationError", + operationName: "break", + operationIndex: 0, + errorType: "RangeError", + message: expect.stringContaining("invalid operation"), + } satisfies Partial); + }); + + it("exposes the helpers through the test namespace", () => { + expect(test).toEqual({ + createOrchestrationHarness, + runActivity, + runEntity, + runOrchestrator, + }); + }); +}); From e77ae8ebaf26dee5cafdf5f2187b6cb98af95652 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 10:22:21 -0700 Subject: [PATCH 2/6] fix(durable-functions): harden testing lifecycle Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- packages/azure-functions-durable/README.md | 4 + .../src/testing/index.ts | 217 +++++++++++++++--- .../test/unit/testing.spec.ts | 153 +++++++++++- 3 files changed, 338 insertions(+), 36 deletions(-) diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index f0bf77ee..7ddfd661 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -220,6 +220,10 @@ Durable timers are supported with **real wall-clock delays**. The current core i no virtual clock or timer-advance API, so use short timer delays in tests. The harness does not add a synthetic fast-forward operation. +Future `startAt` values are rejected because the current in-memory backend enqueues scheduled starts +immediately instead of deferring their execution. Past `startAt` values are accepted and start +immediately. + ### Entities `runEntity` executes a classic or core-native entity batch directly through the existing entity diff --git a/packages/azure-functions-durable/src/testing/index.ts b/packages/azure-functions-durable/src/testing/index.ts index 519f9be3..8feb1076 100644 --- a/packages/azure-functions-durable/src/testing/index.ts +++ b/packages/azure-functions-durable/src/testing/index.ts @@ -19,6 +19,8 @@ const DEFAULT_TIMEOUT_MS = 10_000; const DEFAULT_ORCHESTRATOR_NAME = "orchestrator"; const DEFAULT_ENTITY_NAME = "entity"; const DEFAULT_ENTITY_KEY = "test"; +const HARNESS_DISPOSED_MESSAGE = "The orchestration harness has been disposed."; +const ACTIVITY_CANCELLED = Symbol("activityCancelled"); /** An Azure Functions activity handler used by the testing helpers. */ export type TestActivityHandler = ( @@ -67,6 +69,7 @@ export interface OrchestrationTestResult { export interface OrchestrationStartOptions { input?: TInput; instanceId?: string; + /** A past time is accepted, but future scheduled starts are not supported by the in-memory backend. */ startAt?: Date; } @@ -221,10 +224,22 @@ class InMemoryOrchestrationHarness implements OrchestrationHarness { private readonly backend = new InMemoryOrchestrationBackend(); private readonly worker = new TestOrchestrationWorker(this.backend); private readonly client = new TestOrchestrationClient(this.backend); - private started = false; + private readonly runs = new Set>(); + private readonly activityCancellation: Promise; + private cancelActivities!: () => void; + private transition: Promise = Promise.resolve(); + private workerStartPromise: Promise | undefined; + private workerStarted = false; + private startRequested = false; + private activityCancellationRequested = false; private disposed = false; + private disposePromise: Promise | undefined; - constructor(private readonly timeoutMs: number) {} + constructor(private readonly timeoutMs: number) { + this.activityCancellation = new Promise((resolve) => { + this.cancelActivities = () => resolve(ACTIVITY_CANCELLED); + }); + } registerOrchestrator(name: string, handler: OrchestrationHandler): OrchestrationHarness { this.ensureCanRegister(); @@ -235,39 +250,66 @@ class InMemoryOrchestrationHarness implements OrchestrationHarness { registerActivity(name: string, handler: TestActivityHandler): OrchestrationHarness { this.ensureCanRegister(); this.worker.addNamedActivity(name, async (_context, input) => { - return await runActivity(handler, input as TInput, { functionName: name }); + if (this.activityCancellationRequested) { + throw createHarnessDisposedError(); + } + const result = await Promise.race([ + runActivity(handler, input as TInput, { functionName: name }), + this.activityCancellation, + ]); + if (result === ACTIVITY_CANCELLED) { + throw createHarnessDisposedError(); + } + return result; }); return this; } - async start( + start( name: string, options: OrchestrationStartOptions = {}, ): Promise> { - this.ensureNotDisposed(); - if (!this.started) { - await this.worker.start(); - this.started = true; - } - - const instanceId = await this.client.scheduleNewOrchestration( - name, - options.input, - options.instanceId, - options.startAt, - ); - return new InMemoryOrchestrationRun(instanceId, this.client, this.timeoutMs); + this.startRequested = true; + return this.enqueueTransition(async () => { + this.ensureNotDisposed(); + validateStartAt(options.startAt); + await this.ensureWorkerStarted(); + this.ensureNotDisposed(); + + const instanceId = await this.client.scheduleNewOrchestration( + name, + options.input, + options.instanceId, + options.startAt, + ); + this.ensureNotDisposed(); + + const run = new InMemoryOrchestrationRun(instanceId, this.client, this.timeoutMs); + this.runs.add(run as InMemoryOrchestrationRun); + return run; + }); } - async dispose(): Promise { - if (this.disposed) { - return; + dispose(): Promise { + if (!this.disposePromise) { + this.disposed = true; + this.activityCancellationRequested = true; + this.cancelActivities(); + for (const run of this.runs) { + run.markDisposed(); + } + this.disposePromise = this.enqueueTransition(() => this.disposeCore()); } - this.disposed = true; + return this.disposePromise; + } + + private async disposeCore(): Promise { + await this.workerStartPromise?.catch(() => undefined); try { - if (this.started) { + if (this.workerStarted) { await this.worker.stop(); } + await Promise.all([...this.runs].map((run) => run.captureTerminalSnapshot())); } finally { this.backend.reset(); await this.client.stop(); @@ -276,16 +318,34 @@ class InMemoryOrchestrationHarness implements OrchestrationHarness { private ensureCanRegister(): void { this.ensureNotDisposed(); - if (this.started) { + if (this.startRequested) { throw new Error("Orchestrators and activities must be registered before the first run starts."); } } private ensureNotDisposed(): void { if (this.disposed) { - throw new Error("The orchestration harness has been disposed."); + throw createHarnessDisposedError(); } } + + private ensureWorkerStarted(): Promise { + if (!this.workerStartPromise) { + this.workerStartPromise = this.worker.start().then(() => { + this.workerStarted = true; + }); + } + return this.workerStartPromise; + } + + private enqueueTransition(operation: () => Promise): Promise { + const result = this.transition.then(operation); + this.transition = result.then( + () => undefined, + () => undefined, + ); + return result; + } } class InMemoryOrchestrationRun implements OrchestrationRun { @@ -293,6 +353,8 @@ class InMemoryOrchestrationRun implements OrchestrationRun { private currentOutput: TOutput | undefined; private currentCustomStatus: unknown; private currentFailure: OrchestrationTestFailure | undefined; + private terminalResult: OrchestrationTestResult | undefined; + private disposed = false; constructor( readonly instanceId: string, @@ -317,38 +379,71 @@ class InMemoryOrchestrationRun implements OrchestrationRun { } async waitForStart(timeoutMs: number = this.timeoutMs): Promise> { - const state = await this.client.waitForOrchestrationStart(this.instanceId, true, timeoutMs / 1000); + const state = await this.executeWhileActive(() => + this.client.waitForOrchestrationStart(this.instanceId, true, timeoutMs / 1000), + ); return this.applyState(requireState(state, this.instanceId)); } async waitForCompletion(timeoutMs: number = this.timeoutMs): Promise> { - const state = await this.client.waitForOrchestrationCompletion(this.instanceId, true, timeoutMs / 1000); - return this.applyState(requireState(state, this.instanceId)); + if (this.terminalResult) { + return this.terminalResult; + } + if (this.disposed) { + throw createHarnessDisposedError(); + } + + try { + const state = await this.client.waitForOrchestrationCompletion(this.instanceId, true, timeoutMs / 1000); + return this.applyState(requireState(state, this.instanceId)); + } catch (error) { + if (this.disposed) { + if (this.terminalResult) { + return this.terminalResult; + } + throw createHarnessDisposedError(); + } + throw error; + } } async raiseEvent(name: string, data?: TData): Promise { - await this.client.raiseOrchestrationEvent(this.instanceId, name, data); + await this.executeWhileActive(() => this.client.raiseOrchestrationEvent(this.instanceId, name, data)); } async terminate(output?: TOutputData): Promise { - await this.client.terminateOrchestration(this.instanceId, output); + await this.executeWhileActive(() => this.client.terminateOrchestration(this.instanceId, output)); } async suspend(): Promise { - await this.client.suspendOrchestration(this.instanceId); + await this.executeWhileActive(() => this.client.suspendOrchestration(this.instanceId)); await this.refresh(); } async resume(): Promise { - await this.client.resumeOrchestration(this.instanceId); + await this.executeWhileActive(() => this.client.resumeOrchestration(this.instanceId)); await this.refresh(); } async refresh(): Promise> { - const state = await this.client.getOrchestrationState(this.instanceId, true); + const state = await this.executeWhileActive(() => this.client.getOrchestrationState(this.instanceId, true)); return this.applyState(requireState(state, this.instanceId)); } + markDisposed(): void { + this.disposed = true; + } + + async captureTerminalSnapshot(): Promise { + if (this.terminalResult) { + return; + } + const state = await this.client.getOrchestrationState(this.instanceId, true); + if (state && isTerminalStatus(state.runtimeStatus)) { + this.applyState(state); + } + } + private applyState(state: OrchestrationState): OrchestrationTestResult { this.currentStatus = toTestStatus(state.runtimeStatus); this.currentOutput = deserialize(state.serializedOutput); @@ -361,13 +456,37 @@ class InMemoryOrchestrationRun implements OrchestrationRun { } : undefined; - return { + const result = { instanceId: this.instanceId, status: this.currentStatus, output: this.currentOutput, customStatus: this.currentCustomStatus, failure: this.currentFailure, }; + if (isTerminalStatus(state.runtimeStatus)) { + this.terminalResult = result; + } + return result; + } + + private async executeWhileActive(operation: () => Promise): Promise { + this.ensureActive(); + try { + const result = await operation(); + this.ensureActive(); + return result; + } catch (error) { + if (this.disposed) { + throw createHarnessDisposedError(); + } + throw error; + } + } + + private ensureActive(): void { + if (this.disposed) { + throw createHarnessDisposedError(); + } } } @@ -378,7 +497,7 @@ function createEntityBatchRequest( // TaskEntityShim reads only these protobuf accessors. Keeping this adapter local avoids exposing // generated protocol types through the durable-functions testing API. const request = { - getEntitystate: () => serializedValue(initialState), + getEntitystate: () => serializedStateValue(initialState), getOperationsList: () => operations.map((operation) => ({ getOperation: () => operation.name, @@ -399,6 +518,10 @@ function serializedValue(value: unknown): { getValue(): string } | undefined { return { getValue: () => serialized }; } +function serializedStateValue(value: unknown): { getValue(): string } | undefined { + return value == null ? undefined : serializedValue(value); +} + function deserialize(value: string | undefined): T | undefined { return value === undefined ? undefined : (JSON.parse(value) as T); } @@ -427,8 +550,32 @@ function toTestStatus(status: OrchestrationStatus): OrchestrationTestStatus { case OrchestrationStatus.CONTINUED_AS_NEW: return "Running"; default: - throw new Error(`Unexpected orchestration status value: ${String(status)}.`); + return "Running"; + } +} + +function isTerminalStatus(status: OrchestrationStatus): boolean { + return ( + status === OrchestrationStatus.COMPLETED || + status === OrchestrationStatus.FAILED || + status === OrchestrationStatus.TERMINATED + ); +} + +function validateStartAt(startAt: Date | undefined): void { + if (!startAt) { + return; } + if (Number.isNaN(startAt.getTime())) { + throw new TypeError("startAt must be a valid Date."); + } + if (startAt.getTime() > Date.now()) { + throw new Error("Future startAt values are not supported by the in-memory orchestration harness."); + } +} + +function createHarnessDisposedError(): Error { + return new Error(HARNESS_DISPOSED_MESSAGE); } /** Namespace-style access matching the rest of the Durable Functions API. */ diff --git a/packages/azure-functions-durable/test/unit/testing.spec.ts b/packages/azure-functions-durable/test/unit/testing.spec.ts index 6556077c..edcd1629 100644 --- a/packages/azure-functions-durable/test/unit/testing.spec.ts +++ b/packages/azure-functions-durable/test/unit/testing.spec.ts @@ -2,7 +2,7 @@ // Licensed under the MIT License. import { InvocationContext } from "@azure/functions"; -import { TaskEntity } from "@microsoft/durabletask-js"; +import { TaskEntity, TestOrchestrationWorker } from "@microsoft/durabletask-js"; import { EntityOperationError, createOrchestrationHarness, @@ -86,6 +86,101 @@ describe("durable-functions/testing", () => { }); }); + it("does not hang cleanup when an activity never settles", async () => { + let activityStarted!: () => void; + let releaseActivity!: () => void; + const started = new Promise((resolve) => { + activityStarted = resolve; + }); + const neverSettlingActivity = new Promise((resolve) => { + releaseActivity = resolve; + }); + const orchestrator: OrchestrationHandler = function* ( + context: OrchestrationContext, + ): Generator { + yield context.df.callActivity("never"); + }; + + const execution = runOrchestrator(orchestrator, { + timeoutMs: 25, + activities: { + never: () => { + activityStarted(); + return neverSettlingActivity; + }, + }, + }); + await started; + + const outcome = await Promise.race([ + execution.then( + () => "settled", + () => "settled", + ), + new Promise<"hung">((resolve) => setTimeout(() => resolve("hung"), 250)), + ]); + + releaseActivity(); + await execution.catch(() => undefined); + expect(outcome).toBe("settled"); + }); + + it("serializes concurrent harness starts through one worker startup", async () => { + const harness = createOrchestrationHarness(); + harness.registerOrchestrator("ready", async (_context, input) => input); + + try { + const [first, second] = await Promise.all([ + harness.start("ready", { instanceId: "first", input: 1 }), + harness.start("ready", { instanceId: "second", input: 2 }), + ]); + + await expect(first.waitForCompletion()).resolves.toMatchObject({ output: 1 }); + await expect(second.waitForCompletion()).resolves.toMatchObject({ output: 2 }); + } finally { + await harness.dispose(); + } + }); + + it("stops a worker when disposal races its startup", async () => { + let releaseStart!: () => void; + const startGate = new Promise((resolve) => { + releaseStart = resolve; + }); + const startSpy = jest.spyOn(TestOrchestrationWorker.prototype, "start").mockImplementation(async () => startGate); + const stopSpy = jest.spyOn(TestOrchestrationWorker.prototype, "stop").mockResolvedValue(); + const harness = createOrchestrationHarness(); + harness.registerOrchestrator("ready", async () => "done"); + + try { + const starting = harness.start("ready"); + await Promise.resolve(); + const disposing = harness.dispose(); + releaseStart(); + + await expect(starting).rejects.toThrow("The orchestration harness has been disposed."); + await disposing; + expect(startSpy).toHaveBeenCalledTimes(1); + expect(stopSpy).toHaveBeenCalledTimes(1); + } finally { + startSpy.mockRestore(); + stopSpy.mockRestore(); + } + }); + + it("rejects future scheduled starts that the in-memory backend cannot defer", async () => { + const harness = createOrchestrationHarness(); + harness.registerOrchestrator("ready", async () => "done"); + + try { + await expect(harness.start("ready", { startAt: new Date(Date.now() + 60_000) })).rejects.toThrow( + "Future startAt values are not supported", + ); + } finally { + await harness.dispose(); + } + }); + it("drives external events through an orchestration harness", async () => { const harness = createOrchestrationHarness(); harness.registerOrchestrator("approval", function* (context: OrchestrationContext): Generator< @@ -170,6 +265,42 @@ describe("durable-functions/testing", () => { } }); + it("retains terminal results but rejects other run operations after disposal", async () => { + const harness = createOrchestrationHarness({ timeoutMs: 20 }); + harness.registerOrchestrator("ready", async () => ({ done: true })); + const run = await harness.start<{ done: boolean }>("ready"); + const completed = await run.waitForCompletion(); + + await harness.dispose(); + + await expect(run.waitForCompletion()).resolves.toEqual(completed); + expect(run.output).toEqual({ done: true }); + await expect(run.waitForStart()).rejects.toThrow("The orchestration harness has been disposed."); + await expect(run.refresh()).rejects.toThrow("The orchestration harness has been disposed."); + await expect(run.raiseEvent("ignored")).rejects.toThrow("The orchestration harness has been disposed."); + await expect(run.terminate()).rejects.toThrow("The orchestration harness has been disposed."); + await expect(run.suspend()).rejects.toThrow("The orchestration harness has been disposed."); + await expect(run.resume()).rejects.toThrow("The orchestration harness has been disposed."); + }); + + it("rejects nonterminal run operations immediately after disposal", async () => { + const harness = createOrchestrationHarness({ timeoutMs: 20 }); + harness.registerOrchestrator("waiting", function* (context: OrchestrationContext): Generator< + unknown, + void, + unknown + > { + yield context.df.waitForExternalEvent("never"); + }); + const run = await harness.start("waiting"); + await run.waitForStart(); + + await harness.dispose(); + + await expect(run.waitForCompletion()).rejects.toThrow("The orchestration harness has been disposed."); + await expect(run.refresh()).rejects.toThrow("The orchestration harness has been disposed."); + }); + it("runs a classic entity batch and deserializes state and operation results", async () => { const entity: EntityHandler = (context) => { const current = context.df.getState(() => 0) ?? 0; @@ -215,6 +346,26 @@ describe("durable-functions/testing", () => { expect(result).toEqual({ state: 4, results: [4] }); }); + it("treats null initial entity state as absent while preserving null operation input", async () => { + const entity: EntityHandler = (context) => { + context.df.return({ + input: context.df.getInput(), + isNewlyConstructed: context.df.isNewlyConstructed, + state: context.df.getState(() => 7), + }); + }; + + const result = await runEntity(entity, { + initialState: null, + operations: [{ name: "inspect", input: null }], + }); + + expect(result).toEqual({ + state: undefined, + results: [{ input: null, isNewlyConstructed: true, state: 7 }], + }); + }); + it("surfaces entity operation failures without committing their state", async () => { const entity: EntityHandler = (context) => { context.df.setState(99); From 8ae48c54152e7417fa474f95cb71475c75fd867e Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 10:48:31 -0700 Subject: [PATCH 3/6] docs(durable-functions): clarify activity timeout behavior Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- packages/azure-functions-durable/README.md | 5 +++ .../src/testing/index.ts | 18 ++++++++++ .../test/unit/testing.spec.ts | 33 ++++++++++--------- 3 files changed, 40 insertions(+), 16 deletions(-) diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index 7ddfd661..29b68ecc 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -188,6 +188,11 @@ expect(orchestrationResult.output).toBe("Hello, World!"); Failed orchestrations return `status: "Failed"` with plain `failure` details instead of requiring manual parsing of core state. +An orchestration timeout or harness disposal stops the helper from waiting for an in-flight activity +and bounds worker shutdown, but it **does not cancel JavaScript activity code that has already +started**. That code may continue running timers, I/O, or other side effects after the helper +returns. Use finite activity stubs, or stubs that cancel their own external resources. + ### Interactive orchestration tests Use a harness when a test needs to raise events, terminate, suspend, or resume an instance: diff --git a/packages/azure-functions-durable/src/testing/index.ts b/packages/azure-functions-durable/src/testing/index.ts index 8feb1076..a265bf53 100644 --- a/packages/azure-functions-durable/src/testing/index.ts +++ b/packages/azure-functions-durable/src/testing/index.ts @@ -76,11 +76,19 @@ export interface OrchestrationStartOptions { /** Options for a one-shot orchestrator test. */ export interface OrchestratorTestOptions extends OrchestrationStartOptions { activities?: Readonly>>; + /** + * Maximum time to wait for orchestration completion. Timing out bounds harness shutdown but + * cannot cancel activity code that has already started. + */ timeoutMs?: number; } /** Options for an interactive orchestration harness. */ export interface OrchestrationHarnessOptions { + /** + * Default wait timeout. Timing out bounds harness waiting but cannot cancel activity code that + * has already started. + */ timeoutMs?: number; } @@ -111,6 +119,10 @@ export interface OrchestrationHarness { name: string, options?: OrchestrationStartOptions, ): Promise>; + /** + * Stops harness processing. Already-running activity code is abandoned, not cancelled, and may + * continue its own timers, I/O, or side effects. + */ dispose(): Promise; } @@ -119,6 +131,9 @@ export interface OrchestrationHarness { * * @remarks Durable timers use real wall-clock time. The current core in-memory backend does not * expose a virtual clock or timer-advance API, so tests should schedule short timer delays. + * Timeout or disposal stops the harness from waiting for an activity and bounds worker shutdown, + * but JavaScript cannot forcibly cancel activity code that has already started. Such code may + * continue timers, I/O, or side effects; use finite or independently cancellable activity stubs. */ export function createOrchestrationHarness(options: OrchestrationHarnessOptions = {}): OrchestrationHarness { return new InMemoryOrchestrationHarness(options.timeoutMs ?? DEFAULT_TIMEOUT_MS); @@ -129,6 +144,9 @@ export function createOrchestrationHarness(options: OrchestrationHarnessOptions * * @remarks Durable timers use real wall-clock time. The current core in-memory backend does not * expose a virtual clock or timer-advance API, so tests should schedule short timer delays. + * A timeout stops waiting and bounds worker shutdown, but JavaScript cannot forcibly cancel + * activity code that has already started. Such code may continue timers, I/O, or side effects; use + * finite or independently cancellable activity stubs. */ export async function runOrchestrator( handler: OrchestrationHandler, diff --git a/packages/azure-functions-durable/test/unit/testing.spec.ts b/packages/azure-functions-durable/test/unit/testing.spec.ts index edcd1629..4b03d5b5 100644 --- a/packages/azure-functions-durable/test/unit/testing.spec.ts +++ b/packages/azure-functions-durable/test/unit/testing.spec.ts @@ -86,15 +86,16 @@ describe("durable-functions/testing", () => { }); }); - it("does not hang cleanup when an activity never settles", async () => { + it("bounds cleanup without cancelling already-running activity code", async () => { let activityStarted!: () => void; - let releaseActivity!: () => void; + let activityFinished!: () => void; const started = new Promise((resolve) => { activityStarted = resolve; }); - const neverSettlingActivity = new Promise((resolve) => { - releaseActivity = resolve; + const finished = new Promise((resolve) => { + activityFinished = resolve; }); + let sideEffectCompleted = false; const orchestrator: OrchestrationHandler = function* ( context: OrchestrationContext, ): Generator { @@ -106,23 +107,23 @@ describe("durable-functions/testing", () => { activities: { never: () => { activityStarted(); - return neverSettlingActivity; + return new Promise((resolve) => { + setTimeout(() => { + sideEffectCompleted = true; + activityFinished(); + resolve(); + }, 250); + }); }, }, }); await started; - const outcome = await Promise.race([ - execution.then( - () => "settled", - () => "settled", - ), - new Promise<"hung">((resolve) => setTimeout(() => resolve("hung"), 250)), - ]); - - releaseActivity(); - await execution.catch(() => undefined); - expect(outcome).toBe("settled"); + await expect(execution).rejects.toThrow("Timeout waiting for orchestration"); + expect(sideEffectCompleted).toBe(false); + + await finished; + expect(sideEffectCompleted).toBe(true); }); it("serializes concurrent harness starts through one worker startup", async () => { From dd918ecf568f6fd8aecd600acb9fff92d0e539f6 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 11:17:27 -0700 Subject: [PATCH 4/6] fix(durable-functions): expose testing types to node resolution Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- packages/azure-functions-durable/package.json | 7 ++++ .../test/unit/compat-exports.spec.ts | 34 +++++++++++++++++++ 2 files changed, 41 insertions(+) diff --git a/packages/azure-functions-durable/package.json b/packages/azure-functions-durable/package.json index 9ae3e354..614016ac 100644 --- a/packages/azure-functions-durable/package.json +++ b/packages/azure-functions-durable/package.json @@ -7,6 +7,13 @@ "description": "Azure Functions Durable provider for the Durable Task JavaScript SDK", "main": "./dist/index.js", "types": "./dist/index.d.ts", + "typesVersions": { + "*": { + "testing": [ + "./dist/testing/index.d.ts" + ] + } + }, "exports": { ".": { "types": "./dist/index.d.ts", diff --git a/packages/azure-functions-durable/test/unit/compat-exports.spec.ts b/packages/azure-functions-durable/test/unit/compat-exports.spec.ts index 1db3f4dd..2e382ba4 100644 --- a/packages/azure-functions-durable/test/unit/compat-exports.spec.ts +++ b/packages/azure-functions-durable/test/unit/compat-exports.spec.ts @@ -1,6 +1,10 @@ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. +import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "fs"; +import { tmpdir } from "os"; +import { join } from "path"; +import ts from "typescript"; import type { ActivityHandler, DurableClient, @@ -31,6 +35,36 @@ describe("v3 compatibility type aliases", () => { require: "./dist/testing/index.js", import: "./dist/testing/index.js", }); + expect(packageJson.typesVersions).toEqual({ + "*": { + testing: ["./dist/testing/index.d.ts"], + }, + }); + }); + + it("resolves testing declarations with classic Node module resolution", () => { + const consumerRoot = mkdtempSync(join(tmpdir(), "durable-functions-types-")); + const packageRoot = join(consumerRoot, "node_modules", "durable-functions"); + const declarationPath = join(packageRoot, "dist", "testing", "index.d.ts"); + const consumerPath = join(consumerRoot, "consumer.ts"); + + try { + mkdirSync(join(packageRoot, "dist", "testing"), { recursive: true }); + writeFileSync(join(packageRoot, "package.json"), JSON.stringify(packageJson)); + writeFileSync(declarationPath, "export declare function runOrchestrator(): Promise;"); + writeFileSync(consumerPath, 'import { runOrchestrator } from "durable-functions/testing";'); + + const resolved = ts.resolveModuleName( + "durable-functions/testing", + consumerPath, + { moduleResolution: ts.ModuleResolutionKind.Node10 }, + ts.sys, + ).resolvedModule; + + expect(resolved?.resolvedFileName.replace(/\\/g, "/")).toBe(declarationPath.replace(/\\/g, "/")); + } finally { + rmSync(consumerRoot, { recursive: true, force: true }); + } }); }); From 5f15a90df44381440902ce19c0ae68ed20dd1fbe Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 5 Aug 2026 11:18:54 -0700 Subject: [PATCH 5/6] fix(durable-functions): await testing handlers on cleanup Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- packages/azure-functions-durable/README.md | 14 +++- .../src/testing/index.ts | 53 ++++-------- .../test/unit/testing.spec.ts | 84 ++++++++++++++----- 3 files changed, 86 insertions(+), 65 deletions(-) diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index 29b68ecc..b4715a04 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -188,10 +188,11 @@ expect(orchestrationResult.output).toBe("Hello, World!"); Failed orchestrations return `status: "Failed"` with plain `failure` details instead of requiring manual parsing of core state. -An orchestration timeout or harness disposal stops the helper from waiting for an in-flight activity -and bounds worker shutdown, but it **does not cancel JavaScript activity code that has already -started**. That code may continue running timers, I/O, or other side effects after the helper -returns. Use finite activity stubs, or stubs that cancel their own external resources. +`runOrchestrator` intentionally has no forced timeout. It returns only after the orchestration +reaches a terminal state and worker cleanup finishes, so activity code cannot keep mutating test +state after the helper returns. Arbitrary JavaScript promises cannot be forcibly cancelled: if an +orchestrator or activity never settles, the helper also remains pending and the test runner's own +timeout applies. ### Interactive orchestration tests @@ -221,6 +222,11 @@ try { } ``` +Harness wait timeouts are observation-only: a timed-out `waitForStart()` or +`waitForCompletion()` call leaves the explicitly owned harness running. `dispose()` waits for +in-flight orchestrator and activity handlers to settle before it reports completion; if +non-cooperative user code never settles, disposal also remains pending. + Durable timers are supported with **real wall-clock delays**. The current core in-memory backend has no virtual clock or timer-advance API, so use short timer delays in tests. The harness does not add a synthetic fast-forward operation. diff --git a/packages/azure-functions-durable/src/testing/index.ts b/packages/azure-functions-durable/src/testing/index.ts index a265bf53..46ace84b 100644 --- a/packages/azure-functions-durable/src/testing/index.ts +++ b/packages/azure-functions-durable/src/testing/index.ts @@ -20,7 +20,6 @@ const DEFAULT_ORCHESTRATOR_NAME = "orchestrator"; const DEFAULT_ENTITY_NAME = "entity"; const DEFAULT_ENTITY_KEY = "test"; const HARNESS_DISPOSED_MESSAGE = "The orchestration harness has been disposed."; -const ACTIVITY_CANCELLED = Symbol("activityCancelled"); /** An Azure Functions activity handler used by the testing helpers. */ export type TestActivityHandler = ( @@ -76,18 +75,13 @@ export interface OrchestrationStartOptions { /** Options for a one-shot orchestrator test. */ export interface OrchestratorTestOptions extends OrchestrationStartOptions { activities?: Readonly>>; - /** - * Maximum time to wait for orchestration completion. Timing out bounds harness shutdown but - * cannot cancel activity code that has already started. - */ - timeoutMs?: number; } /** Options for an interactive orchestration harness. */ export interface OrchestrationHarnessOptions { /** - * Default wait timeout. Timing out bounds harness waiting but cannot cancel activity code that - * has already started. + * Default timeout for observing a run. A timeout rejects only that wait; the harness remains live + * and owned by the caller until {@link OrchestrationHarness.dispose} finishes. */ timeoutMs?: number; } @@ -120,8 +114,10 @@ export interface OrchestrationHarness { options?: OrchestrationStartOptions, ): Promise>; /** - * Stops harness processing. Already-running activity code is abandoned, not cancelled, and may - * continue its own timers, I/O, or side effects. + * Stops harness processing after in-flight orchestrator and activity handlers settle. + * + * @remarks Arbitrary JavaScript promises cannot be forcibly cancelled. If user code never + * settles, disposal also remains pending. */ dispose(): Promise; } @@ -131,9 +127,9 @@ export interface OrchestrationHarness { * * @remarks Durable timers use real wall-clock time. The current core in-memory backend does not * expose a virtual clock or timer-advance API, so tests should schedule short timer delays. - * Timeout or disposal stops the harness from waiting for an activity and bounds worker shutdown, - * but JavaScript cannot forcibly cancel activity code that has already started. Such code may - * continue timers, I/O, or side effects; use finite or independently cancellable activity stubs. + * Wait timeouts reject only the observation call; the harness remains live and must still be + * disposed. Disposal awaits in-flight orchestrator and activity handlers because arbitrary + * JavaScript promises cannot be forcibly cancelled. */ export function createOrchestrationHarness(options: OrchestrationHarnessOptions = {}): OrchestrationHarness { return new InMemoryOrchestrationHarness(options.timeoutMs ?? DEFAULT_TIMEOUT_MS); @@ -144,15 +140,15 @@ export function createOrchestrationHarness(options: OrchestrationHarnessOptions * * @remarks Durable timers use real wall-clock time. The current core in-memory backend does not * expose a virtual clock or timer-advance API, so tests should schedule short timer delays. - * A timeout stops waiting and bounds worker shutdown, but JavaScript cannot forcibly cancel - * activity code that has already started. Such code may continue timers, I/O, or side effects; use - * finite or independently cancellable activity stubs. + * This helper intentionally has no forced timeout: it awaits terminal orchestration execution and + * worker cleanup before returning. Arbitrary JavaScript promises cannot be forcibly cancelled, so + * non-cooperative orchestrator or activity code remains subject to the test runner's own timeout. */ export async function runOrchestrator( handler: OrchestrationHandler, options: OrchestratorTestOptions = {}, ): Promise> { - const harness = createOrchestrationHarness({ timeoutMs: options.timeoutMs }); + const harness = createOrchestrationHarness({ timeoutMs: 0 }); harness.registerOrchestrator(DEFAULT_ORCHESTRATOR_NAME, handler); for (const [name, activity] of Object.entries(options.activities ?? {})) { harness.registerActivity(name, activity); @@ -243,21 +239,14 @@ class InMemoryOrchestrationHarness implements OrchestrationHarness { private readonly worker = new TestOrchestrationWorker(this.backend); private readonly client = new TestOrchestrationClient(this.backend); private readonly runs = new Set>(); - private readonly activityCancellation: Promise; - private cancelActivities!: () => void; private transition: Promise = Promise.resolve(); private workerStartPromise: Promise | undefined; private workerStarted = false; private startRequested = false; - private activityCancellationRequested = false; private disposed = false; private disposePromise: Promise | undefined; - constructor(private readonly timeoutMs: number) { - this.activityCancellation = new Promise((resolve) => { - this.cancelActivities = () => resolve(ACTIVITY_CANCELLED); - }); - } + constructor(private readonly timeoutMs: number) {} registerOrchestrator(name: string, handler: OrchestrationHandler): OrchestrationHarness { this.ensureCanRegister(); @@ -268,17 +257,7 @@ class InMemoryOrchestrationHarness implements OrchestrationHarness { registerActivity(name: string, handler: TestActivityHandler): OrchestrationHarness { this.ensureCanRegister(); this.worker.addNamedActivity(name, async (_context, input) => { - if (this.activityCancellationRequested) { - throw createHarnessDisposedError(); - } - const result = await Promise.race([ - runActivity(handler, input as TInput, { functionName: name }), - this.activityCancellation, - ]); - if (result === ACTIVITY_CANCELLED) { - throw createHarnessDisposedError(); - } - return result; + return await runActivity(handler, input as TInput, { functionName: name }); }); return this; } @@ -311,8 +290,6 @@ class InMemoryOrchestrationHarness implements OrchestrationHarness { dispose(): Promise { if (!this.disposePromise) { this.disposed = true; - this.activityCancellationRequested = true; - this.cancelActivities(); for (const run of this.runs) { run.markDisposed(); } diff --git a/packages/azure-functions-durable/test/unit/testing.spec.ts b/packages/azure-functions-durable/test/unit/testing.spec.ts index 4b03d5b5..1a48bca0 100644 --- a/packages/azure-functions-durable/test/unit/testing.spec.ts +++ b/packages/azure-functions-durable/test/unit/testing.spec.ts @@ -71,22 +71,7 @@ describe("durable-functions/testing", () => { }); }); - it("stops one-shot resources when waiting times out", async () => { - const orchestrator: OrchestrationHandler = function* ( - context: OrchestrationContext, - ): Generator { - yield context.df.waitForExternalEvent("never"); - }; - - await expect(runOrchestrator(orchestrator, { timeoutMs: 10 })).rejects.toThrow("Timeout waiting for orchestration"); - - await expect(runOrchestrator(async () => "still runs", { timeoutMs: 1000 })).resolves.toMatchObject({ - status: "Completed", - output: "still runs", - }); - }); - - it("bounds cleanup without cancelling already-running activity code", async () => { + it("does not return before a delayed activity settles", async () => { let activityStarted!: () => void; let activityFinished!: () => void; const started = new Promise((resolve) => { @@ -95,35 +80,56 @@ describe("durable-functions/testing", () => { const finished = new Promise((resolve) => { activityFinished = resolve; }); - let sideEffectCompleted = false; + const mutations: string[] = []; const orchestrator: OrchestrationHandler = function* ( context: OrchestrationContext, ): Generator { yield context.df.callActivity("never"); }; + let executionSettled = false; const execution = runOrchestrator(orchestrator, { - timeoutMs: 25, activities: { never: () => { activityStarted(); return new Promise((resolve) => { setTimeout(() => { - sideEffectCompleted = true; + mutations.push("activity"); activityFinished(); resolve(); - }, 250); + }, 100); }); }, }, + }).finally(() => { + executionSettled = true; }); await started; - await expect(execution).rejects.toThrow("Timeout waiting for orchestration"); - expect(sideEffectCompleted).toBe(false); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(executionSettled).toBe(false); await finished; - expect(sideEffectCompleted).toBe(true); + await expect(execution).resolves.toMatchObject({ status: "Completed" }); + expect(mutations).toEqual(["activity"]); + + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(mutations).toEqual(["activity"]); + }); + + it("does not return before a delayed async orchestrator settles", async () => { + let settled = false; + const execution = runOrchestrator(async () => { + await new Promise((resolve) => setTimeout(resolve, 75)); + settled = true; + return "done"; + }); + + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(settled).toBe(false); + + await expect(execution).resolves.toMatchObject({ status: "Completed", output: "done" }); + expect(settled).toBe(true); }); it("serializes concurrent harness starts through one worker startup", async () => { @@ -302,6 +308,38 @@ describe("durable-functions/testing", () => { await expect(run.refresh()).rejects.toThrow("The orchestration harness has been disposed."); }); + it("does not finish disposal while activity code is still running", async () => { + let activityStarted!: () => void; + const started = new Promise((resolve) => { + activityStarted = resolve; + }); + let activityFinished = false; + const harness = createOrchestrationHarness({ timeoutMs: 20 }); + harness.registerActivity("slow", async () => { + activityStarted(); + await new Promise((resolve) => setTimeout(resolve, 100)); + activityFinished = true; + }); + harness.registerOrchestrator("slow", function* (context: OrchestrationContext): Generator { + yield context.df.callActivity("slow"); + }); + + const run = await harness.start("slow"); + await started; + await expect(run.waitForCompletion()).rejects.toThrow("Timeout waiting for orchestration"); + + let disposalSettled = false; + const disposal = harness.dispose().finally(() => { + disposalSettled = true; + }); + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(disposalSettled).toBe(false); + expect(activityFinished).toBe(false); + + await disposal; + expect(activityFinished).toBe(true); + }); + it("runs a classic entity batch and deserializes state and operation results", async () => { const entity: EntityHandler = (context) => { const current = context.df.getState(() => 0) ?? 0; From 1ae9a278f3a7957dca4917c29e46b95eadc3578c Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 6 Aug 2026 10:21:52 -0700 Subject: [PATCH 6/6] refactor(durable-functions): reduce testing API to a minimal surface Cut the `durable-functions/testing` entry point from 19 exported symbols to 4 by reusing what the package and core SDK already ship. - Drop `runActivity` and `TestActivityHandler`. Activities are plain Functions handlers, so they are called directly; only the `InvocationContext` needs a factory (`createActivityContext`). - Drop `OrchestrationHarness`, `createOrchestrationHarness`, and `runEntity`. Interactive scenarios and entities are driven with `TestOrchestrationWorker` / `TestOrchestrationClient` plus the already public `wrapOrchestrator` / `wrapEntity`, which the README now shows. This removes the hand-rolled protobuf entity batch request. - Drop `OrchestrationTestFailure` and `OrchestrationTestStatus` in favour of core `TaskFailureDetails` and the package's `OrchestrationRuntimeStatus`. - Map results through the existing `toDurableOrchestrationStatus`, so a test observes exactly what `client.getStatus()` returns. This also fixes a latent bug: the previous `JSON.parse` of `serializedOutput` throws a `SyntaxError` for a failed instance whose output is a plain error string. - Drop `startAt`, which only accepted past values and was a no-op. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- packages/azure-functions-durable/CHANGELOG.md | 9 +- packages/azure-functions-durable/README.md | 149 ++--- .../src/testing/index.ts | 613 ++---------------- .../test/unit/testing.spec.ts | 533 ++++----------- 4 files changed, 286 insertions(+), 1018 deletions(-) diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index e8bc5e4c..05fc47f7 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -2,9 +2,12 @@ ### New -- Added a first-class `durable-functions/testing` entry point with one-shot activity, - orchestration, and entity helpers plus an interactive in-memory orchestration harness for events, - termination, suspension, resumption, and real-time durable timers. +- Added a `durable-functions/testing` entry point with `runOrchestrator`, which runs an orchestrator + to a terminal state against inline activity implementations on the in-memory backend and always + releases its worker, and `createActivityContext` for invoking activity handlers directly. + Interactive scenarios (external events, termination, suspend/resume) and entity batches are + covered by driving the `@microsoft/durabletask-js` in-memory test stack with `wrapOrchestrator` / + `wrapEntity`; see the README. ### Fixes diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index b4715a04..19694078 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -152,119 +152,102 @@ app.http("startHello", { ## Testing -Import the first-class test helpers from `durable-functions/testing`. They create the compatibility -wrappers and core in-memory components internally, deserialize outputs, and clean up workers after -one-shot runs. +`durable-functions/testing` provides one helper for the common case — running an orchestrator to +completion against fake activities — plus a factory for the activity invocation context. Everything +else is already covered by the in-memory test stack in `@microsoft/durabletask-js`. -### Activities and one-shot orchestrations +### Activities + +Activity handlers are ordinary Azure Functions handlers with no durable state, so call them directly +and pass a context: + +```typescript +import type { InvocationContext } from "@azure/functions"; +import { createActivityContext } from "durable-functions/testing"; + +const sayHello = (name: string, context: InvocationContext) => `${context.functionName}: Hello, ${name}!`; + +expect(await sayHello("World", createActivityContext("sayHello"))).toBe("sayHello: Hello, World!"); +``` + +### Orchestrations + +`runOrchestrator` registers the orchestrator and the supplied activities on an in-memory worker, runs +the instance to a terminal state, and stops the worker before returning: ```typescript import type { OrchestrationContext } from "durable-functions"; -import { runActivity, runOrchestrator } from "durable-functions/testing"; +import { OrchestrationRuntimeStatus } from "durable-functions"; +import { runOrchestrator } from "durable-functions/testing"; const helloOrchestrator = function* (context: OrchestrationContext) { const name = context.df.getInput(); return yield context.df.callActivity("sayHello", name); }; -const activityOutput = await runActivity( - (name: string, context) => `${context.functionName}: Hello, ${name}!`, - "World", - { functionName: "sayHello" }, -); - -const orchestrationResult = await runOrchestrator(helloOrchestrator, { +const result = await runOrchestrator(helloOrchestrator, { input: "World", activities: { sayHello: (name: unknown) => `Hello, ${String(name)}!`, }, }); -expect(activityOutput).toBe("sayHello: Hello, World!"); -expect(orchestrationResult.status).toBe("Completed"); -expect(orchestrationResult.output).toBe("Hello, World!"); +expect(result.runtimeStatus).toBe(OrchestrationRuntimeStatus.Completed); +expect(result.output).toBe("Hello, World!"); ``` -Failed orchestrations return `status: "Failed"` with plain `failure` details instead of requiring -manual parsing of core state. +`runtimeStatus`, `output`, and `customStatus` are produced by the same mapping `client.getStatus()` +applies at runtime, so a test asserts on the values a deployed client would observe. A failed run +returns `OrchestrationRuntimeStatus.Failed` together with `failure` (`errorType`, `message`, +`stackTrace`). -`runOrchestrator` intentionally has no forced timeout. It returns only after the orchestration -reaches a terminal state and worker cleanup finishes, so activity code cannot keep mutating test -state after the helper returns. Arbitrary JavaScript promises cannot be forcibly cancelled: if an -orchestrator or activity never settles, the helper also remains pending and the test runner's own -timeout applies. +`runOrchestrator` intentionally has no forced timeout. It returns only after the orchestration is +terminal and the worker has drained, so activity code cannot keep mutating test state afterwards. +Arbitrary JavaScript promises cannot be cancelled: if a handler never settles, the helper stays +pending and the test runner's own timeout applies. -### Interactive orchestration tests +Durable timers run on **real wall-clock delays** — the in-memory backend has no virtual clock, so +keep timer delays short in tests. -Use a harness when a test needs to raise events, terminate, suspend, or resume an instance: +### Interactive scenarios and entities -```typescript -import type { OrchestrationContext } from "durable-functions"; -import { createOrchestrationHarness } from "durable-functions/testing"; +External events, termination, suspend/resume, and entity batches are not wrapped. Drive the core +in-memory stack directly and register Durable Functions handlers with `wrapOrchestrator` (or +`wrapEntity`): -const approvalOrchestrator = function* (context: OrchestrationContext) { - const approved = yield context.df.waitForExternalEvent("approved"); - return { approved }; -}; - -const harness = createOrchestrationHarness(); -harness.registerOrchestrator("approval", approvalOrchestrator); +```typescript +import { + InMemoryOrchestrationBackend, + TestOrchestrationClient, + TestOrchestrationWorker, +} from "@microsoft/durabletask-js"; +import { toDurableOrchestrationStatus, wrapOrchestrator } from "durable-functions"; + +const backend = new InMemoryOrchestrationBackend(); +const worker = new TestOrchestrationWorker(backend); +const client = new TestOrchestrationClient(backend); + +worker.addNamedOrchestrator( + "approval", + wrapOrchestrator(function* (context) { + return { approved: yield context.df.waitForExternalEvent("approved") }; + }), +); +await worker.start(); try { - const run = await harness.start("approval", { input: { orderId: "42" } }); - await run.waitForStart(); - await run.raiseEvent("approved", true); + const instanceId = await client.scheduleNewOrchestration("approval", undefined, "approval-1"); + await client.waitForOrchestrationStart(instanceId, true, 10); + await client.raiseOrchestrationEvent(instanceId, "approved", true); - const result = await run.waitForCompletion(); - expect(result.output).toEqual({ approved: true }); + const state = await client.waitForOrchestrationCompletion(instanceId, true, 10); + expect(toDurableOrchestrationStatus(state!).output).toEqual({ approved: true }); } finally { - await harness.dispose(); + await worker.stop(); + backend.reset(); } ``` -Harness wait timeouts are observation-only: a timed-out `waitForStart()` or -`waitForCompletion()` call leaves the explicitly owned harness running. `dispose()` waits for -in-flight orchestrator and activity handlers to settle before it reports completion; if -non-cooperative user code never settles, disposal also remains pending. - -Durable timers are supported with **real wall-clock delays**. The current core in-memory backend has -no virtual clock or timer-advance API, so use short timer delays in tests. The harness does not add a -synthetic fast-forward operation. - -Future `startAt` values are rejected because the current in-memory backend enqueues scheduled starts -immediately instead of deferring their execution. Past `startAt` values are accepted and start -immediately. - -### Entities - -`runEntity` executes a classic or core-native entity batch directly through the existing entity -executor, preserving per-operation rollback semantics: - -```typescript -import type { EntityHandler } from "durable-functions"; -import { runEntity } from "durable-functions/testing"; - -const counterEntity: EntityHandler = (context) => { - const state = context.df.getState(() => 0) ?? 0; - if (context.df.operationName === "add") { - context.df.setState(state + (context.df.getInput() ?? 0)); - } else if (context.df.operationName === "get") { - context.df.return(state); - } -}; - -const result = await runEntity(counterEntity, { - initialState: 0, - operations: [{ name: "add", input: 5 }, { name: "get" }], -}); - -expect(result.state).toBe(5); -expect(result.results).toEqual([undefined, 5]); -``` - -This is a direct, host-free entity unit seam. End-to-end entity messaging through the -orchestration harness is not supported by the current core in-memory backend. - ### Client (starter) functions `app.client.*` is sugar for the client-starter above — it adds the `durableClient` input binding and diff --git a/packages/azure-functions-durable/src/testing/index.ts b/packages/azure-functions-durable/src/testing/index.ts index 46ace84b..6111564a 100644 --- a/packages/azure-functions-durable/src/testing/index.ts +++ b/packages/azure-functions-durable/src/testing/index.ts @@ -3,580 +3,107 @@ import { InvocationContext } from "@azure/functions"; import { - EntityInstanceId, InMemoryOrchestrationBackend, - OrchestrationState, - OrchestrationStatus, - TaskEntityShim, TestOrchestrationClient, TestOrchestrationWorker, } from "@microsoft/durabletask-js"; -import type { EntityHandler as RegisteredEntityHandler, OrchestrationHandler } from "../app"; -import { wrapEntity } from "../entity-context"; +import type { TaskFailureDetails } from "@microsoft/durabletask-js"; +import type { ActivityHandler, OrchestrationHandler } from "../app"; import { wrapOrchestrator } from "../orchestration-context"; +import { OrchestrationRuntimeStatus, toDurableOrchestrationStatus } from "../orchestration-status"; -const DEFAULT_TIMEOUT_MS = 10_000; -const DEFAULT_ORCHESTRATOR_NAME = "orchestrator"; -const DEFAULT_ENTITY_NAME = "entity"; -const DEFAULT_ENTITY_KEY = "test"; -const HARNESS_DISPOSED_MESSAGE = "The orchestration harness has been disposed."; +const ORCHESTRATOR_NAME = "orchestrator"; +const DEFAULT_ACTIVITY_NAME = "activity"; -/** An Azure Functions activity handler used by the testing helpers. */ -export type TestActivityHandler = ( - input: TInput, - context: InvocationContext, -) => TOutput | Promise; - -/** Options for running an activity directly. */ -export interface ActivityTestOptions { - /** Function name exposed on the generated invocation context. */ - functionName?: string; - /** Invocation context to pass instead of creating one. */ - context?: InvocationContext; -} - -/** Runs one activity invocation without starting a Functions host. */ -export async function runActivity( - handler: TestActivityHandler, - input: TInput, - options: ActivityTestOptions = {}, -): Promise { - const context = options.context ?? new InvocationContext({ functionName: options.functionName ?? "activity" }); - return await handler(input, context); -} - -/** Plain failure details returned for a failed orchestration. */ -export interface OrchestrationTestFailure { - errorType: string; - message: string; - stackTrace?: string; -} - -/** Runtime states exposed by the Functions testing API. */ -export type OrchestrationTestStatus = "Pending" | "Running" | "Completed" | "Failed" | "Terminated" | "Suspended"; - -/** A deserialized snapshot of an orchestration run. */ -export interface OrchestrationTestResult { - instanceId: string; - status: OrchestrationTestStatus; - output?: TOutput; - customStatus?: unknown; - failure?: OrchestrationTestFailure; +/** + * Creates the {@link InvocationContext} an activity handler receives at runtime. + * + * @remarks Activities are ordinary Azure Functions handlers with no durable state, so they are + * tested by calling them directly: `await sayHello("World", createActivityContext("sayHello"))`. + */ +export function createActivityContext(functionName: string = DEFAULT_ACTIVITY_NAME): InvocationContext { + return new InvocationContext({ functionName }); } -/** Options shared by one-shot and harness orchestration starts. */ -export interface OrchestrationStartOptions { +/** Options for {@link runOrchestrator}. */ +export interface OrchestratorTestOptions { + /** Input passed to the orchestrator. */ input?: TInput; + /** Instance id to schedule under. Defaults to a generated id. */ instanceId?: string; - /** A past time is accepted, but future scheduled starts are not supported by the in-memory backend. */ - startAt?: Date; -} - -/** Options for a one-shot orchestrator test. */ -export interface OrchestratorTestOptions extends OrchestrationStartOptions { - activities?: Readonly>>; -} - -/** Options for an interactive orchestration harness. */ -export interface OrchestrationHarnessOptions { - /** - * Default timeout for observing a run. A timeout rejects only that wait; the harness remains live - * and owned by the caller until {@link OrchestrationHarness.dispose} finishes. - */ - timeoutMs?: number; -} - -/** A running orchestration controlled through an in-memory harness. */ -export interface OrchestrationRun { - readonly instanceId: string; - readonly status: OrchestrationTestStatus; - readonly output: TOutput | undefined; - readonly customStatus: unknown; - readonly failure: OrchestrationTestFailure | undefined; - waitForStart(timeoutMs?: number): Promise>; - waitForCompletion(timeoutMs?: number): Promise>; - raiseEvent(name: string, data?: TData): Promise; - terminate(output?: TOutputData): Promise; - suspend(): Promise; - resume(): Promise; - refresh(): Promise>; -} - -/** In-memory harness for interactive orchestration tests. */ -export interface OrchestrationHarness { - registerOrchestrator(name: string, handler: OrchestrationHandler): OrchestrationHarness; - registerActivity( - name: string, - handler: TestActivityHandler, - ): OrchestrationHarness; - start( - name: string, - options?: OrchestrationStartOptions, - ): Promise>; - /** - * Stops harness processing after in-flight orchestrator and activity handlers settle. - * - * @remarks Arbitrary JavaScript promises cannot be forcibly cancelled. If user code never - * settles, disposal also remains pending. - */ - dispose(): Promise; + /** Activity implementations the orchestrator may call, keyed by activity name. */ + activities?: Readonly>; } /** - * Creates a host-free orchestration harness backed by the real replay executor. + * The terminal state of an orchestration run. * - * @remarks Durable timers use real wall-clock time. The current core in-memory backend does not - * expose a virtual clock or timer-advance API, so tests should schedule short timer delays. - * Wait timeouts reject only the observation call; the harness remains live and must still be - * disposed. Disposal awaits in-flight orchestrator and activity handlers because arbitrary - * JavaScript promises cannot be forcibly cancelled. + * @remarks `runtimeStatus`, `output`, and `customStatus` are produced by the same mapping + * `DurableFunctionsClient.getStatus` applies at runtime, so a test asserts on exactly the values a + * deployed client would observe. */ -export function createOrchestrationHarness(options: OrchestrationHarnessOptions = {}): OrchestrationHarness { - return new InMemoryOrchestrationHarness(options.timeoutMs ?? DEFAULT_TIMEOUT_MS); +export interface OrchestrationTestResult { + instanceId: string; + runtimeStatus: OrchestrationRuntimeStatus; + output?: TOutput; + customStatus?: unknown; + /** Populated when the orchestration failed. */ + failure?: TaskFailureDetails; } /** - * Runs one orchestrator to a terminal state and always releases its worker. + * Runs one orchestrator to a terminal state on the in-memory backend and always releases its worker. + * + * @remarks Durable timers use real wall-clock time, because the in-memory backend has no virtual + * clock; tests should schedule short delays. There is no forced timeout: the helper returns only + * once the orchestration is terminal and the worker has drained, so activity code cannot keep + * mutating test state after it returns. Arbitrary JavaScript promises cannot be cancelled, so a + * handler that never settles leaves this helper pending and the test runner's timeout applies. * - * @remarks Durable timers use real wall-clock time. The current core in-memory backend does not - * expose a virtual clock or timer-advance API, so tests should schedule short timer delays. - * This helper intentionally has no forced timeout: it awaits terminal orchestration execution and - * worker cleanup before returning. Arbitrary JavaScript promises cannot be forcibly cancelled, so - * non-cooperative orchestrator or activity code remains subject to the test runner's own timeout. + * Interactive scenarios (external events, terminate, suspend/resume) are covered by driving + * `TestOrchestrationWorker` and `TestOrchestrationClient` from `@microsoft/durabletask-js` directly + * and registering handlers with {@link wrapOrchestrator}; see the package README. */ export async function runOrchestrator( handler: OrchestrationHandler, options: OrchestratorTestOptions = {}, ): Promise> { - const harness = createOrchestrationHarness({ timeoutMs: 0 }); - harness.registerOrchestrator(DEFAULT_ORCHESTRATOR_NAME, handler); + const backend = new InMemoryOrchestrationBackend(); + const worker = new TestOrchestrationWorker(backend); + const client = new TestOrchestrationClient(backend); + + worker.addNamedOrchestrator(ORCHESTRATOR_NAME, wrapOrchestrator(handler)); for (const [name, activity] of Object.entries(options.activities ?? {})) { - harness.registerActivity(name, activity); + worker.addNamedActivity(name, async (_context, input) => activity(input, createActivityContext(name))); } + await worker.start(); try { - const run = await harness.start(DEFAULT_ORCHESTRATOR_NAME, options); - return await run.waitForCompletion(); - } finally { - await harness.dispose(); - } -} - -/** One operation in a standalone entity batch. */ -export interface EntityTestOperation { - name: string; - input?: TInput; -} - -/** Options for a standalone entity batch. */ -export interface EntityTestOptions { - initialState?: TState; - operations: ReadonlyArray; - entityName?: string; - entityKey?: string; -} - -/** Deserialized state and operation return values from an entity batch. */ -export interface EntityTestResult { - state: TState | undefined; - results: Array; -} - -/** Error thrown when an entity operation in a test batch fails. */ -export class EntityOperationError extends Error { - readonly operationName: string; - readonly operationIndex: number; - readonly errorType: string; - readonly stackTrace?: string; - - constructor(operationName: string, operationIndex: number, failure: OrchestrationTestFailure) { - super(`Entity operation '${operationName}' at index ${operationIndex} failed: ${failure.message}`); - this.name = "EntityOperationError"; - this.operationName = operationName; - this.operationIndex = operationIndex; - this.errorType = failure.errorType; - this.stackTrace = failure.stackTrace; - } -} - -/** - * Runs an entity batch directly through the core entity executor. - * - * @remarks This is a standalone unit seam. It does not enqueue entity messages in the orchestration - * harness; end-to-end in-memory entity routing is not available in the current core test backend. - */ -export async function runEntity( - handler: RegisteredEntityHandler, - options: EntityTestOptions, -): Promise> { - const entityName = options.entityName ?? DEFAULT_ENTITY_NAME; - const entityKey = options.entityKey ?? DEFAULT_ENTITY_KEY; - const entityId = new EntityInstanceId(entityName, entityKey); - const executor = new TaskEntityShim(wrapEntity(handler)(), entityId); - const request = createEntityBatchRequest(options.initialState, options.operations); - const response = await executor.executeAsync(request); - - const results = response.getResultsList().map((operationResult, index) => { - const failureDetails = operationResult.getFailure()?.getFailuredetails(); - if (failureDetails) { - throw new EntityOperationError(options.operations[index].name, index, { - errorType: failureDetails.getErrortype(), - message: failureDetails.getErrormessage(), - stackTrace: failureDetails.getStacktrace()?.getValue(), - }); + const instanceId = await client.scheduleNewOrchestration(ORCHESTRATOR_NAME, options.input, options.instanceId); + // A zero timeout disables the backend's wait timer, per the no-forced-timeout remark above. + const state = await client.waitForOrchestrationCompletion(instanceId, true, 0); + if (!state) { + throw new Error(`Orchestration '${instanceId}' was not found.`); } - return deserialize(operationResult.getSuccess()?.getResult()?.getValue()); - }); - - return { - state: deserialize(response.getEntitystate()?.getValue()), - results, - }; -} - -class InMemoryOrchestrationHarness implements OrchestrationHarness { - private readonly backend = new InMemoryOrchestrationBackend(); - private readonly worker = new TestOrchestrationWorker(this.backend); - private readonly client = new TestOrchestrationClient(this.backend); - private readonly runs = new Set>(); - private transition: Promise = Promise.resolve(); - private workerStartPromise: Promise | undefined; - private workerStarted = false; - private startRequested = false; - private disposed = false; - private disposePromise: Promise | undefined; - - constructor(private readonly timeoutMs: number) {} - - registerOrchestrator(name: string, handler: OrchestrationHandler): OrchestrationHarness { - this.ensureCanRegister(); - this.worker.addNamedOrchestrator(name, wrapOrchestrator(handler)); - return this; - } - - registerActivity(name: string, handler: TestActivityHandler): OrchestrationHarness { - this.ensureCanRegister(); - this.worker.addNamedActivity(name, async (_context, input) => { - return await runActivity(handler, input as TInput, { functionName: name }); - }); - return this; - } - start( - name: string, - options: OrchestrationStartOptions = {}, - ): Promise> { - this.startRequested = true; - return this.enqueueTransition(async () => { - this.ensureNotDisposed(); - validateStartAt(options.startAt); - await this.ensureWorkerStarted(); - this.ensureNotDisposed(); - - const instanceId = await this.client.scheduleNewOrchestration( - name, - options.input, - options.instanceId, - options.startAt, - ); - this.ensureNotDisposed(); - - const run = new InMemoryOrchestrationRun(instanceId, this.client, this.timeoutMs); - this.runs.add(run as InMemoryOrchestrationRun); - return run; - }); - } - - dispose(): Promise { - if (!this.disposePromise) { - this.disposed = true; - for (const run of this.runs) { - run.markDisposed(); - } - this.disposePromise = this.enqueueTransition(() => this.disposeCore()); - } - return this.disposePromise; - } - - private async disposeCore(): Promise { - await this.workerStartPromise?.catch(() => undefined); - try { - if (this.workerStarted) { - await this.worker.stop(); - } - await Promise.all([...this.runs].map((run) => run.captureTerminalSnapshot())); - } finally { - this.backend.reset(); - await this.client.stop(); - } - } - - private ensureCanRegister(): void { - this.ensureNotDisposed(); - if (this.startRequested) { - throw new Error("Orchestrators and activities must be registered before the first run starts."); - } - } - - private ensureNotDisposed(): void { - if (this.disposed) { - throw createHarnessDisposedError(); - } - } - - private ensureWorkerStarted(): Promise { - if (!this.workerStartPromise) { - this.workerStartPromise = this.worker.start().then(() => { - this.workerStarted = true; - }); - } - return this.workerStartPromise; - } - - private enqueueTransition(operation: () => Promise): Promise { - const result = this.transition.then(operation); - this.transition = result.then( - () => undefined, - () => undefined, - ); - return result; - } -} - -class InMemoryOrchestrationRun implements OrchestrationRun { - private currentStatus: OrchestrationTestStatus = "Pending"; - private currentOutput: TOutput | undefined; - private currentCustomStatus: unknown; - private currentFailure: OrchestrationTestFailure | undefined; - private terminalResult: OrchestrationTestResult | undefined; - private disposed = false; - - constructor( - readonly instanceId: string, - private readonly client: TestOrchestrationClient, - private readonly timeoutMs: number, - ) {} - - get status(): OrchestrationTestStatus { - return this.currentStatus; - } - - get output(): TOutput | undefined { - return this.currentOutput; - } - - get customStatus(): unknown { - return this.currentCustomStatus; - } - - get failure(): OrchestrationTestFailure | undefined { - return this.currentFailure; - } - - async waitForStart(timeoutMs: number = this.timeoutMs): Promise> { - const state = await this.executeWhileActive(() => - this.client.waitForOrchestrationStart(this.instanceId, true, timeoutMs / 1000), - ); - return this.applyState(requireState(state, this.instanceId)); - } - - async waitForCompletion(timeoutMs: number = this.timeoutMs): Promise> { - if (this.terminalResult) { - return this.terminalResult; - } - if (this.disposed) { - throw createHarnessDisposedError(); - } - - try { - const state = await this.client.waitForOrchestrationCompletion(this.instanceId, true, timeoutMs / 1000); - return this.applyState(requireState(state, this.instanceId)); - } catch (error) { - if (this.disposed) { - if (this.terminalResult) { - return this.terminalResult; - } - throw createHarnessDisposedError(); - } - throw error; - } - } - - async raiseEvent(name: string, data?: TData): Promise { - await this.executeWhileActive(() => this.client.raiseOrchestrationEvent(this.instanceId, name, data)); - } - - async terminate(output?: TOutputData): Promise { - await this.executeWhileActive(() => this.client.terminateOrchestration(this.instanceId, output)); - } - - async suspend(): Promise { - await this.executeWhileActive(() => this.client.suspendOrchestration(this.instanceId)); - await this.refresh(); - } - - async resume(): Promise { - await this.executeWhileActive(() => this.client.resumeOrchestration(this.instanceId)); - await this.refresh(); - } - - async refresh(): Promise> { - const state = await this.executeWhileActive(() => this.client.getOrchestrationState(this.instanceId, true)); - return this.applyState(requireState(state, this.instanceId)); - } - - markDisposed(): void { - this.disposed = true; - } - - async captureTerminalSnapshot(): Promise { - if (this.terminalResult) { - return; - } - const state = await this.client.getOrchestrationState(this.instanceId, true); - if (state && isTerminalStatus(state.runtimeStatus)) { - this.applyState(state); - } - } - - private applyState(state: OrchestrationState): OrchestrationTestResult { - this.currentStatus = toTestStatus(state.runtimeStatus); - this.currentOutput = deserialize(state.serializedOutput); - this.currentCustomStatus = deserialize(state.serializedCustomStatus); - this.currentFailure = state.failureDetails - ? { - errorType: state.failureDetails.errorType, - message: state.failureDetails.message, - stackTrace: state.failureDetails.stackTrace, - } - : undefined; - - const result = { - instanceId: this.instanceId, - status: this.currentStatus, - output: this.currentOutput, - customStatus: this.currentCustomStatus, - failure: this.currentFailure, + const status = toDurableOrchestrationStatus(state); + return { + instanceId: status.instanceId, + runtimeStatus: status.runtimeStatus, + output: status.output as TOutput | undefined, + customStatus: status.customStatus, + failure: state.failureDetails + ? { + errorType: state.failureDetails.errorType, + message: state.failureDetails.message, + stackTrace: state.failureDetails.stackTrace, + } + : undefined, }; - if (isTerminalStatus(state.runtimeStatus)) { - this.terminalResult = result; - } - return result; - } - - private async executeWhileActive(operation: () => Promise): Promise { - this.ensureActive(); - try { - const result = await operation(); - this.ensureActive(); - return result; - } catch (error) { - if (this.disposed) { - throw createHarnessDisposedError(); - } - throw error; - } - } - - private ensureActive(): void { - if (this.disposed) { - throw createHarnessDisposedError(); - } - } -} - -function createEntityBatchRequest( - initialState: TState | undefined, - operations: ReadonlyArray, -): Parameters[0] { - // TaskEntityShim reads only these protobuf accessors. Keeping this adapter local avoids exposing - // generated protocol types through the durable-functions testing API. - const request = { - getEntitystate: () => serializedStateValue(initialState), - getOperationsList: () => - operations.map((operation) => ({ - getOperation: () => operation.name, - getInput: () => serializedValue(operation.input), - })), - }; - return request as Parameters[0]; -} - -function serializedValue(value: unknown): { getValue(): string } | undefined { - if (value === undefined) { - return undefined; - } - const serialized = JSON.stringify(value); - if (serialized === undefined) { - throw new TypeError("Test values must be JSON-serializable."); - } - return { getValue: () => serialized }; -} - -function serializedStateValue(value: unknown): { getValue(): string } | undefined { - return value == null ? undefined : serializedValue(value); -} - -function deserialize(value: string | undefined): T | undefined { - return value === undefined ? undefined : (JSON.parse(value) as T); -} - -function requireState(state: OrchestrationState | undefined, instanceId: string): OrchestrationState { - if (!state) { - throw new Error(`Orchestration '${instanceId}' was not found.`); - } - return state; -} - -function toTestStatus(status: OrchestrationStatus): OrchestrationTestStatus { - switch (status) { - case OrchestrationStatus.PENDING: - return "Pending"; - case OrchestrationStatus.RUNNING: - return "Running"; - case OrchestrationStatus.COMPLETED: - return "Completed"; - case OrchestrationStatus.FAILED: - return "Failed"; - case OrchestrationStatus.TERMINATED: - return "Terminated"; - case OrchestrationStatus.SUSPENDED: - return "Suspended"; - case OrchestrationStatus.CONTINUED_AS_NEW: - return "Running"; - default: - return "Running"; - } -} - -function isTerminalStatus(status: OrchestrationStatus): boolean { - return ( - status === OrchestrationStatus.COMPLETED || - status === OrchestrationStatus.FAILED || - status === OrchestrationStatus.TERMINATED - ); -} - -function validateStartAt(startAt: Date | undefined): void { - if (!startAt) { - return; - } - if (Number.isNaN(startAt.getTime())) { - throw new TypeError("startAt must be a valid Date."); - } - if (startAt.getTime() > Date.now()) { - throw new Error("Future startAt values are not supported by the in-memory orchestration harness."); + } finally { + // Stopping the worker drains the in-flight orchestrator/activity handler; resetting the backend + // clears durable timers that would otherwise keep the process alive after the test. + await worker.stop(); + backend.reset(); } } - -function createHarnessDisposedError(): Error { - return new Error(HARNESS_DISPOSED_MESSAGE); -} - -/** Namespace-style access matching the rest of the Durable Functions API. */ -export const test = { - createOrchestrationHarness, - runActivity, - runEntity, - runOrchestrator, -}; diff --git a/packages/azure-functions-durable/test/unit/testing.spec.ts b/packages/azure-functions-durable/test/unit/testing.spec.ts index 1a48bca0..d234fb52 100644 --- a/packages/azure-functions-durable/test/unit/testing.spec.ts +++ b/packages/azure-functions-durable/test/unit/testing.spec.ts @@ -2,435 +2,190 @@ // Licensed under the MIT License. import { InvocationContext } from "@azure/functions"; -import { TaskEntity, TestOrchestrationWorker } from "@microsoft/durabletask-js"; import { - EntityOperationError, - createOrchestrationHarness, - runActivity, - runEntity, - runOrchestrator, - test, -} from "../../src/testing"; -import type { EntityHandler, OrchestrationContext, OrchestrationHandler } from "../../src"; + InMemoryOrchestrationBackend, + TestOrchestrationClient, + TestOrchestrationWorker, +} from "@microsoft/durabletask-js"; +import { OrchestrationRuntimeStatus, toDurableOrchestrationStatus, wrapOrchestrator } from "../../src"; +import type { OrchestrationContext, OrchestrationHandler } from "../../src"; +import { createActivityContext, runOrchestrator } from "../../src/testing"; describe("durable-functions/testing", () => { - it("runs an activity with a Functions invocation context", async () => { - const result = await runActivity( - (input: string, context: InvocationContext) => `${context.functionName}:${input}`, - "World", - { functionName: "sayHello" }, - ); - - expect(result).toBe("sayHello:World"); - }); - - it("runs a classic orchestrator with inline Functions-style activities", async () => { - const orchestrator: OrchestrationHandler = function* ( - context: OrchestrationContext, - ): Generator { - const name = context.df.getInput(); - return yield context.df.callActivity("sayHello", name); - }; + describe("createActivityContext", () => { + it("builds the invocation context an activity handler receives", async () => { + const sayHello = (name: string, context: InvocationContext) => `${context.functionName}: Hello, ${name}!`; - const result = await runOrchestrator(orchestrator, { - input: "World", - activities: { - sayHello: (input: unknown) => `Hello, ${String(input)}!`, - }, - instanceId: "one-shot", - }); - - expect(result).toMatchObject({ - instanceId: "one-shot", - status: "Completed", - output: "Hello, World!", + expect(await sayHello("World", createActivityContext("sayHello"))).toBe("sayHello: Hello, World!"); + expect(createActivityContext().functionName).toBe("activity"); }); }); - it("returns deserialized custom status and failure details", async () => { - const orchestrator: OrchestrationHandler = function* ( - context: OrchestrationContext, - ): Generator { - context.df.setCustomStatus({ phase: "starting" }); - yield context.df.callActivity("fail"); - }; + describe("runOrchestrator", () => { + it("runs a classic orchestrator against inline activities", async () => { + const orchestrator: OrchestrationHandler = function* ( + context: OrchestrationContext, + ): Generator { + const name = context.df.getInput(); + return yield context.df.callActivity("sayHello", name); + }; - const result = await runOrchestrator(orchestrator, { - activities: { - fail: () => { - throw new TypeError("activity failed"); + const result = await runOrchestrator(orchestrator, { + input: "World", + instanceId: "one-shot", + activities: { + sayHello: (name: unknown) => `Hello, ${String(name)}!`, }, - }, - }); + }); - expect(result.status).toBe("Failed"); - expect(result.customStatus).toEqual({ phase: "starting" }); - expect(result.failure).toMatchObject({ - errorType: expect.any(String), - message: expect.stringContaining("activity failed"), + expect(result).toEqual({ + instanceId: "one-shot", + runtimeStatus: OrchestrationRuntimeStatus.Completed, + output: "Hello, World!", + customStatus: undefined, + failure: undefined, + }); }); - }); - it("does not return before a delayed activity settles", async () => { - let activityStarted!: () => void; - let activityFinished!: () => void; - const started = new Promise((resolve) => { - activityStarted = resolve; - }); - const finished = new Promise((resolve) => { - activityFinished = resolve; - }); - const mutations: string[] = []; - const orchestrator: OrchestrationHandler = function* ( - context: OrchestrationContext, - ): Generator { - yield context.df.callActivity("never"); - }; + it("names each activity's invocation context after the registered activity", async () => { + const orchestrator: OrchestrationHandler = function* ( + context: OrchestrationContext, + ): Generator { + return yield context.df.callActivity("whoAmI"); + }; - let executionSettled = false; - const execution = runOrchestrator(orchestrator, { - activities: { - never: () => { - activityStarted(); - return new Promise((resolve) => { - setTimeout(() => { - mutations.push("activity"); - activityFinished(); - resolve(); - }, 100); - }); + const result = await runOrchestrator(orchestrator, { + activities: { + whoAmI: (_input: unknown, context: InvocationContext) => context.functionName, }, - }, - }).finally(() => { - executionSettled = true; - }); - await started; - - await new Promise((resolve) => setTimeout(resolve, 50)); - expect(executionSettled).toBe(false); - - await finished; - await expect(execution).resolves.toMatchObject({ status: "Completed" }); - expect(mutations).toEqual(["activity"]); - - await new Promise((resolve) => setTimeout(resolve, 25)); - expect(mutations).toEqual(["activity"]); - }); - - it("does not return before a delayed async orchestrator settles", async () => { - let settled = false; - const execution = runOrchestrator(async () => { - await new Promise((resolve) => setTimeout(resolve, 75)); - settled = true; - return "done"; - }); - - await new Promise((resolve) => setTimeout(resolve, 25)); - expect(settled).toBe(false); - - await expect(execution).resolves.toMatchObject({ status: "Completed", output: "done" }); - expect(settled).toBe(true); - }); - - it("serializes concurrent harness starts through one worker startup", async () => { - const harness = createOrchestrationHarness(); - harness.registerOrchestrator("ready", async (_context, input) => input); - - try { - const [first, second] = await Promise.all([ - harness.start("ready", { instanceId: "first", input: 1 }), - harness.start("ready", { instanceId: "second", input: 2 }), - ]); - - await expect(first.waitForCompletion()).resolves.toMatchObject({ output: 1 }); - await expect(second.waitForCompletion()).resolves.toMatchObject({ output: 2 }); - } finally { - await harness.dispose(); - } - }); + }); - it("stops a worker when disposal races its startup", async () => { - let releaseStart!: () => void; - const startGate = new Promise((resolve) => { - releaseStart = resolve; + expect(result.output).toBe("whoAmI"); }); - const startSpy = jest.spyOn(TestOrchestrationWorker.prototype, "start").mockImplementation(async () => startGate); - const stopSpy = jest.spyOn(TestOrchestrationWorker.prototype, "stop").mockResolvedValue(); - const harness = createOrchestrationHarness(); - harness.registerOrchestrator("ready", async () => "done"); - try { - const starting = harness.start("ready"); - await Promise.resolve(); - const disposing = harness.dispose(); - releaseStart(); + it("reports failures as a terminal status with custom status and failure details", async () => { + const orchestrator: OrchestrationHandler = function* ( + context: OrchestrationContext, + ): Generator { + context.df.setCustomStatus({ phase: "starting" }); + yield context.df.callActivity("fail"); + }; - await expect(starting).rejects.toThrow("The orchestration harness has been disposed."); - await disposing; - expect(startSpy).toHaveBeenCalledTimes(1); - expect(stopSpy).toHaveBeenCalledTimes(1); - } finally { - startSpy.mockRestore(); - stopSpy.mockRestore(); - } - }); - - it("rejects future scheduled starts that the in-memory backend cannot defer", async () => { - const harness = createOrchestrationHarness(); - harness.registerOrchestrator("ready", async () => "done"); - - try { - await expect(harness.start("ready", { startAt: new Date(Date.now() + 60_000) })).rejects.toThrow( - "Future startAt values are not supported", - ); - } finally { - await harness.dispose(); - } - }); - - it("drives external events through an orchestration harness", async () => { - const harness = createOrchestrationHarness(); - harness.registerOrchestrator("approval", function* (context: OrchestrationContext): Generator< - unknown, - { approved: boolean }, - boolean - > { - const approved = yield context.df.waitForExternalEvent("approved"); - return { approved }; - }); - - try { - const run = await harness.start<{ approved: boolean }>("approval", { - instanceId: "approval-1", + const result = await runOrchestrator(orchestrator, { + activities: { + fail: () => { + throw new TypeError("activity failed"); + }, + }, }); - await run.waitForStart(); - - expect(run.status).toBe("Running"); - - await run.raiseEvent("approved", true); - const result = await run.waitForCompletion(); - - expect(result.output).toEqual({ approved: true }); - expect(run.output).toEqual({ approved: true }); - } finally { - await harness.dispose(); - } - }); - it("supports real-time durable timers", async () => { - const harness = createOrchestrationHarness({ timeoutMs: 1000 }); - harness.registerOrchestrator("timer", function* (context: OrchestrationContext): Generator< - unknown, - string, - unknown - > { - yield context.df.createTimer(new Date(context.df.currentUtcDateTime.getTime() + 10)); - return "timer fired"; - }); - - try { - const run = await harness.start("timer"); - await expect(run.waitForCompletion()).resolves.toMatchObject({ - status: "Completed", - output: "timer fired", + expect(result.runtimeStatus).toBe(OrchestrationRuntimeStatus.Failed); + expect(result.customStatus).toEqual({ phase: "starting" }); + expect(result.failure).toMatchObject({ + errorType: expect.any(String), + message: expect.stringContaining("activity failed"), }); - } finally { - await harness.dispose(); - } - }); - - it("terminates, suspends, and resumes orchestration runs", async () => { - const harness = createOrchestrationHarness(); - harness.registerOrchestrator("interactive", function* (context: OrchestrationContext): Generator< - unknown, - string, - unknown - > { - yield context.df.waitForExternalEvent("finish"); - return "completed"; }); - try { - const suspended = await harness.start("interactive", { instanceId: "suspended" }); - await suspended.waitForStart(); - await suspended.suspend(); - expect(suspended.status).toBe("Suspended"); - await suspended.resume(); - expect(suspended.status).toBe("Running"); - - const terminated = await harness.start("interactive", { instanceId: "terminated" }); - await terminated.waitForStart(); - await terminated.terminate({ reason: "test" }); - const result = await terminated.waitForCompletion(); - - expect(result).toMatchObject({ - status: "Terminated", - output: { reason: "test" }, + it("does not return before a delayed activity settles", async () => { + let activityStarted!: () => void; + const started = new Promise((resolve) => { + activityStarted = resolve; + }); + const mutations: string[] = []; + const orchestrator: OrchestrationHandler = function* ( + context: OrchestrationContext, + ): Generator { + yield context.df.callActivity("slow"); + }; + + let executionSettled = false; + const execution = runOrchestrator(orchestrator, { + activities: { + slow: async () => { + activityStarted(); + await new Promise((resolve) => setTimeout(resolve, 100)); + mutations.push("activity"); + }, + }, + }).finally(() => { + executionSettled = true; }); - } finally { - await harness.dispose(); - } - }); - - it("retains terminal results but rejects other run operations after disposal", async () => { - const harness = createOrchestrationHarness({ timeoutMs: 20 }); - harness.registerOrchestrator("ready", async () => ({ done: true })); - const run = await harness.start<{ done: boolean }>("ready"); - const completed = await run.waitForCompletion(); - - await harness.dispose(); - - await expect(run.waitForCompletion()).resolves.toEqual(completed); - expect(run.output).toEqual({ done: true }); - await expect(run.waitForStart()).rejects.toThrow("The orchestration harness has been disposed."); - await expect(run.refresh()).rejects.toThrow("The orchestration harness has been disposed."); - await expect(run.raiseEvent("ignored")).rejects.toThrow("The orchestration harness has been disposed."); - await expect(run.terminate()).rejects.toThrow("The orchestration harness has been disposed."); - await expect(run.suspend()).rejects.toThrow("The orchestration harness has been disposed."); - await expect(run.resume()).rejects.toThrow("The orchestration harness has been disposed."); - }); - - it("rejects nonterminal run operations immediately after disposal", async () => { - const harness = createOrchestrationHarness({ timeoutMs: 20 }); - harness.registerOrchestrator("waiting", function* (context: OrchestrationContext): Generator< - unknown, - void, - unknown - > { - yield context.df.waitForExternalEvent("never"); - }); - const run = await harness.start("waiting"); - await run.waitForStart(); - - await harness.dispose(); - - await expect(run.waitForCompletion()).rejects.toThrow("The orchestration harness has been disposed."); - await expect(run.refresh()).rejects.toThrow("The orchestration harness has been disposed."); - }); - - it("does not finish disposal while activity code is still running", async () => { - let activityStarted!: () => void; - const started = new Promise((resolve) => { - activityStarted = resolve; - }); - let activityFinished = false; - const harness = createOrchestrationHarness({ timeoutMs: 20 }); - harness.registerActivity("slow", async () => { - activityStarted(); - await new Promise((resolve) => setTimeout(resolve, 100)); - activityFinished = true; - }); - harness.registerOrchestrator("slow", function* (context: OrchestrationContext): Generator { - yield context.df.callActivity("slow"); - }); - - const run = await harness.start("slow"); - await started; - await expect(run.waitForCompletion()).rejects.toThrow("Timeout waiting for orchestration"); - - let disposalSettled = false; - const disposal = harness.dispose().finally(() => { - disposalSettled = true; - }); - await new Promise((resolve) => setTimeout(resolve, 25)); - expect(disposalSettled).toBe(false); - expect(activityFinished).toBe(false); - - await disposal; - expect(activityFinished).toBe(true); - }); - it("runs a classic entity batch and deserializes state and operation results", async () => { - const entity: EntityHandler = (context) => { - const current = context.df.getState(() => 0) ?? 0; - switch (context.df.operationName) { - case "add": - context.df.setState(current + (context.df.getInput() ?? 0)); - break; - case "get": - context.df.return(current); - break; - } - }; + await started; + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(executionSettled).toBe(false); - const result = await runEntity(entity, { - initialState: 2, - entityName: "Counter", - entityKey: "Key", - operations: [{ name: "add", input: 3 }, { name: "get" }], - }); + await expect(execution).resolves.toMatchObject({ runtimeStatus: OrchestrationRuntimeStatus.Completed }); + expect(mutations).toEqual(["activity"]); - expect(result).toEqual({ - state: 5, - results: [undefined, 5], + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(mutations).toEqual(["activity"]); }); - }); - it("runs a core-native entity factory without exposing the core executor", async () => { - class Counter extends TaskEntity { - protected initializeState(): number { - return 0; - } + it("does not return before a delayed async orchestrator settles", async () => { + let settled = false; + const execution = runOrchestrator(async () => { + await new Promise((resolve) => setTimeout(resolve, 75)); + settled = true; + return "done"; + }); - add(value: number): number { - this.state += value; - return this.state; - } - } + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(settled).toBe(false); - const result = await runEntity(() => new Counter(), { - operations: [{ name: "add", input: 4 }], + await expect(execution).resolves.toMatchObject({ + runtimeStatus: OrchestrationRuntimeStatus.Completed, + output: "done", + }); + expect(settled).toBe(true); }); - expect(result).toEqual({ state: 4, results: [4] }); - }); + it("clears durable timers so a completed run leaves nothing pending", async () => { + const orchestrator: OrchestrationHandler = function* ( + context: OrchestrationContext, + ): Generator { + yield context.df.createTimer(new Date(context.df.currentUtcDateTime.getTime() + 10)); + return "timer fired"; + }; - it("treats null initial entity state as absent while preserving null operation input", async () => { - const entity: EntityHandler = (context) => { - context.df.return({ - input: context.df.getInput(), - isNewlyConstructed: context.df.isNewlyConstructed, - state: context.df.getState(() => 7), + await expect(runOrchestrator(orchestrator)).resolves.toMatchObject({ + runtimeStatus: OrchestrationRuntimeStatus.Completed, + output: "timer fired", }); - }; - - const result = await runEntity(entity, { - initialState: null, - operations: [{ name: "inspect", input: null }], - }); - - expect(result).toEqual({ - state: undefined, - results: [{ input: null, isNewlyConstructed: true, state: 7 }], }); }); - it("surfaces entity operation failures without committing their state", async () => { - const entity: EntityHandler = (context) => { - context.df.setState(99); - throw new RangeError("invalid operation"); - }; + // Interactive scenarios intentionally have no dedicated wrapper: the core in-memory stack already + // exposes them. This pins the pattern the README documents. + it("drives external events through the core in-memory stack", async () => { + const backend = new InMemoryOrchestrationBackend(); + const worker = new TestOrchestrationWorker(backend); + const client = new TestOrchestrationClient(backend); - await expect( - runEntity(entity, { - initialState: 1, - operations: [{ name: "break" }], + worker.addNamedOrchestrator( + "approval", + wrapOrchestrator(function* (context: OrchestrationContext): Generator { + const approved = yield context.df.waitForExternalEvent("approved"); + return { approved }; }), - ).rejects.toMatchObject({ - name: "EntityOperationError", - operationName: "break", - operationIndex: 0, - errorType: "RangeError", - message: expect.stringContaining("invalid operation"), - } satisfies Partial); - }); + ); + await worker.start(); - it("exposes the helpers through the test namespace", () => { - expect(test).toEqual({ - createOrchestrationHarness, - runActivity, - runEntity, - runOrchestrator, - }); + try { + const instanceId = await client.scheduleNewOrchestration("approval", undefined, "approval-1"); + await client.waitForOrchestrationStart(instanceId, true, 10); + await client.raiseOrchestrationEvent(instanceId, "approved", true); + const state = await client.waitForOrchestrationCompletion(instanceId, true, 10); + + expect(state).toBeDefined(); + expect(toDurableOrchestrationStatus(state!)).toMatchObject({ + runtimeStatus: OrchestrationRuntimeStatus.Completed, + output: { approved: true }, + }); + } finally { + await worker.stop(); + backend.reset(); + } }); });