oskar/homelab-codex-ws
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

feat(ha): deploy.sh — api write path with drift-abort and verify

Browse source
This commit is contained in:
oskar 2026-07-22 18:21:11 +02:00
parent a8c2071e11
commit adeed6b902
7 changed files with 992 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
docs/backlog.md
View file

@ -31,6 +31,16 @@ dawny kontener piha (docker-exec, archived).
nie skonfigurowany — `config_not_found`, odnotowany w raporcie, nie
twardy błąd). Pełny import `/config` pozostaje poza zasięgiem (HAOS bez
SSH) — patrz DESIGN.md.
4. ✅ ZROBIONE (2026-07-22) — **`scripts/ha/deploy.sh`, adapter `api`, zakres
automations/scripts/scenes**: drift-check (świeży re-import vs. `HEAD`,
dowolna różnica poza plikami z tego deployu = abort z diffem) →
walidacja (lokalny sanity check + `check_config` na instancji) → zapis
per obiekt (`POST /api/config/<domain>/config/<id>`) → verify (GET +
porównanie, bez auto-rollbacku). `--dry-run` zweryfikowany na żywym
`ken` (read-only, bez różnic). Testy offline:
`scripts/ha/tests/test_deploy_api_offline.sh`. Poza zakresem: dashboardy/
helpery (WS, brak mutującej komendy), adapter `docker-exec`, DELETE
obiektów usuniętych z repo (tylko ostrzeżenie).
---

96
scripts/ha/deploy.sh Executable file
View file

@ -0,0 +1,96 @@
#!/usr/bin/env bash
# Write path: repo -> HA instance, for the "api" adapter only (see
# services/home-assistant/DESIGN.md, "Deploy path: adapter per instance",
# "Sync model", "Validation gate").
#
# Scope: automations/scripts/scenes only. Dashboards and helpers are
# WebSocket-only (scripts/ha/lib/ha_ws.py has no mutating command,
# deliberately) — writing those is a separate, not-yet-built task.
#
# Hard sequence enforced in lib/deploy_api.py, no shortcuts: DRIFT-CHECK ->
# VALIDATE (local + live check_config) -> WRITE (per object) -> VERIFY (per
# object). Any drift or validation failure aborts before a single mutating
# call is made. --dry-run runs the same DRIFT-CHECK/VALIDATE steps (both
# read-only in effect) and stops before WRITE/VERIFY.
#
# Instances with status != "active" (e.g. ken-legacy: archived, chelsty-ha:
# offline) or adapter != "api" are refused unconditionally — see
# instances.yaml and DESIGN.md's "Incident log" for why ken-legacy must
# never be a deploy target again.
#
# Usage: deploy.sh <instance> [--dry-run] [file...]
# no file args -> every automations/scripts/scenes file currently in
# services/home-assistant/config/<instance>/
# file args -> only those objects (path as given, repo-root-relative,
# or config-dir-relative all accepted)
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SERVICE_DIR="$(cd "$SCRIPT_DIR/../../services/home-assistant" && pwd)"
REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
INSTANCES_FILE="$SERVICE_DIR/instances.yaml"
usage() {
echo "usage: $0 <instance> [--dry-run] [file...]" >&2
}
if [[ $# -lt 1 ]]; then
usage
exit 2
fi
INSTANCE="$1"
shift
DRY_RUN=0
FILE_ARGS=()
for arg in "$@"; do
case "$arg" in
--dry-run) DRY_RUN=1 ;;
--) ;;
*) FILE_ARGS+=("$arg") ;;
esac
done
if [[ ! -f "$INSTANCES_FILE" ]]; then
echo "error: instances file not found: $INSTANCES_FILE" >&2
exit 1
fi
INSTANCE_CONFIG_OUTPUT="$(python3 "$SCRIPT_DIR/lib/instance_config.py" "$INSTANCES_FILE" "$INSTANCE")" || exit 1
eval "$INSTANCE_CONFIG_OUTPUT"
if [[ "$HA_STATUS" != "active" ]]; then
echo "error: instance '$INSTANCE' has status='$HA_STATUS', not 'active' — deploy.sh refuses" >&2
echo " non-active instances unconditionally, no override flag (see instances.yaml)." >&2
exit 1
fi
if [[ "$HA_ADAPTER" != "api" ]]; then
echo "error: instance '$INSTANCE' uses adapter='$HA_ADAPTER' — deploy.sh only implements" >&2
echo " the 'api' write path (see DESIGN.md, 'Deploy path: adapter per instance')." >&2
exit 1
fi
api_token_file="${HA_TOKEN_PATH/#\~/$HOME}"
if [[ -z "$HA_TOKEN_PATH" || ! -f "$api_token_file" ]]; then
echo "error: instance '$INSTANCE' has no token at '${api_token_file:-<unset>}' — deploy.sh" >&2
echo " has no fallback without a deploy_agent long-lived access token (see" >&2
echo " DESIGN.md, 'Tokens')." >&2
exit 1
fi
CONFIG_DIR="$SERVICE_DIR/config/$INSTANCE"
CONFIG_RELPATH="services/home-assistant/config/$INSTANCE"
if [[ ! -d "$CONFIG_DIR" ]]; then
echo "error: no config/$INSTANCE/ directory — run scripts/ha/import.sh $INSTANCE at least" >&2
echo " once first, so there is a last-known snapshot to drift-check against." >&2
exit 1
fi
echo "== ha deploy: instance=$INSTANCE adapter=$HA_ADAPTER dry_run=$DRY_RUN ==" >&2
python3 "$SCRIPT_DIR/lib/deploy_api.py" \
"$HA_BASE_URL" "$api_token_file" "$REPO_ROOT" "$CONFIG_DIR" "$CONFIG_RELPATH" "$INSTANCE" "$DRY_RUN" \
"${FILE_ARGS[@]}"

