358 lines
13 KiB
Python
358 lines
13 KiB
Python
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.2"
|
|
|
|
|
|
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_uses_writable_runtime_known_hosts(monkeypatch):
|
|
captured = {}
|
|
|
|
class Result:
|
|
returncode = 0
|
|
stdout = "ok"
|
|
stderr = ""
|
|
|
|
def fake_run(args, **_kwargs):
|
|
captured["args"] = args
|
|
return Result()
|
|
|
|
monkeypatch.setattr(app._sp, "run", fake_run)
|
|
assert app._ssh("sascha@example", "true", timeout=1)[0] == 0
|
|
joined = " ".join(captured["args"])
|
|
assert "UserKnownHostsFile=/tmp/butler_known_hosts" in joined
|
|
|
|
|
|
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"
|
|
]
|
|
|
|
|
|
def test_proxy_route_validation_restricts_domain_and_upstream():
|
|
assert app._validate_proxy_route("speed.guck.tv", "127.0.0.1:8080") == (
|
|
"speed.guck.tv", "127.0.0.1:8080", "guck.tv", "speed"
|
|
)
|
|
|
|
for domain, upstream in [
|
|
("guck.tv", "127.0.0.1:8080"),
|
|
("speed.evil.example", "127.0.0.1:8080"),
|
|
("speed.guck.tv", "10.0.0.1:8080"),
|
|
("speed.guck.tv", "127.0.0.1:70000"),
|
|
("speed.guck.tv;rm", "127.0.0.1:8080"),
|
|
]:
|
|
try:
|
|
app._validate_proxy_route(domain, upstream)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError(f"unsafe route accepted: {domain} -> {upstream}")
|
|
|
|
|
|
def test_proxy_route_endpoint_configures_caddy_and_dns(monkeypatch):
|
|
calls = []
|
|
|
|
def fake_caddy(domain, upstream):
|
|
calls.append(("caddy", domain, upstream))
|
|
return {"status": "reloaded", "backup": "/app-config/caddy/Caddyfile.bak-test"}
|
|
|
|
async def fake_dns(zone, name, token_override=None):
|
|
calls.append(("dns", zone, name, token_override))
|
|
return {"zone_id": 123, "records": ["A", "AAAA"]}
|
|
|
|
monkeypatch.setattr(app, "_configure_caddy_route", fake_caddy)
|
|
monkeypatch.setattr(app, "_upsert_dns_records", fake_dns)
|
|
with TestClient(app.app) as client:
|
|
response = client.post(
|
|
"/vps/proxy-route",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
json={"domain": "speed.guck.tv", "upstream": "127.0.0.1:8080", "dns_token": "test-dns-token"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "configured"
|
|
assert calls == [
|
|
("caddy", "speed.guck.tv", "127.0.0.1:8080"),
|
|
("dns", "guck.tv", "speed", "test-dns-token"),
|
|
]
|
|
|
|
|
|
def test_hetzner_token_refreshes_vault_cache_when_missing(monkeypatch):
|
|
reads = iter([None, "refreshed-token"])
|
|
calls = []
|
|
monkeypatch.setattr(app, "_read", lambda _name: next(reads))
|
|
monkeypatch.setattr(app, "_load_vault_cache", lambda: calls.append("reload"))
|
|
monkeypatch.setattr(app, "_ssh", lambda host, command, timeout=600: (calls.append((host, command, timeout)) or (0, "vault-sync: ok", "")))
|
|
|
|
assert app._get_hetzner_dns_token() == "refreshed-token"
|
|
assert calls[0][0] == "sascha@10.4.1.116"
|
|
assert calls[0][1] == "sudo bash /data/stacks/homelab-butler/vault-sync.sh"
|
|
assert calls[1] == "reload"
|
|
|
|
|
|
def test_speedtest_deploy_requires_strong_password_and_uses_git_compose(monkeypatch):
|
|
calls = []
|
|
|
|
async def fake_fetch(repo, path):
|
|
calls.append(("fetch", repo, path))
|
|
return "services:\n speedtest:\n image: ghcr.io/librespeed/speedtest:latest\n"
|
|
|
|
def fake_deploy(compose, password):
|
|
calls.append(("deploy", compose, password))
|
|
return {"status": "deployed", "health": "ok"}
|
|
|
|
monkeypatch.setattr(app, "_fetch_forgejo_text", fake_fetch)
|
|
monkeypatch.setattr(app, "_deploy_speedtest_compose", fake_deploy)
|
|
with TestClient(app.app) as client:
|
|
weak = client.post(
|
|
"/vps/speedtest/deploy",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
json={"stats_password": "short"},
|
|
)
|
|
response = client.post(
|
|
"/vps/speedtest/deploy",
|
|
headers={"Authorization": "Bearer test-token"},
|
|
json={"stats_password": "correct-horse-battery-staple"},
|
|
)
|
|
|
|
assert weak.status_code == 400
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "deployed"
|
|
assert calls == [
|
|
("fetch", "sascha/speedtest", "compose.yaml"),
|
|
("deploy", "services:\n speedtest:\n image: ghcr.io/librespeed/speedtest:latest\n", "correct-horse-battery-staple"),
|
|
]
|