id: api-contract-smoke-test
namespace: company.team
description: |
Run a small, configurable set of HTTP contract checks against an API,
capture status and latency evidence, persist a JSON report, and alert Slack
when any endpoint returns the wrong status or misses its response marker.
triggers:
- id: every_15_minutes
type: io.kestra.plugin.core.trigger.Schedule
description: Run the smoke test every 15 minutes after validating it manually.
cron: "*/15 * * * *"
disabled: true
inputs:
- id: endpoints
type: JSON
displayName: API endpoints
description: |
JSON array of checks. Each item requires name and url; expected_status
defaults to 200. Optional method, headers, body, body_contains, and
timeout_seconds fields are supported. Keep secret values in headers as
Kestra secret expressions.
defaults: |
[
{
"name": "public-health",
"url": "https://api.example.com/health",
"expected_status": 200,
"method": "GET",
"timeout_seconds": 10
}
]
required: true
- id: user_agent
type: STRING
displayName: User agent
description: User-Agent header sent with every request.
defaults: kestra-api-contract-smoke-test
required: true
- id: notify_slack
type: BOOL
displayName: Send Slack alert
description: Post the failed checks to Slack when one or more contracts fail.
defaults: true
required: true
tasks:
- id: run_contract_checks
type: io.kestra.plugin.scripts.python.Script
description: Execute each configured HTTP assertion and collect status, latency,
and failure details.
taskRunner:
type: io.kestra.plugin.core.runner.Process
env:
ENDPOINTS: "{{ inputs.endpoints | toJson }}"
USER_AGENT: "{{ inputs.user_agent }}"
script: |
import http.client
import json
import os
import time
import urllib.error
import urllib.request
raw_endpoints = os.environ["ENDPOINTS"]
endpoints = json.loads(raw_endpoints)
if not isinstance(endpoints, list) or not endpoints:
raise ValueError("endpoints must be a non-empty JSON array")
results = []
for index, check in enumerate(endpoints):
if not isinstance(check, dict):
raise ValueError(f"endpoint at index {index} must be an object")
name = str(check.get("name", "")).strip()
url = str(check.get("url", "")).strip()
if not name or not url:
raise ValueError(f"endpoint at index {index} requires name and url")
method = str(check.get("method", "GET")).upper()
expected_status = int(check.get("expected_status", 200))
timeout = float(check.get("timeout_seconds", 10))
if timeout <= 0:
raise ValueError(f"{name}: timeout_seconds must be positive")
headers = {"User-Agent": os.environ["USER_AGENT"]}
configured_headers = check.get("headers", {})
if configured_headers:
if not isinstance(configured_headers, dict):
raise ValueError(f"{name}: headers must be an object")
headers.update({str(key): str(value) for key, value in configured_headers.items()})
body = check.get("body")
request_body = None
if body is not None:
request_body = json.dumps(body).encode("utf-8")
headers.setdefault("Content-Type", "application/json")
started = time.perf_counter()
status = None
error = None
response_text = ""
try:
request = urllib.request.Request(
url,
data=request_body,
headers=headers,
method=method,
)
try:
response = urllib.request.urlopen(request, timeout=timeout)
except urllib.error.HTTPError as exc:
# a 4xx/5xx is still a response: check it against expected_status
response = exc
with response:
status = response.status
response_text = response.read(4096).decode("utf-8", errors="replace")
except http.client.HTTPException as exc:
# the message can echo the reply or the URL, so only the failure type is kept
error = type(exc).__name__
except OSError as exc:
error = f"{type(exc).__name__}: {exc}"
latency_ms = round((time.perf_counter() - started) * 1000, 1)
marker = check.get("body_contains")
status_ok = status == expected_status
marker_ok = not marker or str(marker) in response_text
passed = status_ok and marker_ok and error is None
reasons = []
if not status_ok:
reasons.append(f"expected HTTP {expected_status}, got {status or 'no response'}")
if not marker_ok:
reasons.append(f"response did not contain {marker!r}")
if error:
reasons.append(error)
results.append({
"name": name,
"url": url,
"method": method,
"expected_status": expected_status,
"actual_status": status,
"latency_ms": latency_ms,
"passed": passed,
"error": "; ".join(reasons) if reasons else None,
})
failed = [result for result in results if not result["passed"]]
summary = {
"checked_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"total": len(results),
"passed": len(results) - len(failed),
"failed": len(failed),
"results": results,
}
print("::" + json.dumps({"outputs": {
"summary": summary,
"failed_count": len(failed),
"report_markdown": "\n".join(
f"- {'PASS' if result['passed'] else 'FAIL'} `{result['name']}` "
f"({result['latency_ms']} ms)"
+ (f": {result['error']}" if result["error"] else "")
for result in results
),
}}) + "::")
- id: write_report
type: io.kestra.plugin.core.storage.Write
description: Persist the complete contract-check evidence for later inspection.
content: "{{ outputs.run_contract_checks.vars.summary | toJson }}"
extension: ".json"
- id: alert_if_failed
type: io.kestra.plugin.core.flow.If
description: Notify the owning team only when one or more endpoint assertions fail.
condition: "{{ outputs.run_contract_checks.vars.failed_count > 0 and
inputs.notify_slack }}"
then:
- id: send_slack_alert
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Send the failed endpoint names, reasons, and report location to Slack.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: |
API contract smoke test failed in {{ flow.namespace }}.{{ flow.id }} (execution {{ execution.id }}).
{{ outputs.run_contract_checks.vars.report_markdown }}
Evidence: {{ outputs.write_report.uri }}
outputs:
- id: report
type: URI
description: JSON report containing every request, status, latency, and
assertion result.
value: "{{ outputs.write_report.uri }}"
- id: failed_count
type: INT
description: Number of endpoint checks that failed in this execution.
value: "{{ outputs.run_contract_checks.vars.failed_count }}"
