sascha/homelab-butler
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Butler 2.3: deterministische Control Plane für leichte Modelle #1

Merged
sascha merged 6 commits from trulla/butler-2.3-20260722 into main 2026-07-22 10:38:13 +02:00
Conversation 0 Commits 6 Files changed 6 +1014 -134
6 changed files with 1014 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

175
README.md
View file

@ -1,110 +1,127 @@
# Homelab Butler 🤵
Unified API proxy + infrastructure management for Homelab Pfannkuchen.
Unified API proxy and infrastructure management for Homelab Pfannkuchen.
**Base URL:** `http://10.4.1.116:8888`
**Auth:** `Authorization: Bearer ***`
**Version:** 2.1.0
- **Base URL:** `http://10.4.1.116:8888`
- **Authentication:** `Authorization: Bearer <BUTLER_TOKEN>`
- **Version:** 2.3.0
- **Interactive API documentation:** `/docs`
## Service Proxy
## Service proxy
Proxies requests to backend services with automatic authentication:
Requests are proxied as `/{service}/{backend-path}`. Butler adds each backend's authentication automatically and forwards query parameters.
| Service | Backend | Auth |
|---------|---------|------|
| `dockhand` | 10.4.1.116:3000 | Session |
| `sonarr` | 10.2.1.100:8989 | API Key |
| `radarr` | 10.2.1.100:7878 | API Key |
| `seerr` | 10.2.1.100:5055 | API Key |
| `outline` | 10.1.1.100:3000 | Bearer |
| `n8n` | 10.4.1.113:5678 | X-N8N-API-KEY |
| `proxmox` | 10.5.85.11:8006 | PVE Token |
| `homeassistant` | 10.10.1.20:8123 | Bearer |
| `grafana` | 10.1.1.111:3000 | Bearer |
| `uptime` | 159.69.245.190:3001 | Bearer |
| `waha` | 10.4.1.110:3500 | API Key |
| `forgejo` | 10.4.1.116:3001 | Bearer |
| `semaphore` | 10.4.1.116:8090 | Bearer |
| Service | Backend | Authentication |
|---|---|---|
| `dockhand` | `10.4.1.116:3000` | Session |
| `sonarr` / `sonarr1080p` | `10.2.1.100:8989/8990` | API key |
| `radarr` / `radarr1080p` | `10.2.1.100:7878/7879` | API key |
| `seerr` | `10.2.1.100:5055` | API key |
| `outline` | `10.1.1.100:3000` | Bearer |
| `n8n` | `10.4.1.113:5678` | n8n API key |
| `proxmox` | `10.5.85.11:8006` | PVE API token |
| `homeassistant` | `10.10.1.1:8123` | Bearer |
| `grafana` | `10.1.1.111:3000` | Bearer |
| `uptime` | `10.5.85.5:3001` | Web UI only; no supported REST API |
| `waha` | `10.4.1.110:3500` | API key |
| `forgejo` | `10.4.1.116:3001` | Bearer |
| `semaphore` | `10.4.1.116:8090` | Bearer |
| `fileflows` | `10.2.1.104:8268` | Local API, no additional auth |
Usage: `GET/POST/PUT/DELETE /{service}/{path}`
Known secret response fields such as Dockhand's `hawserToken` and `webhookSecret` are redacted before data leaves Butler.
## VM Lifecycle
## Operations
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/vm/list` | GET | All VMs across all 7 Proxmox nodes |
| `/vm/create` | POST | Full VM deployment (~10 min): ISO build, VM create, SSH wait, inventory, Ansible |
| `/vm/status/{vmid}` | GET | VM status (CPU, RAM, uptime) |
| `/vm/{vmid}` | DELETE | Destroy VM |
|---|---:|---|
| `/info` | GET | Secret-free machine-readable context |
| `/status` | GET | Concurrent authenticated functional probes with deterministic states |
| `/overview` | GET | Compact overall verdict and ordered findings for lightweight models; `details=true` adds raw data |
| `/audit` | GET | Recent proxied/management calls |
| `/health/all` | GET | SSH reachability and Docker status for inventory hosts |
| `/backup/status` | GET | Concurrent Borgmatic checks with age, state and summary |
| `/disk/usage` | GET | Root filesystem usage per host |
| `/logs/{host}/{container}` | GET | Docker logs, `tail` query supported |
| `/docker/inspect/{host}/{container}` | GET | Sanitized image, runtime, resources, mounts and state |
| `/docker/restart/{host}/{container}` | POST | Restart container; supports `dry_run=true` |
| `/config/reload` | POST | Reload YAML configuration and credential cache |
### POST /vm/create
```json
{"node": 5, "ip": "10.5.1.115", "hostname": "lychee", "cores": 2, "memory": 4096, "disk": 32}
```
Steps: iso-builder → Proxmox VM → wait SSH → add to pfannkuchen.ini → Ansible base setup (Docker, Borgmatic, Hawser)
## Ansible / Inventory
## VM lifecycle and inventory
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/inventory/host` | POST | Add host to pfannkuchen.ini (idempotent, with group) |
| `/ansible/run` | POST | Run Ansible playbook on host |
|---|---:|---|
| `/vm/list` | GET | All VMs across the seven Proxmox nodes |
| `/vm/create` | POST | VM deployment; supports `dry_run=true` |
| `/vm/status/{vmid}` | GET | CPU, RAM, uptime and state |
| `/vm/destroy/{vmid}` | DELETE | Full lifecycle cleanup; supports `dry_run=true` |
| `/inventory/host` | POST | Create or update inventory host and host vars |
| `/ansible/run` | POST | Run the standard setup for a host |
Example inventory upsert:
### POST /ansible/run
```json
{"hostname": "lychee"}
{"name":"example","ip":"10.5.1.115","group":"auto","user":"sascha"}
```
**Post-Run Automation** (after successful Ansible):
1. **Hawser Token Sync** – Reads `/etc/hawser/config` from VM, syncs token to Dockhand environment
2. **SOPS + .env Setup** – If `compose.yaml` exists in `/app-config/github/{hostname}/`:
- Generates secure secrets (admin password, DB password, secret key)
- Creates `.env` file with service-specific variables
- Encrypts to `.env.enc` using SOPS (Age key from automation1)
- Copies both files to VM's git repo directory
- Stores secrets in Butler vault cache for future reference
Node-7 VMs default to SSH user `chris`; Proxmox nodes default to `root`.
### POST /inventory/host
```json
{"name": "lychee", "ip": "10.5.1.115", "group": "auto"}
```
## TTS / Speech
## TTS
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/tts/speak` | POST | Text-to-speech via Chatterbox |
| `/tts/voices` | GET | Available voices |
| `/tts/health` | GET | Speaker + Chatterbox status |
|---|---:|---|
| `/tts/speak` | POST | Chatterbox/speaker TTS |
| `/tts/voices` | GET | Available Chatterbox voices |
| `/tts/health` | GET | Speaker and Chatterbox status |
### POST /tts/speak
```json
{"text": "Hallo!", "target": "speaker"}
```
- `"target": "speaker"` → plays on Pi5 speaker (10.10.1.166)
- `"target": "telegram"` → generates OGG on hermes, use `MEDIA:/tmp/trulla_voice.ogg`
The speaker endpoint is `10.5.85.2:10800`; Chatterbox runs at `10.2.1.104:8004`.
## Credentials
## Deployment
Reads from Vaultwarden cache (synced by host cron) with flat-file fallback (`/data/api/`).
Required mounts and settings are defined in `compose.yaml`:
## Stack
- `.env` containing `BUTLER_TOKEN`
- `/app-config/kiro/api/` as flat-file credential fallback
- persistent Vaultwarden cache volume
- SSH key mounted read-only at `/root/.ssh`
- `butler.yaml` mounted read-only at `/data/butler.yaml`
```
docker compose build && docker compose up -d
Git is the source of truth. Build/recreate the Compose service only after committing and pushing changes.
## Tests
```bash
python -m pytest -q tests/test_app.py
```
Requires:
- `.env` with `BUTLER_TOKEN`
- `/app-config/kiro/api/` flat-file credentials
- SSH key mount (`/home/sascha/.ssh:/root/.ssh:ro`) for VM operations
Integration Compose definition: `tests/compose.integration.yaml` (binds only to `127.0.0.1:8889`).
## Changelog
### v2.1.0 (22.04.2026)
- ✅ **Hawser Token Auto-Sync** – After `/ansible/run`, reads token from VM and updates Dockhand environment
- ✅ **SOPS + .env Automation** – Auto-generates and encrypts environment files for Git-centric deployments
- ✅ **Service Detection** – Recognizes Paperless-ngx and other services from hostname, generates appropriate env vars
### 2.3.0 — 22.07.2026
### v2.0.0
- Initial unified API proxy release
- Added `/overview`, a compact schema-versioned verdict for lightweight language models.
- Added explicit `healthy`, `degraded`, `auth_failed`, `misconfigured` and `offline` service states.
- Service probes now use the configured backend credentials and run concurrently.
- Backup checks now run with bounded concurrency and a 12-second per-host timeout.
- Backup results include age and severity (`healthy` up to 30 h, `warning` up to 48 h, then `critical`).
- Host and disk collection now run concurrently.
- Added regression tests for classification, authentication, backup age/concurrency and overview output.
### 2.2.0 — 17.07.2026
- Restored and modernized `/info`, `/health/all`, `/backup/status`, `/disk/usage` and Docker log/restart endpoints.
- Fixed stale Home Assistant, Uptime Kuma and speaker addresses.
- Fixed Forgejo credential fallback by deploying the YAML-driven service config.
- Correct SSH defaults for Proxmox nodes and Node-7 VMs.
- Added recursive secret redaction for proxied JSON.
- Forward query parameters through the generic proxy.
- Added validated inventory upserts.
- Added regression tests and a loopback-only integration Compose setup.
### 2.1.1
- Full VM destruction lifecycle cleanup.
### 2.1.0
- YAML service configuration, status/audit endpoints, dry-run support, Hawser token sync and SOPS automation.

