#!/usr/bin/env python3 """Deploy path (repo -> instance) for the HA "api" adapter — see kb/decisions/ha-configs-as-code.md, "Sync model" and "Validation gate". Scope: automations/scripts/scenes only (the domains reachable via `/api/config//config/`). 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 " " <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))