refactor(errors): introduce MissingEnvVarError for missing host env vars
lint / lint (push) Failing after 1m52s
test / unit (pull_request) Failing after 40s
test / integration (pull_request) Successful in 19s

Replace die()/RuntimeError at the point of detection with a typed
MissingEnvVarError, and catch it in the CLI dispatcher alongside
ManifestError. Library code stays free of stderr output; the CLI
renders all user-facing error lines in one place.

Sites updated: egress.egress_resolve_token_values (unset/empty token),
git_gate._provision_dynamic_key and revoke_git_gate_provisioned_keys
(missing forge_token_env), env.resolve_env (unset interpolated var).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-07-08 12:55:30 -04:00
parent 8e89227cb8
commit a8e880fad6
5 changed files with 41 additions and 13 deletions
+8 -5
View File
@@ -36,7 +36,8 @@ from abc import ABC
from dataclasses import dataclass
from pathlib import Path
from .log import die, info
from .errors import MissingEnvVarError
from .log import info
from .manifest import ManifestBottle, ManifestGitEntry
@@ -564,9 +565,10 @@ def _provision_dynamic_key(
pk = entry.Key
token = os.environ.get(pk.forge_token_env)
if token is None:
die(
raise MissingEnvVarError(
pk.forge_token_env,
f"git-gate.repos[{entry.Name!r}] key.forge_token_env"
f" = {pk.forge_token_env!r}: env var is not set"
f" = {pk.forge_token_env!r}: env var is not set",
)
api_url = pk.api_url or f"https://{entry.UpstreamHost}"
provisioner = get_provisioner(pk.provider, token, api_url)
@@ -608,10 +610,11 @@ def revoke_git_gate_provisioned_keys(bottle: ManifestBottle, stage_dir: Path) ->
key_id = id_file.read_text().strip()
token = os.environ.get(pk.forge_token_env)
if token is None:
raise RuntimeError(
raise MissingEnvVarError(
pk.forge_token_env,
f"git-gate.repos[{entry.Name!r}] key.forge_token_env"
f" = {pk.forge_token_env!r}: env var is not set;"
f" cannot revoke deploy key {key_id}"
f" cannot revoke deploy key {key_id}",
)
api_url = pk.api_url or f"https://{entry.UpstreamHost}"
provisioner = get_provisioner(pk.provider, token, api_url)