"""Tests for Executor container_restart dispatch — the no-SSH remediation path. Covers kb/phases/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 os import sys import time from pathlib import Path from unittest.mock import MagicMock 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") # Must be patched too: _ensure_dirs() creates it, and the module default # points at the real /opt/homelab. monkeypatch.setattr(executor_module, "DEPLOY_DISPATCH_DIR", actions / "deploy") 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--) 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_synchronous_running_actions(tmp_path, monkeypatch): """Only dispatched types (container_restart, redeploy) are reconciled here. disk_cleanup/alert_only resolve inside _execute_action and never linger.""" ex = _setup_executor(tmp_path, monkeypatch) running_path = tmp_path / "actions" / "running" / "disk-cleanup-x.json" running_path.write_text(json.dumps({ "action_id": "disk-cleanup-x", "type": "disk_cleanup", "node": "piha", "started_at": time.time() - 999_999, })) ex._reconcile_running_actions() assert running_path.exists() # untouched 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")) # --------------------------------------------------------------------------- # Inbox permissions (dispatch leak, 2026-08-06) # # The per-node inbox is written here but drained by the target node over rsync # --remove-source-files, authenticating as a different user that is only a # group member. Unlinking the fetched file needs write permission on the # containing dir; at 0o755 it silently fails and the node re-pulls the same # action forever. # --------------------------------------------------------------------------- @pytest.fixture def restrictive_umask(): """0o022 masks the group-write bit out of mkdir(mode=0o775) — the reason the fix cannot rely on the mode argument alone.""" old = os.umask(0o022) yield os.umask(old) def _mode(path: Path) -> int: return path.stat().st_mode & 0o777 def test_dispatch_inbox_is_group_writable(tmp_path, monkeypatch, restrictive_umask): ex = _setup_executor(tmp_path, monkeypatch) ex._execute_action(_write_approved(tmp_path, "cr-perm", node="piha")) inbox = tmp_path / "actions" / "dispatch" / "piha" assert (inbox / "cr-perm.json").exists() assert _mode(inbox) == 0o775 def test_deploy_inbox_is_group_writable(tmp_path, monkeypatch, restrictive_umask): """Same defect, same rsync-pull drain — the deploy runner's inbox needs it too.""" ex = _setup_executor(tmp_path, monkeypatch) ex._execute_action( _write_approved(tmp_path, "rd-perm", node="piha", action_type="redeploy") ) inbox = tmp_path / "actions" / "deploy" / "piha" assert (inbox / "rd-perm.json").exists() assert _mode(inbox) == 0o775 def test_existing_inbox_is_repaired_in_place(tmp_path, monkeypatch, restrictive_umask): """Inboxes already on disk fleet-wide were created at 0o755 by an earlier build; dispatching to one must fix it rather than inherit it.""" ex = _setup_executor(tmp_path, monkeypatch) inbox = tmp_path / "actions" / "dispatch" / "piha" inbox.mkdir(parents=True) os.chmod(inbox, 0o755) ex._execute_action(_write_approved(tmp_path, "cr-repair", node="piha")) assert _mode(inbox) == 0o775 def test_dispatch_survives_unsettable_mode(tmp_path, monkeypatch, restrictive_umask): """A chmod failure (inbox owned by another user) must not cost us the dispatch — the action still executes on the node; only the source delete stays broken, and the node warns about that.""" ex = _setup_executor(tmp_path, monkeypatch) monkeypatch.setattr( executor_module.os, "chmod", MagicMock(side_effect=PermissionError("Operation not permitted")), ) ex._execute_action(_write_approved(tmp_path, "cr-chmod-fail", node="piha")) assert (tmp_path / "actions" / "dispatch" / "piha" / "cr-chmod-fail.json").exists() assert _exists(tmp_path, "running", "cr-chmod-fail")