Postman Newman API Contract Testing with Kestra

id: newman-postman-api-regression-gate
namespace: company.qa

inputs:
  - id: collection_url
    type: STRING
    displayName: Postman Collection URL
    description: Public URL or raw Git repository link pointing to the exported
      Postman collection JSON file.
    defaults: https://raw.githubusercontent.com/postmanlabs/postman-collection/develop/examples/simple.json

  - id: environment_name
    type: STRING
    displayName: Target Environment
    description: Deployment tier being evaluated (e.g., staging, canary, production).
    defaults: staging

  - id: max_failed_assertions_allowed
    type: INT
    displayName: Max Allowed Failed Assertions
    description: Tolerance threshold for failed test assertions before failing the
      quality gate.
    defaults: 0

  - id: fail_pipeline_on_error
    type: BOOL
    displayName: Fail Execution on Regression
    description: Terminate the execution with a FAILED status when assertions breach
      tolerance.
    defaults: true

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Send a test summary report card to Slack. Requires
      SLACK_WEBHOOK_URL secret.
    defaults: true

concurrency:
  limit: 3

triggers:
  - id: post_deploy_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Webhook endpoint called by CI/CD release pipelines (GitHub Actions,
      ArgoCD, Helm) post-deployment.
    key: run-api-tests

tasks:
  - id: run_newman_tests
    type: io.kestra.plugin.scripts.shell.Commands
    description: Execute Postman collection test suite using Newman CLI in a
      throwaway container and export JSON report.
    containerImage: postman/newman:alpine
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    outputFiles:
      - newman-report.json
    commands:
      - |
        set -euo pipefail
        COLLECTION="{{ inputs.collection_url }}"
        echo "Starting Newman test execution for collection: ${COLLECTION}"

        # Run Newman with JSON reporter; suppress exit code so Kestra can parse assertions in subsequent task
        newman run "${COLLECTION}" \
          --reporters cli,json \
          --reporter-json-export newman-report.json \
          --suppress-exit-code 1 || true

        if [ ! -s newman-report.json ]; then
          echo "Failed to generate Newman test report!" >&2
          exit 1
        fi
        echo "Newman run completed. Report size: $(wc -c < newman-report.json) bytes."

  - id: parse_test_results
    type: io.kestra.plugin.scripts.python.Script
    description: Parse Newman JSON report to extract total assertions, failure
      diagnostics, latency percentiles, and pass rate.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    inputFiles:
      newman-report.json: "{{ outputs.run_newman_tests.outputFiles['newman-report.json'] }}"
    beforeCommands:
      - pip install --no-cache-dir kestra
    script: |
      import json
      from kestra import Kestra

      with open("newman-report.json", "r") as f:
          data = json.load(f)

      run_data = data.get("run", {})
      stats = run_data.get("stats", {})

      requests_total = stats.get("requests", {}).get("total", 0)
      assertions_total = stats.get("assertions", {}).get("total", 0)
      assertions_failed = stats.get("assertions", {}).get("failed", 0)
      assertions_passed = assertions_total - assertions_failed

      pass_rate = round((assertions_passed / assertions_total * 100), 1) if assertions_total > 0 else 100.0

      # Extract response timings
      timings = run_data.get("timings", {})
      response_avg_ms = round(timings.get("responseAverage", 0), 1)

      # Extract failure diagnostics
      failures = run_data.get("failures", [])
      failure_items = []
      for f in failures:
          parent_name = f.get("parent", {}).get("name", "Request")
          assertion_name = f.get("source", {}).get("name") or f.get("error", {}).get("test", "Assertion")
          error_msg = f.get("error", {}).get("message", "Test assertion failed")
          failure_items.append(f"• `{parent_name}`: *{assertion_name}* -> {error_msg}")

      failure_summary = "\n".join(failure_items[:10])
      if len(failure_items) > 10:
          failure_summary += f"\n... and {len(failure_items) - 10} more assertion failures."

      print(f"Tests: {assertions_total}, Passed: {assertions_passed}, Failed: {assertions_failed}, Pass Rate: {pass_rate}%")

      max_allowed = int("{{ inputs.max_failed_assertions_allowed }}")
      gate_passed = assertions_failed <= max_allowed

      Kestra.outputs({
          "total_requests": requests_total,
          "total_assertions": assertions_total,
          "failed_assertions": assertions_failed,
          "pass_rate_percent": pass_rate,
          "response_avg_ms": response_avg_ms,
          "gate_passed": gate_passed,
          "failure_summary": failure_summary or "All assertions passed successfully."
      })

  - id: evaluate_quality_gate
    type: io.kestra.plugin.core.flow.If
    description: Branch based on whether failed test assertions exceed tolerance.
    condition: "{{ (outputs.parse_test_results.vars.failed_assertions | number) >
      (inputs.max_failed_assertions_allowed | number) }}"
    then:
      - id: notify_slack_failure
        type: io.kestra.plugin.core.flow.If
        description: Post failure breakdown to Slack if notifications are enabled.
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack_failure_card
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Send structured test regression card to Slack QA/DevOps channel.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "API Test Quality Gate FAILED in `{{ inputs.environment_name }}`!\n• Total Requests: {{ outputs.parse_test_results.vars.total_requests }}\n• Failed Assertions: {{ outputs.parse_test_results.vars.failed_assertions }} / {{ outputs.parse_test_results.vars.total_assertions }} (Pass rate: {{ outputs.parse_test_results.vars.pass_rate_percent }}%)\n• Avg Latency: {{ outputs.parse_test_results.vars.response_avg_ms }}ms\n\nFailures:\n{{ outputs.parse_test_results.vars.failure_summary }}\n\nExecution: {{ execution.id }}"
              }

      - id: check_fail_pipeline
        type: io.kestra.plugin.core.flow.If
        description: Halt workflow execution when fail_pipeline_on_error is enabled.
        condition: "{{ inputs.fail_pipeline_on_error }}"
        then:
          - id: fail_execution
            type: io.kestra.plugin.core.execution.Fail
            description: Halt downstream deployment due to failed API contract assertions.
            errorMessage: "API regression gate failed: {{
              outputs.parse_test_results.vars.failed_assertions }} assertion(s)
              failed, exceeding allowed threshold of {{
              inputs.max_failed_assertions_allowed }}."
    else:
      - id: log_test_success
        type: io.kestra.plugin.core.log.Log
        description: Log successful test execution metrics.
        message: >-
          API contract quality gate PASSED in '{{ inputs.environment_name }}'.
          Total Assertions: {{ outputs.parse_test_results.vars.total_assertions
          }}, Failed: {{ outputs.parse_test_results.vars.failed_assertions }}.
          Pass Rate: {{ outputs.parse_test_results.vars.pass_rate_percent }}%,
          Avg Response Time: {{ outputs.parse_test_results.vars.response_avg_ms
          }}ms.

      - id: notify_slack_success
        type: io.kestra.plugin.core.flow.If
        description: Post success notification to Slack if enabled.
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack_success_card
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post smoke test pass confirmation to Slack.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "API Quality Gate PASSED in `{{ inputs.environment_name }}`: All {{ outputs.parse_test_results.vars.total_assertions }} assertions passed (Pass rate: {{ outputs.parse_test_results.vars.pass_rate_percent }}%, Avg Latency: {{ outputs.parse_test_results.vars.response_avg_ms }}ms). Execution: {{ execution.id }}."
              }