671
app.py
View file

@ -1,14 +1,17 @@
"""Homelab Butler v2.1 – Unified API proxy for Pfannkuchen homelab.
Reads service config from butler.yaml, credentials from Vaultwarden cache with flat-file fallback."""
import os, json, asyncio, logging, time
import os, json, asyncio, logging, time, base64, re, subprocess, ipaddress
from datetime import datetime, timezone
import httpx, yaml
from typing import Literal
from pydantic import BaseModel
from fastapi import FastAPI, Request, HTTPException, Depends, Query
from fastapi.responses import JSONResponse, RedirectResponse
from contextlib import asynccontextmanager
log = logging.getLogger("butler")
VERSION = "2.3.0"
API_DIR = os.environ.get("API_KEY_DIR", "/data/api")
VAULT_CACHE_DIR = os.environ.get("VAULT_CACHE_DIR", "/data/vault-cache")
@ -37,6 +40,7 @@ def _load_config():
SERVICES: dict = {}
VM_CFG: dict = {}
TTS_CFG: dict = {}
_load_config()
# --- Audit log ---
@ -92,9 +96,45 @@ async def lifespan(app: FastAPI):
yield
task.cancel()
app = FastAPI(title="Homelab Butler", version="2.1.0", lifespan=lifespan,
app = FastAPI(title="Homelab Butler", version=VERSION, lifespan=lifespan,
description="Unified API proxy + infrastructure management. AI agents: see GET / for self-onboarding.")
OverviewState = Literal["healthy", "warning", "critical"]
class OverviewCounts(BaseModel):
critical: int
warning: int
healthy: int
class OverviewFinding(BaseModel):
severity: OverviewState
code: str
target: str
message: str
age_hours: float | None = None
pct: int | None = None
class OverviewModelContract(BaseModel):
instruction: str
severity_order: list[OverviewState]
class OverviewResponse(BaseModel):
schema_version: Literal[1]
generated: datetime
overall_state: OverviewState
action_required: bool
summary: OverviewCounts
components: dict[str, OverviewCounts]
findings: list[OverviewFinding]
model_contract: OverviewModelContract
details: dict | None = None
# --- Credential reading (vault-first, file-fallback) ---
def _read(name):
@ -162,6 +202,54 @@ def _get_key(cfg):
return _vault_cache[vault_key]
return _read(cfg.get("key_file", ""))
SENSITIVE_RESPONSE_FIELDS = {
"hawsertoken", "webhooksecret", "accesstoken", "refreshtoken",
"password", "secret", "apikey", "api_key", "privatekey",
}
def _redact_response(value, extra_fields=None):
"""Recursively redact known secret fields in proxied JSON responses."""
sensitive = set(SENSITIVE_RESPONSE_FIELDS)
sensitive.update(str(x).lower() for x in (extra_fields or []))
if isinstance(value, dict):
return {
key: "[REDACTED]" if str(key).lower() in sensitive
else _redact_response(item, sensitive)
for key, item in value.items()
}
if isinstance(value, list):
return [_redact_response(item, sensitive) for item in value]
return value
def _inventory_hosts(text: str) -> list[dict]:
"""Parse Ansible inventory host lines and apply Pfannkuchen SSH defaults."""
hosts = []
seen = set()
for raw in text.splitlines():
line = raw.strip()
if not line or line.startswith(("#", "[")) or "ansible_host=" not in line:
continue
parts = line.split()
name = parts[0]
attrs = {k: v for k, v in (p.split("=", 1) for p in parts[1:] if "=" in p)}
ip = attrs.get("ansible_host")
if not ip or name in seen:
continue
user = attrs.get("ansible_user")
if not user:
if name.startswith("node"):
user = "root"
elif ip.startswith("10.7.1."):
user = "chris"
else:
user = "sascha"
hosts.append({"name": name, "ip": ip, "user": user})
seen.add(name)
return hosts
# --- Routes ---
@app.get("/")
@ -171,7 +259,7 @@ async def root():
for name, cfg in SERVICES.items():
svc_list[name] = {"url": cfg.get("url", ""), "auth": cfg.get("auth", ""), "description": cfg.get("description", "")}
return {
"service": "homelab-butler", "version": "2.1.0",
"service": "homelab-butler", "version": VERSION,
"docs": "/docs",
"openapi": "/openapi.json",
"services": svc_list,
@ -188,6 +276,7 @@ async def root():
"tts_voices": "GET /tts/voices",
"tts_health": "GET /tts/health",
"status": "GET /status - health of all backends",
"overview": "GET /overview?details=false - deterministic homelab verdict for small models",
"audit": "GET /audit - recent API calls",
},
"vault_items": len(_vault_cache),
@ -195,23 +284,96 @@ async def root():
@app.get("/health")
async def health():
return {"status": "ok", "vault_items": len(_vault_cache), "services": len(SERVICES), "version": "2.1.0"}
return {"status": "ok", "vault_items": len(_vault_cache), "services": len(SERVICES), "version": VERSION}
def _classify_http_status(status_code: int, expected: set[int]) -> str:
"""Return a deterministic service state suitable for small models."""
if status_code in expected:
return "healthy"
if status_code in (401, 403):
return "auth_failed"
if status_code == 404:
return "misconfigured"
return "degraded"
def _service_auth(cfg: dict) -> dict:
"""Build secret-bearing request data without ever returning it from an endpoint."""
auth_type = cfg.get("auth", "none")
headers = {}
cookies = {}
base_url = cfg.get("url")
if auth_type == "apikey":
headers["X-Api-Key"] = _get_key(cfg) or ""
elif auth_type == "apikey_urlfile":
base_url, key = _parse_url_key(cfg.get("key_file", ""))
headers["X-Api-Key"] = key or ""
elif auth_type == "bearer":
headers["Authorization"] = f"Bearer {_get_key(cfg) or ''}"
elif auth_type == "n8n":
headers["X-N8N-API-KEY"] = _get_key(cfg) or ""
elif auth_type == "proxmox":
pv = _parse_kv("proxmox")
headers["Authorization"] = f"PVEAPIToken={pv.get('tokenid', '')}={pv.get('secret', '')}"
return {"base_url": base_url, "headers": headers, "cookies": cookies}
async def _collect_service_status() -> dict:
"""Run authenticated functional probes concurrently and classify their result."""
started = time.monotonic()
async with httpx.AsyncClient(verify=False, timeout=5, follow_redirects=True) as client:
async def probe(name: str, cfg: dict):
probe_started = time.monotonic()
try:
request_data = _service_auth(cfg)
base_url = request_data["base_url"]
if not base_url:
return name, {
"reachable": False, "status": "misconfigured", "message": "No service URL configured"
}
if cfg.get("auth") == "session":
request_data["cookies"] = await _dockhand_login(client) or {}
health_path = cfg.get("health_path", "")
target = f"{base_url.rstrip('/')}/{health_path.lstrip('/')}" if health_path else base_url
expected = {int(code) for code in cfg.get("health_expected", range(200, 400))}
response = await client.get(
target,
headers=request_data["headers"],
cookies=request_data["cookies"],
)
state = _classify_http_status(response.status_code, expected)
messages = {
"healthy": "Functional probe succeeded",
"auth_failed": "Configured credentials were rejected",
"misconfigured": "Configured health route was not found",
"degraded": "Backend returned an unexpected HTTP status",
}
return name, {
"reachable": True,
"status": state,
"http": response.status_code,
"latency_ms": round((time.monotonic() - probe_started) * 1000),
"message": messages[state],
}
except Exception as exc:
return name, {
"reachable": False,
"status": "offline",
"latency_ms": round((time.monotonic() - probe_started) * 1000),
"error": type(exc).__name__,
"message": "Backend could not be reached",
}
pairs = await asyncio.gather(*(probe(name, cfg) for name, cfg in SERVICES.items()))
results = dict(pairs)
results["_meta"] = {"duration_ms": round((time.monotonic() - started) * 1000)}
return results
@app.get("/status")
async def status(_=Depends(_verify)):
"""Health check all configured backend services."""
results = {}
async with httpx.AsyncClient(verify=False, timeout=5) as c:
for name, cfg in SERVICES.items():
url = cfg.get("url")
if not url:
results[name] = {"status": "no_url"}
continue
try:
r = await c.get(url, follow_redirects=True)
results[name] = {"status": "ok", "http": r.status_code}
except Exception as e:
results[name] = {"status": "offline", "error": type(e).__name__}
"""Authenticated functional health check for all configured backends."""
results = await _collect_service_status()
_audit("/status", "GET", 200)
return results
@ -227,6 +389,383 @@ async def config_reload(_=Depends(_verify)):
_load_vault_cache()
return {"config_services": len(SERVICES), "vault_items": len(_vault_cache)}
@app.get("/info")
async def info(_=Depends(_verify)):
"""Secret-free machine-readable context for AI agents and operators."""
return {
"service": "homelab-butler",
"version": VERSION,
"generated": datetime.now(timezone.utc).isoformat(),
"services": {
name: {
"url": cfg.get("url"),
"auth": cfg.get("auth"),
"description": cfg.get("description", ""),
}
for name, cfg in SERVICES.items()
},
"endpoints": {
"status": "/status",
"overview": "/overview?details=false",
"audit": "/audit",
"host_health": "/health/all",
"backups": "/backup/status",
"disk": "/disk/usage",
"logs": "/logs/{host}/{container}?tail=200",
"inspect": "/docker/inspect/{host}/{container}",
"docs": "/docs",
},
"rules": [
"Backend services are accessed through Butler or dedicated MCP servers",
"VMs only; no LXC",
"Docker Compose is stored in Git; no docker run",
"Persistent volumes live under /app-config",
"Node 7 VM SSH user is chris",
],
}
def _get_inventory_hosts() -> list[dict]:
rc, out, _err = _ssh(
AUTOMATION1,
"python3 -c \"print(open('/app-config/ansible/pfannkuchen.ini').read())\"",
timeout=15,
)
return _inventory_hosts(out) if rc == 0 else []
def _find_inventory_host(name: str) -> dict | None:
return next((host for host in _get_inventory_hosts() if host["name"] == name), None)
async def _get_inventory_hosts_async() -> list[dict]:
return await asyncio.to_thread(_get_inventory_hosts)
async def _collect_health_all(concurrency: int = 10) -> dict:
"""Collect SSH and container health concurrently with bounded fan-out."""
semaphore = asyncio.Semaphore(concurrency)
async def inspect_host(host: dict):
async with semaphore:
rc, out, err = await asyncio.to_thread(
_ssh,
f'{host["user"]}@{host["ip"]}',
"echo __BUTLER_OK__; (sudo -n docker ps --format '{{.Names}}: {{.Status}}' 2>/dev/null || docker ps --format '{{.Names}}: {{.Status}}' 2>/dev/null) | head -30",
10,
)
lines = out.strip().splitlines()
return host["name"], {
"ip": host["ip"],
"user": host["user"],
"reachable": rc == 0 and bool(lines) and lines[0] == "__BUTLER_OK__",
"containers": lines[1:] if lines and lines[0] == "__BUTLER_OK__" else [],
"error": err.strip()[:200] if rc != 0 else None,
}
hosts = await _get_inventory_hosts_async()
pairs = await asyncio.gather(*(inspect_host(host) for host in hosts))
return dict(pairs)
@app.get("/health/all")
async def health_all(_=Depends(_verify)):
"""SSH reachability and Docker status for all inventory hosts."""
return await _collect_health_all()
def _parse_backup_time(value: str | None) -> datetime | None:
if not value:
return None
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc)
except (TypeError, ValueError):
return None
def _backup_item(rc: int, out: str, err: str, now: datetime | None = None) -> dict:
"""Normalize borgmatic output into one small-model-friendly state object."""
item = {"state": "unknown", "ok": False, "last_backup": None, "age_hours": None}
if rc != 0 or not out.strip():
if err:
item["error"] = err.strip()[:200]
return item
try:
data = json.loads(out)
archives = data[0].get("archives", []) if isinstance(data, list) and data else []
if not archives:
item["error"] = "no archives returned"
return item
last = archives[-1]
started_at = _parse_backup_time(last.get("start"))
if not started_at:
item["error"] = "invalid backup timestamp"
return item
age_hours = max(0, ((now or datetime.now(timezone.utc)) - started_at).total_seconds() / 3600)
state = "healthy" if age_hours <= 30 else "warning" if age_hours <= 48 else "critical"
return {
"state": state,
"ok": state == "healthy",
"last_backup": last.get("start"),
"age_hours": round(age_hours, 1),
"name": last.get("name"),
}
except (json.JSONDecodeError, TypeError, IndexError, KeyError):
item["error"] = "invalid borgmatic JSON"
return item
async def _collect_backup_status(concurrency: int = 10) -> dict:
"""Query VM backups concurrently; one slow host no longer blocks all others serially."""
semaphore = asyncio.Semaphore(concurrency)
hosts = [host for host in await _get_inventory_hosts_async() if not host["name"].startswith("node")]
async def inspect_backup(host: dict):
async with semaphore:
rc, out, err = await asyncio.to_thread(
_ssh,
f'{host["user"]}@{host["ip"]}',
"sudo -n borgmatic list --last 1 --json 2>/dev/null",
12,
)
return host["name"], _backup_item(rc, out, err)
pairs = await asyncio.gather(*(inspect_backup(host) for host in hosts))
results = dict(pairs)
summary = {"total": len(results), "healthy": 0, "warning": 0, "critical": 0, "unknown": 0}
for item in results.values():
summary[item["state"]] += 1
return {"summary": summary, "hosts": results}
@app.get("/backup/status")
async def backup_status(_=Depends(_verify)):
"""Latest Borgmatic archive, age and severity for all VM inventory hosts."""
return await _collect_backup_status()
async def _collect_disk_usage(concurrency: int = 10) -> dict:
semaphore = asyncio.Semaphore(concurrency)
async def inspect_disk(host: dict):
async with semaphore:
rc, out, _err = await asyncio.to_thread(
_ssh, f'{host["user"]}@{host["ip"]}', "df -P / | tail -1", 10
)
parts = out.split()
if rc != 0 or len(parts) < 6:
return host["name"], None
return host["name"], {
"size_kib": int(parts[1]), "used_kib": int(parts[2]),
"avail_kib": int(parts[3]), "pct": parts[4], "mount": parts[5],
}
hosts = await _get_inventory_hosts_async()
pairs = await asyncio.gather(*(inspect_disk(host) for host in hosts))
return {name: item for name, item in pairs if item is not None}
@app.get("/disk/usage")
async def disk_usage(_=Depends(_verify)):
"""Root filesystem usage for all reachable inventory hosts."""
return await _collect_disk_usage()
def _add_component(summary: dict, bucket: dict, state: str):
normalized = state if state in ("healthy", "warning", "critical") else "warning"
summary[normalized] += 1
bucket[normalized] += 1
@app.get("/overview", response_model=OverviewResponse, response_model_exclude_none=True)
async def overview(details: bool = Query(False), _=Depends(_verify)):
"""Compact deterministic homelab verdict designed for small language models."""
services, hosts, backups, disks = await asyncio.gather(
_collect_service_status(),
_collect_health_all(),
_collect_backup_status(),
_collect_disk_usage(),
)
summary = {"critical": 0, "warning": 0, "healthy": 0}
component_summary = {
name: {"critical": 0, "warning": 0, "healthy": 0}
for name in ("services", "hosts", "backups", "disks")
}
findings = []
for name in sorted(hosts):
item = hosts[name]
containers = item.get("containers", [])
bad_container = next(
(line for line in containers if "unhealthy" in line.lower() or "restarting" in line.lower()),
None,
)
if not item.get("reachable"):
state = "critical"
findings.append({
"severity": "critical", "code": "host_unreachable", "target": name,
"message": "Host is not reachable over SSH",
})
elif bad_container:
state = "critical"
findings.append({
"severity": "critical", "code": "container_unhealthy", "target": name,
"message": bad_container[:200],
})
else:
state = "healthy"
_add_component(summary, component_summary["hosts"], state)
service_states = {
"healthy": "healthy", "degraded": "warning", "offline": "critical",
"auth_failed": "critical", "misconfigured": "critical",
}
service_codes = {
"degraded": "service_degraded", "offline": "service_offline",
"auth_failed": "service_auth_failed", "misconfigured": "service_misconfigured",
}
for name in sorted(key for key in services if not key.startswith("_")):
item = services[name]
raw_state = item.get("status", "degraded")
state = service_states.get(raw_state, "warning")
_add_component(summary, component_summary["services"], state)
if state != "healthy":
findings.append({
"severity": state,
"code": service_codes.get(raw_state, "service_degraded"),
"target": name,
"message": item.get("message", "Service health probe failed"),
})
for name in sorted(backups.get("hosts", {})):
item = backups["hosts"][name]
raw_state = item.get("state", "unknown")
state = "healthy" if raw_state == "healthy" else "critical" if raw_state in ("critical", "unknown") else "warning"
_add_component(summary, component_summary["backups"], state)
if state != "healthy":
findings.append({
"severity": state,
"code": f"backup_{raw_state}",
"target": name,
"message": "Backup is missing, stale or could not be verified",
"age_hours": item.get("age_hours"),
})
for name in sorted(disks):
pct = int(str(disks[name].get("pct", "0")).rstrip("%") or 0)
state = "critical" if pct >= 90 else "warning" if pct >= 80 else "healthy"
_add_component(summary, component_summary["disks"], state)
if state != "healthy":
findings.append({
"severity": state,
"code": "disk_critical" if state == "critical" else "disk_high",
"target": name,
"message": f"Root filesystem usage is {pct}%",
"pct": pct,
})
overall_state = "critical" if summary["critical"] else "warning" if summary["warning"] else "healthy"
response = {
"schema_version": 1,
"generated": datetime.now(timezone.utc).isoformat(),
"overall_state": overall_state,
"action_required": overall_state != "healthy",
"summary": summary,
"components": component_summary,
"findings": findings,
"model_contract": {
"instruction": "Report overall_state, then findings in the returned order. Do not infer missing facts.",
"severity_order": ["critical", "warning", "healthy"],
},
}
if details:
response["details"] = {"services": services, "hosts": hosts, "backups": backups, "disks": disks}
_audit("/overview", "GET", 200, f"state={overall_state} findings={len(findings)}")
return response
@app.get("/logs/{host}/{container}")
async def docker_logs(host: str, container: str, tail: int = Query(200, ge=1, le=20000), _=Depends(_verify)):
"""Read Docker logs from an inventory host."""
if not re.fullmatch(r"[A-Za-z0-9_.-]+", host) or not re.fullmatch(r"[A-Za-z0-9_.-]+", container):
raise HTTPException(400, "Invalid host or container name")
target = _find_inventory_host(host)
if not target:
raise HTTPException(404, f"Host {host} not found")
rc, out, err = _ssh(
f'{target["user"]}@{target["ip"]}',
f"sudo -n docker logs {container} --tail {tail} 2>&1 || docker logs {container} --tail {tail} 2>&1",
timeout=30,
)
if rc != 0:
raise HTTPException(502, (err or out).strip()[:500])
return {"host": host, "container": container, "tail": tail, "output": out}
@app.get("/docker/inspect/{host}/{container}")
async def docker_inspect(host: str, container: str, _=Depends(_verify)):
"""Return a sanitized runtime/resource summary for a Docker container."""
if not re.fullmatch(r"[A-Za-z0-9_.-]+", host) or not re.fullmatch(r"[A-Za-z0-9_.-]+", container):
raise HTTPException(400, "Invalid host or container name")
target = _find_inventory_host(host)
if not target:
raise HTTPException(404, f"Host {host} not found")
rc, out, err = _ssh(
f'{target["user"]}@{target["ip"]}',
f"sudo -n docker inspect {container}",
timeout=20,
)
if rc != 0:
raise HTTPException(502, (err or out).strip()[:500])
try:
raw = json.loads(out)[0]
except (json.JSONDecodeError, IndexError, TypeError):
raise HTTPException(502, "Invalid docker inspect response")
host_cfg = raw.get("HostConfig", {})
cfg = raw.get("Config", {})
state = raw.get("State", {})
return {
"name": raw.get("Name", "").lstrip("/"),
"image": cfg.get("Image"),
"state": {
"status": state.get("Status"), "running": state.get("Running"),
"started_at": state.get("StartedAt"), "exit_code": state.get("ExitCode"),
"oom_killed": state.get("OOMKilled"), "restart_count": raw.get("RestartCount"),
},
"runtime": host_cfg.get("Runtime"),
"resources": {
"memory": host_cfg.get("Memory"), "memory_reservation": host_cfg.get("MemoryReservation"),
"nano_cpus": host_cfg.get("NanoCpus"), "device_requests": host_cfg.get("DeviceRequests"),
},
"restart_policy": host_cfg.get("RestartPolicy"),
"log_config": host_cfg.get("LogConfig"),
"environment_keys": sorted(item.split("=", 1)[0] for item in cfg.get("Env", []) if "=" in item),
"mounts": [
{"type": mount.get("Type"), "source": mount.get("Source"), "destination": mount.get("Destination"), "rw": mount.get("RW")}
for mount in raw.get("Mounts", [])
],
}
@app.post("/docker/restart/{host}/{container}")
async def docker_restart(host: str, container: str, _=Depends(_verify), dry_run: bool = Query(False)):
"""Restart a named Docker container, with optional dry-run."""
if not re.fullmatch(r"[A-Za-z0-9_.-]+", host) or not re.fullmatch(r"[A-Za-z0-9_.-]+", container):
raise HTTPException(400, "Invalid host or container name")
target = _find_inventory_host(host)
if not target:
raise HTTPException(404, f"Host {host} not found")
if dry_run:
return {"dry_run": True, "host": host, "container": container}
rc, out, err = _ssh(f'{target["user"]}@{target["ip"]}', f"sudo -n docker restart {container}", timeout=45)
_audit(f"/docker/restart/{host}/{container}", "POST", 200 if rc == 0 else 502)
if rc != 0:
raise HTTPException(502, (err or out).strip()[:500])
return {"success": True, "output": out.strip()}
@app.post("/vault/reload")
async def vault_reload(_=Depends(_verify)):
_load_vault_cache()
@ -234,7 +773,6 @@ async def vault_reload(_=Depends(_verify)):
# --- VM Lifecycle Endpoints ---
from pydantic import BaseModel
import subprocess as _sp
AUTOMATION1 = VM_CFG.get("automation_host", "sascha@10.5.85.5") if VM_CFG else "sascha@10.5.85.5"
@ -249,9 +787,12 @@ class VMCreate(BaseModel):
disk: int = 32
def _ssh(host, cmd, timeout=600):
try:
r = _sp.run(["ssh","-o","ConnectTimeout=10","-o","StrictHostKeyChecking=accept-new",host,cmd],
capture_output=True, text=True, timeout=timeout)
return r.returncode, r.stdout, r.stderr
except _sp.TimeoutExpired:
return 124, "", f"SSH command timed out after {timeout} seconds"
def _pve_auth():
pv = _parse_kv("proxmox")
@ -497,36 +1038,67 @@ print('removed')
@app.post("/inventory/host")
async def inventory_host(request: Request, _=Depends(_verify)):
"""Create or update an Ansible inventory host idempotently."""
body = await request.json()
name, ip = body["name"], body["ip"]
name, ip = body.get("name", ""), body.get("ip", "")
group = body.get("group", "auto")
user = body.get("user", "sascha")
if not all(re.fullmatch(r"[A-Za-z0-9_.-]+", value) for value in (name, group, user)):
raise HTTPException(400, "Invalid name, group, or user")
try:
ipaddress.ip_address(ip)
except ValueError:
raise HTTPException(400, "Invalid IP address")
ini = "/app-config/ansible/pfannkuchen.ini"
# Add host to group in pfannkuchen.ini (idempotent)
add_cmd = f"""python3 -c "
lines = open('{ini}').readlines()
# Check if host already exists
if any('{name} ' in l or '{name}\\n' in l for l in lines):
print('already exists')
else:
# Find the group and insert after it
out, found = [], False
for l in lines:
out.append(l)
if l.strip() == '[{group}]':
found = True
elif found and (l.startswith('[') or l.strip() == ''):
out.insert(-1, '{name} ansible_host={ip}\\n')
found = False
if found: # group was last
out.append('{name} ansible_host={ip}\\n')
open('{ini}','w').writelines(out)
print('added to [{group}]')
" """
rc, out, _ = _ssh(AUTOMATION1, add_cmd, timeout=30)
# Also create host_vars
_ssh(AUTOMATION1, f"mkdir -p /app-config/ansible/host_vars/{name} && printf 'ansible_host: {ip}\\nansible_user: {user}\\n' > /app-config/ansible/host_vars/{name}/vars.yml", timeout=30)
return {"status": "ok", "name": name, "ip": ip, "group": group, "result": out.strip()}
host_line = f"{name} ansible_host={ip} ansible_user={user}"
script = f'''lines = open({ini!r}).readlines()
name = {name!r}
host_line = {host_line!r}
group = {group!r}
updated = False
for idx, line in enumerate(lines):
parts = line.split()
if parts and parts[0] == name and "ansible_host=" in line:
lines[idx] = host_line + "\\n"
updated = True
break
if not updated:
insert_at = None
in_group = False
for idx, line in enumerate(lines):
if line.strip() == "[" + group + "]":
in_group = True
insert_at = idx + 1
continue
if in_group and line.startswith("["):
break
if in_group:
insert_at = idx + 1
if insert_at is None:
lines.extend(["\\n[" + group + "]\\n", host_line + "\\n"])
else:
lines.insert(insert_at, host_line + "\\n")
open({ini!r}, "w").writelines(lines)
print("updated" if updated else "added")'''
encoded_script = base64.b64encode(script.encode()).decode()
rc, out, err = _ssh(
AUTOMATION1,
f"python3 -c \"import base64;exec(base64.b64decode('{encoded_script}'))\"",
timeout=30,
)
if rc != 0:
raise HTTPException(502, err.strip()[:500])
vars_script = (
f"mkdir -p /app-config/ansible/host_vars/{name} && "
f"printf 'ansible_host: {ip}\\nansible_user: {user}\\n' > "
f"/app-config/ansible/host_vars/{name}/vars.yml"
)
rc2, _out2, err2 = _ssh(AUTOMATION1, vars_script, timeout=30)
if rc2 != 0:
raise HTTPException(502, err2.strip()[:500])
_audit("/inventory/host", "POST", 200, f"{name} {ip} {user}")
return {"status": "ok", "name": name, "ip": ip, "group": group, "user": user, "result": out.strip()}
@app.post("/ansible/run")
async def ansible_run(request: Request, _=Depends(_verify)):
@ -749,17 +1321,20 @@ async def proxy(service: str, path: str, request: Request, _=Depends(_verify)):
target = f"{base_url}/{path}"
body = await request.body()
async with httpx.AsyncClient(verify=False, timeout=30.0) as client:
timeout = float(cfg.get("timeout", 30))
async with httpx.AsyncClient(verify=False, timeout=timeout) as client:
resp = await client.request(method=request.method, url=target,
headers=headers, cookies=cookies, content=body)
headers=headers, cookies=cookies, content=body,
params=request.query_params)
if auth_type == "session" and resp.status_code == 401:
_dockhand_cookie = None
await _dockhand_login(client)
resp = await client.request(method=request.method, url=target,
headers=headers, cookies=_dockhand_cookie or {}, content=body)
headers=headers, cookies=_dockhand_cookie or {}, content=body,
params=request.query_params)
try:
data = resp.json()
data = _redact_response(resp.json(), cfg.get("redact_response_fields"))
except Exception:
data = resp.text
_audit(f"/{service}/{path}", request.method, resp.status_code)

