Connection smoke test

A staged smoke test mirroring the console's "Test connection" flow (Integration bundle). Stages run in order and each reports what actually happened:

Stage	Request	Pass condition
`reachability`	`GET /health`	200, `status: ok`, no failing checks. Abort on failure.
`telemetry_subsystem`	`GET /health/telemetry`	200 and healthy body.
`topology_subsystem`	`GET /health/topology`	200 and healthy body.
`auth_topology`	`GET /topology?freshness_seconds=900` with `x-api-key`	200; extract `tenant_key`, distinct node count, newest `observed_at`.

Stage 4 interpretation, matching the console: 401 means the key was rejected; 503 means the tenant data plane is unavailable (possibly a disabled tenant); 200 with zero nodes means authenticated but no telemetry in the last 15 minutes, so start the fleet agent; 200 with nodes means telemetry is flowing. With no key set, stages 1 to 3 still run and the auth stage is reported as not tested, never faked.

Set CONSTELLATION_API_BASE and optionally CONSTELLATION_API_TOKEN (common setup).

TypeScript
Python
Go
cURL

const BASE = (process.env.CONSTELLATION_API_BASE ?? 'https://api.constellation.space').replace(/\/$/, '')
const TOKEN = process.env.CONSTELLATION_API_TOKEN

type Stage = { stage: string; ok: boolean; detail: string; latencyMs: number }
const stages: Stage[] = []

type Health = { status?: string; checks?: Record<string, { ok?: boolean }> }
type Entity = { entity_id?: string; observed_at?: string; tags?: Record<string, string> }
type Topology = { tenant_key?: string; entities?: Entity[] }

async function probe(path: string, headers?: Record<string, string>) {
  const started = Date.now()
  try {
    const res = await fetch(`${BASE}${path}`, { headers, signal: AbortSignal.timeout(8000) })
    const json = await res.json().catch(() => null)
    return { status: res.status, json, latencyMs: Date.now() - started, error: null as string | null }
  } catch {
    return { status: 0, json: null, latencyMs: Date.now() - started,
      error: 'Network error or timeout. Check the base URL and egress.' }
  }
}

function healthStage(name: string, label: string, r: Awaited<ReturnType<typeof probe>>) {
  const body = r.json as Health | null
  const failing = Object.entries(body?.checks ?? {}).filter(([, c]) => c.ok === false).map(([n]) => n)
  const ok = r.error == null && r.status === 200 && body?.status === 'ok' && failing.length === 0
  const detail = r.error ?? (ok ? `${label} healthy.` :
    `${label} degraded (status ${r.status}${failing.length ? `, failing: ${failing.join(', ')}` : ''}).`)
  stages.push({ stage: name, ok, detail, latencyMs: r.latencyMs })
  return ok
}

// Stage 1: reachability. Abort if the platform itself is down.
if (!healthStage('reachability', 'Platform API', await probe('/health'))) {
  report()
  process.exit(2)
}

// Stages 2 and 3: subsystems.
healthStage('telemetry_subsystem', 'Telemetry subsystem', await probe('/health/telemetry'))
healthStage('topology_subsystem', 'Topology subsystem', await probe('/health/topology'))

// Stage 4: authenticated data plane.
if (TOKEN) {
  const r = await probe('/topology?freshness_seconds=900', { 'x-api-key': TOKEN })
  if (r.error) {
    stages.push({ stage: 'auth_topology', ok: false, detail: r.error, latencyMs: r.latencyMs })
  } else if (r.status === 401) {
    stages.push({ stage: 'auth_topology', ok: false,
      detail: 'API key rejected (401). Verify the tenant x-api-key.', latencyMs: r.latencyMs })
  } else if (r.status === 503) {
    stages.push({ stage: 'auth_topology', ok: false,
      detail: 'Tenant data plane unavailable (503). The tenant may be disabled.', latencyMs: r.latencyMs })
  } else if (r.status !== 200) {
    stages.push({ stage: 'auth_topology', ok: false,
      detail: `Unexpected status ${r.status}.`, latencyMs: r.latencyMs })
  } else {
    const topo = r.json as Topology
    const nodes = new Set((topo.entities ?? []).map((e) => e.tags?.node_id ?? e.entity_id).filter(Boolean))
    const newest = (topo.entities ?? []).map((e) => e.observed_at ?? '').sort().at(-1) || null
    stages.push({ stage: 'auth_topology', ok: true, latencyMs: r.latencyMs,
      detail: nodes.size > 0
        ? `API key accepted (tenant ${topo.tenant_key}). Telemetry flowing: ${nodes.size} nodes, newest sample ${newest}.`
        : `API key accepted (tenant ${topo.tenant_key}), but no telemetry in the last 15 minutes. Start the fleet agent to begin ingest.` })
  }
} else {
  console.log('note: CONSTELLATION_API_TOKEN not set; authenticated stage not tested')
}

