homelab-codex-ws/services/control-plane/tests/test_incident_lifecycle.py

"""Tests for incident lifecycle: auto-resolve, orphan detection, timestamp parsing."""
from __future__ import annotations

import json
import sys
import time
from pathlib import Path

import pytest

# Observer lives outside the control-plane package; add scripts/ to path.
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "scripts"))
from observer.observer import Observer, _parse_ts, _atomic_write_json


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

@pytest.fixture(autouse=True)
def _redirect_observer_paths(tmp_path, monkeypatch):
    """Redirect every observer.observer runtime path into a per-test tmp_path.

    Uses monkeypatch so the patches are reverted after each test — crucially
    including OBSERVER_STATE_FILE, which observer.observer derives from STATE_DIR
    *at import time* (OBSERVER_STATE_FILE = STATE_DIR / "observer_checkpoint.json").

    The previous _make_observer_simple helper patched STATE_DIR but never
    OBSERVER_STATE_FILE, so run_once() / _save_checkpoint() wrote the checkpoint to
    the real /opt/homelab/state/observer_checkpoint.json. That stale file leaked
    node_checkpoints (tmp paths tagged with a pytest run number) across tests AND
    across pytest runs; the string comparison `file_path > node_checkpoints[node]`
    in run_once then skipped or kept events depending purely on run-number ordering,
    making the test_run_once_* cases flaky. Redirecting + restoring every path here
    isolates each test from disk and from sibling tests.
    """
    import observer.observer as obs_mod

    world = tmp_path / "world"
    state = tmp_path / "state"
    events = tmp_path / "events"
    logs = tmp_path / "logs"
    repo = tmp_path / "repo"

    for d in (world, state, events, logs, repo / "inventory", repo / "hosts"):
        d.mkdir(parents=True, exist_ok=True)

    # Minimal topology so inventory isn't empty (avoids prune-guard early-return).
    (repo / "inventory" / "topology.yaml").write_text(
        "nodes:\n  vps:\n    roles: [control-plane]\n    connectivity: {}\n"
    )

    monkeypatch.setattr(obs_mod, "WORLD_DIR", world)
    monkeypatch.setattr(obs_mod, "STATE_DIR", state)
    monkeypatch.setattr(obs_mod, "EVENTS_DIR", events)
    monkeypatch.setattr(obs_mod, "LOGS_DIR", logs)
    monkeypatch.setattr(obs_mod, "INVENTORY_TOPOLOGY", repo / "inventory" / "topology.yaml")
    monkeypatch.setattr(obs_mod, "REPO_ROOT", repo)
    monkeypatch.setattr(obs_mod, "FAILED_EVENTS_DIR", state / "observer_failed_events")
    monkeypatch.setattr(obs_mod, "OBSERVER_STATE_FILE", state / "observer_checkpoint.json")


def _make_observer_simple(tmp_path: Path) -> Observer:
    """Return an Observer. All runtime paths are redirected (and restored) by the
    autouse _redirect_observer_paths fixture, which shares this test's tmp_path."""
    return Observer()


# ---------------------------------------------------------------------------
# 1. _parse_ts — timestamp normalisation
# ---------------------------------------------------------------------------

def test_parse_ts_int():
    ts = int(time.time()) - 3600
    assert abs(_parse_ts(ts) - ts) < 1


def test_parse_ts_float():
    ts = time.time() - 100.5
    assert abs(_parse_ts(ts) - ts) < 0.01


def test_parse_ts_iso_string():
    # ISO format as emitted by events.py / stability-agent
    from datetime import datetime, timezone
    iso = "2026-06-01T00:03:22Z"
    expected = datetime(2026, 6, 1, 0, 3, 22, tzinfo=timezone.utc).timestamp()
    result = _parse_ts(iso)
    assert result > 0
    assert isinstance(result, float)
    assert abs(result - expected) < 1


def test_parse_ts_none_returns_zero():
    assert _parse_ts(None) == 0.0


def test_parse_ts_garbage_returns_zero():
    assert _parse_ts("not-a-date") == 0.0


def test_parse_ts_zero_int():
    assert _parse_ts(0) == 0.0


# ---------------------------------------------------------------------------
# 2. Lifecycle: service_healthy event resolves linked incident
# ---------------------------------------------------------------------------