26
butler.yaml
View file

@ -8,6 +8,8 @@ services:
auth: session
vault_key: dockhand_password
description: "Docker management UI"
health_path: "/api/health"
redact_response_fields: [hawserToken, webhookSecret]
sonarr:
url: "http://10.2.1.100:8989"
@ -15,6 +17,7 @@ services:
key_file: sonarr
vault_key: sonarr_uhd_key
description: "TV show management (UHD)"
health_path: "/ping"
sonarr1080p:
auth: apikey_urlfile
@ -27,6 +30,7 @@ services:
key_file: radarr
vault_key: radarr_uhd_key
description: "Movie management (UHD)"
health_path: "/ping"
radarr1080p:
auth: apikey_urlfile
@ -39,6 +43,7 @@ services:
key_file: seer
vault_key: seerr_api_key
description: "Media request management"
health_path: "/api/v1/status"
outline:
url: "http://10.1.1.100:3000"
@ -53,6 +58,7 @@ services:
key_file: n8n
vault_key: n8n_api_key
description: "Workflow automation"
health_path: "/healthz"
proxmox:
url: "https://10.5.85.11:8006"
@ -65,6 +71,7 @@ services:
key_file: homeassistent
vault_key: ha_token
description: "Home automation"
health_path: "/api/"
grafana:
url: "http://10.1.1.111:3000"
@ -72,13 +79,14 @@ services:
key_file: grafana
vault_key: grafana_api_key
description: "Monitoring dashboards"
health_path: "/api/health"
uptime:
url: "http://159.69.245.190:3001"
url: "http://10.5.85.5:3001"
auth: bearer
key_file: uptime
vault_key: uptime_api_key
description: "Uptime monitoring"
description: "Uptime Kuma web UI on automation1 (no supported REST API)"
waha:
url: "http://10.4.1.110:3500"
@ -86,6 +94,8 @@ services:
key_file: waha_api_key
vault_key: waha_api_key
description: "WhatsApp API"
health_path: "/api/sessions"
health_expected: [200]
forgejo:
url: "http://10.4.1.116:3001"
@ -93,6 +103,7 @@ services:
key_file: forgejo
vault_key: forgejo_token
description: "Git server (Gitea fork)"
health_path: "/api/healthz"
semaphore:
url: "http://10.4.1.116:8090"
@ -100,6 +111,15 @@ services:
key_file: semaphore
vault_key: semaphore_token
description: "Ansible UI/API"
health_path: "/api/projects"
health_expected: [200]
fileflows:
url: "http://10.2.1.104:8268"
auth: none
description: "Media processing and transcoding"
health_path: "/"
timeout: 300
# VM lifecycle settings
vm:
@ -114,7 +134,7 @@ vm:
# TTS settings
tts:
speaker_url: "http://10.10.1.166:10800"
speaker_url: "http://10.5.85.2:10800"
chatterbox_url: "http://10.2.1.104:8004/tts"
chatterbox_health_url: "http://10.2.1.104:8004/api/model-info"
default_voice: "deep_thought.mp3"