function report() {
  for (const s of stages) {
    console.log(`${s.ok ? 'PASS' : 'FAIL'}  ${s.stage.padEnd(20)} ${s.latencyMs}ms  ${s.detail}`)
  }
}
report()
process.exit(stages.every((s) => s.ok) ? 0 : 1)

import os
import sys
import time

import requests

BASE = os.environ.get("CONSTELLATION_API_BASE", "https://api.constellation.space").rstrip("/")
TOKEN = os.environ.get("CONSTELLATION_API_TOKEN")

stages = []


def probe(path, headers=None):
    started = time.monotonic()
    try:
        resp = requests.get(f"{BASE}{path}", headers=headers, timeout=8)
        try:
            body = resp.json()
        except ValueError:
            body = None
        return resp.status_code, body, int((time.monotonic() - started) * 1000), None
    except requests.RequestException:
        latency = int((time.monotonic() - started) * 1000)
        return 0, None, latency, "Network error or timeout. Check the base URL and egress."


def health_stage(name, label, status, body, latency, error):
    checks = (body or {}).get("checks") or {}
    failing = [n for n, c in checks.items() if isinstance(c, dict) and c.get("ok") is False]
    ok = error is None and status == 200 and (body or {}).get("status") == "ok" and not failing
    if error:
        detail = error
    elif ok:
        detail = f"{label} healthy."
    else:
        suffix = f", failing: {', '.join(failing)}" if failing else ""
        detail = f"{label} degraded (status {status}{suffix})."
    stages.append({"stage": name, "ok": ok, "detail": detail, "latency_ms": latency})
    return ok


def report():
    for s in stages:
        mark = "PASS" if s["ok"] else "FAIL"
        print(f"{mark}  {s['stage']:<20} {s['latency_ms']}ms  {s['detail']}")


# Stage 1: reachability. Abort if the platform itself is down.
if not health_stage("reachability", "Platform API", *probe("/health")):
    report()
    sys.exit(2)

# Stages 2 and 3: subsystems.
health_stage("telemetry_subsystem", "Telemetry subsystem", *probe("/health/telemetry"))
health_stage("topology_subsystem", "Topology subsystem", *probe("/health/topology"))

# Stage 4: authenticated data plane.
if TOKEN:
    status, body, latency, error = probe("/topology?freshness_seconds=900",
                                         headers={"x-api-key": TOKEN})
    if error:
        stage = {"ok": False, "detail": error}
    elif status == 401:
        stage = {"ok": False, "detail": "API key rejected (401). Verify the tenant x-api-key."}
    elif status == 503:
        stage = {"ok": False,
                 "detail": "Tenant data plane unavailable (503). The tenant may be disabled."}
    elif status != 200:
        stage = {"ok": False, "detail": f"Unexpected status {status}."}
    else:
        entities = (body or {}).get("entities") or []
        nodes = {e.get("tags", {}).get("node_id") or e.get("entity_id")
                 for e in entities if e.get("tags", {}).get("node_id") or e.get("entity_id")}
        newest = max((e.get("observed_at") or "" for e in entities), default=None)
        tenant = (body or {}).get("tenant_key")
        if nodes:
            detail = (f"API key accepted (tenant {tenant}). Telemetry flowing: "
                      f"{len(nodes)} nodes, newest sample {newest}.")
        else:
            detail = (f"API key accepted (tenant {tenant}), but no telemetry in the last "
                      "15 minutes. Start the fleet agent to begin ingest.")
        stage = {"ok": True, "detail": detail}
    stages.append({"stage": "auth_topology", "latency_ms": latency, **stage})
