orchestrata_ultimate.promotion.env_profile

Reference › orchestrata_ultimate › promotion › env_profile

The environment-profile checks behind the validate stage of a promotion. A profile is the env/<env>.env file of an environment.

PromotionError is the exception raised whenever a promotion gate is not satisfied; the stage is refused and nothing proceeds. missing_required lists the required keys that are absent or blank, with no literal fallback (RUL-003), because a stage cannot guess an endpoint. check_subnet is the mis-edit guard: every infrastructure host in the profile (the Redis host and the graph addresses) must sit on the subnet mandated for that environment, and each violation is returned as a message. probe_infra makes a TCP connection to the Redis host and port and to the graph endpoint, defaulting to port 7687 when the address names none, and returns one message per unreachable endpoint. The helpers _extract_host, _graph_host_port, _probe_tcp, _utc_now and _as_utc parse addresses and format timestamps for the ledger.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PromotionError

Bases: RuntimeError · Line 47

A promotion gate was not satisfied — the stage is refused.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/promotion/env_profile.py

"""Environment-profile helpers for ``scripts/promote_env.py``.

Pure gate checks over an ``env/<env>.env`` profile dict (no I/O beyond what
the caller passes in, except the TCP reachability probe), plus the
``PromotionError`` a gate raises when it is not satisfied. Split out of
``scripts/promote_env.py`` (RUL-021 file budget) — behaviour-neutral, every
name here is re-exported from that script unchanged.
"""

from __future__ import annotations

import socket
from datetime import UTC, datetime

# The ordered promotion pipeline. Index in this list defines what "advance"
# means; a stage may only be entered once the prior stage has cleared its gate.
PIPELINE: tuple[str, ...] = ("dev", "test", "prod")

# Each env's mandated infra subnet. A profile whose REDIS_HOST/graph URI does
# not sit on the expected subnet is a mis-edit (e.g. dev pointing at prod IPs)
# and is refused — the deploy-path twin of conftest's prod-subnet guard.
ENV_SUBNET: dict[str, str] = {
    "dev": "10.53.3.",
    "test": "10.53.2.",
    "prod": "10.53.1.",
}

# Keys whose value carries an infra host/URI we can subnet-check + TCP-probe.
_HOST_KEY = "REDIS_HOST"
_REDIS_PORT_KEY = "REDIS_PORT"
_GRAPH_KEYS = ("KG_ROUTING_URI", "MEMGRAPH_URI")

# Required keys — kept aligned with scripts/validate_env.py REQUIRED_KEYS. We do
# not import them to avoid a heavy dependency; validate_env is still invoked as
# the authoritative preflight in `validate`, this list is the fast local gate.
REQUIRED_KEYS: tuple[str, ...] = (
    "REDIS_HOST",
    "REDIS_PORT",
    "KG_ROUTING_URI",
    "KG_DATABASE",
    "MEMGRAPH_URI",
    "CXP_GRPC_PORT",
    "KG_GRPC_PORT",
)


class PromotionError(RuntimeError):
    """A promotion gate was not satisfied — the stage is refused."""


# --------------------------------------------------------------------------- #
# Pure helpers (no I/O beyond the caller-supplied profile dict)
# --------------------------------------------------------------------------- #


def _as_utc(stamp: str) -> str:
    """Render any ledger timestamp as UTC, whatever convention wrote it.

    Entries written before `record` used an aware clock are NAIVE LOCAL times.
    Slicing to [:19] dropped the offset from the aware ones, so both printed in
    the same shape and a reader comparing them was silently four hours out.
    Naive stamps are interpreted as local — which is what they were — and
    converted, so one column means one thing.
    """
    if not stamp:
        return "-"
    try:
        parsed = datetime.fromisoformat(stamp)
    except ValueError:
        return stamp[:19]
    if parsed.tzinfo is None:
        parsed = parsed.astimezone()
    return parsed.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")


def _utc_now() -> str:
    return datetime.now(UTC).isoformat()


def _extract_host(uri_or_host: str) -> str:
    """Return the bare host from a ``scheme://host:port`` URI or a plain host."""
    value = uri_or_host
    if "://" in value:
        value = value.split("://", 1)[1]
    value = value.split("@")[-1].split("/", 1)[0]
    if ":" in value:
        value = value.rsplit(":", 1)[0]
    return value


def _graph_host_port(uri: str) -> tuple[str, int]:
    host = _extract_host(uri)
    port = 7687
    tail = uri.split("://", 1)[-1]
    if ":" in tail:
        maybe = tail.rsplit(":", 1)[-1].split("/", 1)[0]
        if maybe.isdigit():
            port = int(maybe)
    return host, port


def missing_required(profile: dict[str, str]) -> list[str]:
    """REQUIRED keys absent or empty in *profile* (RUL-003 fast local gate)."""
    return [k for k in REQUIRED_KEYS if not profile.get(k, "").strip()]


def check_subnet(env: str, profile: dict[str, str]) -> list[str]:
    """Return subnet-violation messages (empty when every infra host in
    *profile* sits on *env*'s mandated subnet)."""
    expected = ENV_SUBNET[env]
    violations: list[str] = []
    for key in (_HOST_KEY, *_GRAPH_KEYS):
        value = profile.get(key, "")
        if not value:
            continue
        host = _extract_host(value)
        if expected not in host:
            violations.append(
                f"{key}={value!r} (host {host}) is not on the {env} subnet {expected}x"
            )
    return violations


def _probe_tcp(host: str, port: int, timeout: float = 3.0) -> bool:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def probe_infra(profile: dict[str, str]) -> list[str]:
    """TCP-probe the profile's redis + graph endpoints. Returns a list of
    unreachable-endpoint messages (empty when everything answers)."""
    failures: list[str] = []
    redis_host = profile[_HOST_KEY]
    redis_port = int(profile[_REDIS_PORT_KEY])
    if not _probe_tcp(redis_host, redis_port):
        failures.append(f"redis {redis_host}:{redis_port} unreachable")
    # Direct index, not .get(KEY, default): load_profile has already validated
    # KG_ROUTING_URI is present and non-empty (RUL-003 — no literal fallback).
    graph_uri = profile[_GRAPH_KEYS[0]]
    gh, gp = _graph_host_port(graph_uri)
    if not _probe_tcp(gh, gp):
        failures.append(f"graph {gh}:{gp} unreachable")
    return failures