[vitest-pool-workers] Workflows test support (#10494)

* Add Workflow instances DOs (Engines) to vitest-pool-workers worker

* Get waitUntil working

* skip sleep

* skip sleep works now

* InstanceModifier is now inside worflows-shared with direct access to the engine

* add dynamic engine binding

* add forceEventTimeout test support

* Strip out engine bindings from user facing env

* Prevent waitUntil from crashing tests after 5s

* Add mockEvent

* fix: prevent forceEventTimeot from not working if a mockEvent gets called afterwards

* Add mockStepResult

* Allow list of sleep steps in disableSleeps()

* Add mockStepError and cleanUp. Improve waitForStepResult and waitForStatus

* Add forceStepTimeout

* Make mockEvent not crash if instance has not been created yet

* Update abort for testing

* prevent the same step from being mocked multiple times

* check edge cases for wairForStatus - prevents user from waiting indefinitely

* Add workflows plugin name to loopback

* Fix force step timeout

* Make it possible to import from workflows-shared

* Allow step errors and timeouts to be set and be performed in the correct order

* Add introspectWorkflow API with proxies

* Expose APIs with some documentation

* Add workflow tests with introspection

* Fix test

* Remove Vite comment

* Fixing documentation

* fixtures READMEs updated

* Add changeset

* Prettier fix

* Update bundle.mjs file

* Put workflows-shared as a devDependency

* Change waitUntil timeouts to 1 second

* Fix createBatch interception from calling this.create

* Add review comment suggestions

* Apply suggestions from code review - changelog file

Co-authored-by: Somhairle MacLeòid <[email protected]>

* Add review suggestions (miniflare do plugin, dispose introspectors)

* Support isolated storage and remove Engine DO bindings to runnerWorker

* Update changeset

* Add team review suggestions

* Update fixtures example to match a simple workflow example

* Remove @ts-expect-error and add some documentation

* Change cleanup to dispose. Apply suggestions.

---------

Co-authored-by: Sid Chatterjee <[email protected]>
Co-authored-by: Somhairle MacLeòid <[email protected]>

Author: 3 people

.changeset/nice-carrots-sink.md

@@ -0,0 +1,5 @@
+---
+"miniflare": patch
+---
+
+Include workflow bining name in workflow plugin.

.changeset/public-women-exist.md

@@ -0,0 +1,11 @@
+---
+"@cloudflare/vitest-pool-workers": minor
+---
+
+Add Workflows test support to the `cloudflare:test` module.
+
+The `cloudflare:test` module has two new APIs:
+
+- `introspectWorkflowInstance`
+- `introspectWorkflow`
+  which allow changing the behavior of one or multiple Workflow instances created during tests.

fixtures/vitest-pool-workers-examples/README.md

@@ -10,6 +10,7 @@ This directory contains example projects tested with `@cloudflare/vitest-pool-wo
 | [📦 kv-r2-caches](kv-r2-caches)                                                    | Isolated tests using KV, R2 and the Cache API             |
 | [📚 d1](d1)                                                                        | Isolated tests using D1 with migrations                   |
 | [📌 durable-objects](durable-objects)                                              | Isolated tests using Durable Objects with direct access   |
+| [🔁 workflows](workflows)                                                          | Tests using Workflows                                     |
 | [🚥 queues](queues)                                                                | Tests using Queue producers and consumers                 |
 | [🚰 pipelines](pipelines)                                                          | Tests using Pipelines                                     |
 | [🚀 hyperdrive](hyperdrive)                                                        | Tests using Hyperdrive with a Vitest managed TCP server   |

fixtures/vitest-pool-workers-examples/workflows/README.md

@@ -1,4 +1,9 @@
 # 🔁 workflows
 
-Unit testing of workflows themselves is not possible yet - but you can still
-trigger them in unit tests.
+This Worker includes a ModeratorWorkflow that serves as a template for an automated content moderation process.
+The testing suite uses workflow mocking to validate the logic of each step.
+
+| Test                                            | Overview                                                                                  |
+| ----------------------------------------------- | ----------------------------------------------------------------------------------------- |
+| [integration.test.ts](test/integration.test.ts) | Tests on the Worker's endpoints, ensuring that workflows are created and run correctly.   |
+| [unit.test.ts](test/unit.test.ts)               | Tests on the internal logic of each workflow. It uses mocking to test steps in isolation. |

fixtures/vitest-pool-workers-examples/workflows/src/env.d.ts

@@ -1,3 +1,6 @@
-interface Env {
-	TEST_WORKFLOW: Workflow;
+declare namespace Cloudflare {
+	interface Env {
+		MODERATOR: Workflow;
+	}
 }
+interface Env extends Cloudflare.Env {}

fixtures/vitest-pool-workers-examples/workflows/src/index.ts

@@ -1,28 +1,89 @@
 import {
-	WorkerEntrypoint,
 	WorkflowEntrypoint,
 	WorkflowEvent,
 	WorkflowStep,
 } from "cloudflare:workers";
 
-export class TestWorkflow extends WorkflowEntrypoint<Env> {
+export class ModeratorWorkflow extends WorkflowEntrypoint<Env> {
 	async run(_event: Readonly<WorkflowEvent<unknown>>, step: WorkflowStep) {
-		console.log("ola");
-		return "test-workflow";
+		await step.sleep("sleep for a while", "10 seconds");
+
+		// Get an initial analysis from an AI model
+		const aiResult = await step.do("AI content scan", async () => {
+			// Call to an workers-ai to scan the text content and return a violation score
+
+			// Simulated score:
+			const violationScore = Math.floor(Math.random() * 100);
+
+			return { violationScore: violationScore };
+		});
+
+		// Triage based on the AI score
+		if (aiResult.violationScore < 10) {
+			await step.do("auto approve content", async () => {
+				// API call to set app content status to "approved"
+				return { status: "auto_approved" };
+			});
+			return { status: "auto_approved" };
+		}
+		if (aiResult.violationScore > 90) {
+			await step.do("auto reject content", async () => {
+				// API call to set app content status to "rejected"
+				return { status: "auto_rejected" };
+			});
+			return { status: "auto_rejected" };
+		}
+
+		// If the score is ambiguous, require human review
+		type EventPayload = {
+			moderatorAction: string;
+		};
+		const eventPayload = await step.waitForEvent<EventPayload>("human review", {
+			type: "moderation-decision",
+			timeout: "1 day",
+		});
+
+		if (eventPayload) {
+			// The moderator responded in time.
+			const decision = eventPayload.payload.moderatorAction; // e.g., "approve" or "reject"
+			await step.do("apply moderator decision", async () => {
+				// API call to update content status based on the decision
+				return { status: "moderated", decision: decision };
+			});
+			return { status: "moderated", decision: decision };
+		}
 	}
 }
 
-export default class TestNamedEntrypoint extends WorkerEntrypoint<Env> {
-	async fetch(request: Request) {
-		const maybeId = new URL(request.url).searchParams.get("id");
+export default {
+	async fetch(request: Request, env: Env) {
+		const url = new URL(request.url);
+		const maybeId = url.searchParams.get("id");
 		if (maybeId !== null) {
-			const instance = await this.env.TEST_WORKFLOW.get(maybeId);
+			const instance = await env.MODERATOR.get(maybeId);
 
 			return Response.json(await instance.status());
 		}
 
-		const workflow = await this.env.TEST_WORKFLOW.create();
+		if (url.pathname === "/moderate") {
+			const workflow = await env.MODERATOR.create();
+			return Response.json({
+				id: workflow.id,
+				details: await workflow.status(),
+			});
+		}
 
-		return new Response(JSON.stringify({ id: workflow.id }));
-	}
-}
+		if (url.pathname === "/moderate-batch") {
+			const workflows = await env.MODERATOR.createBatch([
+				{},
+				{ id: "321" },
+				{},
+			]);
+
+			const ids = workflows.map((workflow) => workflow.id);
+			return Response.json({ ids: ids });
+		}
+
+		return new Response("Not found", { status: 404 });
+	},
+};

fixtures/vitest-pool-workers-examples/workflows/test/env.d.ts

@@ -1,6 +1,4 @@
 declare module "cloudflare:test" {
 	// Controls the type of `import("cloudflare:test").env`
-	interface ProvidedEnv extends Env {
-		TEST_WORKFLOW: Workflow;
-	}
+	interface ProvidedEnv extends Env {}
 }

fixtures/vitest-pool-workers-examples/workflows/test/integration-self.test.ts

@@ -0,0 +1,62 @@
+import { env, introspectWorkflow, SELF } from "cloudflare:test";
+import { expect, it } from "vitest";
+
+const STATUS_COMPLETE = "complete";
+const STEP_NAME = "AI content scan";
+const mockResult = { violationScore: 0 };
+
+// This example implicitly disposes the Workflow instance
+it("workflow should be able to reach the end and be successful", async () => {
+	// CONFIG with `await using` to ensure Workflow instances cleanup:
+	await using introspector = await introspectWorkflow(env.MODERATOR);
+	await introspector.modifyAll(async (m) => {
+		await m.disableSleeps();
+		await m.mockStepResult({ name: STEP_NAME }, mockResult);
+	});
+
+	await SELF.fetch(`https://mock-worker.local/moderate`);
+
+	const instances = introspector.get();
+	expect(instances.length).toBe(1);
+
+	// ASSERTIONS:
+	const instance = instances[0];
+	expect(await instance.waitForStepResult({ name: STEP_NAME })).toEqual(
+		mockResult
+	);
+	await expect(instance.waitForStatus(STATUS_COMPLETE)).resolves.not.toThrow();
+
+	// DISPOSE: ensured by `await using`
+});
+
+// This example explicitly disposes the Workflow instances
+it("workflow batch should be able to reach the end and be successful", async () => {
+	// CONFIG:
+	let introspector = await introspectWorkflow(env.MODERATOR);
+	try {
+		await introspector.modifyAll(async (m) => {
+			await m.disableSleeps();
+			await m.mockStepResult({ name: STEP_NAME }, mockResult);
+		});
+
+		await SELF.fetch(`https://mock-worker.local/moderate-batch`);
+
+		const instances = introspector.get();
+		expect(instances.length).toBe(3);
+
+		// ASSERTIONS:
+		for (const instance of instances) {
+			expect(await instance.waitForStepResult({ name: STEP_NAME })).toEqual(
+				mockResult
+			);
+			await expect(
+				instance.waitForStatus(STATUS_COMPLETE)
+			).resolves.not.toThrow();
+		}
+	} finally {
+		// DISPOSE:
+		// Workflow introspector should be disposed the end of each test, if no `await using` dyntax is used
+		// Also disposes all intercepted instances
+		await introspector.dispose();
+	}
+});