else:
    print("note: CONSTELLATION_API_TOKEN not set; authenticated stage not tested")

report()
sys.exit(0 if all(s["ok"] for s in stages) else 1)

package main

import (
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"strings"
	"time"
)

type stage struct {
	Name      string
	OK        bool
	Detail    string
	LatencyMs int64
}

type health struct {
	Status string                     `json:"status"`
	Checks map[string]map[string]any  `json:"checks"`
}

type entity struct {
	EntityID   string            `json:"entity_id"`
	ObservedAt string            `json:"observed_at"`
	Tags       map[string]string `json:"tags"`
}

type topology struct {
	TenantKey string   `json:"tenant_key"`
	Entities  []entity `json:"entities"`
}

var client = &http.Client{Timeout: 8 * time.Second}

func probe(base, path string, headers map[string]string) (int, []byte, int64, string) {
	started := time.Now()
	req, _ := http.NewRequest(http.MethodGet, base+path, nil)
	for k, v := range headers {
		req.Header.Set(k, v)
	}
	resp, err := client.Do(req)
	latency := time.Since(started).Milliseconds()
	if err != nil {
		return 0, nil, latency, "Network error or timeout. Check the base URL and egress."
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	return resp.StatusCode, body, latency, ""
}

func healthStage(name, label string, status int, body []byte, latency int64, netErr string) stage {
	if netErr != "" {
		return stage{name, false, netErr, latency}
	}
	var h health
	_ = json.Unmarshal(body, &h)
	var failing []string
	for checkName, c := range h.Checks {
		if ok, present := c["ok"].(bool); present && !ok {
			failing = append(failing, checkName)
		}
	}
	if status == 200 && h.Status == "ok" && len(failing) == 0 {
		return stage{name, true, label + " healthy.", latency}
	}
	detail := fmt.Sprintf("%s degraded (status %d", label, status)
	if len(failing) > 0 {
		detail += ", failing: " + strings.Join(failing, ", ")
	}
	return stage{name, false, detail + ").", latency}
}

func main() {
	base := strings.TrimRight(getenv("CONSTELLATION_API_BASE", "https://api.constellation.space"), "/")
	token := os.Getenv("CONSTELLATION_API_TOKEN")
	var stages []stage

	// Stage 1: reachability. Abort if the platform itself is down.
	s := healthStage("reachability", "Platform API", probe(base, "/health", nil))
	stages = append(stages, s)
	if !s.OK {
		report(stages)
		os.Exit(2)
	}

	// Stages 2 and 3: subsystems.
	stages = append(stages,
		healthStage("telemetry_subsystem", "Telemetry subsystem", probe(base, "/health/telemetry", nil)),
		healthStage("topology_subsystem", "Topology subsystem", probe(base, "/health/topology", nil)))

	// Stage 4: authenticated data plane.
	if token != "" {
		status, body, latency, netErr := probe(base, "/topology?freshness_seconds=900",
			map[string]string{"x-api-key": token})
		var auth stage
		switch {
		case netErr != "":
			auth = stage{"auth_topology", false, netErr, latency}
		case status == 401:
			auth = stage{"auth_topology", false, "API key rejected (401). Verify the tenant x-api-key.", latency}
		case status == 503:
			auth = stage{"auth_topology", false, "Tenant data plane unavailable (503). The tenant may be disabled.", latency}
		case status != 200:
			auth = stage{"auth_topology", false, fmt.Sprintf("Unexpected status %d.", status), latency}
		default:
			var topo topology
			_ = json.Unmarshal(body, &topo)
			nodes := map[string]struct{}{}
			newest := ""
			for _, e := range topo.Entities {
				id := e.Tags["node_id"]
				if id == "" {
					id = e.EntityID
				}
				if id != "" {
					nodes[id] = struct{}{}
				}
				if e.ObservedAt > newest {
					newest = e.ObservedAt
				}
			}
			detail := fmt.Sprintf("API key accepted (tenant %s), but no telemetry in the last 15 minutes. "+
				"Start the fleet agent to begin ingest.", topo.TenantKey)
			if len(nodes) > 0 {
				detail = fmt.Sprintf("API key accepted (tenant %s). Telemetry flowing: %d nodes, newest sample %s.",
					topo.TenantKey, len(nodes), newest)
			}
			auth = stage{"auth_topology", true, detail, latency}
		}
		stages = append(stages, auth)
	} else {
		fmt.Println("note: CONSTELLATION_API_TOKEN not set; authenticated stage not tested")
	}

	report(stages)
	for _, s := range stages {
		if !s.OK {
			os.Exit(1)
		}
	}
}

func report(stages []stage) {
	for _, s := range stages {
		mark := "PASS"
		if !s.OK {
			mark = "FAIL"
		}
		fmt.Printf("%s  %-20s %dms  %s\n", mark, s.Name, s.LatencyMs, s.Detail)
	}
}

func getenv(key, fallback string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return fallback
}

#!/usr/bin/env bash
set -uo pipefail

BASE="${CONSTELLATION_API_BASE:-https://api.constellation.space}"
BASE="${BASE%/}"
TOKEN="${CONSTELLATION_API_TOKEN:-}"
fail=0

healthy() {
  jq -e '.status == "ok" and ([.checks // {} | .[] | select(.ok == false)] | length == 0)' \
    >/dev/null 2>&1 <<<"$1"
}

health_stage() {  # $1 stage name, $2 label, $3 path
  local body
  if body=$(curl -sS --max-time 8 "$BASE$3") && healthy "$body"; then
    echo "PASS  $1  $2 healthy."
  else
    echo "FAIL  $1  $2 unreachable or degraded: ${body:-no response}"
    fail=1
    return 1
  fi
}

# Stage 1: reachability. Abort if the platform itself is down.
health_stage "reachability" "Platform API" "/health" || exit 2

# Stages 2 and 3: subsystems.
health_stage "telemetry_subsystem" "Telemetry subsystem" "/health/telemetry"
health_stage "topology_subsystem" "Topology subsystem" "/health/topology"

# Stage 4: authenticated data plane.
if [[ -n "$TOKEN" ]]; then
  status=$(curl -sS -o /tmp/smoke-topo.json -w '%{http_code}' --max-time 8 \
    -H "x-api-key: $TOKEN" \
    "$BASE/topology?freshness_seconds=900") || status=0
  case "$status" in
    200)
      tenant=$(jq -r '.tenant_key' /tmp/smoke-topo.json)
      nodes=$(jq '[.entities[] | (.tags.node_id // .entity_id)] | unique | length' /tmp/smoke-topo.json)
      newest=$(jq -r '[.entities[].observed_at] | max // "n/a"' /tmp/smoke-topo.json)
      if ((nodes > 0)); then
        echo "PASS  auth_topology  API key accepted (tenant $tenant). Telemetry flowing: $nodes nodes, newest sample $newest."
      else
        echo "PASS  auth_topology  API key accepted (tenant $tenant), but no telemetry in the last 15 minutes. Start the fleet agent."
      fi
      ;;
    401) echo "FAIL  auth_topology  API key rejected (401). Verify the tenant x-api-key."; fail=1 ;;
    503) echo "FAIL  auth_topology  Tenant data plane unavailable (503). The tenant may be disabled."; fail=1 ;;
    *)   echo "FAIL  auth_topology  Unexpected status $status."; fail=1 ;;
  esac
else
  echo "note: CONSTELLATION_API_TOKEN not set; authenticated stage not tested"
fi

exit "$fail"

Expected output for a fully connected tenant:

PASS  reachability         112ms  Platform API healthy.
PASS  telemetry_subsystem  148ms  Telemetry subsystem healthy.
PASS  topology_subsystem   139ms  Topology subsystem healthy.
PASS  auth_topology        201ms  API key accepted (tenant acme-sat). Telemetry flowing: 24 nodes, newest sample 2026-07-09T14:31:55Z.

Integration bundle: where this flow comes from, and the Bearer versus x-api-key gap to check when stage 4 fails with 401.
Monitor the platform: the same probes shaped for cron with exit codes.
Health API reference: what each health endpoint proves.

Related​

Related