scripts.promote_env

Reference › scripts › promote_env

scripts/promote_env.py is the environment promotion tool: it advances the code through the dev, test and prod environments (PIPELINE) behind gates. A human runs it. It is a thin front end over Promoter in orchestrata_ultimate.promotion.promoter and the profiles in orchestrata_ultimate.promotion.env_profile.

build_parser defines four subcommands, each dispatched in main to a Promoter method. validate <env> checks a profile and infrastructure reachability. test <env> runs the suite pinned to that environment; the default command is scripts/run_tests.py (see run_tests), and trailing arguments override it. promote --to <env> is the gated deploy that advances, restarts and verifies; --yes confirms a prod deployment, --dry-run prints the stages and what each would do without touching anything, --skip-graphs skips bootstrapping the graphs, and --skip-ingest skips re-ingesting workspace projects into the knowledge graph. status reports the current state. A PromotionError is printed as REFUSED: and the exit code is 2; otherwise the exit code is the method's own result.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

#!/usr/bin/env python3
"""Environment promotion tool — advance the code through dev -> test -> prod.

The three environments are ISOLATED Incus infra fleets (see INFRASTRUCTURE.md
and the ``env/*.env`` endpoint profiles):

    dev   10.53.3.x    single infra servers, developer scratch
    test  10.53.2.x   single infra servers, the authoritative suite target
    prod  10.53.1.x   the live HA orchgraph cluster — never a test target

"Promotion" is a GATED progression: you cannot advance to ``test`` until
``dev`` validates, and you cannot advance to ``prod`` until the full suite runs
GREEN against the isolated ``test`` env. Each stage is physically pinned to its
own ``env/<env>.env`` profile, so a stage can never talk to the wrong fleet —
the same guarantee ``tests/conftest.py`` enforces for pytest, extended to the
deploy path.

RUL-003: every infra endpoint is read from the profile with NO literal
fallback. A missing/empty required key fails the stage loudly rather than
silently defaulting to some other fleet.

Subcommands
    validate <env>            profile present + required keys + infra reachable
    test <env>                run the suite pinned to <env> (refused for prod)
    promote --to <env>        gated advance; records the promotion in the ledger
    status                    show the promotion ledger

Ledger: ``reports/promotion-state.json`` (per-env validate / test / promote
records with UTC timestamps).
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path

from orchestrata_ultimate.promotion.env_profile import ENV_SUBNET as ENV_SUBNET
from orchestrata_ultimate.promotion.env_profile import PIPELINE as PIPELINE
from orchestrata_ultimate.promotion.env_profile import REQUIRED_KEYS as REQUIRED_KEYS
from orchestrata_ultimate.promotion.env_profile import (
    PromotionError as PromotionError,
)
from orchestrata_ultimate.promotion.env_profile import _as_utc as _as_utc
from orchestrata_ultimate.promotion.env_profile import (
    _extract_host as _extract_host,
)
from orchestrata_ultimate.promotion.env_profile import (
    _graph_host_port as _graph_host_port,
)
from orchestrata_ultimate.promotion.env_profile import _probe_tcp as _probe_tcp
from orchestrata_ultimate.promotion.env_profile import _utc_now as _utc_now
from orchestrata_ultimate.promotion.env_profile import check_subnet as check_subnet
from orchestrata_ultimate.promotion.env_profile import (
    missing_required as missing_required,
)
from orchestrata_ultimate.promotion.env_profile import probe_infra as probe_infra
from orchestrata_ultimate.promotion.promoter import Promoter as Promoter

REPO_ROOT = Path(__file__).resolve().parent.parent


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="promote_env.py",
        description="Gated dev -> test -> prod environment promotion.",
    )
    sub = p.add_subparsers(dest="command", required=True)

    pv = sub.add_parser("validate", help="check a profile + infra reachability")
    pv.add_argument("env", choices=PIPELINE)

    pt = sub.add_parser("test", help="run the suite pinned to an env")
    pt.add_argument("env", choices=PIPELINE)
    pt.add_argument(
        "cmd_override",
        nargs=argparse.REMAINDER,
        help=f"override command (default: {sys.executable} scripts/run_tests.py)",
    )

    pp = sub.add_parser(
        "promote", help="gated DEPLOY to a target env (advance, restart, verify)"
    )
    pp.add_argument("--to", dest="target", required=True, choices=PIPELINE)
    pp.add_argument("--yes", action="store_true", help="confirm prod deployment")
    pp.add_argument(
        "--dry-run",
        action="store_true",
        help="print the stages and what each WOULD do; touch nothing",
    )
    pp.add_argument(
        "--skip-graphs",
        action="store_true",
        help="do not bootstrap the graphs (services are still deployed+verified)",
    )
    pp.add_argument(
        "--skip-ingest",
        action="store_true",
        help="do not re-ingest workspace projects into the KG",
    )

    sub.add_parser("status", help="show the promotion ledger")
    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    promoter = Promoter()
    try:
        if args.command == "validate":
            return promoter.validate(args.env)
        if args.command == "test":
            return promoter.test(args.env, args.cmd_override or None)
        if args.command == "promote":
            return promoter.promote(
                args.target,
                args.yes,
                execute=not args.dry_run,
                skip_graphs=args.skip_graphs,
                skip_ingest=args.skip_ingest,
            )
        if args.command == "status":
            return promoter.status()
    except PromotionError as exc:
        print(f"REFUSED: {exc}", file=sys.stderr)
        return 2
    return 1


if __name__ == "__main__":
    raise SystemExit(main())