Workflow Studio

Test Suites

Define automated tests for your workflows — mock trigger payloads, stub node outputs, assert on execution results, and run suites from CI.

Workflow test suites let you verify that a workflow executes the right nodes and produces the right outputs for a given trigger payload — without running against real external systems. Manage them at /workflows/[id]/tests.

Test Structure

Each test defines: a trigger payload, optional node stubs (mock outputs for specific nodes), and assertions on the run result.

Test: P1 incident triggers page-oncall

{
  "name": "P1 incident triggers page-oncall",
  "description": "Verify that a newly created P1 incident pages the on-call engineer",

  "trigger": {
    "type": "record_event",
    "payload": {
      "table": "incidents",
      "action": "created",
      "record": {
        "id": "inc_test_01",
        "number": "INC-TEST-1",
        "title": "API Gateway down",
        "severity": "P1",
        "status": "open",
        "assigned_team": "platform-engineering"
      }
    }
  },

  "nodeStubs": {
    "get-oncall-user": {
      "output": { "userId": "usr_oncall_01", "name": "Alice Engineer" }
    },
    "send-page": {
      "output": { "sent": true, "messageId": "msg_test_01" }
    }
  },

  "assertions": [
    { "type": "status",       "expected": "completed" },
    { "type": "node_ran",     "nodeId": "get-oncall-user" },
    { "type": "node_ran",     "nodeId": "send-page" },
    { "type": "node_skipped", "nodeId": "send-p2-alert" },
    { "type": "output",       "path": "$.notified", "expected": true },
    { "type": "node_input",   "nodeId": "send-page", "path": "$.userId", "expected": "usr_oncall_01" }
  ]
}

Trigger Payloads by Trigger Type

record_event

json

{
  "type": "record_event",
  "payload": {
    "table": "incidents",
    "action": "created",          // created | updated | deleted
    "record": { /* full record */ },
    "previous": null              // null for created; previous state for updated
  }
}

webhook

json

{
  "type": "webhook",
  "payload": {
    "headers": { "content-type": "application/json", "x-github-event": "push" },
    "body": { /* webhook body */ },
    "query": { }
  }
}

manual

json

{
  "type": "manual",
  "payload": {
    "input": { /* any input object */ }
  }
}

Assertion Types

Type	Config	Description
status	{ expected: "completed"\|"failed"\|... }	Assert the run ended in a specific status.
node_ran	{ nodeId }	Assert a specific node executed.
node_skipped	{ nodeId }	Assert a node was not executed.
node_input	{ nodeId, path, expected }	Assert the input object passed to a node matches via JSONPath.
node_output	{ nodeId, path, expected }	Assert the output of a node matches via JSONPath.
output	{ path, expected }	Assert the run's final output matches via JSONPath.
variable	{ name, expected }	Assert a vars.* variable has a specific value at end of run.
duration	{ max: 5000 }	Assert run completed within N milliseconds.
error_message	{ contains: "substring" }	Assert the error message contains a string (for expected failures).

Running Tests

From the UI: /workflows/[id]/tests → Run All, or click a test row → Run.

bash

# Run all tests for a workflow
flowos workflow test run --workflow incident-auto-escalation

# Run a specific test
flowos workflow test run --workflow incident-auto-escalation --test "P1 incident triggers page-oncall"

# Run all workflow test suites
flowos workflow test run --all --env staging

# CI output (JUnit XML)
flowos workflow test run --all --reporter junit --output results.xml

CI Integration

Add workflow tests to your pipeline before deploying to production:

.github/workflows/flowos-test.yml

- name: Run workflow tests
  run: |
    flowos auth login --token ${{ secrets.FLOWOS_API_TOKEN }}
    flowos workflow test run --all --reporter junit --output test-results.xml

- name: Upload test results
  uses: actions/upload-artifact@v4
  with:
    name: workflow-test-results
    path: test-results.xml

Use node stubs to make tests deterministic. If your workflow calls an external API node, stub its output with a fixed response — this prevents test failures caused by external service downtime.