2dac154d78
Executor nie ma klienta ssh ani klucza do floty (uid 1000 homelab, brak ~/.ssh, brak resolucji nazw wezlow) — container_restart przez subprocess ssh failowal w 6ms na kazdej probie. Zamiast dodawac SSH do executora, kierunek jest odwrocony: executor zapisuje zlecenie do /opt/homelab/actions/dispatch/<node>/<action_id>.json, a node-agent na docelowym wezle (ktory ma dzialajacy docker.sock i juz ma klucz SSH do VPS uzywany do shippingu eventow) sam je odbiera i wykonuje lokalnie. - executor: _dispatch_container_restart pisze zlecenie zamiast ssh; _reconcile_running_actions konsumuje zwrotne action_result eventy i timeoutuje akcje bez odpowiedzi (ACTION_TIMEOUT_SECS, domyslnie 300s). redeploy/disk_cleanup/alert_only bez zmian. - node-agent: nowy krok w petli — rsync-pull wlasnej podkatalogu dispatch z VPS (ten sam klucz co _ship_events_to_vps, w przeciwnym kierunku; no-op na VPS, gdzie katalog jest lokalny), walidacja (node_name, whitelist tylko container_restart, odmowa restartu wlasnego kontenera), wykonanie przez docker SDK, raport jako event action_result (istniejacy kanal shippingu). Idempotencja przez znacznik w /opt/homelab/state/processed-actions/. - 26 nowych testow (10 executor, 16 node-agent), pelny suite obu serwisow 183/183 zielony. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
237 lines
8.8 KiB
Python
237 lines
8.8 KiB
Python
"""Tests for Executor container_restart dispatch — the no-SSH remediation path.
|
|
|
|
Covers docs/backlog.md "PROJEKT: remediacja bez SSH": the executor never
|
|
shells out to SSH for container_restart. Instead it drops a dispatch file for
|
|
the target node's node-agent to pick up, and later resolves the action from
|
|
either a matching action_result event or a timeout — never leaving it running
|
|
forever.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
import executor as executor_module
|
|
from executor import Executor
|
|
|
|
|
|
def _setup_executor(tmp_path: Path, monkeypatch) -> Executor:
|
|
actions = tmp_path / "actions"
|
|
events = tmp_path / "events"
|
|
repo = tmp_path / "repo"
|
|
state = tmp_path / "state"
|
|
for d in (actions, events, repo, state):
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
|
|
monkeypatch.setattr(executor_module, "ACTIONS_DIR", actions)
|
|
monkeypatch.setattr(executor_module, "EVENTS_DIR", events)
|
|
monkeypatch.setattr(executor_module, "DISPATCH_DIR", actions / "dispatch")
|
|
monkeypatch.setattr(executor_module, "REPO_ROOT", repo)
|
|
return Executor()
|
|
|
|
|
|
def _write_approved(tmp_path, action_id, node="piha", service="zigbee2mqtt",
|
|
container_name=None, action_type="container_restart"):
|
|
action = {
|
|
"action_id": action_id,
|
|
"type": action_type,
|
|
"node": node,
|
|
"service": service,
|
|
"container_name": container_name or service,
|
|
"status": "approved",
|
|
"timestamp": time.time(),
|
|
}
|
|
path = tmp_path / "actions" / "approved" / f"{action_id}.json"
|
|
path.write_text(json.dumps(action))
|
|
return path
|
|
|
|
|
|
def _write_action_result_event(tmp_path, node, action_id, success, error="", ts=None):
|
|
ts = int(ts if ts is not None else time.time())
|
|
node_dir = tmp_path / "events" / node
|
|
node_dir.mkdir(parents=True, exist_ok=True)
|
|
event = {
|
|
"id": f"evt-{node}-{ts}-action_result-{action_id}",
|
|
"timestamp": ts,
|
|
"type": "action_result",
|
|
"node": node,
|
|
"payload": {"action_id": action_id, "success": success, "error": error, "node": node},
|
|
}
|
|
path = node_dir / f"evt-{node}-{ts}-action_result-{action_id}.json"
|
|
path.write_text(json.dumps(event))
|
|
return path
|
|
|
|
|
|
def _read(tmp_path, state, action_id):
|
|
return json.loads((tmp_path / "actions" / state / f"{action_id}.json").read_text())
|
|
|
|
|
|
def _exists(tmp_path, state, action_id):
|
|
return (tmp_path / "actions" / state / f"{action_id}.json").exists()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatch: no SSH, writes to DISPATCH_DIR, stays in running
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_container_restart_dispatched_not_ssh(tmp_path, monkeypatch):
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
called = {"ssh": False}
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
if cmd and cmd[0] == "ssh":
|
|
called["ssh"] = True
|
|
raise AssertionError("executor must never invoke subprocess for container_restart")
|
|
|
|
monkeypatch.setattr(executor_module.subprocess, "run", fake_run)
|
|
|
|
action_file = _write_approved(tmp_path, "cr-1", node="piha", container_name="zigbee2mqtt")
|
|
ex._execute_action(action_file)
|
|
|
|
assert not called["ssh"]
|
|
dispatch_file = tmp_path / "actions" / "dispatch" / "piha" / "cr-1.json"
|
|
assert dispatch_file.exists()
|
|
payload = json.loads(dispatch_file.read_text())
|
|
assert payload["action_id"] == "cr-1"
|
|
assert payload["container_name"] == "zigbee2mqtt"
|
|
assert payload["node"] == "piha"
|
|
|
|
# Stays in running — not resolved synchronously.
|
|
assert _exists(tmp_path, "running", "cr-1")
|
|
assert not _exists(tmp_path, "completed", "cr-1")
|
|
assert not _exists(tmp_path, "failed", "cr-1")
|
|
|
|
|
|
def test_container_restart_with_no_node_does_not_crash(tmp_path, monkeypatch):
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
action_file = _write_approved(tmp_path, "cr-none", node="", container_name="x")
|
|
|
|
ex._execute_action(action_file) # must not raise
|
|
|
|
assert list((tmp_path / "actions" / "dispatch").glob("*/cr-none.json")) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reconcile: action_result event resolves the action
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_reconcile_moves_to_completed_on_success_result(tmp_path, monkeypatch):
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
action_file = _write_approved(tmp_path, "cr-2", node="piha")
|
|
ex._execute_action(action_file)
|
|
|
|
_write_action_result_event(tmp_path, "piha", "cr-2", success=True)
|
|
ex._reconcile_running_actions()
|
|
|
|
assert _exists(tmp_path, "completed", "cr-2")
|
|
assert not _exists(tmp_path, "running", "cr-2")
|
|
|
|
|
|
def test_reconcile_moves_to_failed_on_failure_result(tmp_path, monkeypatch):
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
action_file = _write_approved(tmp_path, "cr-3", node="piha")
|
|
ex._execute_action(action_file)
|
|
|
|
_write_action_result_event(tmp_path, "piha", "cr-3", success=False, error="no such container")
|
|
ex._reconcile_running_actions()
|
|
|
|
assert _exists(tmp_path, "failed", "cr-3")
|
|
data = _read(tmp_path, "failed", "cr-3")
|
|
assert data["error"] == "no such container"
|
|
|
|
|
|
def test_reconcile_ignores_stale_result_from_previous_run(tmp_path, monkeypatch):
|
|
"""A same-action_id result left over from an earlier run (before this run's
|
|
started_at) must never resolve the CURRENT run — action_ids are
|
|
deterministic (container-restart-<node>-<service>) and can repeat days
|
|
apart."""
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
|
|
# Stale result from "yesterday"
|
|
stale_ts = int(time.time()) - 86_400
|
|
_write_action_result_event(tmp_path, "piha", "cr-4", success=True, ts=stale_ts)
|
|
|
|
action_file = _write_approved(tmp_path, "cr-4", node="piha")
|
|
ex._execute_action(action_file) # started_at is "now", after the stale event
|
|
|
|
ex._reconcile_running_actions()
|
|
|
|
assert _exists(tmp_path, "running", "cr-4")
|
|
assert not _exists(tmp_path, "completed", "cr-4")
|
|
|
|
|
|
def test_reconcile_no_op_when_no_result_and_not_timed_out(tmp_path, monkeypatch):
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
action_file = _write_approved(tmp_path, "cr-5", node="piha")
|
|
ex._execute_action(action_file)
|
|
|
|
ex._reconcile_running_actions()
|
|
|
|
assert _exists(tmp_path, "running", "cr-5")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Timeout
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_reconcile_times_out_stuck_action(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(executor_module, "ACTION_TIMEOUT_SECS", 5)
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
action_file = _write_approved(tmp_path, "cr-6", node="piha")
|
|
ex._execute_action(action_file)
|
|
|
|
running_path = tmp_path / "actions" / "running" / "cr-6.json"
|
|
data = json.loads(running_path.read_text())
|
|
data["started_at"] = time.time() - 10 # older than the 5s timeout
|
|
running_path.write_text(json.dumps(data))
|
|
|
|
ex._reconcile_running_actions()
|
|
|
|
assert _exists(tmp_path, "failed", "cr-6")
|
|
data = _read(tmp_path, "failed", "cr-6")
|
|
assert "Timed out" in data["error"]
|
|
|
|
|
|
def test_reconcile_does_not_time_out_before_deadline(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(executor_module, "ACTION_TIMEOUT_SECS", 300)
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
action_file = _write_approved(tmp_path, "cr-7", node="piha")
|
|
ex._execute_action(action_file)
|
|
|
|
ex._reconcile_running_actions()
|
|
|
|
assert _exists(tmp_path, "running", "cr-7")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Other action types are untouched by the reconcile loop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_reconcile_ignores_non_container_restart_running_actions(tmp_path, monkeypatch):
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
running_path = tmp_path / "actions" / "running" / "redeploy-x.json"
|
|
running_path.write_text(json.dumps({
|
|
"action_id": "redeploy-x", "type": "redeploy", "node": "piha",
|
|
"started_at": time.time() - 999_999,
|
|
}))
|
|
|
|
ex._reconcile_running_actions()
|
|
|
|
assert running_path.exists() # untouched: redeploy resolves synchronously elsewhere
|
|
|
|
|
|
def test_alert_only_action_resolves_synchronously_not_via_dispatch(tmp_path, monkeypatch):
|
|
ex = _setup_executor(tmp_path, monkeypatch)
|
|
action_file = _write_approved(tmp_path, "alert-1", node="piha", action_type="alert_only")
|
|
|
|
ex._execute_action(action_file)
|
|
|
|
assert _exists(tmp_path, "completed", "alert-1")
|
|
assert not (tmp_path / "actions" / "dispatch").exists() or \
|
|
not list((tmp_path / "actions" / "dispatch").glob("**/*.json"))
|