#!/usr/bin/env python3 from __future__ import annotations import json from dataclasses import dataclass from pathlib import Path from typing import Callable CONFIG_PATH = Path(__file__).resolve().parents[2] / "deployment-config.json" AUTHORITY_MODES = { "latest_main", "explicit_debug", "final_pack_lineage", } TOPOLOGY_KINDS = {"single", "bluegreen", "multi_slot", "multi_version"} class DeploymentConfigError(ValueError): pass @dataclass(frozen=True) class RepoOperations: require_commands: Callable[..., object] clone_or_update_repo: Callable[[str, Path, str], object] resolve_repo_ref: Callable[[Path, str], str] checkout_resolved_ref: Callable[[Path, str], object] resolve_pack_repo_commit_runtime_refs: Callable[[Path, Path, Path], dict[str, str]] error_type: type[Exception] = DeploymentConfigError @dataclass(frozen=True) class DeploymentPlan: environment: str authority_mode: str release_refs_source: str release_class: bool downstream_ref: str upstream_ref: str official_app_packs_ref: str def runtime_ref_env(self) -> dict[str, str]: return { "DOWNSTREAM_REF": self.downstream_ref, "UPSTREAM_REF": self.upstream_ref, "OFFICIAL_APP_PACKS_REF": self.official_app_packs_ref, "DEPLOYMENT_AUTHORITY_MODE": self.authority_mode, "RELEASE_REFS_SOURCE": self.release_refs_source, "RELEASE_CLASS": "true" if self.release_class else "false", } @dataclass(frozen=True) class PrepareRefRequest: downstream_ref: str upstream_ref: str official_app_packs_ref: str def _fail(message: str, error_type: type[Exception] = DeploymentConfigError) -> None: raise error_type(message) def load_deployment_config( path: Path | None = None, *, error_type: type[Exception] = DeploymentConfigError, ) -> dict[str, object]: config_path = path or CONFIG_PATH try: payload = json.loads(config_path.read_text()) except FileNotFoundError as exc: raise error_type(f"deployment config not found: {config_path}") from exc except json.JSONDecodeError as exc: raise error_type(f"deployment config is not valid JSON: {config_path}: {exc}") from exc validate_deployment_config(payload, error_type=error_type) return payload def validate_deployment_config( payload: object, *, error_type: type[Exception] = DeploymentConfigError, ) -> None: errors: list[str] = [] if not isinstance(payload, dict): _fail("deployment config must be a JSON object", error_type) if payload.get("schema_version") != 1: errors.append("schema_version must be 1") repositories = payload.get("repositories") if not isinstance(repositories, dict): errors.append("repositories must be an object") else: for key in ("downstream", "upstream", "official_app_packs"): value = str(repositories.get(key, "")).strip() if not value: errors.append(f"repositories.{key} must be non-empty") environments = payload.get("environments") if not isinstance(environments, dict) or not environments: errors.append("environments must be a non-empty object") elif isinstance(environments, dict): for env_name, env_config in environments.items(): errors.extend(_validate_environment(str(env_name), env_config)) if errors: _fail("invalid deployment config: " + "; ".join(errors), error_type) def _validate_environment(env_name: str, env_config: object) -> list[str]: errors: list[str] = [] if not isinstance(env_config, dict): return [f"environments.{env_name} must be an object"] host_class = str(env_config.get("host_class", "")).strip() if host_class not in {"offline", "online"}: errors.append(f"environments.{env_name}.host_class must be offline or online") topology = env_config.get("topology") if not isinstance(topology, dict): errors.append(f"environments.{env_name}.topology must be an object") else: kind = str(topology.get("kind", "")).strip() if kind not in TOPOLOGY_KINDS: errors.append(f"environments.{env_name}.topology.kind is invalid") targets = topology.get("targets") if not isinstance(targets, list) or not all(str(item).strip() for item in targets): errors.append(f"environments.{env_name}.topology.targets must be a non-empty string array") if kind == "bluegreen" and (not isinstance(targets, list) or len(targets) < 2): errors.append(f"environments.{env_name}.topology.targets must include at least two slots") if kind == "multi_slot" and (not isinstance(targets, list) or len(targets) < 2): errors.append(f"environments.{env_name}.topology.targets must include at least two slots") target_config = topology.get("target_config") if target_config is not None: if not isinstance(target_config, dict): errors.append(f"environments.{env_name}.topology.target_config must be an object") elif isinstance(targets, list): for target in targets: target_name = str(target).strip() if target_name not in target_config: errors.append( f"environments.{env_name}.topology.target_config.{target_name} must be present" ) else: errors.extend(_validate_topology_target(env_name, target_name, target_config[target_name])) authority = env_config.get("authority") if not isinstance(authority, dict): errors.append(f"environments.{env_name}.authority must be an object") else: default = str(authority.get("default", "")).strip() if default not in AUTHORITY_MODES: errors.append(f"environments.{env_name}.authority.default is invalid") for key in ("allow_explicit_debug",): if not isinstance(authority.get(key), bool): errors.append(f"environments.{env_name}.authority.{key} must be boolean") database = env_config.get("database") if not isinstance(database, dict): errors.append(f"environments.{env_name}.database must be an object") else: for key in ("kind", "reset"): if not str(database.get(key, "")).strip(): errors.append(f"environments.{env_name}.database.{key} must be non-empty") for key in ("allow_reset_db", "allow_reset_tidb_branch"): if not isinstance(database.get(key), bool): errors.append(f"environments.{env_name}.database.{key} must be boolean") pack = env_config.get("pack") if not isinstance(pack, dict): errors.append(f"environments.{env_name}.pack must be an object") elif not str(pack.get("build", "")).strip(): errors.append(f"environments.{env_name}.pack.build must be non-empty") publish = env_config.get("publish") if not isinstance(publish, dict): errors.append(f"environments.{env_name}.publish must be an object") elif not str(publish.get("manifest", "")).strip(): errors.append(f"environments.{env_name}.publish.manifest must be non-empty") hubspot = env_config.get("hubspot") if hubspot is not None: if not isinstance(hubspot, dict): errors.append(f"environments.{env_name}.hubspot must be an object") else: credential_admission = hubspot.get("credential_admission") if credential_admission is not None: errors.extend( _validate_hubspot_credential_admission( env_name, credential_admission ) ) return errors def _validate_hubspot_credential_admission( env_name: str, credential_admission: object, ) -> list[str]: prefix = f"environments.{env_name}.hubspot.credential_admission" if not isinstance(credential_admission, dict): return [f"{prefix} must be an object"] errors: list[str] = [] mode = str(credential_admission.get("mode", "")).strip() if mode not in {"prod", "nonprod"}: errors.append(f"{prefix}.mode must be prod or nonprod") portal_id = str(credential_admission.get("expected_portal_id", "")).strip() try: portal_id_value = int(portal_id) except ValueError: portal_id_value = 0 if portal_id_value <= 0: errors.append(f"{prefix}.expected_portal_id must be a positive integer") if not str(credential_admission.get("expected_account_type", "")).strip(): errors.append(f"{prefix}.expected_account_type must be non-empty") return errors def _validate_topology_target(env_name: str, target_name: str, target_config: object) -> list[str]: errors: list[str] = [] prefix = f"environments.{env_name}.topology.target_config.{target_name}" if not isinstance(target_config, dict): return [f"{prefix} must be an object"] if not str(target_config.get("backend_port", "")).strip(): errors.append(f"{prefix}.backend_port must be non-empty") tidb = target_config.get("tidb") if tidb is not None: if not isinstance(tidb, dict): errors.append(f"{prefix}.tidb must be an object") else: required_keys = ["host", "port", "user", "cluster_id", "branch_id", "password_env"] if env_name in {"staging", "test"}: required_keys.append("database") for key in required_keys: if not str(tidb.get(key, "")).strip(): errors.append(f"{prefix}.tidb.{key} must be non-empty") hubspot = target_config.get("hubspot") if hubspot is not None: if not isinstance(hubspot, dict): errors.append(f"{prefix}.hubspot must be an object") else: portal_id = str(hubspot.get("portal_id", "")).strip() try: portal_id_value = int(portal_id) except ValueError: portal_id_value = 0 if portal_id_value <= 0: errors.append(f"{prefix}.hubspot.portal_id must be a positive integer") return errors def environment_config( environment: str, *, config_path: Path | None = None, error_type: type[Exception] = DeploymentConfigError, ) -> dict[str, object]: payload = load_deployment_config(config_path, error_type=error_type) environments = payload["environments"] assert isinstance(environments, dict) env_config = environments.get(environment) if not isinstance(env_config, dict): _fail(f"unknown deployment environment: {environment}", error_type) return env_config def hubspot_credential_admission_policy( environment: str, *, config_path: Path | None = None, error_type: type[Exception] = DeploymentConfigError, ) -> dict[str, str]: env_config = environment_config( environment, config_path=config_path, error_type=error_type, ) hubspot = env_config.get("hubspot") if not isinstance(hubspot, dict): _fail(f"environment {environment} has no HubSpot configuration", error_type) policy = hubspot.get("credential_admission") if not isinstance(policy, dict): _fail( f"environment {environment} has no HubSpot credential admission policy", error_type, ) mode = str(policy.get("mode", "")).strip() portal_id = str(policy.get("expected_portal_id", "")).strip() account_type = str(policy.get("expected_account_type", "")).strip() if mode not in {"prod", "nonprod"} or not portal_id or not account_type: _fail( f"environment {environment} has invalid HubSpot credential admission policy", error_type, ) return { "mode": mode, "expected_portal_id": portal_id, "expected_account_type": account_type, } def topology_target_config( environment: str, target: str, *, config_path: Path | None = None, error_type: type[Exception] = DeploymentConfigError, ) -> dict[str, object]: env_config = environment_config(environment, config_path=config_path, error_type=error_type) topology = env_config.get("topology") if not isinstance(topology, dict): _fail(f"environment {environment} has invalid topology config", error_type) target_config = topology.get("target_config") if not isinstance(target_config, dict): _fail(f"environment {environment} has no topology target config", error_type) selected = target_config.get(target) if not isinstance(selected, dict): _fail(f"unknown deployment target for {environment}: {target}", error_type) return selected def repository_urls( *, config_path: Path | None = None, error_type: type[Exception] = DeploymentConfigError, ) -> dict[str, str]: payload = load_deployment_config(config_path, error_type=error_type) repositories = payload["repositories"] assert isinstance(repositories, dict) return { "downstream": str(repositories["downstream"]).strip(), "upstream": str(repositories["upstream"]).strip(), "official_app_packs": str(repositories["official_app_packs"]).strip(), } def _is_mainish(value: str) -> bool: normalized = str(value or "").strip() return normalized in {"", "main", "origin/main", "refs/heads/main"} def select_authority_mode( *, environment: str, env_config: dict[str, object], downstream_ref: str, upstream_ref: str, official_app_packs_ref: str, error_type: type[Exception] = DeploymentConfigError, ) -> str: authority = env_config.get("authority") if not isinstance(authority, dict): _fail(f"environment {environment} has invalid authority config", error_type) explicit_non_main = not ( _is_mainish(downstream_ref) and _is_mainish(upstream_ref) and _is_mainish(official_app_packs_ref) ) explicit_companion_ref = bool( str(upstream_ref or "").strip() or str(official_app_packs_ref or "").strip() ) default_mode = str(authority.get("default", "")).strip() if default_mode == "final_pack_lineage" and not explicit_companion_ref: return "final_pack_lineage" if default_mode == "latest_main" and not explicit_non_main: return "latest_main" if not bool(authority.get("allow_explicit_debug")): _fail(f"environment {environment} does not allow explicit debug refs", error_type) return "explicit_debug" def resolve_prepare_ref_request( *, environment: str, downstream_ref: str, upstream_ref: str, official_app_packs_ref: str, config_path: Path | None = None, error_type: type[Exception] = DeploymentConfigError, ) -> PrepareRefRequest: env_config = environment_config(environment, config_path=config_path, error_type=error_type) authority = env_config.get("authority") if not isinstance(authority, dict): _fail(f"environment {environment} has invalid authority config", error_type) raw_downstream_ref = str(downstream_ref or "").strip() or "main" raw_upstream_ref = str(upstream_ref or "").strip() raw_official_app_packs_ref = str(official_app_packs_ref or "").strip() explicit_companion_ref = bool(raw_upstream_ref or raw_official_app_packs_ref) default_mode = str(authority.get("default", "")).strip() if default_mode == "final_pack_lineage": if explicit_companion_ref: _fail(f"environment {environment} does not allow explicit debug refs", error_type) return PrepareRefRequest( downstream_ref=raw_downstream_ref, upstream_ref="", official_app_packs_ref="", ) return PrepareRefRequest( downstream_ref=raw_downstream_ref, upstream_ref=raw_upstream_ref or "main", official_app_packs_ref=raw_official_app_packs_ref or "main", ) def validate_database_request( *, environment: str, reset_db: bool = False, reset_tidb_branch: bool = False, config_path: Path | None = None, error_type: type[Exception] = DeploymentConfigError, ) -> None: env_config = environment_config(environment, config_path=config_path, error_type=error_type) database = env_config.get("database") if not isinstance(database, dict): _fail(f"environment {environment} has invalid database config", error_type) if reset_db and not bool(database.get("allow_reset_db")): _fail(f"environment {environment} does not allow RESET_DB", error_type) if reset_tidb_branch and not bool(database.get("allow_reset_tidb_branch")): _fail(f"environment {environment} does not allow RESET_TIDB_BRANCH", error_type) def resolve_deployment_plan( *, environment: str, repos_root: Path, downstream_root: Path, upstream_root: Path, app_packs_root: Path, downstream_repo_url: str, upstream_repo_url: str, official_app_packs_repo_url: str, downstream_ref: str, upstream_ref: str, official_app_packs_ref: str, gh_token: str, operations: RepoOperations, config_path: Path | None = None, ) -> DeploymentPlan: env_config = environment_config(environment, config_path=config_path, error_type=operations.error_type) authority_mode = select_authority_mode( environment=environment, env_config=env_config, downstream_ref=downstream_ref, upstream_ref=upstream_ref, official_app_packs_ref=official_app_packs_ref, error_type=operations.error_type, ) operations.require_commands("git") repos_root.mkdir(parents=True, exist_ok=True) operations.clone_or_update_repo(downstream_repo_url, downstream_root, gh_token) operations.clone_or_update_repo(upstream_repo_url, upstream_root, gh_token) operations.clone_or_update_repo(official_app_packs_repo_url, app_packs_root, gh_token) resolved_downstream_ref = operations.resolve_repo_ref(downstream_root, downstream_ref) if authority_mode == "final_pack_lineage": operations.checkout_resolved_ref(downstream_root, resolved_downstream_ref) authoritative_refs = operations.resolve_pack_repo_commit_runtime_refs( downstream_root, upstream_root, app_packs_root, ) upstream_ref = authoritative_refs["UPSTREAM_REF"] official_app_packs_ref = authoritative_refs["OFFICIAL_APP_PACKS_REF"] resolved_upstream_ref = operations.resolve_repo_ref(upstream_root, upstream_ref) resolved_official_app_packs_ref = operations.resolve_repo_ref(app_packs_root, official_app_packs_ref) release_refs_source = "final_pack_lineage" if authority_mode == "final_pack_lineage" else "explicit" return DeploymentPlan( environment=environment, authority_mode=authority_mode, release_refs_source=release_refs_source, release_class=authority_mode == "final_pack_lineage", downstream_ref=resolved_downstream_ref, upstream_ref=resolved_upstream_ref, official_app_packs_ref=resolved_official_app_packs_ref, ) def resolve_prepare_refs( *, environment: str, repos_root: Path, downstream_root: Path, upstream_root: Path, app_packs_root: Path, downstream_repo_url: str, upstream_repo_url: str, official_app_packs_repo_url: str, downstream_ref: str, upstream_ref: str, official_app_packs_ref: str, gh_token: str, operations: RepoOperations, config_path: Path | None = None, ) -> dict[str, str]: plan = resolve_deployment_plan( environment=environment, repos_root=repos_root, downstream_root=downstream_root, upstream_root=upstream_root, app_packs_root=app_packs_root, downstream_repo_url=downstream_repo_url, upstream_repo_url=upstream_repo_url, official_app_packs_repo_url=official_app_packs_repo_url, downstream_ref=downstream_ref, upstream_ref=upstream_ref, official_app_packs_ref=official_app_packs_ref, gh_token=gh_token, operations=operations, config_path=config_path, ) return plan.runtime_ref_env()