Testing Pipelines

Write comprehensive tests for your FlowMason pipelines with assertions and mocks.

FlowMason includes a testing framework that integrates with VSCode’s Test Explorer, making it easy to write, run, and debug pipeline tests.

Test File Structure

Create a .test.json file alongside your pipeline:

{
  "name": "Content Summarizer Tests",
  "pipeline": "pipelines/content-summarizer.pipeline.json",
  "tests": [
    {
      "name": "summarizes article successfully",
      "input": {
        "url": "https://example.com/article"
      },
      "assertions": [
        { "path": "output.summary", "type": "string" },
        { "path": "output.summary.length", "greaterThan": 0 }
      ]
    }
  ]
}

Running Tests

CLI

# Run all tests
fm test

# Run tests for a specific pipeline
fm test pipelines/content-summarizer.test.json

# Run with coverage
fm test --coverage

VSCode

• Open the Test Explorer panel
• Click the play button next to any test or test file
• Use Cmd+; A to run all tests

Assertions

FlowMason supports 9 assertion types for comprehensive testing:

Type Checks

{ "path": "output.summary", "type": "string" }
{ "path": "output.count", "type": "number" }
{ "path": "output.items", "type": "array" }
{ "path": "output.config", "type": "object" }
{ "path": "output.enabled", "type": "boolean" }

Equality

{ "path": "output.status", "equals": "success" }
{ "path": "output.code", "equals": 200 }
{ "path": "output.config", "deepEquals": { "key": "value" } }

Comparisons

{ "path": "output.count", "greaterThan": 0 }
{ "path": "output.count", "lessThan": 100 }
{ "path": "output.score", "greaterThanOrEqual": 0.5 }
{ "path": "output.score", "lessThanOrEqual": 1.0 }

String Matching

{ "path": "output.text", "contains": "summary" }
{ "path": "output.text", "startsWith": "The" }
{ "path": "output.text", "endsWith": "." }
{ "path": "output.url", "matches": "^https://" }

Array Operations

{ "path": "output.items", "hasLength": 5 }
{ "path": "output.items", "hasMinLength": 1 }
{ "path": "output.items", "hasMaxLength": 10 }
{ "path": "output.tags", "includes": "important" }
{ "path": "output.tags", "includesAll": ["tag1", "tag2"] }

Existence

{ "path": "output.metadata", "exists": true }
{ "path": "output.error", "exists": false }
{ "path": "output.optional", "isNull": true }

Snapshot Testing

Compare outputs against saved snapshots:

{ "path": "output", "matchesSnapshot": "summarizer-output" }

Run with --update-snapshots to update:

fm test --update-snapshots

Mocking

Mock external services and LLM responses:

{
  "name": "test with mocks",
  "input": { "url": "https://example.com" },
  "mocks": {
    "http-request": {
      "output": {
        "status": 200,
        "body": { "content": "Test content..." }
      }
    },
    "generator": {
      "output": {
        "text": "Mocked summary response"
      }
    }
  },
  "assertions": [
    { "path": "output.summary", "equals": "Mocked summary response" }
  ]
}

Error Testing

Test that pipelines handle errors correctly:

{
  "name": "handles invalid URL",
  "input": { "url": "not-a-valid-url" },
  "expectError": {
    "type": "VALIDATION:INVALID_INPUT",
    "messageContains": "invalid URL"
  }
}

Stage-Level Testing

Test individual stages:

{
  "name": "extract stage parses correctly",
  "input": { "url": "https://example.com" },
  "stageAssertions": {
    "fetch": [
      { "path": "output.status", "equals": 200 }
    ],
    "extract": [
      { "path": "output", "type": "string" },
      { "path": "output.length", "greaterThan": 100 }
    ]
  }
}

Test Coverage

Generate coverage reports:

fm test --coverage

Output:

Coverage Report
───────────────
Pipeline: content-summarizer.pipeline.json
  Stages covered: 3/3 (100%)

  fetch      ✓ covered
  extract    ✓ covered
  summarize  ✓ covered

Overall: 100% stage coverage

CI/CD Integration

Run tests in your CI pipeline:

# .github/workflows/test.yml
name: Tests
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - run: pip install flowmason
      - run: fm test --coverage

Setup and Teardown

Configure setup and teardown steps for your tests:

{
  "name": "Database Tests",
  "pipeline": "pipelines/data-processor.pipeline.json",
  "setup": {
    "stages": ["seed-database", "create-temp-files"]
  },
  "teardown": {
    "stages": ["cleanup-database", "remove-temp-files"]
  },
  "tests": [...]
}

Per-Test Setup

{
  "name": "requires fresh state",
  "setup": { "stages": ["reset-state"] },
  "input": {...},
  "assertions": [...]
}

Best Practices

1. Test happy paths - Verify normal operation
2. Test error cases - Ensure graceful error handling
3. Mock external services - Don’t make real API calls in tests
4. Use meaningful names - Describe what each test verifies
5. Aim for high coverage - Target 80%+ stage coverage
6. Use snapshots wisely - Ideal for complex structured outputs
7. Isolate tests - Use setup/teardown to prevent test pollution