def test_service_healthy_resolves_active_incident(tmp_path):
    obs = _make_observer_simple(tmp_path)
    inc_id = "inc-111-vps-outline"
    obs.world_state["services"]["vps/outline"] = {
        "node": "vps", "service": "outline",
        "status": "unhealthy", "last_check": None,
        "incident_id": inc_id,
    }
    obs.world_state["incidents"][inc_id] = {
        "id": inc_id, "node": "vps", "service": "outline",
        "status": "active", "trigger_type": "service_unhealthy",
        "started_at": int(time.time()) - 600,
        "last_occurrence": int(time.time()) - 600,
        "occurrence_count": 1, "events": [],
    }

    obs.process_event({
        "type": "service_healthy",
        "node": "vps",
        "service": "outline",
        "severity": "info",
        "timestamp": int(time.time()),
        "payload": {},
    })

    assert obs.world_state["services"]["vps/outline"]["status"] == "healthy"
    assert obs.world_state["services"]["vps/outline"]["incident_id"] is None
    assert obs.world_state["incidents"][inc_id]["status"] == "resolved"


def test_service_healthy_does_not_resolve_other_incidents(tmp_path):
    """service_healthy for service A must not touch incident for service B."""
    obs = _make_observer_simple(tmp_path)
    inc_b = "inc-222-vps-supervisor"
    obs.world_state["services"]["vps/supervisor"] = {
        "node": "vps", "service": "supervisor",
        "status": "unhealthy", "last_check": None,
        "incident_id": inc_b,
    }
    obs.world_state["incidents"][inc_b] = {
        "id": inc_b, "status": "active",
        "last_occurrence": int(time.time()) - 300,
    }

    obs.process_event({
        "type": "service_healthy",
        "node": "vps",
        "service": "outline",   # different service
        "severity": "info",
        "timestamp": int(time.time()),
        "payload": {},
    })

    assert obs.world_state["incidents"][inc_b]["status"] == "active"


# ---------------------------------------------------------------------------
# 3. _prune_stale_world: healthy-service-linked incident → immediate resolve
# ---------------------------------------------------------------------------

def test_prune_resolves_healthy_linked_incident(tmp_path):
    """If a service is healthy but still points at an active incident, resolve it."""
    obs = _make_observer_simple(tmp_path)
    inc_id = "inc-333-vps-outline"
    obs.world_state["services"]["vps/outline"] = {
        "node": "vps", "service": "outline",
        "status": "healthy",          # <-- healthy but incident_id still set
        "last_check": None,
        "incident_id": inc_id,
    }
    obs.world_state["incidents"][inc_id] = {
        "id": inc_id, "status": "active",
        "started_at": int(time.time()) - 7200,
        "last_occurrence": int(time.time()) - 7200,
    }

    obs._prune_stale_world()

    assert obs.world_state["services"]["vps/outline"]["incident_id"] is None
    assert obs.world_state["incidents"][inc_id]["status"] == "resolved"


def test_prune_resolves_healthy_linked_incident_iso_timestamp(tmp_path):
    """Healthy-linked incident with ISO-string last_occurrence must still resolve."""
    obs = _make_observer_simple(tmp_path)
    inc_id = "inc-444-vps-outline"
    obs.world_state["services"]["vps/outline"] = {
        "node": "vps", "service": "outline",
        "status": "healthy", "last_check": None, "incident_id": inc_id,
    }
    obs.world_state["incidents"][inc_id] = {
        "id": inc_id, "status": "active",
        "last_occurrence": "2026-06-01T00:03:22Z",  # ISO string from events.py
    }

    obs._prune_stale_world()   # must not raise TypeError

    assert obs.world_state["incidents"][inc_id]["status"] == "resolved"


# ---------------------------------------------------------------------------
# 4. _prune_stale_world: orphaned incident (no service link) → resolve after 5 min
# ---------------------------------------------------------------------------

def test_prune_resolves_orphaned_incident_old_enough(tmp_path):
    """Orphaned active incident older than 5 min must be auto-resolved."""
    obs = _make_observer_simple(tmp_path)
    inc_id = "inc-555-vps-supervisor"
    # No service entry links to this incident
    obs.world_state["incidents"][inc_id] = {
        "id": inc_id, "status": "active", "node": "vps", "service": "supervisor",
        "last_occurrence": int(time.time()) - 400,   # 6.7 min ago
    }

    obs._prune_stale_world()

    assert obs.world_state["incidents"][inc_id]["status"] == "resolved"


