501 lines
19 KiB
Python
501 lines
19 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Deploy path (repo -> instance) for the HA "api" adapter — see
|
||
|
|
services/home-assistant/DESIGN.md, "Sync model" and "Validation gate".
|
||
|
|
|
||
|
|
Scope: automations/scripts/scenes only (the domains reachable via
|
||
|
|
`/api/config/<domain>/config/<id>`). Dashboards and helpers are WS-only —
|
||
|
|
ha_ws.py has no mutating command, deliberately — and are a separate task.
|
||
|
|
|
||
|
|
Hard sequence, no shortcuts, matching DESIGN.md:
|
||
|
|
|
||
|
|
1. DRIFT-CHECK — re-import the instance's current state (reusing
|
||
|
|
import_api.py's tested fetch/normalize logic) and diff it against the
|
||
|
|
last-known-imported snapshot committed at repo HEAD. Any difference
|
||
|
|
outside the objects this run is about to write aborts immediately —
|
||
|
|
drift is never silently overwritten.
|
||
|
|
2. VALIDATE — local sanity per object (parses, has the domain's required
|
||
|
|
keys) plus a live `check_config` gate on the instance. Either failing
|
||
|
|
aborts before a single write.
|
||
|
|
3. WRITE — one POST per object, body = the repo file parsed back to JSON
|
||
|
|
(the inverse of import_api's GET-json -> normalize -> yaml transform).
|
||
|
|
4. VERIFY — GET the same object back, normalize, compare to what was
|
||
|
|
just written. A mismatch is reported as an error; there is no
|
||
|
|
auto-rollback.
|
||
|
|
|
||
|
|
Every function here is written to be called directly with an injected
|
||
|
|
duck-typed client (see tests/test_deploy_api_offline.sh) — no network, no
|
||
|
|
real HA instance, no live git repo required for the core logic.
|
||
|
|
"""
|
||
|
|
import difflib
|
||
|
|
import os
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
|
||
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
|
|
||
|
|
import ha_api # noqa: E402
|
||
|
|
import ha_write_api # noqa: E402
|
||
|
|
import import_api # noqa: E402
|
||
|
|
from normalize import dump_canonical_yaml, parse_yaml_text # noqa: E402
|
||
|
|
|
||
|
|
DOMAIN_SUBDIRS = ("automations", "scripts", "scenes")
|
||
|
|
DOMAIN_API_NAME = {"automations": "automation", "scripts": "script", "scenes": "scene"}
|
||
|
|
REQUIRED_KEYS = {
|
||
|
|
"automations": ("id",),
|
||
|
|
"scenes": ("id", "entities"),
|
||
|
|
"scripts": ("sequence",),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class DeployScopeError(RuntimeError):
|
||
|
|
"""A file argument can't be resolved to an in-scope object to deploy."""
|
||
|
|
|
||
|
|
|
||
|
|
# --- git-HEAD access (the "last-known-imported snapshot") ------------------
|
||
|
|
|
||
|
|
|
||
|
|
def _run_git(repo_root, args):
|
||
|
|
return subprocess.run(
|
||
|
|
["git", "-C", repo_root] + args, capture_output=True, text=True, check=False
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def git_show(repo_root, relpath):
|
||
|
|
"""Return HEAD's content of relpath, or None if not tracked at HEAD."""
|
||
|
|
proc = _run_git(repo_root, ["show", f"HEAD:{relpath}"])
|
||
|
|
if proc.returncode != 0:
|
||
|
|
return None
|
||
|
|
return proc.stdout
|
||
|
|
|
||
|
|
|
||
|
|
def git_ls_tracked(repo_root, dir_relpath):
|
||
|
|
"""List tracked file paths (relative to repo_root) under dir_relpath at HEAD."""
|
||
|
|
proc = _run_git(repo_root, ["ls-tree", "-r", "--name-only", "HEAD", "--", dir_relpath])
|
||
|
|
if proc.returncode != 0:
|
||
|
|
return []
|
||
|
|
return [line for line in proc.stdout.splitlines() if line]
|
||
|
|
|
||
|
|
|
||
|
|
def load_repo_head_dir(repo_root, dir_relpath):
|
||
|
|
"""{key: canonical_yaml_text} for every *.yaml tracked at HEAD under dir_relpath."""
|
||
|
|
result = {}
|
||
|
|
for relpath in git_ls_tracked(repo_root, dir_relpath):
|
||
|
|
if not relpath.endswith(".yaml"):
|
||
|
|
continue
|
||
|
|
key = os.path.basename(relpath)[: -len(".yaml")]
|
||
|
|
text = git_show(repo_root, relpath)
|
||
|
|
if text is None:
|
||
|
|
continue
|
||
|
|
result[key] = dump_canonical_yaml(parse_yaml_text(text))
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def load_yaml_dir(dir_path):
|
||
|
|
"""{key: canonical_yaml_text} for every *.yaml file in dir_path (working tree)."""
|
||
|
|
result = {}
|
||
|
|
if not os.path.isdir(dir_path):
|
||
|
|
return result
|
||
|
|
for name in sorted(os.listdir(dir_path)):
|
||
|
|
if not name.endswith(".yaml"):
|
||
|
|
continue
|
||
|
|
key = name[: -len(".yaml")]
|
||
|
|
with open(os.path.join(dir_path, name), encoding="utf-8") as f:
|
||
|
|
text = f.read()
|
||
|
|
result[key] = dump_canonical_yaml(parse_yaml_text(text))
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
# --- target resolution ------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def build_targets(config_dir, repo_root, file_args):
|
||
|
|
"""Resolve CLI file arguments to a deploy target list.
|
||
|
|
|
||
|
|
No args -> every automations/scripts/scenes *.yaml file currently in
|
||
|
|
config_dir (working tree) — "wszystkie w scope".
|
||
|
|
|
||
|
|
Explicit args -> each is tried as-is, then repo-root-relative, then
|
||
|
|
config-dir-relative; whichever resolves to an existing file wins. Any
|
||
|
|
file outside config_dir/{automations,scripts,scenes}/ is a hard error
|
||
|
|
(dashboards/helpers are out of scope for this adapter).
|
||
|
|
|
||
|
|
Returns a list of (subdir, key, abs_path), in sorted order for the
|
||
|
|
no-args case and CLI-argument order otherwise.
|
||
|
|
"""
|
||
|
|
if not file_args:
|
||
|
|
targets = []
|
||
|
|
for subdir in DOMAIN_SUBDIRS:
|
||
|
|
dir_path = os.path.join(config_dir, subdir)
|
||
|
|
if not os.path.isdir(dir_path):
|
||
|
|
continue
|
||
|
|
for name in sorted(os.listdir(dir_path)):
|
||
|
|
if name.endswith(".yaml"):
|
||
|
|
targets.append((subdir, name[: -len(".yaml")], os.path.join(dir_path, name)))
|
||
|
|
return targets
|
||
|
|
|
||
|
|
config_dir_abs = os.path.abspath(config_dir)
|
||
|
|
targets = []
|
||
|
|
for arg in file_args:
|
||
|
|
candidates = [arg, os.path.join(repo_root, arg), os.path.join(config_dir, arg)]
|
||
|
|
resolved = next((c for c in candidates if os.path.isfile(c)), None)
|
||
|
|
if resolved is None:
|
||
|
|
raise DeployScopeError(
|
||
|
|
f"file not found: {arg!r} (tried as given, repo-root-relative, "
|
||
|
|
"and config-dir-relative)"
|
||
|
|
)
|
||
|
|
resolved = os.path.abspath(resolved)
|
||
|
|
if not resolved.startswith(config_dir_abs + os.sep):
|
||
|
|
raise DeployScopeError(
|
||
|
|
f"{arg!r} resolves outside {config_dir} — deploy.sh only handles "
|
||
|
|
"this instance's automations/scripts/scenes"
|
||
|
|
)
|
||
|
|
if not resolved.endswith(".yaml"):
|
||
|
|
raise DeployScopeError(f"{arg!r} is not a .yaml file")
|
||
|
|
subdir = os.path.basename(os.path.dirname(resolved))
|
||
|
|
if subdir not in DOMAIN_SUBDIRS:
|
||
|
|
raise DeployScopeError(
|
||
|
|
f"{arg!r} is outside deploy.sh's scope (only automations/scripts/scenes are "
|
||
|
|
"handled here — dashboards/helpers are WS-only and a separate task)"
|
||
|
|
)
|
||
|
|
key = os.path.basename(resolved)[: -len(".yaml")]
|
||
|
|
targets.append((subdir, key, resolved))
|
||
|
|
return targets
|
||
|
|
|
||
|
|
|
||
|
|
# --- 1. drift-check ----------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def drift_check(live_snapshot_dir, repo_root, config_relpath, target_keys):
|
||
|
|
"""Compare a fresh live snapshot (already on disk, canonical) against
|
||
|
|
repo HEAD, for every object NOT in target_keys.
|
||
|
|
|
||
|
|
Returns (aborts, warnings):
|
||
|
|
aborts -- [(subdir, key, reason, live_text_or_None, repo_text_or_None)]
|
||
|
|
content differs, or repo HEAD has an object missing live
|
||
|
|
(deleted out-of-band) -- never silently overwritten.
|
||
|
|
warnings -- [(subdir, key, reason)]
|
||
|
|
object lives on the instance but has no repo file -- DELETE
|
||
|
|
is out of scope for this task, so this is informational only.
|
||
|
|
"""
|
||
|
|
aborts = []
|
||
|
|
warnings = []
|
||
|
|
for subdir in DOMAIN_SUBDIRS:
|
||
|
|
live = load_yaml_dir(os.path.join(live_snapshot_dir, subdir))
|
||
|
|
repo = load_repo_head_dir(repo_root, f"{config_relpath}/{subdir}")
|
||
|
|
for key in sorted(set(live) | set(repo)):
|
||
|
|
if (subdir, key) in target_keys:
|
||
|
|
continue
|
||
|
|
live_val = live.get(key)
|
||
|
|
repo_val = repo.get(key)
|
||
|
|
if live_val is None and repo_val is not None:
|
||
|
|
aborts.append(
|
||
|
|
(
|
||
|
|
subdir,
|
||
|
|
key,
|
||
|
|
"present in repo (HEAD), missing on the live instance — "
|
||
|
|
"deleted out-of-band since the last import",
|
||
|
|
None,
|
||
|
|
repo_val,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
elif live_val is not None and repo_val is None:
|
||
|
|
warnings.append(
|
||
|
|
(
|
||
|
|
subdir,
|
||
|
|
key,
|
||
|
|
"present on the live instance, no repo file — object was removed "
|
||
|
|
"from the repo; DELETE is out of scope here, left untouched on the "
|
||
|
|
"instance",
|
||
|
|
)
|
||
|
|
)
|
||
|
|
elif live_val != repo_val:
|
||
|
|
aborts.append(
|
||
|
|
(
|
||
|
|
subdir,
|
||
|
|
key,
|
||
|
|
"content differs between the live instance and repo HEAD — "
|
||
|
|
"uncaptured drift",
|
||
|
|
live_val,
|
||
|
|
repo_val,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
return aborts, warnings
|
||
|
|
|
||
|
|
|
||
|
|
# --- 2. validation -----------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def local_validate(target_objects):
|
||
|
|
"""target_objects: {(subdir, key): parsed_obj}. Returns [(subdir, key, reason)]."""
|
||
|
|
errors = []
|
||
|
|
for (subdir, key), obj in target_objects.items():
|
||
|
|
if not isinstance(obj, dict):
|
||
|
|
errors.append((subdir, key, "not a mapping at the top level"))
|
||
|
|
continue
|
||
|
|
for required in REQUIRED_KEYS[subdir]:
|
||
|
|
if required not in obj:
|
||
|
|
errors.append((subdir, key, f"missing required key '{required}'"))
|
||
|
|
if subdir in ("automations", "scenes") and "id" in obj and str(obj["id"]) != key:
|
||
|
|
errors.append(
|
||
|
|
(subdir, key, f"'id' field {obj['id']!r} does not match filename '{key}.yaml'")
|
||
|
|
)
|
||
|
|
return errors
|
||
|
|
|
||
|
|
|
||
|
|
def remote_validate(client):
|
||
|
|
"""Live check_config gate. Returns (valid, raw_response)."""
|
||
|
|
return client.check_config()
|
||
|
|
|
||
|
|
|
||
|
|
# --- 3+4. write + verify -----------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def write_and_verify(client, target_objects):
|
||
|
|
"""target_objects: {(subdir, key): parsed_obj}, already locally validated.
|
||
|
|
|
||
|
|
Returns {"written": [...], "write_failed": [...], "verify_failed": [...]}.
|
||
|
|
No auto-rollback on a verify mismatch — writes to other objects continue,
|
||
|
|
and the mismatch is reported so an operator can act on it directly.
|
||
|
|
"""
|
||
|
|
report = {"written": [], "write_failed": [], "verify_failed": []}
|
||
|
|
for (subdir, key), obj in sorted(target_objects.items()):
|
||
|
|
domain = DOMAIN_API_NAME[subdir]
|
||
|
|
try:
|
||
|
|
client.post_config(domain, key, obj)
|
||
|
|
except Exception as exc: # noqa: BLE001 - surface any transport/HTTP error verbatim
|
||
|
|
report["write_failed"].append((subdir, key, str(exc)))
|
||
|
|
continue
|
||
|
|
|
||
|
|
try:
|
||
|
|
live_obj = client.get(f"/api/config/{domain}/config/{key}")
|
||
|
|
except Exception as exc: # noqa: BLE001 - surface any transport error verbatim
|
||
|
|
report["verify_failed"].append((subdir, key, f"verify GET failed: {exc}"))
|
||
|
|
continue
|
||
|
|
|
||
|
|
if live_obj is None:
|
||
|
|
report["verify_failed"].append(
|
||
|
|
(subdir, key, "verify GET returned 404 immediately after a successful write")
|
||
|
|
)
|
||
|
|
continue
|
||
|
|
|
||
|
|
expected_text = dump_canonical_yaml(obj)
|
||
|
|
actual_text = dump_canonical_yaml(live_obj)
|
||
|
|
if expected_text != actual_text:
|
||
|
|
report["verify_failed"].append(
|
||
|
|
(subdir, key, "post-write GET does not match what was written", expected_text, actual_text)
|
||
|
|
)
|
||
|
|
continue
|
||
|
|
|
||
|
|
report["written"].append((subdir, key))
|
||
|
|
return report
|
||
|
|
|
||
|
|
|
||
|
|
# --- orchestration ------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def new_result(instance, dry_run):
|
||
|
|
return {
|
||
|
|
"instance": instance,
|
||
|
|
"dry_run": dry_run,
|
||
|
|
"targets": [],
|
||
|
|
"drift_aborts": [],
|
||
|
|
"drift_warnings": [],
|
||
|
|
"validation_errors": [],
|
||
|
|
"check_config_valid": None,
|
||
|
|
"check_config_raw": None,
|
||
|
|
"written": [],
|
||
|
|
"write_failed": [],
|
||
|
|
"verify_failed": [],
|
||
|
|
"aborted": False,
|
||
|
|
"abort_reason": None,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def run(repo_root, config_dir, config_relpath, instance, file_args, dry_run, client):
|
||
|
|
"""The full DRIFT-CHECK -> VALIDATE -> WRITE -> VERIFY sequence.
|
||
|
|
|
||
|
|
`client` is duck-typed: .get(path), .post_config(domain, id, body),
|
||
|
|
.check_config() -- see ha_write_api.WriteClient for the real
|
||
|
|
implementation and tests/test_deploy_api_offline.sh for the fake used
|
||
|
|
in offline tests.
|
||
|
|
"""
|
||
|
|
result = new_result(instance, dry_run)
|
||
|
|
|
||
|
|
try:
|
||
|
|
targets = build_targets(config_dir, repo_root, file_args)
|
||
|
|
except DeployScopeError as exc:
|
||
|
|
result["aborted"] = True
|
||
|
|
result["abort_reason"] = str(exc)
|
||
|
|
return result
|
||
|
|
|
||
|
|
if not targets:
|
||
|
|
result["aborted"] = True
|
||
|
|
result["abort_reason"] = "no target objects resolved — nothing to deploy"
|
||
|
|
return result
|
||
|
|
|
||
|
|
result["targets"] = [(subdir, key) for subdir, key, _ in targets]
|
||
|
|
target_keys = {(subdir, key) for subdir, key, _ in targets}
|
||
|
|
|
||
|
|
states = client.get("/api/states")
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory(prefix="ha-deploy-drift-") as snapshot_dir:
|
||
|
|
live_report = import_api.new_report()
|
||
|
|
import_api.import_automations(client, states, snapshot_dir, live_report)
|
||
|
|
import_api.import_scripts(client, states, snapshot_dir, live_report)
|
||
|
|
import_api.import_scenes(client, states, snapshot_dir, live_report)
|
||
|
|
aborts, warnings = drift_check(snapshot_dir, repo_root, config_relpath, target_keys)
|
||
|
|
|
||
|
|
result["drift_aborts"] = aborts
|
||
|
|
result["drift_warnings"] = warnings
|
||
|
|
if aborts:
|
||
|
|
result["aborted"] = True
|
||
|
|
result["abort_reason"] = "drift detected outside deploy scope — never overwritten"
|
||
|
|
return result
|
||
|
|
|
||
|
|
target_objects = {}
|
||
|
|
parse_errors = []
|
||
|
|
for subdir, key, path in targets:
|
||
|
|
with open(path, encoding="utf-8") as f:
|
||
|
|
text = f.read()
|
||
|
|
try:
|
||
|
|
target_objects[(subdir, key)] = parse_yaml_text(text)
|
||
|
|
except Exception as exc: # noqa: BLE001 - any parse failure is a validation rejection
|
||
|
|
parse_errors.append((subdir, key, f"does not parse as YAML: {exc}"))
|
||
|
|
|
||
|
|
result["validation_errors"] = parse_errors + local_validate(target_objects)
|
||
|
|
if result["validation_errors"]:
|
||
|
|
result["aborted"] = True
|
||
|
|
result["abort_reason"] = "local validation failed — nothing was written"
|
||
|
|
return result
|
||
|
|
|
||
|
|
valid, raw = remote_validate(client)
|
||
|
|
result["check_config_valid"] = valid
|
||
|
|
result["check_config_raw"] = raw
|
||
|
|
if not valid:
|
||
|
|
result["aborted"] = True
|
||
|
|
result["abort_reason"] = "instance check_config reported invalid — nothing was written"
|
||
|
|
return result
|
||
|
|
|
||
|
|
if dry_run:
|
||
|
|
return result
|
||
|
|
|
||
|
|
write_report = write_and_verify(client, target_objects)
|
||
|
|
result["written"] = write_report["written"]
|
||
|
|
result["write_failed"] = write_report["write_failed"]
|
||
|
|
result["verify_failed"] = write_report["verify_failed"]
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
# --- reporting ----------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def _print_diff(live_text, repo_text, label_live, label_repo):
|
||
|
|
diff = difflib.unified_diff(
|
||
|
|
(repo_text or "").splitlines(keepends=True),
|
||
|
|
(live_text or "").splitlines(keepends=True),
|
||
|
|
fromfile=label_repo,
|
||
|
|
tofile=label_live,
|
||
|
|
)
|
||
|
|
sys.stderr.writelines(diff)
|
||
|
|
|
||
|
|
|
||
|
|
def print_report(result):
|
||
|
|
mode = "DRY-RUN (plan only, no mutating calls)" if result["dry_run"] else "LIVE"
|
||
|
|
print(f"== ha deploy: instance={result['instance']} mode={mode} ==", file=sys.stderr)
|
||
|
|
|
||
|
|
print(f"-> targets ({len(result['targets'])}):", file=sys.stderr)
|
||
|
|
for subdir, key in result["targets"]:
|
||
|
|
print(f" {subdir}/{key}.yaml", file=sys.stderr)
|
||
|
|
|
||
|
|
for subdir, key, reason in result["drift_warnings"]:
|
||
|
|
print(f"-> warning: {subdir}/{key}: {reason}", file=sys.stderr)
|
||
|
|
|
||
|
|
if result["drift_aborts"]:
|
||
|
|
print("-> DRIFT DETECTED — aborting, nothing was written:", file=sys.stderr)
|
||
|
|
for subdir, key, reason, live_text, repo_text in result["drift_aborts"]:
|
||
|
|
print(f" {subdir}/{key}: {reason}", file=sys.stderr)
|
||
|
|
_print_diff(live_text, repo_text, f"live:{subdir}/{key}", f"repo-HEAD:{subdir}/{key}")
|
||
|
|
|
||
|
|
if result["validation_errors"]:
|
||
|
|
print("-> LOCAL VALIDATION FAILED — aborting, nothing was written:", file=sys.stderr)
|
||
|
|
for subdir, key, reason in result["validation_errors"]:
|
||
|
|
print(f" {subdir}/{key}: {reason}", file=sys.stderr)
|
||
|
|
|
||
|
|
if result["check_config_valid"] is not None:
|
||
|
|
status = "valid" if result["check_config_valid"] else "INVALID"
|
||
|
|
print(f"-> check_config: {status}", file=sys.stderr)
|
||
|
|
if not result["check_config_valid"]:
|
||
|
|
print(f" {result['check_config_raw']}", file=sys.stderr)
|
||
|
|
|
||
|
|
if result["aborted"]:
|
||
|
|
print(f"== ABORTED: {result['abort_reason']} ==", file=sys.stderr)
|
||
|
|
return
|
||
|
|
|
||
|
|
if result["dry_run"]:
|
||
|
|
print("== dry-run complete: plan above would be applied on a live run ==", file=sys.stderr)
|
||
|
|
return
|
||
|
|
|
||
|
|
print(f"-> written ({len(result['written'])}):", file=sys.stderr)
|
||
|
|
for subdir, key in result["written"]:
|
||
|
|
print(f" {subdir}/{key}.yaml", file=sys.stderr)
|
||
|
|
|
||
|
|
if result["write_failed"]:
|
||
|
|
print(f"-> write failed ({len(result['write_failed'])}):", file=sys.stderr)
|
||
|
|
for subdir, key, reason in result["write_failed"]:
|
||
|
|
print(f" {subdir}/{key}: {reason}", file=sys.stderr)
|
||
|
|
|
||
|
|
if result["verify_failed"]:
|
||
|
|
print(f"-> verify failed ({len(result['verify_failed'])}):", file=sys.stderr)
|
||
|
|
for entry in result["verify_failed"]:
|
||
|
|
subdir, key, reason = entry[0], entry[1], entry[2]
|
||
|
|
print(f" {subdir}/{key}: {reason}", file=sys.stderr)
|
||
|
|
if len(entry) == 5:
|
||
|
|
_print_diff(entry[3], entry[4], f"written:{subdir}/{key}", f"live-after-write:{subdir}/{key}")
|
||
|
|
|
||
|
|
total_errors = len(result["write_failed"]) + len(result["verify_failed"])
|
||
|
|
print(
|
||
|
|
f"== deploy complete: {len(result['written'])} written, "
|
||
|
|
f"{len(result['write_failed'])} write errors, {len(result['verify_failed'])} verify errors ==",
|
||
|
|
file=sys.stderr,
|
||
|
|
)
|
||
|
|
if total_errors:
|
||
|
|
print("== see errors above — no auto-rollback was performed ==", file=sys.stderr)
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv):
|
||
|
|
if len(argv) < 8:
|
||
|
|
print(
|
||
|
|
"usage: deploy_api.py <base_url> <token_path> <repo_root> <config_dir> "
|
||
|
|
"<config_relpath> <instance> <0|1:dry_run> [file...]",
|
||
|
|
file=sys.stderr,
|
||
|
|
)
|
||
|
|
return 2
|
||
|
|
_, base_url, token_path, repo_root, config_dir, config_relpath, instance, dry_run_flag = argv[:8]
|
||
|
|
file_args = argv[8:]
|
||
|
|
dry_run = dry_run_flag == "1"
|
||
|
|
|
||
|
|
try:
|
||
|
|
token = ha_api.read_token(token_path)
|
||
|
|
except ha_api.HaApiError as exc:
|
||
|
|
print(f"error: {exc}", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
|
||
|
|
client = ha_write_api.WriteClient(base_url, token)
|
||
|
|
|
||
|
|
try:
|
||
|
|
result = run(repo_root, config_dir, config_relpath, instance, file_args, dry_run, client)
|
||
|
|
except Exception as exc: # noqa: BLE001 - surface any transport/HTTP error verbatim
|
||
|
|
print(f"error: deploy failed: {exc}", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
|
||
|
|
print_report(result)
|
||
|
|
|
||
|
|
if result["aborted"] or result["write_failed"] or result["verify_failed"]:
|
||
|
|
return 1
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main(sys.argv))
|