Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0716234d00 | ||
|
|
a8121aa4e9 |
74
docs/LAB_003_BATTERY_DISCONNECT_PACKET.md
Normal file
74
docs/LAB_003_BATTERY_DISCONNECT_PACKET.md
Normal file
@@ -0,0 +1,74 @@
|
||||
# LAB-003 — Truck Battery Disconnect Install Packet
|
||||
|
||||
No battery disconnect switch has been purchased or installed yet.
|
||||
This packet turns the issue into a field-ready purchase / install / validation checklist while preserving what still requires live work.
|
||||
|
||||
## Candidate Store Run
|
||||
|
||||
- AutoZone — Newport or Claremont
|
||||
- Advance Auto Parts — Newport or Claremont
|
||||
- O'Reilly Auto Parts — Newport or Claremont
|
||||
|
||||
## Required Items
|
||||
|
||||
- battery terminal disconnect switch
|
||||
- terminal shim/post riser if needed
|
||||
|
||||
## Selection Criteria
|
||||
|
||||
- Fits the truck battery post without forcing the clamp
|
||||
- Mounts on the negative battery terminal
|
||||
- Physically secure once tightened
|
||||
- no special tools required to operate
|
||||
|
||||
## Live Purchase State
|
||||
|
||||
- Store selected: pending
|
||||
- Part selected: pending
|
||||
- Part cost: pending purchase
|
||||
|
||||
## Installation Target
|
||||
|
||||
- Install location: negative battery terminal
|
||||
- Ready to operate without tools: yes
|
||||
|
||||
## Install Checklist
|
||||
|
||||
- [ ] Verify the truck is off and keys are removed before touching the battery
|
||||
- [ ] Confirm the disconnect fits the negative battery terminal before final tightening
|
||||
- [ ] Install the disconnect on the negative battery terminal
|
||||
- [ ] Tighten until physically secure with no terminal wobble
|
||||
- [ ] Verify the disconnect can be opened and closed by hand
|
||||
|
||||
## Validation Checklist
|
||||
|
||||
- [ ] Leave the truck parked with the disconnect opened for at least 24 hours
|
||||
- [ ] Reconnect the switch by hand the next day
|
||||
- [ ] Truck starts reliably after sitting 24+ hours with switch disconnected
|
||||
- [ ] Receipt or photo of installed switch uploaded to this issue
|
||||
|
||||
## Overnight Verification Log
|
||||
|
||||
- Install completed: False
|
||||
- Physically secure: False
|
||||
- Overnight disconnect duration: pending
|
||||
- Truck started after disconnect: pending
|
||||
- Receipt / photo path: pending
|
||||
|
||||
## Battery Replacement Fallback
|
||||
|
||||
If the truck still fails the overnight test after the disconnect install, replace battery and re-run the 24-hour validation.
|
||||
|
||||
## Missing Live Fields
|
||||
|
||||
- store_selected
|
||||
- part_name
|
||||
- install_completed
|
||||
- physically_secure
|
||||
- overnight_test_hours
|
||||
- truck_started_after_disconnect
|
||||
- receipt_or_photo_path
|
||||
|
||||
## Honest next step
|
||||
|
||||
Buy the disconnect switch, install it on the negative battery terminal, leave the truck disconnected for 24+ hours, and only close the issue after receipt/photo evidence and the overnight start result are attached.
|
||||
@@ -1,14 +0,0 @@
|
||||
fleet_name: timmy-phase-3-config
|
||||
|
||||
targets:
|
||||
- host: ezra
|
||||
config_root: /root/wizards/ezra/home/.hermes
|
||||
files:
|
||||
- config.yaml
|
||||
- dispatch/rules.json
|
||||
|
||||
- host: bezalel
|
||||
config_root: /root/wizards/bezalel/home/.hermes
|
||||
files:
|
||||
- config.yaml
|
||||
- dispatch/rules.json
|
||||
@@ -1,292 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Plan atomic fleet config sync releases.
|
||||
|
||||
Refs: timmy-home #550
|
||||
|
||||
Phase-3 orchestration slice:
|
||||
- define a shared config-sync manifest for fleet hosts
|
||||
- fingerprint the exact config payload into one release id
|
||||
- generate per-host staging paths and atomic symlink-swap promotion metadata
|
||||
- stay dry-run by default so rollout planning is safe to verify locally
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
DEFAULT_INVENTORY_FILE = Path(__file__).resolve().parents[1] / "ansible" / "inventory" / "hosts.ini"
|
||||
|
||||
|
||||
def load_inventory_hosts(path: str | Path) -> dict[str, dict[str, str]]:
|
||||
hosts: dict[str, dict[str, str]] = {}
|
||||
section = None
|
||||
for raw_line in Path(path).read_text(encoding="utf-8").splitlines():
|
||||
line = raw_line.strip()
|
||||
if not line or line.startswith("#") or line.startswith(";"):
|
||||
continue
|
||||
if line.startswith("[") and line.endswith("]"):
|
||||
section = line[1:-1].strip().lower()
|
||||
continue
|
||||
if section != "fleet":
|
||||
continue
|
||||
|
||||
parts = line.split()
|
||||
host = parts[0]
|
||||
metadata = {"host": host}
|
||||
for token in parts[1:]:
|
||||
if "=" not in token:
|
||||
continue
|
||||
key, value = token.split("=", 1)
|
||||
metadata[key] = value
|
||||
hosts[host] = metadata
|
||||
|
||||
if not hosts:
|
||||
raise ValueError("inventory defines no [fleet] hosts")
|
||||
return hosts
|
||||
|
||||
|
||||
|
||||
def load_manifest(path: str | Path) -> dict[str, Any]:
|
||||
data = yaml.safe_load(Path(path).read_text(encoding="utf-8")) or {}
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("manifest must contain a YAML object")
|
||||
data.setdefault("fleet_name", "timmy-fleet-config")
|
||||
data.setdefault("targets", [])
|
||||
if not isinstance(data["targets"], list):
|
||||
raise ValueError("targets must be a list")
|
||||
return data
|
||||
|
||||
|
||||
|
||||
def _normalize_relative_path(value: str) -> str:
|
||||
path = Path(value)
|
||||
if path.is_absolute():
|
||||
raise ValueError(f"sync file path must be relative: {value}")
|
||||
if any(part == ".." for part in path.parts):
|
||||
raise ValueError(f"sync file path may not escape source root: {value}")
|
||||
normalized = path.as_posix()
|
||||
if normalized in {"", "."}:
|
||||
raise ValueError("sync file path may not be empty")
|
||||
return normalized
|
||||
|
||||
|
||||
|
||||
def validate_manifest(manifest: dict[str, Any], inventory_hosts: dict[str, dict[str, str]]) -> None:
|
||||
targets = manifest.get("targets", [])
|
||||
if not targets:
|
||||
raise ValueError("manifest must define at least one sync target")
|
||||
|
||||
seen_hosts: set[str] = set()
|
||||
for target in targets:
|
||||
if not isinstance(target, dict):
|
||||
raise ValueError("each target must be a mapping")
|
||||
|
||||
host = str(target.get("host", "")).strip()
|
||||
if not host:
|
||||
raise ValueError("each target must declare a host")
|
||||
if host in seen_hosts:
|
||||
raise ValueError(f"duplicate target host: {host}")
|
||||
if host not in inventory_hosts:
|
||||
raise ValueError(f"unknown inventory host: {host}")
|
||||
seen_hosts.add(host)
|
||||
|
||||
config_root = str(target.get("config_root", "")).strip()
|
||||
if not config_root:
|
||||
raise ValueError(f"target {host} missing config_root")
|
||||
|
||||
files = target.get("files")
|
||||
if not isinstance(files, list) or not files:
|
||||
raise ValueError(f"target {host} must declare at least one file")
|
||||
|
||||
normalized: list[str] = []
|
||||
for entry in files:
|
||||
normalized.append(_normalize_relative_path(str(entry)))
|
||||
if len(set(normalized)) != len(normalized):
|
||||
raise ValueError(f"target {host} declares duplicate file paths")
|
||||
|
||||
|
||||
|
||||
def _hash_file(path: Path) -> str:
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
|
||||
def _collect_target_files(source_root: Path, rel_paths: list[str]) -> list[dict[str, Any]]:
|
||||
items: list[dict[str, Any]] = []
|
||||
for rel_path in sorted(_normalize_relative_path(path) for path in rel_paths):
|
||||
source_path = source_root / rel_path
|
||||
if not source_path.exists():
|
||||
raise FileNotFoundError(f"missing source file: {rel_path}")
|
||||
if not source_path.is_file():
|
||||
raise ValueError(f"sync source must be a file: {rel_path}")
|
||||
items.append(
|
||||
{
|
||||
"relative_path": rel_path,
|
||||
"source": str(source_path),
|
||||
"sha256": _hash_file(source_path),
|
||||
"size": source_path.stat().st_size,
|
||||
}
|
||||
)
|
||||
return items
|
||||
|
||||
|
||||
|
||||
def compute_release_id(target_payloads: list[dict[str, Any]]) -> str:
|
||||
digest = hashlib.sha256()
|
||||
for target in sorted(target_payloads, key=lambda item: item["host"]):
|
||||
digest.update(target["host"].encode("utf-8"))
|
||||
digest.update(b"\0")
|
||||
for file_item in sorted(target["files"], key=lambda item: item["relative_path"]):
|
||||
digest.update(file_item["relative_path"].encode("utf-8"))
|
||||
digest.update(b"\0")
|
||||
digest.update(file_item["sha256"].encode("utf-8"))
|
||||
digest.update(b"\0")
|
||||
digest.update(str(file_item["size"]).encode("utf-8"))
|
||||
digest.update(b"\0")
|
||||
return digest.hexdigest()[:12]
|
||||
|
||||
|
||||
|
||||
def build_rollout_plan(
|
||||
manifest: dict[str, Any],
|
||||
inventory_hosts: dict[str, dict[str, str]],
|
||||
*,
|
||||
source_root: str | Path,
|
||||
) -> dict[str, Any]:
|
||||
validate_manifest(manifest, inventory_hosts)
|
||||
|
||||
source_root = Path(source_root)
|
||||
if not source_root.exists():
|
||||
raise FileNotFoundError(f"source root not found: {source_root}")
|
||||
if not source_root.is_dir():
|
||||
raise ValueError(f"source root must be a directory: {source_root}")
|
||||
|
||||
staged_targets: list[dict[str, Any]] = []
|
||||
for target in sorted(manifest["targets"], key=lambda item: item["host"]):
|
||||
host = target["host"]
|
||||
files = _collect_target_files(source_root, target["files"])
|
||||
staged_targets.append(
|
||||
{
|
||||
"host": host,
|
||||
"inventory": inventory_hosts[host],
|
||||
"config_root": str(target["config_root"]),
|
||||
"files": files,
|
||||
}
|
||||
)
|
||||
|
||||
release_id = compute_release_id(staged_targets)
|
||||
total_bytes = 0
|
||||
file_count = 0
|
||||
rendered_targets: list[dict[str, Any]] = []
|
||||
for target in staged_targets:
|
||||
config_root = target["config_root"].rstrip("/")
|
||||
stage_root = f"{config_root}/.releases/{release_id}"
|
||||
live_symlink = f"{config_root}/current"
|
||||
previous_symlink = f"{config_root}/previous"
|
||||
file_count += len(target["files"])
|
||||
total_bytes += sum(item["size"] for item in target["files"])
|
||||
rendered_targets.append(
|
||||
{
|
||||
"host": target["host"],
|
||||
"ansible_host": target["inventory"].get("ansible_host", ""),
|
||||
"ansible_user": target["inventory"].get("ansible_user", ""),
|
||||
"config_root": config_root,
|
||||
"stage_root": stage_root,
|
||||
"live_symlink": live_symlink,
|
||||
"previous_symlink": previous_symlink,
|
||||
"files": target["files"],
|
||||
"promote": {
|
||||
"mode": "symlink_swap",
|
||||
"release_id": release_id,
|
||||
"from": stage_root,
|
||||
"to": live_symlink,
|
||||
"backup_link": previous_symlink,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"fleet_name": manifest.get("fleet_name", "timmy-fleet-config"),
|
||||
"source_root": str(source_root),
|
||||
"release_id": release_id,
|
||||
"target_count": len(rendered_targets),
|
||||
"file_count": file_count,
|
||||
"total_bytes": total_bytes,
|
||||
"targets": rendered_targets,
|
||||
}
|
||||
|
||||
|
||||
|
||||
def render_markdown(plan: dict[str, Any]) -> str:
|
||||
lines = [
|
||||
"# Fleet Config Sync Plan",
|
||||
"",
|
||||
f"Fleet: {plan['fleet_name']}",
|
||||
f"Release ID: `{plan['release_id']}`",
|
||||
f"Source root: `{plan['source_root']}`",
|
||||
f"Target count: {plan['target_count']}",
|
||||
f"File count: {plan['file_count']}",
|
||||
f"Total bytes: {plan['total_bytes']}",
|
||||
"",
|
||||
"Atomic promote via symlink swap keeps every host on one named release boundary.",
|
||||
"",
|
||||
"| Host | Address | Stage root | Live symlink | Files |",
|
||||
"|---|---|---|---|---:|",
|
||||
]
|
||||
|
||||
for target in plan["targets"]:
|
||||
lines.append(
|
||||
f"| {target['host']} | {target['ansible_host'] or 'n/a'} | `{target['stage_root']}` | `{target['live_symlink']}` | {len(target['files'])} |"
|
||||
)
|
||||
|
||||
lines.extend(["", "## Target file manifests", ""])
|
||||
for target in plan["targets"]:
|
||||
lines.extend(
|
||||
[
|
||||
f"### {target['host']}",
|
||||
"",
|
||||
f"- Promote: `{target['promote']['from']}` -> `{target['promote']['to']}`",
|
||||
f"- Backup link: `{target['promote']['backup_link']}`",
|
||||
"",
|
||||
"| Relative path | Bytes | SHA256 |",
|
||||
"|---|---:|---|",
|
||||
]
|
||||
)
|
||||
for file_item in target["files"]:
|
||||
lines.append(
|
||||
f"| `{file_item['relative_path']}` | {file_item['size']} | `{file_item['sha256'][:16]}…` |"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines).rstrip() + "\n"
|
||||
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Plan a dry-run atomic config sync release across fleet hosts")
|
||||
parser.add_argument("manifest", help="Path to fleet config sync manifest YAML")
|
||||
parser.add_argument("--inventory", default=str(DEFAULT_INVENTORY_FILE), help="Path to Ansible fleet inventory")
|
||||
parser.add_argument("--source-root", default=".", help="Local source root containing files listed in the manifest")
|
||||
parser.add_argument("--markdown", action="store_true", help="Render markdown instead of JSON")
|
||||
args = parser.parse_args()
|
||||
|
||||
inventory = load_inventory_hosts(args.inventory)
|
||||
manifest = load_manifest(args.manifest)
|
||||
plan = build_rollout_plan(manifest, inventory, source_root=args.source_root)
|
||||
|
||||
if args.markdown:
|
||||
print(render_markdown(plan))
|
||||
else:
|
||||
print(json.dumps(plan, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
267
scripts/lab_003_battery_disconnect_packet.py
Normal file
267
scripts/lab_003_battery_disconnect_packet.py
Normal file
@@ -0,0 +1,267 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Prepare a field-ready install packet for LAB-003 truck battery disconnect work."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
CANDIDATE_STORES = [
|
||||
"AutoZone — Newport or Claremont",
|
||||
"Advance Auto Parts — Newport or Claremont",
|
||||
"O'Reilly Auto Parts — Newport or Claremont",
|
||||
]
|
||||
|
||||
REQUIRED_ITEMS = [
|
||||
"battery terminal disconnect switch",
|
||||
"terminal shim/post riser if needed",
|
||||
]
|
||||
|
||||
SELECTION_CRITERIA = [
|
||||
"Fits the truck battery post without forcing the clamp",
|
||||
"Mounts on the negative battery terminal",
|
||||
"Physically secure once tightened",
|
||||
"no special tools required to operate",
|
||||
]
|
||||
|
||||
INSTALL_CHECKLIST = [
|
||||
"Verify the truck is off and keys are removed before touching the battery",
|
||||
"Confirm the disconnect fits the negative battery terminal before final tightening",
|
||||
"Install the disconnect on the negative battery terminal",
|
||||
"Tighten until physically secure with no terminal wobble",
|
||||
"Verify the disconnect can be opened and closed by hand",
|
||||
]
|
||||
|
||||
VALIDATION_CHECKLIST = [
|
||||
"Leave the truck parked with the disconnect opened for at least 24 hours",
|
||||
"Reconnect the switch by hand the next day",
|
||||
"Truck starts reliably after sitting 24+ hours with switch disconnected",
|
||||
"Receipt or photo of installed switch uploaded to this issue",
|
||||
]
|
||||
|
||||
BATTERY_REPLACEMENT_FOLLOWUP = (
|
||||
"If the truck still fails the overnight test after the disconnect install, "
|
||||
"replace battery and re-run the 24-hour validation."
|
||||
)
|
||||
|
||||
|
||||
def _as_bool(value: Any) -> bool | None:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
text = str(value).strip().lower()
|
||||
if text in {"1", "true", "yes", "y"}:
|
||||
return True
|
||||
if text in {"0", "false", "no", "n"}:
|
||||
return False
|
||||
return None
|
||||
|
||||
|
||||
def build_packet(details: dict[str, Any]) -> dict[str, Any]:
|
||||
store_selected = (details.get("store_selected") or "").strip()
|
||||
part_name = (details.get("part_name") or "").strip()
|
||||
receipt_or_photo_path = (details.get("receipt_or_photo_path") or "").strip()
|
||||
install_completed = _as_bool(details.get("install_completed"))
|
||||
physically_secure = _as_bool(details.get("physically_secure"))
|
||||
truck_started = _as_bool(details.get("truck_started_after_disconnect"))
|
||||
replacement_needed = _as_bool(details.get("replacement_battery_needed"))
|
||||
overnight_test_hours = details.get("overnight_test_hours")
|
||||
part_cost_usd = details.get("part_cost_usd")
|
||||
|
||||
try:
|
||||
overnight_test_hours = int(overnight_test_hours) if overnight_test_hours is not None else None
|
||||
except (TypeError, ValueError):
|
||||
overnight_test_hours = None
|
||||
|
||||
try:
|
||||
part_cost_usd = float(part_cost_usd) if part_cost_usd is not None else None
|
||||
except (TypeError, ValueError):
|
||||
part_cost_usd = None
|
||||
|
||||
missing_fields: list[str] = []
|
||||
if not store_selected:
|
||||
missing_fields.append("store_selected")
|
||||
if not part_name:
|
||||
missing_fields.append("part_name")
|
||||
if install_completed is not True:
|
||||
missing_fields.append("install_completed")
|
||||
if physically_secure is not True:
|
||||
missing_fields.append("physically_secure")
|
||||
if overnight_test_hours is None:
|
||||
missing_fields.append("overnight_test_hours")
|
||||
if truck_started is None:
|
||||
missing_fields.append("truck_started_after_disconnect")
|
||||
if not receipt_or_photo_path:
|
||||
missing_fields.append("receipt_or_photo_path")
|
||||
|
||||
ready_to_operate_without_tools = True
|
||||
|
||||
if replacement_needed is True or truck_started is False:
|
||||
status = "battery_replace_candidate"
|
||||
elif not store_selected or not part_name:
|
||||
status = "pending_parts_run"
|
||||
elif install_completed is not True:
|
||||
status = "pending_install"
|
||||
elif physically_secure is not True or overnight_test_hours is None or truck_started is None or not receipt_or_photo_path:
|
||||
status = "overnight_validation"
|
||||
elif overnight_test_hours >= 24 and truck_started is True:
|
||||
status = "verified"
|
||||
else:
|
||||
status = "overnight_validation"
|
||||
|
||||
return {
|
||||
"candidate_stores": list(CANDIDATE_STORES),
|
||||
"required_items": list(REQUIRED_ITEMS),
|
||||
"selection_criteria": list(SELECTION_CRITERIA),
|
||||
"install_target": "negative battery terminal",
|
||||
"install_checklist": list(INSTALL_CHECKLIST),
|
||||
"validation_checklist": list(VALIDATION_CHECKLIST),
|
||||
"store_selected": store_selected,
|
||||
"part_name": part_name,
|
||||
"part_cost_usd": part_cost_usd,
|
||||
"install_completed": install_completed,
|
||||
"physically_secure": physically_secure,
|
||||
"overnight_test_hours": overnight_test_hours,
|
||||
"truck_started_after_disconnect": truck_started,
|
||||
"receipt_or_photo_path": receipt_or_photo_path,
|
||||
"ready_to_operate_without_tools": ready_to_operate_without_tools,
|
||||
"missing_fields": missing_fields,
|
||||
"battery_replacement_followup": BATTERY_REPLACEMENT_FOLLOWUP,
|
||||
"status": status,
|
||||
}
|
||||
|
||||
|
||||
def render_markdown(packet: dict[str, Any]) -> str:
|
||||
part_cost = packet["part_cost_usd"]
|
||||
cost_line = f"${part_cost:.2f}" if isinstance(part_cost, (int, float)) else "pending purchase"
|
||||
overnight = packet["overnight_test_hours"]
|
||||
overnight_line = f"{overnight} hours" if overnight is not None else "pending"
|
||||
started = packet["truck_started_after_disconnect"]
|
||||
if started is True:
|
||||
started_line = "yes"
|
||||
elif started is False:
|
||||
started_line = "no"
|
||||
else:
|
||||
started_line = "pending"
|
||||
|
||||
lines = [
|
||||
"# LAB-003 — Truck Battery Disconnect Install Packet",
|
||||
"",
|
||||
"No battery disconnect switch has been purchased or installed yet.",
|
||||
"This packet turns the issue into a field-ready purchase / install / validation checklist while preserving what still requires live work.",
|
||||
"",
|
||||
"## Candidate Store Run",
|
||||
"",
|
||||
]
|
||||
lines.extend(f"- {store}" for store in packet["candidate_stores"])
|
||||
lines.extend([
|
||||
"",
|
||||
"## Required Items",
|
||||
"",
|
||||
])
|
||||
lines.extend(f"- {item}" for item in packet["required_items"])
|
||||
lines.extend([
|
||||
"",
|
||||
"## Selection Criteria",
|
||||
"",
|
||||
])
|
||||
lines.extend(f"- {item}" for item in packet["selection_criteria"])
|
||||
lines.extend([
|
||||
"",
|
||||
"## Live Purchase State",
|
||||
"",
|
||||
f"- Store selected: {packet['store_selected'] or 'pending'}",
|
||||
f"- Part selected: {packet['part_name'] or 'pending'}",
|
||||
f"- Part cost: {cost_line}",
|
||||
"",
|
||||
"## Installation Target",
|
||||
"",
|
||||
f"- Install location: {packet['install_target']}",
|
||||
f"- Ready to operate without tools: {'yes' if packet['ready_to_operate_without_tools'] else 'no'}",
|
||||
"",
|
||||
"## Install Checklist",
|
||||
"",
|
||||
])
|
||||
lines.extend(f"- [ ] {item}" for item in packet["install_checklist"])
|
||||
lines.extend([
|
||||
"",
|
||||
"## Validation Checklist",
|
||||
"",
|
||||
])
|
||||
lines.extend(f"- [ ] {item}" for item in packet["validation_checklist"])
|
||||
lines.extend([
|
||||
"",
|
||||
"## Overnight Verification Log",
|
||||
"",
|
||||
f"- Install completed: {packet['install_completed'] if packet['install_completed'] is not None else 'pending'}",
|
||||
f"- Physically secure: {packet['physically_secure'] if packet['physically_secure'] is not None else 'pending'}",
|
||||
f"- Overnight disconnect duration: {overnight_line}",
|
||||
f"- Truck started after disconnect: {started_line}",
|
||||
f"- Receipt / photo path: {packet['receipt_or_photo_path'] or 'pending'}",
|
||||
"",
|
||||
"## Battery Replacement Fallback",
|
||||
"",
|
||||
packet['battery_replacement_followup'],
|
||||
"",
|
||||
"## Missing Live Fields",
|
||||
"",
|
||||
])
|
||||
if packet["missing_fields"]:
|
||||
lines.extend(f"- {field}" for field in packet["missing_fields"])
|
||||
else:
|
||||
lines.append("- none")
|
||||
lines.extend([
|
||||
"",
|
||||
"## Honest next step",
|
||||
"",
|
||||
"Buy the disconnect switch, install it on the negative battery terminal, leave the truck disconnected for 24+ hours, and only close the issue after receipt/photo evidence and the overnight start result are attached.",
|
||||
"",
|
||||
])
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Prepare the LAB-003 battery disconnect install packet")
|
||||
parser.add_argument("--store-selected", default="")
|
||||
parser.add_argument("--part-name", default="")
|
||||
parser.add_argument("--part-cost-usd", type=float, default=None)
|
||||
parser.add_argument("--install-completed", action="store_true")
|
||||
parser.add_argument("--physically-secure", action="store_true")
|
||||
parser.add_argument("--overnight-test-hours", type=int, default=None)
|
||||
parser.add_argument("--truck-started-after-disconnect", choices=["yes", "no"], default=None)
|
||||
parser.add_argument("--receipt-or-photo-path", default="")
|
||||
parser.add_argument("--replacement-battery-needed", action="store_true")
|
||||
parser.add_argument("--output", default=None)
|
||||
parser.add_argument("--json", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
packet = build_packet(
|
||||
{
|
||||
"store_selected": args.store_selected,
|
||||
"part_name": args.part_name,
|
||||
"part_cost_usd": args.part_cost_usd,
|
||||
"install_completed": args.install_completed,
|
||||
"physically_secure": args.physically_secure,
|
||||
"overnight_test_hours": args.overnight_test_hours,
|
||||
"truck_started_after_disconnect": args.truck_started_after_disconnect,
|
||||
"receipt_or_photo_path": args.receipt_or_photo_path,
|
||||
"replacement_battery_needed": args.replacement_battery_needed,
|
||||
}
|
||||
)
|
||||
rendered = json.dumps(packet, indent=2) if args.json else render_markdown(packet)
|
||||
|
||||
if args.output:
|
||||
output_path = Path(args.output).expanduser()
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_text(rendered, encoding="utf-8")
|
||||
print(f"Battery disconnect packet written to {output_path}")
|
||||
else:
|
||||
print(rendered)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,122 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT_PATH = ROOT / "scripts" / "fleet_config_sync.py"
|
||||
HOSTS_FILE = ROOT / "ansible" / "inventory" / "hosts.ini"
|
||||
EXAMPLE_MANIFEST = ROOT / "docs" / "fleet-config-sync.example.yaml"
|
||||
|
||||
|
||||
def _load_module(path: Path, name: str):
|
||||
assert path.exists(), f"missing {path.relative_to(ROOT)}"
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
assert spec and spec.loader
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def _write_source_tree(tmp_path: Path) -> Path:
|
||||
source_root = tmp_path / "source"
|
||||
(source_root / "dispatch").mkdir(parents=True)
|
||||
(source_root / "config.yaml").write_text("model: local\nroute: hybrid\n", encoding="utf-8")
|
||||
(source_root / "dispatch" / "rules.json").write_text('{"lane":"allegro"}\n', encoding="utf-8")
|
||||
return source_root
|
||||
|
||||
|
||||
def test_example_manifest_targets_known_fleet_hosts() -> None:
|
||||
mod = _load_module(SCRIPT_PATH, "fleet_config_sync")
|
||||
assert EXAMPLE_MANIFEST.exists(), "missing docs/fleet-config-sync.example.yaml"
|
||||
|
||||
inventory = mod.load_inventory_hosts(HOSTS_FILE)
|
||||
manifest = mod.load_manifest(EXAMPLE_MANIFEST)
|
||||
mod.validate_manifest(manifest, inventory)
|
||||
|
||||
assert [target["host"] for target in manifest["targets"]] == ["ezra", "bezalel"]
|
||||
|
||||
|
||||
def test_build_rollout_plan_stages_one_release_for_all_hosts(tmp_path: Path) -> None:
|
||||
mod = _load_module(SCRIPT_PATH, "fleet_config_sync")
|
||||
source_root = _write_source_tree(tmp_path)
|
||||
inventory = {
|
||||
"ezra": {"host": "ezra", "ansible_host": "143.198.27.163"},
|
||||
"bezalel": {"host": "bezalel", "ansible_host": "67.205.155.108"},
|
||||
}
|
||||
manifest = {
|
||||
"fleet_name": "phase-3-config-sync",
|
||||
"targets": [
|
||||
{
|
||||
"host": "ezra",
|
||||
"config_root": "/root/wizards/ezra/home/.hermes",
|
||||
"files": ["config.yaml", "dispatch/rules.json"],
|
||||
},
|
||||
{
|
||||
"host": "bezalel",
|
||||
"config_root": "/root/wizards/bezalel/home/.hermes",
|
||||
"files": ["config.yaml", "dispatch/rules.json"],
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
plan = mod.build_rollout_plan(manifest, inventory, source_root=source_root)
|
||||
|
||||
assert plan["fleet_name"] == "phase-3-config-sync"
|
||||
assert len(plan["release_id"]) == 12
|
||||
assert plan["target_count"] == 2
|
||||
assert plan["file_count"] == 4
|
||||
assert plan["total_bytes"] > 0
|
||||
|
||||
for target in plan["targets"]:
|
||||
assert target["stage_root"].endswith(f"/.releases/{plan['release_id']}")
|
||||
assert target["live_symlink"].endswith("/current")
|
||||
assert target["promote"]["release_id"] == plan["release_id"]
|
||||
assert {item["relative_path"] for item in target["files"]} == {"config.yaml", "dispatch/rules.json"}
|
||||
|
||||
|
||||
def test_validate_manifest_rejects_unknown_inventory_host(tmp_path: Path) -> None:
|
||||
mod = _load_module(SCRIPT_PATH, "fleet_config_sync")
|
||||
source_root = _write_source_tree(tmp_path)
|
||||
manifest = {
|
||||
"targets": [
|
||||
{
|
||||
"host": "unknown-wizard",
|
||||
"config_root": "/srv/wizard/config",
|
||||
"files": ["config.yaml"],
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
try:
|
||||
mod.build_rollout_plan(manifest, {"ezra": {"host": "ezra"}}, source_root=source_root)
|
||||
except ValueError as exc:
|
||||
assert "unknown inventory host" in str(exc)
|
||||
assert "unknown-wizard" in str(exc)
|
||||
else:
|
||||
raise AssertionError("build_rollout_plan should reject unknown inventory hosts")
|
||||
|
||||
|
||||
def test_render_markdown_mentions_atomic_promote_and_targets(tmp_path: Path) -> None:
|
||||
mod = _load_module(SCRIPT_PATH, "fleet_config_sync")
|
||||
source_root = _write_source_tree(tmp_path)
|
||||
manifest = {
|
||||
"fleet_name": "phase-3-config-sync",
|
||||
"targets": [
|
||||
{
|
||||
"host": "ezra",
|
||||
"config_root": "/root/wizards/ezra/home/.hermes",
|
||||
"files": ["config.yaml"],
|
||||
}
|
||||
],
|
||||
}
|
||||
inventory = {"ezra": {"host": "ezra", "ansible_host": "143.198.27.163"}}
|
||||
|
||||
plan = mod.build_rollout_plan(manifest, inventory, source_root=source_root)
|
||||
report = mod.render_markdown(plan)
|
||||
|
||||
assert plan["release_id"] in report
|
||||
assert "Atomic promote via symlink swap" in report
|
||||
assert "ezra" in report
|
||||
assert "/root/wizards/ezra/home/.hermes/current" in report
|
||||
88
tests/test_lab_003_battery_disconnect_packet.py
Normal file
88
tests/test_lab_003_battery_disconnect_packet.py
Normal file
@@ -0,0 +1,88 @@
|
||||
from pathlib import Path
|
||||
import importlib.util
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
SCRIPT_PATH = ROOT / "scripts" / "lab_003_battery_disconnect_packet.py"
|
||||
DOC_PATH = ROOT / "docs" / "LAB_003_BATTERY_DISCONNECT_PACKET.md"
|
||||
|
||||
|
||||
def load_module(path: Path, name: str):
|
||||
assert path.exists(), f"missing {path.relative_to(ROOT)}"
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
assert spec and spec.loader
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class TestLab003BatteryDisconnectPacket(unittest.TestCase):
|
||||
def test_packet_defaults_to_parts_run_and_tracks_issue_specific_requirements(self):
|
||||
mod = load_module(SCRIPT_PATH, "lab_003_battery_disconnect_packet")
|
||||
packet = mod.build_packet({})
|
||||
|
||||
self.assertEqual(packet["status"], "pending_parts_run")
|
||||
self.assertEqual(packet["install_target"], "negative battery terminal")
|
||||
self.assertIn("battery terminal disconnect switch", packet["required_items"])
|
||||
self.assertIn("terminal shim/post riser if needed", packet["required_items"])
|
||||
self.assertIn("AutoZone", packet["candidate_stores"][0])
|
||||
self.assertIn("no special tools required to operate", packet["selection_criteria"])
|
||||
self.assertIn("overnight_test_hours", packet["missing_fields"])
|
||||
self.assertIn("receipt_or_photo_path", packet["missing_fields"])
|
||||
|
||||
def test_packet_marks_verified_after_successful_24h_validation_with_proof(self):
|
||||
mod = load_module(SCRIPT_PATH, "lab_003_battery_disconnect_packet")
|
||||
packet = mod.build_packet(
|
||||
{
|
||||
"store_selected": "AutoZone - Newport",
|
||||
"part_name": "Knob-style battery disconnect switch",
|
||||
"part_cost_usd": 24.99,
|
||||
"install_completed": True,
|
||||
"physically_secure": True,
|
||||
"overnight_test_hours": 26,
|
||||
"truck_started_after_disconnect": True,
|
||||
"receipt_or_photo_path": "evidence/lab-003-installed-switch.jpg",
|
||||
}
|
||||
)
|
||||
|
||||
self.assertEqual(packet["status"], "verified")
|
||||
self.assertEqual(packet["missing_fields"], [])
|
||||
self.assertTrue(packet["ready_to_operate_without_tools"])
|
||||
|
||||
def test_packet_flags_battery_replace_candidate_when_overnight_test_fails(self):
|
||||
mod = load_module(SCRIPT_PATH, "lab_003_battery_disconnect_packet")
|
||||
packet = mod.build_packet(
|
||||
{
|
||||
"store_selected": "O'Reilly - Claremont",
|
||||
"part_name": "Knob-style battery disconnect switch",
|
||||
"install_completed": True,
|
||||
"physically_secure": True,
|
||||
"overnight_test_hours": 24,
|
||||
"truck_started_after_disconnect": False,
|
||||
}
|
||||
)
|
||||
|
||||
self.assertEqual(packet["status"], "battery_replace_candidate")
|
||||
self.assertIn("battery_replacement_followup", packet)
|
||||
self.assertIn("replace battery", packet["battery_replacement_followup"].lower())
|
||||
|
||||
def test_repo_contains_grounded_lab_003_packet_doc(self):
|
||||
self.assertTrue(DOC_PATH.exists(), "missing committed LAB-003 packet doc")
|
||||
text = DOC_PATH.read_text(encoding="utf-8")
|
||||
for snippet in (
|
||||
"# LAB-003 — Truck Battery Disconnect Install Packet",
|
||||
"No battery disconnect switch has been purchased or installed yet.",
|
||||
"negative battery terminal",
|
||||
"AutoZone",
|
||||
"Advance",
|
||||
"O'Reilly",
|
||||
"terminal shim/post riser if needed",
|
||||
"Truck starts reliably after sitting 24+ hours with switch disconnected",
|
||||
"Receipt or photo of installed switch uploaded to this issue",
|
||||
):
|
||||
self.assertIn(snippet, text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user