17147856d1
- Remove unused imports (app.py, stego_routes.py, killswitch.py, etc.) - Sort import blocks (I001) - Add missing os import in stego_routes.py (F821) - Rename shadowed Click commands to avoid F811 (status→chain_status, show→chain_show) - Rename uppercase locals R→earth_r, _HAS_QRCODE_READ→_has_qrcode_read (N806) - Suppress false-positive F821 for get_username (closure scope) - Use datetime.UTC alias (UP017) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
297 lines
9.7 KiB
Python
297 lines
9.7 KiB
Python
"""
|
|
Tests for dead man's switch background enforcement and CLI command.
|
|
|
|
Covers:
|
|
- _deadman_enforcement_loop: does not call execute_purge when not armed
|
|
- _deadman_enforcement_loop: calls execute_purge when armed and overdue
|
|
- _deadman_enforcement_loop: exits after firing so execute_purge is not called twice
|
|
- _start_deadman_thread: returns a live daemon thread
|
|
- check-deadman CLI: exits 0 when disarmed
|
|
- check-deadman CLI: exits 0 and prints OK when armed but not overdue
|
|
- check-deadman CLI: exits 0 and prints OVERDUE warning when past interval but in grace
|
|
- check-deadman CLI: exits 2 when fully expired (past interval + grace)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
# ── Fixtures ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture()
|
|
def soosef_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
|
"""Redirect soosef paths to a tmp directory."""
|
|
import soosef.paths as paths
|
|
|
|
data_dir = tmp_path / ".soosef"
|
|
data_dir.mkdir()
|
|
monkeypatch.setattr(paths, "BASE_DIR", data_dir)
|
|
return data_dir
|
|
|
|
|
|
def _write_deadman_state(
|
|
state_file: Path,
|
|
*,
|
|
armed: bool,
|
|
last_checkin: datetime,
|
|
interval_hours: int = 24,
|
|
grace_hours: int = 2,
|
|
) -> None:
|
|
state_file.parent.mkdir(parents=True, exist_ok=True)
|
|
state = {
|
|
"armed": armed,
|
|
"last_checkin": last_checkin.isoformat(),
|
|
"interval_hours": interval_hours,
|
|
"grace_hours": grace_hours,
|
|
}
|
|
state_file.write_text(json.dumps(state))
|
|
|
|
|
|
# ── Unit tests: enforcement loop ─────────────────────────────────────────────
|
|
|
|
|
|
def test_enforcement_loop_no_op_when_disarmed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
"""Loop should not call check() when the switch is not armed."""
|
|
from soosef.cli import _deadman_enforcement_loop
|
|
from soosef.fieldkit import deadman as deadman_mod
|
|
|
|
# Redirect the module-level DEADMAN_STATE constant so DeadmanSwitch() default is our tmp file
|
|
state_file = tmp_path / "deadman.json"
|
|
monkeypatch.setattr(deadman_mod, "DEADMAN_STATE", state_file)
|
|
|
|
check_calls = []
|
|
|
|
def fake_check(self):
|
|
check_calls.append("fired")
|
|
|
|
monkeypatch.setattr(deadman_mod.DeadmanSwitch, "check", fake_check)
|
|
|
|
iterations = [0]
|
|
|
|
def one_shot_sleep(n):
|
|
iterations[0] += 1
|
|
if iterations[0] >= 2:
|
|
raise StopIteration("stop test loop")
|
|
|
|
monkeypatch.setattr(time, "sleep", one_shot_sleep)
|
|
|
|
with pytest.raises(StopIteration):
|
|
_deadman_enforcement_loop(interval_seconds=0)
|
|
|
|
assert check_calls == []
|
|
|
|
|
|
def test_enforcement_loop_fires_when_overdue(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
"""Loop must call DeadmanSwitch.check() when armed and past interval + grace."""
|
|
from soosef.cli import _deadman_enforcement_loop
|
|
from soosef.fieldkit import deadman as deadman_mod
|
|
|
|
state_file = tmp_path / "deadman.json"
|
|
monkeypatch.setattr(deadman_mod, "DEADMAN_STATE", state_file)
|
|
|
|
last_checkin = datetime.now(UTC) - timedelta(hours=100)
|
|
_write_deadman_state(
|
|
state_file,
|
|
armed=True,
|
|
last_checkin=last_checkin,
|
|
interval_hours=24,
|
|
grace_hours=2,
|
|
)
|
|
|
|
check_calls = []
|
|
|
|
def fake_check(self):
|
|
check_calls.append("fired")
|
|
|
|
monkeypatch.setattr(deadman_mod.DeadmanSwitch, "check", fake_check)
|
|
monkeypatch.setattr(time, "sleep", lambda n: None)
|
|
|
|
_deadman_enforcement_loop(interval_seconds=0)
|
|
|
|
assert len(check_calls) == 1
|
|
|
|
|
|
def test_enforcement_loop_exits_after_firing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
"""After firing, the loop must return and not call check() a second time."""
|
|
from soosef.cli import _deadman_enforcement_loop
|
|
from soosef.fieldkit import deadman as deadman_mod
|
|
|
|
state_file = tmp_path / "deadman.json"
|
|
monkeypatch.setattr(deadman_mod, "DEADMAN_STATE", state_file)
|
|
|
|
last_checkin = datetime.now(UTC) - timedelta(hours=100)
|
|
_write_deadman_state(state_file, armed=True, last_checkin=last_checkin)
|
|
|
|
check_calls = []
|
|
|
|
def fake_check(self):
|
|
check_calls.append("fired")
|
|
|
|
monkeypatch.setattr(deadman_mod.DeadmanSwitch, "check", fake_check)
|
|
monkeypatch.setattr(time, "sleep", lambda n: None)
|
|
|
|
_deadman_enforcement_loop(interval_seconds=0)
|
|
|
|
# Called exactly once — loop exited after firing
|
|
assert len(check_calls) == 1
|
|
|
|
|
|
def test_enforcement_loop_tolerates_exceptions(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
"""Transient errors in check() must not kill the loop."""
|
|
from soosef.cli import _deadman_enforcement_loop
|
|
from soosef.fieldkit import deadman as deadman_mod
|
|
|
|
state_file = tmp_path / "deadman.json"
|
|
monkeypatch.setattr(deadman_mod, "DEADMAN_STATE", state_file)
|
|
|
|
call_count = [0]
|
|
|
|
def counting_sleep(n):
|
|
call_count[0] += 1
|
|
if call_count[0] >= 3:
|
|
raise StopIteration("stop test loop")
|
|
|
|
monkeypatch.setattr(time, "sleep", counting_sleep)
|
|
|
|
error_calls = [0]
|
|
|
|
def flaky_is_armed(self):
|
|
error_calls[0] += 1
|
|
if error_calls[0] == 1:
|
|
raise OSError("state file temporarily unreadable")
|
|
return False # not armed — loop just skips
|
|
|
|
monkeypatch.setattr(deadman_mod.DeadmanSwitch, "is_armed", flaky_is_armed)
|
|
|
|
with pytest.raises(StopIteration):
|
|
_deadman_enforcement_loop(interval_seconds=0)
|
|
|
|
# Should have survived the first exception and continued
|
|
assert call_count[0] >= 2
|
|
|
|
|
|
# ── Unit tests: _start_deadman_thread ────────────────────────────────────────
|
|
|
|
|
|
def test_start_deadman_thread_is_daemon(monkeypatch: pytest.MonkeyPatch):
|
|
"""Thread must be a daemon so it dies with the process."""
|
|
# Patch the loop to exit immediately so the thread doesn't hang in tests
|
|
import soosef.cli as cli_mod
|
|
from soosef.cli import _start_deadman_thread
|
|
|
|
monkeypatch.setattr(cli_mod, "_deadman_enforcement_loop", lambda interval_seconds: None)
|
|
|
|
t = _start_deadman_thread(interval_seconds=60)
|
|
assert t is not None
|
|
assert t.daemon is True
|
|
assert t.name == "deadman-enforcement"
|
|
t.join(timeout=2)
|
|
|
|
|
|
# ── CLI integration: check-deadman ───────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture()
|
|
def cli_runner():
|
|
return CliRunner()
|
|
|
|
|
|
def test_check_deadman_disarmed(
|
|
tmp_path: Path, cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
"""check-deadman exits 0 and prints helpful message when not armed."""
|
|
from soosef.cli import main
|
|
from soosef.fieldkit import deadman as deadman_mod
|
|
|
|
# Point at an empty tmp dir so the real ~/.soosef/fieldkit/deadman.json isn't read
|
|
state_file = tmp_path / "deadman.json"
|
|
monkeypatch.setattr(deadman_mod, "DEADMAN_STATE", state_file)
|
|
|
|
result = cli_runner.invoke(main, ["fieldkit", "check-deadman"])
|
|
assert result.exit_code == 0
|
|
assert "not armed" in result.output
|
|
|
|
|
|
def test_check_deadman_armed_ok(
|
|
tmp_path: Path, cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
"""check-deadman exits 0 when armed and check-in is current."""
|
|
from soosef.cli import main
|
|
from soosef.fieldkit import deadman as deadman_mod
|
|
|
|
state_file = tmp_path / "deadman.json"
|
|
monkeypatch.setattr(deadman_mod, "DEADMAN_STATE", state_file)
|
|
|
|
last_checkin = datetime.now(UTC) - timedelta(hours=1)
|
|
_write_deadman_state(
|
|
state_file,
|
|
armed=True,
|
|
last_checkin=last_checkin,
|
|
interval_hours=24,
|
|
grace_hours=2,
|
|
)
|
|
|
|
result = cli_runner.invoke(main, ["fieldkit", "check-deadman"])
|
|
assert result.exit_code == 0
|
|
assert "OK" in result.output
|
|
|
|
|
|
def test_check_deadman_overdue_in_grace(
|
|
tmp_path: Path, cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
"""check-deadman exits 0 but prints OVERDUE warning when past interval but in grace."""
|
|
from soosef.cli import main
|
|
from soosef.fieldkit import deadman as deadman_mod
|
|
|
|
state_file = tmp_path / "deadman.json"
|
|
monkeypatch.setattr(deadman_mod, "DEADMAN_STATE", state_file)
|
|
|
|
# Past 24h interval but within 26h total (grace=2)
|
|
last_checkin = datetime.now(UTC) - timedelta(hours=25)
|
|
_write_deadman_state(
|
|
state_file,
|
|
armed=True,
|
|
last_checkin=last_checkin,
|
|
interval_hours=24,
|
|
grace_hours=2,
|
|
)
|
|
|
|
result = cli_runner.invoke(main, ["fieldkit", "check-deadman"])
|
|
# Not yet fired (grace not expired), so exit code is 0
|
|
assert result.exit_code == 0
|
|
assert "OVERDUE" in result.output
|
|
|
|
|
|
def test_check_deadman_fires_when_expired(
|
|
tmp_path: Path, cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
"""check-deadman exits 2 when the switch has fully expired."""
|
|
from soosef.cli import main
|
|
from soosef.fieldkit import deadman as deadman_mod
|
|
|
|
state_file = tmp_path / "deadman.json"
|
|
monkeypatch.setattr(deadman_mod, "DEADMAN_STATE", state_file)
|
|
|
|
last_checkin = datetime.now(UTC) - timedelta(hours=100)
|
|
_write_deadman_state(
|
|
state_file,
|
|
armed=True,
|
|
last_checkin=last_checkin,
|
|
interval_hours=24,
|
|
grace_hours=2,
|
|
)
|
|
|
|
# Patch check() so we don't invoke the real killswitch during tests
|
|
monkeypatch.setattr(deadman_mod.DeadmanSwitch, "check", lambda self: None)
|
|
|
|
result = cli_runner.invoke(main, ["fieldkit", "check-deadman"])
|
|
assert result.exit_code == 2
|
|
assert "killswitch triggered" in result.output.lower() or "expired" in result.output.lower()
|