500
scripts/ha/lib/deploy_api.py Executable file
View file

@ -0,0 +1,500 @@
#!/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))

59
scripts/ha/lib/ha_write_api.py Executable file
View file

@ -0,0 +1,59 @@
#!/usr/bin/env python3
"""Write-capable REST client for the HA "api" adapter's deploy path (see
services/home-assistant/DESIGN.md, "Sync model", "Validation gate").
Deliberately kept out of ha_api.py, whose module docstring guarantees that
module is read-only by construction — the import path relies on that
guarantee staying true. This module is the only place under scripts/ha/
that ever issues a mutating call to a live HA instance, and it exposes only
the two operations DESIGN.md's deploy sequence requires:
- check_config() -> POST /api/config/core/check_config (validation gate)
- post_config(...) -> POST /api/config/<domain>/config/<id> (per-object write)
Same token discipline as ha_api.py: callers pass a token_path (via
ha_api.read_token), never a raw token in argv or a log line; the
Authorization header is set in-process via `requests`.
"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import requests # noqa: E402
import ha_api # noqa: E402
class WriteClient(ha_api.Client):
"""Extends the read-only Client with the two mutating calls deploy.sh needs."""
def post_config(self, domain, object_id, body):
"""POST `body` (a dict) to /api/config/<domain>/config/<object_id>.
Returns the parsed JSON response. Raises requests.HTTPError on a
non-2xx status — deploy_api.py treats a failed write as a
per-object error, never a silent skip.
"""
path = f"/api/config/{domain}/config/{object_id}"
resp = requests.post(
self._base_url + path, headers=self._headers(), json=body, timeout=self._timeout
)
resp.raise_for_status()
return resp.json()
def check_config(self):
"""POST /api/config/core/check_config — HA's built-in config validator.
Returns (valid: bool, raw: dict). Only raises on a transport/HTTP
failure; an "invalid" result is a normal outcome the caller gates
on, not an exception.
"""
resp = requests.post(
self._base_url + "/api/config/core/check_config",
headers=self._headers(),
timeout=self._timeout,
)
resp.raise_for_status()
data = resp.json()
return data.get("result") == "valid", data

277
scripts/ha/tests/test_deploy_api_offline.sh Executable file
View file

@ -0,0 +1,277 @@
#!/usr/bin/env bash
# Offline tests for the deploy write path (scripts/ha/lib/deploy_api.py,
# ha_write_api.py): clean deploy, drift-abort, validation rejecting broken
# YAML, and verify detecting a post-write mismatch. No network access, no
# HA instance needed -- REST calls are replaced with a fake in-memory
# client (get/post_config/check_config); "repo HEAD" is a real temporary
# git repo so git_show/git_ls_tracked exercise the actual git plumbing.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
HA_LIB_DIR="$SCRIPT_DIR/../lib"
FIXTURES_DIR="$SCRIPT_DIR/fixtures"
python3 - "$HA_LIB_DIR" "$FIXTURES_DIR" <<'PYEOF'
import json
import os
import subprocess
import sys
import tempfile
lib_dir, fixtures_dir = sys.argv[1:3]
sys.path.insert(0, lib_dir)
import deploy_api # noqa: E402
from normalize import dump_canonical_yaml # noqa: E402
fail = False
def check(condition, message):
global fail
if condition:
print(f"PASS: {message}")
else:
print(f"FAIL: {message}", file=sys.stderr)
fail = True
def load_fixture(name):
with open(os.path.join(fixtures_dir, name), encoding="utf-8") as f:
return json.load(f)
STATES = load_fixture("api_states.json")
AUTOMATION_111 = load_fixture("api_automation_config_111.json")
SCRIPT_DEMO = load_fixture("api_script_config_demo_script.json")
SCENE_222 = load_fixture("api_scene_config_222.json")
CONFIG_RELPATH = "services/home-assistant/config/testinst"
class FakeClient:
"""Duck-types ha_write_api.WriteClient: .get, .post_config, .check_config."""
def __init__(self, get_responses, check_config_result=(True, {"result": "valid"})):
self._get_responses = dict(get_responses)
self._check_config_result = check_config_result
self.posted = []
self.check_config_calls = 0
def get(self, path):
if path == "/api/states":
return STATES
return self._get_responses.get(path)
def post_config(self, domain, object_id, body):
self.posted.append((domain, object_id, body))
return {"result": "ok"}
def check_config(self):
self.check_config_calls += 1
return self._check_config_result
def init_repo_with_committed_objects(repo_root):
"""Commit automations/111, scripts/demo_script, scenes/222 at HEAD, exactly
as import.sh would have written them (canonical YAML, split one-object-
per-file)."""
config_dir = os.path.join(repo_root, CONFIG_RELPATH)
for subdir, key, obj in (
("automations", "111", AUTOMATION_111),
("scripts", "demo_script", SCRIPT_DEMO),
("scenes", "222", SCENE_222),
):
d = os.path.join(config_dir, subdir)
os.makedirs(d, exist_ok=True)
with open(os.path.join(d, f"{key}.yaml"), "w", encoding="utf-8") as f:
f.write(dump_canonical_yaml(obj))
subprocess.run(["git", "init", "-q", repo_root], check=True)
subprocess.run(["git", "-C", repo_root, "add", "-A"], check=True)
subprocess.run(
[
"git",
"-C",
repo_root,
"-c",
"user.email=test@example.com",
"-c",
"user.name=test",
"commit",
"-q",
"-m",
"initial import",
],
check=True,
)
return config_dir
# --- Case A: clean deploy -- all three committed objects redeployed unchanged ---
with tempfile.TemporaryDirectory() as repo_root:
config_dir = init_repo_with_committed_objects(repo_root)
get_responses = {
"/api/config/automation/config/111": AUTOMATION_111,
"/api/config/script/config/demo_script": SCRIPT_DEMO,
"/api/config/scene/config/222": SCENE_222,
}
client = FakeClient(get_responses)
result = deploy_api.run(repo_root, config_dir, CONFIG_RELPATH, "testinst", [], False, client)
check(not result["aborted"], "clean deploy: not aborted")
check(result["drift_aborts"] == [], "clean deploy: no drift aborts (all objects are targets)")
check(result["validation_errors"] == [], "clean deploy: no validation errors")
check(result["check_config_valid"] is True, "clean deploy: check_config reported valid")
check(len(result["written"]) == 3, "clean deploy: all three objects written")
check(result["write_failed"] == [], "clean deploy: no write failures")
check(result["verify_failed"] == [], "clean deploy: no verify failures")
check(
set(client.posted[i][0:2] for i in range(3))
== {("automation", "111"), ("script", "demo_script"), ("scene", "222")},
"clean deploy: posted exactly the three expected (domain, id) pairs",
)
# --- Case B: drift-abort -- a non-target object diverged on the live instance ---
with tempfile.TemporaryDirectory() as repo_root:
config_dir = init_repo_with_committed_objects(repo_root)
drifted_automation = dict(AUTOMATION_111)
drifted_automation["alias"] = "Evil override made directly in the HA UI"
get_responses = {
"/api/config/automation/config/111": drifted_automation, # drift: differs from repo HEAD
"/api/config/script/config/demo_script": SCRIPT_DEMO,
"/api/config/scene/config/222": SCENE_222,
}
client = FakeClient(get_responses)
# Only redeploying the script -- automation/111 and scene/222 are NOT
# targets, so any difference there must abort the whole run.
result = deploy_api.run(
repo_root,
config_dir,
CONFIG_RELPATH,
"testinst",
["scripts/demo_script.yaml"],
False,
client,
)
check(result["aborted"], "drift-abort: run is aborted")
check(
any(a[0] == "automations" and a[1] == "111" for a in result["drift_aborts"]),
"drift-abort: automations/111 reported as drifted",
)
check(client.posted == [], "drift-abort: no write was ever attempted")
check(client.check_config_calls == 0, "drift-abort: check_config was never reached")
# --- Case C: local validation rejects an object missing a required key ---
with tempfile.TemporaryDirectory() as repo_root:
config_dir = init_repo_with_committed_objects(repo_root)
broken_dir = os.path.join(config_dir, "scripts")
with open(os.path.join(broken_dir, "broken_script.yaml"), "w", encoding="utf-8") as f:
f.write(dump_canonical_yaml({"alias": "broken, no sequence key"}))
# Deliberately left uncommitted -- it's a new object being deployed for
# the first time, which is exactly the case target-exclusion is for.
# The other three committed objects must match live state exactly so
# drift-check passes cleanly and the run reaches local validation.
get_responses = {
"/api/config/automation/config/111": AUTOMATION_111,
"/api/config/script/config/demo_script": SCRIPT_DEMO,
"/api/config/scene/config/222": SCENE_222,
}
client = FakeClient(get_responses)
result = deploy_api.run(
repo_root,
config_dir,
CONFIG_RELPATH,
"testinst",
["scripts/broken_script.yaml"],
False,
client,
)
check(result["aborted"], "broken YAML: run is aborted")
check(
any(e[0] == "scripts" and e[1] == "broken_script" for e in result["validation_errors"]),
"broken YAML: scripts/broken_script reported as a validation error",
)
check(client.posted == [], "broken YAML: no write was ever attempted")
check(client.check_config_calls == 0, "broken YAML: check_config was never reached")
with tempfile.TemporaryDirectory() as repo_root:
config_dir = init_repo_with_committed_objects(repo_root)
unparseable_dir = os.path.join(config_dir, "scripts")
with open(os.path.join(unparseable_dir, "unparseable.yaml"), "w", encoding="utf-8") as f:
f.write("alias: [unterminated flow sequence\n")
get_responses = {
"/api/config/automation/config/111": AUTOMATION_111,
"/api/config/script/config/demo_script": SCRIPT_DEMO,
"/api/config/scene/config/222": SCENE_222,
}
client = FakeClient(get_responses)
result = deploy_api.run(
repo_root,
config_dir,
CONFIG_RELPATH,
"testinst",
["scripts/unparseable.yaml"],
False,
client,
)
check(result["aborted"], "unparseable YAML: run is aborted, not a Python traceback")
check(
any(e[0] == "scripts" and e[1] == "unparseable" for e in result["validation_errors"]),
"unparseable YAML: reported as a validation error naming the file",
)
# --- Case D: verify detects a post-write mismatch (no auto-rollback) ---
with tempfile.TemporaryDirectory() as repo_root:
config_dir = init_repo_with_committed_objects(repo_root)
get_responses = {
"/api/config/automation/config/111": AUTOMATION_111,
# verify GET for the script comes back different from what was written --
# e.g. the instance silently coerced/rejected a field.
"/api/config/script/config/demo_script": {**SCRIPT_DEMO, "alias": "mutated_by_instance"},
"/api/config/scene/config/222": SCENE_222,
}
client = FakeClient(get_responses)
result = deploy_api.run(repo_root, config_dir, CONFIG_RELPATH, "testinst", [], False, client)
check(not result["aborted"], "verify-mismatch: run completes (write/verify errors are reported, not an abort)")
check(
any(v[0] == "scripts" and v[1] == "demo_script" for v in result["verify_failed"]),
"verify-mismatch: scripts/demo_script reported as a verify failure",
)
check(
any(w[0] == "automations" and w[1] == "111" for w in result["written"])
and any(w[0] == "scenes" and w[1] == "222" for w in result["written"]),
"verify-mismatch: the other two objects still wrote and verified successfully (no auto-rollback)",
)
check(
("script", "demo_script") in {(p[0], p[1]) for p in client.posted},
"verify-mismatch: the mismatched object was still posted (verify runs after write, not before)",
)
sys.exit(1 if fail else 0)
PYEOF

10
services/home-assistant/DESIGN.md
View file

@ -1,15 +1,17 @@
# Home Assistant configs-as-code — design decisions
Status: **skeleton only**. No deploy path is implemented yet — this document
and the accompanying structure/import tooling are phase 0/1 scaffolding.
See `docs/backlog.md` for the tracking entry.
Status: **phase 1 (partial)**. `scripts/ha/deploy.sh` implements the write
path for the `api` adapter's automations/scripts/scenes scope (see "Deploy
path" and "Sync model" below); dashboards/helpers and the `docker-exec`
adapter have no write path yet. See `docs/backlog.md` for the tracking
entry.
## Phasing
| Phase | Scope |
|---|---|
| **0 — Snapshot** | Import-only tooling (this skeleton). Pull `/config` + `.storage` from each instance into the repo, read-only. No deploy, no write path back to HA. |
| **1 — Repo + CC** | Repo is the reviewable source of truth. Changes are authored in the repo (by a human or Claude Code) and pushed manually via the docker-exec / api adapters described below. Deploy has a hard drift-abort (see Sync model). |
| **1 — Repo + CC** | Repo is the reviewable source of truth. Changes are authored in the repo (by a human or Claude Code) and pushed manually via the docker-exec / api adapters described below. Deploy has a hard drift-abort (see Sync model). **Partially built**: `scripts/ha/deploy.sh` covers the `api` adapter's automations/scripts/scenes scope; `docker-exec` deploy and dashboards/helpers writes are still open. |
| **2 — MCP read-only** | Expose HA state (entities, areas, config) to agents via an MCP server in read-only mode — either a self-hosted MCP or `hass-mcp` — to let agents reason about the live instance without touching import/deploy paths. Undecided which (see Open questions). |
| **3 — Agents** | Agents propose changes (automations, scripts, scenes) through the same reviewable repo path used by humans; the human-in-the-loop approval flow from `services/control-plane/` (pending → approved → executed) governs anything destructive. Telegram becomes a first-class interface alongside CC. |

48
services/home-assistant/README.md
View file

@ -1,8 +1,11 @@
# home-assistant (configs-as-code)
**Status: skeleton.** Structure and read-only import tooling only — no
deploy path exists yet. See `DESIGN.md` for the full phasing, adapter, sync,
and validation model, and for the open questions still blocking phase 2/3.
**Status: phase 1 (partial).** Read-only import tooling, plus a deploy
(repo -> instance) write path for the `api` adapter's automations/scripts/
scenes scope only (`scripts/ha/deploy.sh`) — see "Deploy" below. Dashboards,
helpers, and the `docker-exec` adapter have no write path yet. See
`DESIGN.md` for the full phasing, adapter, sync, and validation model, and
for the open questions still blocking phase 2/3.
## Layout
@ -87,16 +90,53 @@ naming the package to install — it does not crash with a raw traceback, and
it does not silently produce an incomplete `storage-export/` without saying
so.
## Deploy
```bash
scripts/ha/deploy.sh ken --dry-run # plan only, read-only
scripts/ha/deploy.sh ken --dry-run automations/111.yaml # plan for one object
scripts/ha/deploy.sh ken # deploy everything in scope
scripts/ha/deploy.sh ken scripts/notify_email_ntfy.yaml # deploy one object
```
Repo -> instance, `api` adapter only, automations/scripts/scenes only
(dashboards/helpers are WebSocket-only — `ha_ws.py` has no mutating
command, deliberately, and that write path doesn't exist yet). Refuses any
instance with `status != active` (see `instances.yaml`) or an adapter other
than `api` — no override flag.
Hard sequence, per `DESIGN.md`'s "Sync model" and "Validation gate", any
failure aborts before the next step:
1. **drift-check** — re-imports the instance's current state and diffs it
against the last commit (`HEAD`) for every object *not* being deployed
this run. Any difference aborts with a diff printed; drift is never
silently overwritten.
2. **validate** — local sanity (YAML parses, required keys present per
domain) plus a live `check_config` gate on the instance.
3. **write** — one `POST /api/config/<domain>/config/<id>` per object.
4. **verify** — GETs the same object back and compares it to what was
written. A mismatch is reported (which object, what differs); there is
no auto-rollback.
`--dry-run` runs steps 1-2 (both read-only in effect) and stops before any
write, printing the same plan a live run would act on.
## Tests
```bash
scripts/ha/tests/test_split_normalize.sh
scripts/ha/tests/test_normalize_tags.sh
scripts/ha/tests/test_import_api_offline.sh
scripts/ha/tests/test_deploy_api_offline.sh
```
All offline — no network, no HA instance required. `test_import_api_offline.sh`
covers the `api` adapter: normalization of real-shaped API responses (saved
as fixtures under `tests/fixtures/`), idempotency, stale-file cleanup, and
the missing-`websocket-client` error path (via `sys.modules` injection, not
an actual network call).
an actual network call). `test_deploy_api_offline.sh` covers the deploy
write path: a clean deploy, drift-abort, local validation rejecting broken/
unparseable YAML, and verify detecting a post-write mismatch — using a real
temporary git repo for the `HEAD` comparison and a fake in-memory client for
`get`/`post_config`/`check_config` (no network).