Add workflow tests with introspection

Author: pombosilva

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

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

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

@@ -7,22 +7,79 @@ import {
 
 export class TestWorkflow extends WorkflowEntrypoint<Env> {
 	async run(_event: Readonly<WorkflowEvent<unknown>>, step: WorkflowStep) {
-		console.log("ola");
+		console.log("Starting running...");
+
+		await step.do("step one", async () => {
+			return "result of step one";
+		});
+
 		return "test-workflow";
 	}
 }
 
-export default class TestNamedEntrypoint extends WorkerEntrypoint<Env> {
-	async fetch(request: Request) {
+export class TestLongWorkflow extends WorkflowEntrypoint<Env, Params> {
+	async run(event: Readonly<WorkflowEvent<unknown>>, step: WorkflowStep) {
+		console.log("Starting running...");
+
+		await step.sleep("sleep for a while", "10 seconds");
+
+		await step.do(
+			"my step",
+			{
+				retries: {
+					limit: 5,
+					delay: 50,
+					backoff: "constant",
+				},
+				timeout: "1 second",
+			},
+			async () => {
+				console.log("if my outcome gets mocked, this shouldn't be logging");
+				return "result of my step";
+			}
+		);
+
+		await step.sleep("sleep for a day", "8 hours");
+
+		if (event.payload === "run event") {
+			await step.waitForEvent("my event", {
+				type: "event",
+				timeout: "10 seconds",
+			});
+		}
+
+		await step.sleep("sleep for a while", "5 seconds");
+
+		return "test-workflow";
+	}
+}
+
+export default {
+	async fetch(request: Request, env: Env) {
 		const maybeId = new URL(request.url).searchParams.get("id");
 		if (maybeId !== null) {
-			const instance = await this.env.TEST_WORKFLOW.get(maybeId);
+			const instance = await env.TEST_WORKFLOW.get(maybeId);
 
 			return Response.json(await instance.status());
 		}
 
-		const workflow = await this.env.TEST_WORKFLOW.create();
+		if (request.url.endsWith("/long-workflow")) {
+			const workflow = await env.TEST_LONG_WORKFLOW.create();
+			return new Response(JSON.stringify({ id: workflow.id }));
+		}
+
+		if (request.url.endsWith("/long-workflow-batch")) {
+			const workflows = await env.TEST_LONG_WORKFLOW.createBatch([
+				{},
+				{ id: "321" },
+				{},
+			]);
+			const ids = workflows.map((workflow) => workflow.id);
+			return new Response(JSON.stringify({ ids: ids }));
+		}
+
+		const workflow = await env.TEST_WORKFLOW.create();
 
 		return new Response(JSON.stringify({ id: workflow.id }));
-	}
-}
+	},
+};

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

@@ -1,22 +1,107 @@
-import { SELF } from "cloudflare:test";
-import { expect, it, vi } from "vitest";
+import {
+	env,
+	introspectWorkflow,
+	introspectWorkflowInstance,
+	SELF,
+} from "cloudflare:test";
+import {
+	afterEach,
+	assert,
+	beforeEach,
+	describe,
+	expect,
+	it,
+	vi,
+} from "vitest";
 
-it("should be able to trigger a workflow", async () => {
-	const res = await SELF.fetch("https://mock-worker.local");
+describe("Test Workflow", () => {
+	it("should be able to trigger a workflow", async () => {
+		const res = await SELF.fetch("https://mock-worker.local");
 
-	expect(res.status).toBe(200);
+		expect(res.status).toBe(200);
+	});
+
+	it("workflow should reach the end and be successful", async () => {
+		const res = await SELF.fetch("https://mock-worker.local");
+
+		const json = await res.json<{ id: string }>();
+
+		await vi.waitUntil(async () => {
+			const res = await SELF.fetch(`https://mock-worker.local?id=${json.id}`);
+
+			const statusJson = await res.json<{ status: string }>();
+			console.log(statusJson);
+			return statusJson.status === "complete";
+		}, 1000);
+	});
+
+	it("workflow should reach the end and be successful with introspector", async () => {
+		const introspector = await introspectWorkflow(env.TEST_WORKFLOW);
+
+		await SELF.fetch("https://mock-worker.local");
+
+		const instances = introspector.get();
+		assert(instances.length === 1);
+
+		const instance = instances[0];
+		await instance.waitForStatus("complete");
+		await instance.cleanUp();
+	});
 });
 
-it("workflow should reach the end and be successful", async () => {
-	const res = await SELF.fetch("https://mock-worker.local");
+describe("Test long Workflow", () => {
+	const STEP_NAME = "my step";
+	const mockResult = "mocked result";
+
+	let introspector: Awaited<ReturnType<typeof introspectWorkflow>>;
+	let instances: Awaited<ReturnType<typeof introspectWorkflowInstance>>[];
+
+	beforeEach(async () => {
+		// CONFIG:
+		introspector = await introspectWorkflow(env.TEST_LONG_WORKFLOW);
+		introspector.modifyAll(async (m) => {
+			await m.disableSleeps();
+			await m.mockStepResult({ name: STEP_NAME }, mockResult);
+		});
+	});
+
+	afterEach(async () => {
+		// CLEANUP:
+		// Instance introspectors should be clean to prevent persisted state across tests
+		for (const instance of instances) {
+			instance.cleanUp();
+		}
+
+		// Workflow introspector should be cleaned at the end of/after each test
+		introspector.cleanUp();
+	});
+
+	it("workflow should be able to introspect and reach the end and be successful", async () => {
+		await SELF.fetch("https://mock-worker.local/long-workflow");
+
+		instances = introspector.get();
+		assert(instances.length === 1);
+
+		// ASSERTIONS:
+		const instance = instances[0];
+		expect(await instance.waitForStepResult({ name: STEP_NAME })).toEqual(
+			mockResult
+		);
+		await instance.waitForStatus("complete");
+	});
 
-	const json = await res.json<{ id: string }>();
+	it("workflow batch should be able to introspect and reach the end and be successful", async () => {
+		await SELF.fetch("https://mock-worker.local/long-workflow-batch");
 
-	await vi.waitUntil(async () => {
-		const res = await SELF.fetch(`https://mock-worker.local?id=${json.id}`);
+		instances = introspector.get();
+		assert(instances.length === 3);
 
-		const statusJson = await res.json<{ status: string }>();
-		console.log(statusJson);
-		return statusJson.status === "complete";
-	}, 1000);
+		// ASSERTIONS:
+		for (const instance of instances) {
+			expect(await instance.waitForStepResult({ name: STEP_NAME })).toEqual(
+				mockResult
+			);
+			await instance.waitForStatus("complete");
+		}
+	});
 });