Testing Pipelines

Testing ensures your pipelines work correctly and continue to work as you make changes. FlowMason’s testing framework integrates with VSCode’s Test Explorer.

What You’ll Learn

• Write pipeline tests
• Use assertions to verify outputs
• Mock external services and LLMs
• Run tests and check coverage

Creating Your First Test

Create a test file alongside your pipeline:

pipelines/content-summarizer.test.json:

{
  "name": "Content Summarizer Tests",
  "pipeline": "pipelines/content-summarizer.pipeline.json",
  "tests": []
}

Step 1: Write a Basic Test

Add a test case:

{
  "tests": [
    {
      "name": "summarizes article successfully",
      "input": {
        "url": "https://example.com/article"
      },
      "mocks": {
        "http-request": {
          "output": {
            "status": 200,
            "body": {
              "content": "This is a sample article about technology trends in 2025."
            }
          }
        }
      },
      "assertions": [
        { "path": "output.text", "type": "string" },
        { "path": "output.text.length", "greaterThan": 0 }
      ]
    }
  ]
}

Step 2: Understanding Assertions

Type Assertions

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

Equality Assertions

{ "path": "output.status", "equals": "success" }
{ "path": "output.code", "equals": 200 }

Comparison Assertions

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

String Assertions

{ "path": "output.text", "contains": "summary" }
{ "path": "output.url", "startsWith": "https://" }
{ "path": "output.email", "matches": "^[\\w.-]+@[\\w.-]+$" }

Array Assertions

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

Existence Assertions

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

Snapshot Assertions

Capture and compare complex outputs:

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

Update snapshots when output format changes:

fm test --update-snapshots

Step 3: Mocking External Services

Mock HTTP responses to avoid real network calls:

{
  "mocks": {
    "http-request": {
      "output": {
        "status": 200,
        "body": {
          "content": "Test content for summarization"
        }
      }
    }
  }
}

Mock AI responses for predictable tests:

{
  "mocks": {
    "generator": {
      "output": {
        "text": "• Point 1\n• Point 2\n• Point 3"
      }
    }
  }
}

Step 4: Testing Error Cases

Test that errors are handled correctly:

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

Test network failures:

{
  "name": "handles network timeout",
  "input": {
    "url": "https://slow-server.example.com"
  },
  "mocks": {
    "http-request": {
      "error": {
        "type": "CONNECTIVITY:TIMEOUT",
        "message": "Request timed out"
      }
    }
  },
  "expectError": {
    "type": "CONNECTIVITY:TIMEOUT"
  }
}

Step 5: Stage-Level Assertions

Test individual stages:

{
  "name": "extract parses content correctly",
  "input": {
    "url": "https://example.com/article"
  },
  "mocks": {
    "http-request": {
      "output": {
        "status": 200,
        "body": { "content": "Test content" }
      }
    }
  },
  "stageAssertions": {
    "fetch": [
      { "path": "output.status", "equals": 200 }
    ],
    "extract": [
      { "path": "output", "equals": "Test content" }
    ]
  }
}

Step 6: Running Tests

Using CLI

# Run all tests
fm test

# Run specific test file
fm test pipelines/content-summarizer.test.json

# Run with coverage
fm test --coverage

# Update snapshots
fm test --update-snapshots

Using VSCode

1. Open Test Explorer (Cmd+Shift+T)
2. Click play button next to any test
3. Or use shortcuts:
   • Cmd+; A - Run all tests
   • Cmd+; F - Run tests in current file
   • Cmd+; L - Re-run last tests

Step 7: Check Coverage

Run tests with coverage:

fm test --coverage

Output:

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

  fetch      ✓ covered (2 tests)
  extract    ✓ covered (2 tests)
  summarize  ✓ covered (1 test)

Overall: 100% stage coverage

Complete Test File

Here’s a comprehensive test file:

{
  "name": "Content Summarizer Tests",
  "pipeline": "pipelines/content-summarizer.pipeline.json",
  "tests": [
    {
      "name": "summarizes article successfully",
      "input": { "url": "https://example.com/article" },
      "mocks": {
        "http-request": {
          "output": { "status": 200, "body": { "content": "Sample content" } }
        },
        "generator": {
          "output": { "text": "• Summary point" }
        }
      },
      "assertions": [
        { "path": "output.text", "type": "string" },
        { "path": "output.text", "contains": "Summary" }
      ]
    },
    {
      "name": "handles 404 errors",
      "input": { "url": "https://example.com/missing" },
      "mocks": {
        "http-request": {
          "output": { "status": 404, "body": {} }
        }
      },
      "stageAssertions": {
        "fetch": [
          { "path": "output.status", "equals": 404 }
        ]
      }
    },
    {
      "name": "validates URL format",
      "input": { "url": "invalid-url" },
      "expectError": {
        "type": "VALIDATION:INVALID_INPUT"
      }
    }
  ]
}

CI/CD Integration

Run FlowMason tests in your CI pipeline:

# .github/workflows/test.yml
name: FlowMason Tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install FlowMason
        run: pip install flowmason

      - name: Run tests with coverage
        run: fm test --coverage

      - name: Upload coverage report
        uses: actions/upload-artifact@v4
        with:
          name: coverage-report
          path: coverage/

Best Practices

1. Test happy paths first - Verify normal operation
2. Test edge cases - Empty inputs, large data, special characters
3. Test error handling - Network failures, invalid data, timeouts
4. Use mocks consistently - Don’t make real API calls in tests
5. Aim for 80%+ coverage - Cover all critical paths
6. Keep tests focused - One scenario per test
7. Use snapshots wisely - Ideal for complex structured outputs
8. Integrate with CI - Run tests on every push

What’s Next?

You now know how to write comprehensive tests. In the final tutorial, you’ll learn to create custom components.

Key concepts covered:

• Test file structure
• Assertion types
• Mocking services
• Error testing
• Coverage tracking

Continue to Working with Components.