def test_prune_does_not_resolve_orphaned_incident_too_recent(tmp_path):
    """Orphaned incident younger than 5 min must stay active (guard against race)."""
    obs = _make_observer_simple(tmp_path)
    inc_id = "inc-666-vps-supervisor"
    obs.world_state["incidents"][inc_id] = {
        "id": inc_id, "status": "active",
        "last_occurrence": int(time.time()) - 60,   # 1 min ago — within guard
    }

    obs._prune_stale_world()

    assert obs.world_state["incidents"][inc_id]["status"] == "active"


def test_prune_resolves_orphaned_incident_iso_timestamp(tmp_path):
    """Orphaned incident with ISO-string last_occurrence must resolve correctly."""
    obs = _make_observer_simple(tmp_path)
    inc_id = "inc-777-vps-outline"
    # ISO timestamp well in the past (2026-06-01)
    obs.world_state["incidents"][inc_id] = {
        "id": inc_id, "status": "active",
        "last_occurrence": "2026-06-01T00:03:22Z",
    }

    obs._prune_stale_world()   # must not raise TypeError

    assert obs.world_state["incidents"][inc_id]["status"] == "resolved"


def test_prune_does_not_touch_linked_incident(tmp_path):
    """An active incident still linked from a non-healthy service must stay active."""
    obs = _make_observer_simple(tmp_path)
    inc_id = "inc-888-vps-outline"
    obs.world_state["services"]["vps/outline"] = {
        "node": "vps", "service": "outline",
        "status": "unhealthy",   # <-- still unhealthy
        "last_check": None,
        "incident_id": inc_id,
    }
    obs.world_state["incidents"][inc_id] = {
        "id": inc_id, "status": "active",
        "last_occurrence": int(time.time()) - 3600,
    }

    obs._prune_stale_world()

    assert obs.world_state["incidents"][inc_id]["status"] == "active"


# ---------------------------------------------------------------------------
# 5. 7-day stale incident prune with ISO resolved_at
# ---------------------------------------------------------------------------

def test_prune_removes_old_resolved_incident_iso_resolved_at(tmp_path):
    """Resolved incidents with ISO-string resolved_at older than 7 days must be pruned."""
    obs = _make_observer_simple(tmp_path)
    inc_id = "inc-old-resolved"
    obs.world_state["incidents"][inc_id] = {
        "id": inc_id, "status": "resolved",
        "resolved_at": "2026-05-01T00:00:00Z",  # >7 days before 2026-06-03
    }

    obs._prune_stale_world()

    assert inc_id not in obs.world_state["incidents"]


def test_prune_keeps_recently_resolved_incident(tmp_path):
    """Resolved incidents within 7 days must be kept."""
    obs = _make_observer_simple(tmp_path)
    inc_id = "inc-recent-resolved"
    obs.world_state["incidents"][inc_id] = {
        "id": inc_id, "status": "resolved",
        "resolved_at": time.time() - 86400,  # 1 day ago
    }

    obs._prune_stale_world()

    assert inc_id in obs.world_state["incidents"]


def test_run_once_quarantines_bad_event_and_processes_next_for_same_node(tmp_path):
    """A malformed event file must not wedge a node forever."""
    obs = _make_observer_simple(tmp_path)

    import observer.observer as obs_mod

    topology = obs_mod.INVENTORY_TOPOLOGY
    topology.write_text(
        "nodes:\n"
        "  lustro:\n"
        "    roles: [edge]\n"
        "    connectivity: {}\n"
    )
    obs.inventory = obs._load_inventory()

    bad_dir = obs_mod.EVENTS_DIR / "lustro"
    bad_dir.mkdir(parents=True, exist_ok=True)
    bad_event = bad_dir / "evt-lustro-1-bad.json"
    bad_event.write_text("{not-json")

    good_event = bad_dir / "evt-lustro-2-good.json"
    good_event.write_text(json.dumps({
        "id": "evt-lustro-2-good",
        "timestamp": int(time.time()),
        "date": "2026-06-10T00:00:00Z",
        "type": "node_health",
        "severity": "info",
        "node": "lustro",
        "service": "",
        "message": "ok",
        "payload": {"disk_pct": 1, "mem_pct": 2, "cpu_pct": 3},
    }))

    obs.run_once()

    quarantined = obs_mod.FAILED_EVENTS_DIR / "lustro" / bad_event.name
    assert quarantined.exists()
    assert not bad_event.exists()
    assert obs.world_state["nodes"]["lustro"]["status"] == "online"
    assert obs.node_checkpoints["lustro"] == str(good_event)