2
requirements.txt
View file

@ -1,4 +1,4 @@
httpx
fastapi
uvicorn[standard]
pyyaml
PyYAML

23
tests/compose.integration.yaml Normal file
View file

@ -0,0 +1,23 @@
services:
homelab-butler-test:
build:
context: ..
container_name: homelab-butler-test
restart: "no"
ports:
- "127.0.0.1:8889:8888"
volumes:
- /app-config/kiro/api:/data/api:ro
- butler-vault-cache:/data/vault-cache:ro
- /home/sascha/.ssh:/root/.ssh:ro
- ../butler.yaml:/data/butler.yaml:ro
environment:
- API_KEY_DIR=/data/api
- VAULT_CACHE_DIR=/data/vault-cache
- BUTLER_CONFIG=/data/butler.yaml
- BUTLER_TOKEN=${BUTLER_TOKEN}
volumes:
butler-vault-cache:
external: true
name: homelab-butler_vault-cache

245
tests/test_app.py Normal file
View file

@ -0,0 +1,245 @@
import os
import asyncio
import time
from datetime import datetime, timedelta, timezone
os.environ.setdefault("BUTLER_TOKEN", "test-token")
from fastapi.testclient import TestClient
import app
def test_redact_response_recursively():
source = {
"id": 6,
"hawserToken": "secret-value",
"nested": [{"webhookSecret": "also-secret", "name": "tdarr"}],
}
result = app._redact_response(source)
assert result["hawserToken"] == "[REDACTED]"
assert result["nested"][0]["webhookSecret"] == "[REDACTED]"
assert result["nested"][0]["name"] == "tdarr"
def test_inventory_defaults_node7_and_explicit_user():
inventory = """
[node]
node2 ansible_host=10.5.85.12
[apps]
tdarr ansible_host=10.2.1.104
emby-chris ansible_host=10.7.1.106
special ansible_host=10.7.1.200 ansible_user=operator
"""
hosts = {item["name"]: item for item in app._inventory_hosts(inventory)}
assert hosts["node2"]["user"] == "root"
assert hosts["tdarr"]["user"] == "sascha"
assert hosts["emby-chris"]["user"] == "chris"
assert hosts["special"]["user"] == "operator"
def test_health_exposes_current_version():
with TestClient(app.app) as client:
response = client.get("/health")
assert response.status_code == 200
assert response.json()["version"] == app.VERSION == "2.3.0"
def test_invalid_log_target_is_rejected_before_ssh():
with TestClient(app.app) as client:
response = client.get(
"/logs/tdarr/fileflows;rm",
headers={"Authorization": "Bearer test-token"},
)
assert response.status_code == 400
def test_inventory_rejects_invalid_ip_before_ssh():
with TestClient(app.app) as client:
response = client.post(
"/inventory/host",
headers={"Authorization": "Bearer test-token"},
json={"name": "bad-host", "ip": "not-an-ip", "group": "auto"},
)
assert response.status_code == 400
def test_inventory_upsert_uses_base64_script(monkeypatch):
calls = []
def fake_ssh(host, command, timeout=600):
calls.append((host, command, timeout))
return 0, "updated", ""
monkeypatch.setattr(app, "_ssh", fake_ssh)
with TestClient(app.app) as client:
response = client.post(
"/inventory/host",
headers={"Authorization": "Bearer test-token"},
json={"name": "pfannkuchen", "ip": "46.225.230.72", "group": "vps", "user": "root"},
)
assert response.status_code == 200
assert "base64.b64decode" in calls[0][1]
assert "\\nname =" not in calls[0][1]
def test_docker_inspect_returns_sanitized_summary(monkeypatch):
raw = [{
"Name": "/fileflows",
"Config": {"Image": "revenz/fileflows:26.06", "Env": ["API_TOKEN=secret", "TZ=Europe/Berlin"]},
"State": {"Status": "running", "Running": True, "OOMKilled": False, "ExitCode": 0},
"RestartCount": 0,
"HostConfig": {"Runtime": "nvidia", "Memory": 0, "MemoryReservation": 0, "NanoCpus": 0,
"DeviceRequests": [], "RestartPolicy": {"Name": "always"}, "LogConfig": {"Type": "json-file"}},
"Mounts": [{"Type": "bind", "Source": "/app-config/fileflows-data", "Destination": "/app/Data", "RW": True}],
}]
monkeypatch.setattr(app, "_find_inventory_host", lambda _name: {"user": "sascha", "ip": "10.2.1.104"})
monkeypatch.setattr(app, "_ssh", lambda *args, **kwargs: (0, __import__("json").dumps(raw), ""))
with TestClient(app.app) as client:
response = client.get(
"/docker/inspect/tdarr/fileflows",
headers={"Authorization": "Bearer test-token"},
)
assert response.status_code == 200
result = response.json()
assert result["runtime"] == "nvidia"
assert result["environment_keys"] == ["API_TOKEN", "TZ"]
assert "secret" not in response.text
def test_classify_http_status_distinguishes_auth_and_route_errors():
assert app._classify_http_status(200, {200}) == "healthy"
assert app._classify_http_status(401, {200}) == "auth_failed"
assert app._classify_http_status(403, {200}) == "auth_failed"
assert app._classify_http_status(404, {200}) == "misconfigured"
assert app._classify_http_status(429, {200}) == "degraded"
assert app._classify_http_status(503, {200}) == "degraded"
def test_service_auth_headers_support_lightweight_health_probes(monkeypatch):
monkeypatch.setattr(app, "_get_key", lambda _cfg: "test-secret")
assert app._service_auth({"auth": "bearer"})["headers"] == {
"Authorization": "Bearer test-secret"
}
assert app._service_auth({"auth": "apikey"})["headers"] == {
"X-Api-Key": "test-secret"
}
assert app._service_auth({"auth": "n8n"})["headers"] == {
"X-N8N-API-KEY": "test-secret"
}
def _archive_json(start, name="archive"):
return __import__("json").dumps([{"archives": [{"start": start, "name": name}]}])
def test_backup_item_reports_age_and_severity():
now = datetime.now(timezone.utc)
recent = (now - timedelta(hours=4)).isoformat()
warning = (now - timedelta(hours=36)).isoformat()
critical = (now - timedelta(hours=60)).isoformat()
assert app._backup_item(0, _archive_json(recent), "")["state"] == "healthy"
assert app._backup_item(0, _archive_json(warning), "")["state"] == "warning"
assert app._backup_item(0, _archive_json(critical), "")["state"] == "critical"
assert app._backup_item(1, "", "timeout")["state"] == "unknown"
def test_ssh_timeout_is_normalized_instead_of_crashing_collection(monkeypatch):
def timeout(*_args, **_kwargs):
raise app._sp.TimeoutExpired(cmd=["ssh"], timeout=1)
monkeypatch.setattr(app._sp, "run", timeout)
rc, out, err = app._ssh("sascha@example", "true", timeout=1)
assert (rc, out) == (124, "")
assert "timed out" in err.lower()
def test_service_login_failure_is_isolated(monkeypatch):
monkeypatch.setattr(app, "SERVICES", {
"dockhand": {"url": "http://dockhand.invalid", "auth": "session", "health_path": "/api/health"}
})
async def failed_login(_client):
raise RuntimeError("login failed")
monkeypatch.setattr(app, "_dockhand_login", failed_login)
result = asyncio.run(app._collect_service_status())
assert result["dockhand"]["status"] == "offline"
assert result["dockhand"]["reachable"] is False
def test_backup_collection_runs_hosts_concurrently(monkeypatch):
active = 0
max_active = 0
monkeypatch.setattr(app, "_get_inventory_hosts", lambda: [
{"name": f"vm-{index}", "user": "sascha", "ip": f"10.1.1.{index}"}
for index in range(1, 5)
])
def fake_ssh(*_args, **_kwargs):
nonlocal active, max_active
active += 1
max_active = max(max_active, active)
time.sleep(0.05)
active -= 1
return 0, _archive_json(datetime.now(timezone.utc).isoformat()), ""
monkeypatch.setattr(app, "_ssh", fake_ssh)
result = asyncio.run(app._collect_backup_status(concurrency=4))
assert max_active > 1
assert result["summary"] == {
"total": 4, "healthy": 4, "warning": 0, "critical": 0, "unknown": 0
}
def test_overview_openapi_has_stable_enums_and_schema():
schema = app.app.openapi()
response_schema = schema["paths"]["/overview"]["get"]["responses"]["200"]["content"]["application/json"]["schema"]
assert response_schema["$ref"].endswith("/OverviewResponse")
overview_schema = schema["components"]["schemas"]["OverviewResponse"]
assert overview_schema["properties"]["overall_state"]["enum"] == ["healthy", "warning", "critical"]
finding_schema = schema["components"]["schemas"]["OverviewFinding"]
assert finding_schema["properties"]["severity"]["enum"] == ["healthy", "warning", "critical"]
def test_overview_is_compact_deterministic_and_light_model_friendly(monkeypatch):
async def service_data():
return {
"ok": {"status": "healthy", "reachable": True, "http": 200},
"bad-auth": {"status": "auth_failed", "reachable": True, "http": 401},
}
async def host_data():
return {
"vm1": {"reachable": True, "containers": ["app: Up 1 hour (healthy)"]},
"vm2": {"reachable": False, "containers": [], "error": "timeout"},
}
async def backup_data():
return {
"summary": {"total": 1, "healthy": 0, "warning": 1, "critical": 0, "unknown": 0},
"hosts": {"vm1": {"state": "warning", "age_hours": 36}},
}
async def disk_data():
return {"vm1": {"pct": "85%"}}
monkeypatch.setattr(app, "_collect_service_status", service_data)
monkeypatch.setattr(app, "_collect_health_all", host_data)
monkeypatch.setattr(app, "_collect_backup_status", backup_data)
monkeypatch.setattr(app, "_collect_disk_usage", disk_data)
with TestClient(app.app) as client:
response = client.get("/overview", headers={"Authorization": "Bearer test-token"})
assert response.status_code == 200
result = response.json()
assert result["schema_version"] == 1
assert result["overall_state"] == "critical"
assert result["summary"] == {"critical": 2, "warning": 2, "healthy": 2}
assert [item["code"] for item in result["findings"]] == [
"host_unreachable", "service_auth_failed", "backup_warning", "disk_high"
]