errors:
  - id: on_runner_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Catch unhandled container runner failures or network drops.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Newman test runner encountered an unexpected error in flow `{{ flow.id }}` for environment `{{ inputs.environment_name }}`. Check Docker logs. Execution: {{ execution.id }}."
      }

outputs:
  - id: total_assertions
    type: INT
    description: Total number of test assertions evaluated across all requests.
    value: "{{ outputs.parse_test_results.vars.total_assertions }}"
  - id: failed_assertions
    type: INT
    description: Count of test assertions that failed.
    value: "{{ outputs.parse_test_results.vars.failed_assertions }}"
  - id: pass_rate_percent
    type: FLOAT
    description: Percentage of passing test assertions.
    value: "{{ outputs.parse_test_results.vars.pass_rate_percent }}"
  - id: gate_passed
    type: BOOL
    description: Indicates whether the quality gate passed based on allowed thresholds.
    value: "{{ outputs.parse_test_results.vars.gate_passed }}"

API integration and contract regression tests are frequently authored by developers and QA engineers in Postman collections. However, integrating these tests into continuous delivery pipelines often introduces hurdles:

Raw CLI output logs from CI scripts are cluttered and hard to parse during rapid release rollouts.
Lack of automated gating: broken endpoints often go unnoticed until end users report errors in staging or production.
Missing multi-stage orchestration: post-deploy smoke tests should automatically notify on-call teams with structured assertion diagnostics rather than generic job failure notifications.