# ---------------------------------------------------------------------------
# 7. Node liveness — 3-state (fresh/stale/dead) authoritative classification
# ---------------------------------------------------------------------------
# Default TTLs (liveness.py): fresh <=180s, dead >600s.  "vps" is in the test
# topology so it survives stale-node pruning.

def _node_events(obs_mod, node="vps"):
    """List event files the observer emitted into <node>/ during a prune."""
    d = obs_mod.EVENTS_DIR / node
    return sorted(d.glob("*.json")) if d.exists() else []


def test_prune_keeps_fresh_node_online(tmp_path):
    obs = _make_observer_simple(tmp_path)
    obs.world_state["nodes"]["vps"] = {
        "status": "online", "last_seen": time.time() - 60, "roles": [],
    }
    obs._prune_stale_world()
    assert obs.world_state["nodes"]["vps"]["status"] == "online"
    assert obs.world_state["nodes"]["vps"]["liveness"] == "fresh"


def test_prune_marks_node_stale_between_ttls(tmp_path):
    """180s < age <= 600s → stale/degraded, not yet dead."""
    obs = _make_observer_simple(tmp_path)
    obs.world_state["nodes"]["vps"] = {
        "status": "online", "last_seen": time.time() - 300, "roles": [],
    }
    obs._prune_stale_world()
    assert obs.world_state["nodes"]["vps"]["status"] == "stale"
    assert obs.world_state["nodes"]["vps"]["liveness"] == "stale"


def test_prune_marks_node_dead_after_down_ttl(tmp_path):
    """age > 600s → dead/offline."""
    obs = _make_observer_simple(tmp_path)
    obs.world_state["nodes"]["vps"] = {
        "status": "online", "last_seen": time.time() - 700, "roles": [],
    }
    obs._prune_stale_world()
    assert obs.world_state["nodes"]["vps"]["status"] == "offline"
    assert obs.world_state["nodes"]["vps"]["liveness"] == "dead"


def test_prune_unknown_when_no_last_seen(tmp_path):
    """last_seen=None → UNKNOWN: status untouched, no liveness flip, no event."""
    obs = _make_observer_simple(tmp_path)
    import observer.observer as obs_mod
    obs.world_state["nodes"]["vps"] = {
        "status": "online", "last_seen": None, "roles": [],
    }
    obs._prune_stale_world()
    assert obs.world_state["nodes"]["vps"]["status"] == "online"
    assert "liveness" not in obs.world_state["nodes"]["vps"]
    assert _node_events(obs_mod) == []


def test_prune_marks_node_dead_with_iso_last_seen(tmp_path):
    """3-state classification works when last_seen is an ISO-8601 string."""
    obs = _make_observer_simple(tmp_path)
    from datetime import datetime, timezone
    stale_ts = datetime.fromtimestamp(
        time.time() - 700, tz=timezone.utc
    ).isoformat()
    obs.world_state["nodes"]["vps"] = {
        "status": "online", "last_seen": stale_ts, "roles": [],
    }
    obs._prune_stale_world()
    assert obs.world_state["nodes"]["vps"]["status"] == "offline"


# --- Transition emission --------------------------------------------------

def test_transition_emits_node_offline_event(tmp_path):
    """fresh -> dead emits a node_offline event tagged source=observer."""
    obs = _make_observer_simple(tmp_path)
    import observer.observer as obs_mod
    obs.world_state["nodes"]["vps"] = {
        "status": "online", "liveness": "fresh",
        "last_seen": time.time() - 700, "roles": [],
    }
    obs._prune_stale_world()

    events = _node_events(obs_mod)
    assert len(events) == 1
    ev = json.loads(events[0].read_text())
    assert ev["type"] == "node_offline"
    assert ev["node"] == "vps"
    assert ev["source"] == "observer"
    assert ev["payload"]["affected_node"] == "vps"
    assert ev["payload"]["to"] == "dead"


