scripts.validate_env

Reference › scripts › validate_env

scripts/validate_env.py is an environment-contract preflight: it fails fast when required infrastructure keys are missing, before any test-infrastructure work starts. A human or a launcher runs it; it has no internal imports. See check_env_contract for the static gate on the same contract.

validate_required returns the names of required keys that are absent from the environment, missing_optional returns the optional ones, and _report prints the findings. _resolve_env reads the real process environment unless a mapping is passed, which the tests use. The only option is --quiet, which suppresses the report. The exit code is 1 when any required key is missing and 0 otherwise.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Constants

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/validate_env.py

#!/usr/bin/env python3
"""Environment-contract preflight — fail fast on missing REQUIRED infra keys.

Part of the reproducibility plan Phase R2 (plans/reproducibility-plan.md):
`.env.example` documents every key the code reads; this script gates on the
subset that blocks test-infra work if unset (Redis/graph(orchgraph)/Memgraph/
CXP/KG connection keys). Everything else (tuning knobs, API keys) is OPTIONAL and
only produces a warning. Values are read at CALL TIME via an injectable
``env`` mapping (defaults to ``os.environ``) — no bare ``os.environ["X"]``,
no monkeypatch.

Usage:
    python scripts/validate_env.py          # print report, exit 0/1
    python scripts/validate_env.py --quiet  # exit code only, no report
"""

from __future__ import annotations

import argparse
import os
import sys
from collections.abc import Mapping

# REQUIRED: infra-connection keys with no safe default that would silently
# leave the suite pointed at nothing (or localhost, which has no service on
# this fleet). Kept in sync with .env.example's REQUIRED section.
REQUIRED_KEYS: tuple[str, ...] = (
    "REDIS_HOST",
    "REDIS_PORT",
    # KG_ROUTING_URI/KG_DATABASE, not NEO4J_*: those keys were retired with
    # the Neo4j containers on 2026-07-27 (migration plan Step 5.1).
    "KG_ROUTING_URI",
    "KG_DATABASE",
    "MEMGRAPH_URI",
    "CXP_GRPC_PORT",
    "KG_GRPC_PORT",
)

# OPTIONAL: tuning knobs + API keys/secrets with safe code-level defaults.
# Documented here (rather than exhaustively re-listing .env.example) so the
# report can name what it checked; missing/empty entries only warn.
#
# KG_ENGINE/KG_URI/KG_WORKER_COUNT are the rest of the operational-KG seam
# (KG_ROUTING_URI/KG_DATABASE are REQUIRED above, not repeated here — a key
# is gated by exactly one of the two lists, never both).
OPTIONAL_KEYS: tuple[str, ...] = (
    "REDIS_DB",
    "KG_URI",
    "KG_ENGINE",
    "KG_WORKER_COUNT",
    "MEMGRAPH_URIS",
    "MEMGRAPH_USERNAME",
    "MEMGRAPH_PASSWORD",
    "MEMGRAPH_HA_CONTROLLER_URL",
    "MEMGRAPH_CHUNK_LABEL",
    "MEMGRAPH_CODESTRAL_CHUNK_LABEL",
    "MEMGRAPH_VECTOR_INDEX_NAME",
    "CXP_GRPC_HOST",
    "CXP_GRPC_ENDPOINT",
    "CXP_WS_URI",
    "GATEWAY_HOST",
    "GATEWAY_PORT",
    "GATEWAY_HTTP_PORT",
    "GATEWAY_ADMIN_TOKEN",
    "JWT_SECRET",
    "DEEPSEEK_API_KEY",
    "OPENROUTER_API_KEY",
    "ANTHROPIC_API_KEY",
    "OPENAI_API_KEY",
    "GITLAB_TOKEN",
    "LOG_LEVEL",
    "PROLOG_STACK_LIMIT",
    "ORCHESTRATA_WORKSPACE",
)


def _resolve_env(env: Mapping[str, str] | None) -> Mapping[str, str]:
    return os.environ if env is None else env


def validate_required(env: Mapping[str, str] | None = None) -> list[str]:
    """Return the REQUIRED keys that are missing or empty in *env*.

    An empty string counts as missing (a key set to ``""`` is just as
    useless to a connection string as an unset key). Order follows
    :data:`REQUIRED_KEYS`.
    """
    resolved = _resolve_env(env)
    return [key for key in REQUIRED_KEYS if not resolved.get(key, "").strip()]


def missing_optional(env: Mapping[str, str] | None = None) -> list[str]:
    """Return the OPTIONAL keys that are missing or empty in *env* (report-only)."""
    resolved = _resolve_env(env)
    return [key for key in OPTIONAL_KEYS if not resolved.get(key, "").strip()]


def _report(env: Mapping[str, str] | None, *, quiet: bool) -> list[str]:
    missing_required = validate_required(env)
    if quiet:
        return missing_required

    print("=== Environment contract preflight (scripts/validate_env.py) ===")
    if missing_required:
        print(f"FAIL: {len(missing_required)} REQUIRED key(s) missing/empty:")
        for key in missing_required:
            print(f"  - {key}")
    else:
        print(f"OK: all {len(REQUIRED_KEYS)} REQUIRED keys present.")

    optional_gaps = missing_optional(env)
    if optional_gaps:
        print(
            f"WARN: {len(optional_gaps)} OPTIONAL key(s) unset (using code defaults):"
        )
        for key in optional_gaps:
            print(f"  - {key}")

    print("See .env.example for the full documented key list.")
    return missing_required


def main(argv: list[str] | None = None, *, env: Mapping[str, str] | None = None) -> int:
    """CLI entry point.

    ``env`` defaults to ``os.environ`` (real process environment) and is
    exposed only so tests can exercise the exit-code contract without
    mutating the real environment.
    """
    parser = argparse.ArgumentParser(
        description="Preflight-check REQUIRED environment keys before test infra work."
    )
    parser.add_argument(
        "--quiet", action="store_true", help="Suppress the report; exit code only."
    )
    args = parser.parse_args(argv)

    missing_required = _report(env, quiet=args.quiet)
    return 1 if missing_required else 0


if __name__ == "__main__":
    sys.exit(main())