This blueprint bridges Postman and continuous delivery:

Triggers on demand via an event-driven Webhook endpoint designed for CI/CD post-deployment steps.
Executes the target Postman collection inside an isolated postman/newman:alpine container via Kestra's Docker task runner.
Parses Newman's JSON execution report in Python to extract total requests, test assertions, failure causes, and latency averages.
Enforces a configurable quality gate: if assertions fail beyond tolerance, it halts pipeline promotion and delivers an actionable failure card to Slack.

How it works

post_deploy_webhook receives incoming release triggers from GitHub Actions, GitLab CI, ArgoCD, or Helm hooks with collection_url and environment_name.
run_newman_tests launches postman/newman:alpine on Docker to execute the collection with --reporters cli,json and generates newman-report.json.
parse_test_results runs a Python script to parse newman-report.json, calculating pass rates, average response latency, and extracting failed test names with error messages.
evaluate_quality_gate (io.kestra.plugin.core.flow.If) branches:
- If failed assertions exceed max_failed_assertions_allowed: posts an alert card to Slack and halts the execution via Fail.
- If compliant: logs pass metrics and dispatches a confirmation card.
An errors block catches Docker runner crashes or network timeouts.

What you get

Instant Post-Deploy Verification: Automated smoke and regression testing before promoting releases to production.
Actionable Slack Diagnostics: Alerts pinpoint the exact failing request, assertion name, and error message.
Universal CI/CD Compatibility: Simple webhook trigger seamlessly integrates into any CD pipeline via curl.
Zero Credential Exposure: Webhook tokens and Slack URLs are secured using Kestra namespace secrets.

Prerequisites

Docker daemon access configured on the Kestra worker.
An exported Postman collection JSON file accessible via HTTP/HTTPS (or GitHub raw URL).
An incoming Slack webhook URL for notifications.

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook URL.

Quick start

Configure SLACK_WEBHOOK_URL in your Kestra namespace secrets.
Set collection_url to your Postman collection JSON URL (e.g., hosted on GitHub or an internal asset server).

Call the post_deploy_webhook from your CI/CD pipeline post-deployment:

curl -X POST "https://kestra.example.com/api/v1/executions/webhook/company.qa/newman-postman-api-regression-gate/run-api-tests" \
  -H "Content-Type: application/json" \
  -d '{
    "collection_url": "https://raw.githubusercontent.com/org/repo/main/tests/api-tests.json",
    "environment_name": "staging"
  }'

Review the execution history and Slack channel for the test assertion breakdown.

How to extend

Postman Environment Injection: Add an input for Postman environment files (--environment <env.json>) to test across multiple staging endpoints.
Multi-Collection Matrix: Wrap run_newman_tests inside an io.kestra.plugin.core.flow.Loop to run multiple microservice test suites in sequence or parallel.
Jira Issue Creation: Automatically create bug tickets in Jira for failing assertions.

Newman (Postman CLI) Automated API Contract Regression Gate

How it works

What you get

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Newman (Postman CLI) Automated API Contract Regression Gate

How it works

What you get

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Memcached Slab Eviction Sentinel

Scheduled MinIO Bucket Retention Cleanup

Prune noncurrent object versions billed forever behind a versioned bucket

Incremental, idempotent lake ingestion from MinIO with DuckDB

Catch MinIO Replication Drift Before Objects Go Missing on the Target

Meilisearch Indexing Queue Staleness Sentinel

More Core Blueprints

HTTP Redirect Chain Audit

Automated Incident Response Threat Intelligence Alerting

Block a release when k6 shows it is slower than the last good one

HubSpot Stalled Deal Pipeline Sentinel

Jenkins Stuck Build Watchdog

Infrastructure Automation workflow example: Getting started with Kestra

New to Kestra?