def test_transition_emits_recovery_on_dead_to_fresh(tmp_path):
    """dead -> fresh emits node_online (recovery is as visible as the fall)."""
    obs = _make_observer_simple(tmp_path)
    import observer.observer as obs_mod
    obs.world_state["nodes"]["vps"] = {
        "status": "offline", "liveness": "dead",
        "last_seen": time.time() - 10, "roles": [],
    }
    obs._prune_stale_world()

    assert obs.world_state["nodes"]["vps"]["status"] == "online"
    events = _node_events(obs_mod)
    assert len(events) == 1
    ev = json.loads(events[0].read_text())
    assert ev["type"] == "node_online"
    assert ev["payload"]["from"] == "dead"
    assert ev["payload"]["to"] == "fresh"


def test_no_emission_on_baseline_classification(tmp_path):
    """First classification (prev liveness=None) is a baseline, not a transition."""
    obs = _make_observer_simple(tmp_path)
    import observer.observer as obs_mod
    obs.world_state["nodes"]["vps"] = {
        "status": "online", "last_seen": time.time() - 700, "roles": [],
    }
    obs._prune_stale_world()
    assert obs.world_state["nodes"]["vps"]["status"] == "offline"  # still classified
    assert _node_events(obs_mod) == []                              # but no event


def test_no_emission_when_liveness_unchanged(tmp_path):
    """Still-dead node must not re-emit on every cycle."""
    obs = _make_observer_simple(tmp_path)
    import observer.observer as obs_mod
    obs.world_state["nodes"]["vps"] = {
        "status": "offline", "liveness": "dead",
        "last_seen": time.time() - 5000, "roles": [],
    }
    obs._prune_stale_world()
    assert _node_events(obs_mod) == []


def test_remote_node_uses_wider_ttls(tmp_path):
    """A remote-role node at age 700s is still only stale (remote dead TTL=3600)."""
    obs = _make_observer_simple(tmp_path)
    import observer.observer as obs_mod
    # Add chelsty-infra to topology so it survives stale-node pruning.
    obs_mod.INVENTORY_TOPOLOGY.write_text(
        "nodes:\n"
        "  vps:\n    roles: [control-plane]\n    connectivity: {}\n"
        "  chelsty-infra:\n    roles: [remote, infra]\n    connectivity: {}\n"
    )
    obs.inventory = obs._load_inventory()
    obs.world_state["nodes"]["chelsty-infra"] = {
        "status": "online", "last_seen": time.time() - 1200,
        "roles": ["remote", "infra"],
    }
    obs._prune_stale_world()
    # 1200s > default dead (600) but < remote dead (3600) → stale, not dead.
    assert obs.world_state["nodes"]["chelsty-infra"]["liveness"] == "stale"


# --- Observer skips re-ingesting its own transition events ----------------

def test_run_once_skips_observer_emitted_events(tmp_path):
    """An observer-source event must NOT be re-ingested (would reset last_seen)."""
    obs = _make_observer_simple(tmp_path)
    import observer.observer as obs_mod

    vps_dir = obs_mod.EVENTS_DIR / "vps"
    vps_dir.mkdir(parents=True, exist_ok=True)
    # Seed world with a dead vps so we can detect an unwanted resurrection.
    obs.world_state["nodes"]["vps"] = {
        "status": "offline", "liveness": "dead",
        "last_seen": time.time() - 5000, "roles": [],
    }
    # An observer-emitted node_online event with a fresh timestamp — if it were
    # ingested, process_event would set last_seen=now and flip vps back online.
    ev_path = vps_dir / "evt-vps-9999999999-node_online-node.json"
    ev_path.write_text(json.dumps({
        "id": "evt-vps-9999999999-node_online-node",
        "timestamp": int(time.time()),
        "type": "node_online",
        "severity": "info",
        "node": "vps",
        "service": None,
        "source": "observer",
        "message": "synthetic",
        "payload": {"affected_node": "vps"},
    }))

    obs.run_once()

    # Not ingested: last_seen stays old → vps stays dead/offline.
    assert obs.world_state["nodes"]["vps"]["status"] == "offline"
    # Checkpoint still advanced over the skipped file.
    assert obs.node_checkpoints.get("vps") == str(ev_path)