Generic recipes (any CI, any container)

About this example

Two equivalent recipes that use only the REST API — no DataNivra SDK or CLI install:

export DATANIVRA_BASE_URL=https://api.example.test
export DATANIVRA_TOKEN=...            # or DATANIVRA_ID_TOKEN + DATANIVRA_TENANT + DATANIVRA_WORKLOAD_IDENTITY_ID
export DATANIVRA_IDEMPOTENCY_KEY="ci-$RUN_ID-$ATTEMPT"   # stable per run attempt
./datanivra-test-data.sh ../request.json -- ./scripts/run-integration-tests.sh
python3 datanivra_test_data.py ../request.json -- ./scripts/run-integration-tests.sh

Both: request the manifest's spec (POST /v1/test-data/requests), poll GET /v1/test-data/requests/{id} with back-off until READY, run the test command, and release (POST .../release) from a trap/finally — on success, failure and Ctrl-C/SIGTERM. Exit codes: the test command's status, 24 denied, 25 pending approval at timeout, 26 ended, 27 timeout.

Never logged: the token (the shell recipe keeps it in a 0600 header file, not on the command line), connection details, rows. Use these when your runner has network access to the approved targets (see ../self-hosted-runner/README.md).

#!/usr/bin/env python3
"""Generic CI recipe (ADR-0018) — Python standard library only, REST only, no SDK.

    DATANIVRA_BASE_URL=https://api.example.test DATANIVRA_TOKEN=... \
      python3 datanivra_test_data.py ../request.json -- ./scripts/run-integration-tests.sh

(sign in) -> request -> wait READY -> run the test command -> release in ``finally``.
Exit codes: the test command's status; 24 denied by the profile; 25 still pending a human approval;
26 ended without READY; 27 timed out; 1 other API errors.

Credentials: ``DATANIVRA_TOKEN`` (a session), or a workload identity exchange with
``DATANIVRA_ID_TOKEN`` (your CI's OIDC token) + ``DATANIVRA_TENANT`` +
``DATANIVRA_WORKLOAD_IDENTITY_ID``. Only ids, states and reason codes are printed (stderr); the token
is never printed, logged or placed on a command line. DataNivra never returns rows or credentials.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Callable, Mapping, Sequence
from typing import Any

TERMINAL = {"RELEASED", "REJECTED", "CANCELLED", "FAILED"}
EXIT_DENIED, EXIT_PENDING, EXIT_ENDED, EXIT_TIMEOUT = 24, 25, 26, 27

# (method, url, headers, body) -> (status, parsed JSON body). Injectable for tests.
Send = Callable[[str, str, Mapping[str, str], bytes | None], tuple[int, Any]]


LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})


def check_base_url(url: str) -> str:
    """https, or http only to a loopback host; no userinfo, query or fragment (fail closed)."""
    parts = urllib.parse.urlsplit(url)
    host = (parts.hostname or "").lower()
    loopback_http = parts.scheme == "http" and host in LOOPBACK_HOSTS
    if (parts.scheme != "https" and not loopback_http) or not host:
        raise SystemExit("DATANIVRA_BASE_URL must be https:// (the token must not travel unencrypted)")
    if parts.username or parts.password or parts.query or parts.fragment:
        raise SystemExit("DATANIVRA_BASE_URL must not contain credentials, a query or a fragment")
    return url.rstrip("/")


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """Never follow redirects: a 3xx would forward the bearer token to another origin."""

    def redirect_request(self, *args: Any, **kwargs: Any) -> None:
        return None


_OPENER = urllib.request.build_opener(_NoRedirect)


def urllib_send(method: str, url: str, headers: Mapping[str, str], body: bytes | None) -> tuple[int, Any]:
    """One HTTPS request with the standard library; redirects are refused, never followed."""
    check_base_url(url.split("/v1/", 1)[0])
    request = urllib.request.Request(url, data=body, method=method, headers=dict(headers))  # noqa: S310
    try:
        with _OPENER.open(request, timeout=30) as response:
            return response.status, json.loads(response.read() or b"null")
    except urllib.error.HTTPError as error:
        if 300 <= error.code < 400:
            return error.code, {"code": "REDIRECT_REFUSED"}
        return error.code, json.loads(error.read() or b"null")


class ApiError(Exception):
    """A non-2xx answer: HTTP status and the API error code (never a submitted value)."""

    def __init__(self, status: int, payload: Any) -> None:
        self.status = status
        self.code = payload.get("code", "ERROR") if isinstance(payload, dict) else "ERROR"
        self.reasons = (
            [d.get("code") for d in payload.get("details", [])] if isinstance(payload, dict) else []
        )
        super().__init__(f"HTTP {status} {self.code}")


class Api:
    """Minimal client for the five Test Data API calls this recipe needs."""

    def __init__(self, base_url: str, token: str, send: Send = urllib_send) -> None:
        self.base, self._token, self._send = check_base_url(base_url), token, send

    def call(self, method: str, path: str, body: Any = None, key: str | None = None) -> Any:
        headers = {"Authorization": f"Bearer {self._token}", "Content-Type": "application/json"}
        if key:
            headers["Idempotency-Key"] = key
        data = json.dumps(body).encode() if body is not None else None
        status, payload = self._send(method, self.base + path, headers, data)
        if status >= 400:
            raise ApiError(status, payload)
        return payload


def exchange_workload_token(base_url: str, env: Mapping[str, str], send: Send = urllib_send) -> str:
    """CI OIDC token -> short workload session token (``POST /v1/auth/workload/token``)."""
    body = {
        "id_token": env["DATANIVRA_ID_TOKEN"],
        "tenant_slug": env["DATANIVRA_TENANT"],
        "workload_identity_id": env["DATANIVRA_WORKLOAD_IDENTITY_ID"],
    }
    status, payload = send(
        "POST",
        check_base_url(base_url) + "/v1/auth/workload/token",
        {"Content-Type": "application/json"},
        json.dumps(body).encode(),
    )
    if status != 200:
        raise ApiError(status, payload)
    return str(payload["access_token"])


def run(
    manifest: Mapping[str, Any],
    test_command: Sequence[str],
    *,
    api: Api,
    idempotency_key: str,
    timeout_seconds: float = 1200,
    sleep: Callable[[float], None] = time.sleep,
    clock: Callable[[], float] = time.monotonic,
    runner: Callable[[Sequence[str]], int] = lambda cmd: subprocess.call(list(cmd)),  # noqa: S603
    log: Callable[[str], None] = lambda line: sys.stderr.write(line + "\n") and None,
) -> int:
    """The whole recipe; returns the process exit code. Always releases what it requested."""
    if manifest.get("kind") != "TestDataRequest":
        raise SystemExit("expected a TestDataRequest manifest")
    try:
        created = api.call("POST", "/v1/test-data/requests", manifest["spec"], key=idempotency_key)
    except ApiError as error:
        if error.code == "TEST_DATA_REQUEST_DENIED":
            log(f"datanivra: denied: {', '.join(r for r in error.reasons if r)}")
            return EXIT_DENIED
        log(f"datanivra: request failed: {error}")
        return 1
    request_id = created["id"]
    log(f"datanivra: request {request_id} decision {created['decision']}")
    try:
        deadline, delay = clock() + timeout_seconds, 2.0
        while True:
            current = api.call("GET", f"/v1/test-data/requests/{request_id}")
            state = current["state"]
            if state == "READY":
                break
            if state in TERMINAL:
                log(
                    f"datanivra: request ended as {state} ({current.get('state_reason_code') or 'no reason'})"
                )
                return EXIT_ENDED
            if clock() >= deadline:
                log(f"datanivra: still {state} after {timeout_seconds:.0f}s")
                return EXIT_PENDING if state == "PENDING_APPROVAL" else EXIT_TIMEOUT
            sleep(delay)
            delay = min(delay * 2, 30.0)
        log(f"datanivra: READY environment {current.get('environment_id')}")
        return runner(test_command)
    finally:
        try:
            api.call(
                "POST", f"/v1/test-data/requests/{request_id}/release", {"reason_code": "TEST_RUN_FINISHED"}
            )
            log(f"datanivra: released {request_id}")
        except ApiError as error:  # never mask the test result; the lease TTL still tears down
            log(f"datanivra: release failed ({error}); the lease ends at its TTL")


def main(argv: Sequence[str], env: Mapping[str, str] = os.environ) -> int:
    """Command-line entry point (see the module docstring)."""
    if len(argv) < 3 or argv[1] != "--":
        raise SystemExit("usage: datanivra_test_data.py REQUEST.json -- TEST COMMAND...")
    base_url = env.get("DATANIVRA_BASE_URL") or ""
    token = env.get("DATANIVRA_TOKEN") or exchange_workload_token(base_url, env)
    with open(argv[0], encoding="utf-8") as handle:
        manifest = json.load(handle)
    key = env.get("DATANIVRA_IDEMPOTENCY_KEY") or f"py-{int(time.time())}-{os.getpid()}"
    return run(manifest, list(argv[2:]), api=Api(base_url, token), idempotency_key=key)


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

#!/bin/sh
# Generic CI recipe (ADR-0018): POSIX sh + curl + jq against the DataNivra REST API, no SDK.
#
#   DATANIVRA_BASE_URL=https://api.example.test DATANIVRA_TOKEN=... \
#     ./datanivra-test-data.sh ../request.json -- ./scripts/run-integration-tests.sh
#
# Steps: (sign in) -> request -> wait READY -> run your test command -> release (always, via trap).
# Exit codes: the test command's status; 24 denied by the profile; 25 still pending a human approval;
# 26 ended without READY; 27 timed out; 1 other API errors.
#
# Credentials: DATANIVRA_TOKEN (a session), or a workload identity exchange with
# DATANIVRA_ID_TOKEN (your CI's OIDC token) + DATANIVRA_TENANT + DATANIVRA_WORKLOAD_IDENTITY_ID.
# The bearer token is written to a 0600 header file, never put on a command line or printed.
# Only ids, states and reason codes are printed; DataNivra never returns rows or credentials.
set -eu

manifest=${1:?usage: datanivra-test-data.sh REQUEST.json -- TEST COMMAND...}
shift
[ "${1:-}" = "--" ] && shift
: "${DATANIVRA_BASE_URL:?set DATANIVRA_BASE_URL}"
# The token must never travel unencrypted or to another origin: https only (plain http only to a
# loopback control plane), no credentials/query/fragment in the URL; curl never follows redirects
# here (no -L), and a 3xx is treated as a failure.
case "$DATANIVRA_BASE_URL" in
  *@* | *\?* | *\#*) echo "datanivra: DATANIVRA_BASE_URL must not contain credentials, a query or a fragment" >&2; exit 2 ;;
  https://?*) ;;
  http://localhost | http://localhost:[0-9]* | http://localhost/* | http://127.0.0.1 | http://127.0.0.1:[0-9]* | http://127.0.0.1/*) ;;
  *) echo "datanivra: DATANIVRA_BASE_URL must be https://" >&2; exit 2 ;;
esac
timeout=${DATANIVRA_WAIT_SECONDS:-1200}
key=${DATANIVRA_IDEMPOTENCY_KEY:-"sh-$(date +%s)-$"}

workdir=$(mktemp -d)
chmod 700 "$workdir"
headers="$workdir/headers"
request_id=""

cleanup() {
  status=$?
  if [ -n "$request_id" ]; then
    curl -sS -o /dev/null -X POST -H @"$headers" \
      --data '{"reason_code":"TEST_RUN_FINISHED"}' \
      "$DATANIVRA_BASE_URL/v1/test-data/requests/$request_id/release" || true
    echo "datanivra: released $request_id" >&2
  fi
  rm -rf "$workdir"
  exit "$status"
}
trap cleanup EXIT INT TERM

if [ -z "${DATANIVRA_TOKEN:-}" ]; then
  : "${DATANIVRA_ID_TOKEN:?set DATANIVRA_TOKEN or DATANIVRA_ID_TOKEN}"
  : "${DATANIVRA_TENANT:?}" "${DATANIVRA_WORKLOAD_IDENTITY_ID:?}"
  export DATANIVRA_ID_TOKEN DATANIVRA_TENANT DATANIVRA_WORKLOAD_IDENTITY_ID
  # jq reads the token from its environment, so it never appears in a process argument list.
  jq -n '{id_token: env.DATANIVRA_ID_TOKEN, tenant_slug: env.DATANIVRA_TENANT,
          workload_identity_id: env.DATANIVRA_WORKLOAD_IDENTITY_ID}' > "$workdir/exchange.json"
  DATANIVRA_TOKEN=$(curl -sS --fail -H 'Content-Type: application/json' --data @"$workdir/exchange.json" \
    "$DATANIVRA_BASE_URL/v1/auth/workload/token" | jq -r .access_token)
fi
umask 077
printf 'Authorization: Bearer %s\nContent-Type: application/json\n' "$DATANIVRA_TOKEN" > "$headers"

jq '.spec' "$manifest" > "$workdir/body.json"
http=$(curl -sS -o "$workdir/created.json" -w '%{http_code}' -X POST -H @"$headers" \
  -H "Idempotency-Key: $key" --data @"$workdir/body.json" "$DATANIVRA_BASE_URL/v1/test-data/requests")
case "$http" in
  200 | 201) ;;
  403)
    echo "datanivra: denied: $(jq -r '[.details[].code] | join(", ")' "$workdir/created.json")" >&2
    exit 24
    ;;
  *)
    echo "datanivra: request failed: HTTP $http $(jq -r .code "$workdir/created.json")" >&2
    exit 1
    ;;
esac
request_id=$(jq -r .id "$workdir/created.json")
echo "datanivra: request $request_id decision $(jq -r .decision "$workdir/created.json")" >&2

deadline=$(($(date +%s) + timeout))
delay=2
while :; do
  curl -sS --fail -H @"$headers" "$DATANIVRA_BASE_URL/v1/test-data/requests/$request_id" > "$workdir/state.json"
  state=$(jq -r .state "$workdir/state.json")
  case "$state" in
    READY) break ;;
    RELEASED | REJECTED | CANCELLED | FAILED)
      echo "datanivra: request ended as $state ($(jq -r '.state_reason_code // "no reason"' "$workdir/state.json"))" >&2
      exit 26
      ;;
  esac
  if [ "$(date +%s)" -ge "$deadline" ]; then
    echo "datanivra: still $state after ${timeout}s" >&2
    [ "$state" = PENDING_APPROVAL ] && exit 25
    exit 27
  fi
  sleep "$delay"
  [ "$delay" -lt 30 ] && delay=$((delay * 2))
done

echo "datanivra: READY environment $(jq -r .environment_id "$workdir/state.json")" >&2
"$@"