homelab-codex-ws/services/node-agent/tests/test_cleanup_control_plane_fs.py

"""Tests for NodeAgent._cleanup_control_plane_fs() event-retention sweep.

Regression coverage for the event-flood root cause: this method's event
cleanup previously compared str(file_path) against the observer's checkpoint
value with <=, which only worked for the old lexical-path checkpoint format.
Once observer.py migrated node_checkpoints to int epoch timestamps, every
comparison raised a TypeError (caught silently by the broad except), so this
retention sweep was a no-op — the direct cause of the 358k-file backlog that
paralyzed the supervisor's reconcile().
"""
from __future__ import annotations

import json
import os
import shutil
import time

import pytest

import node_agent


@pytest.fixture(autouse=True)
def _isolated_runtime_state():
    """EVENTS_DIR/STATE_DIR are a single temp dir shared for the whole test
    session (conftest.py sets RUNTIME_PATH once at import time). Clear them
    before and after each test here so tests that write real event/checkpoint
    files don't leak state into each other or into other test modules."""
    def _clear():
        if node_agent.EVENTS_DIR.exists():
            shutil.rmtree(node_agent.EVENTS_DIR)
        node_agent.EVENTS_DIR.mkdir(parents=True, exist_ok=True)
        checkpoint = node_agent.STATE_DIR / "observer_checkpoint.json"
        checkpoint.unlink(missing_ok=True)

    _clear()
    yield
    _clear()


def _write_event(events_dir, node, ts, etype, svc="svc"):
    """Write an event file whose filename AND on-disk mtime both reflect `ts`.

    _cleanup_control_plane_fs's age check reads f.stat().st_mtime (real
    filesystem mtime), not the filename-embedded timestamp — matching
    production, where a file's mtime is set once at the moment node-agent
    writes it. Tests must set mtime explicitly to simulate an "old" event.
    """
    node_dir = events_dir / node
    node_dir.mkdir(parents=True, exist_ok=True)
    p = node_dir / f"evt-{node}-{ts}-{etype}-{svc}.json"
    p.write_text("{}")
    os.utime(p, (ts, ts))
    return p


def _write_checkpoint(state_dir, node_checkpoints):
    (state_dir / "observer_checkpoint.json").write_text(
        json.dumps({"node_checkpoints": node_checkpoints})
    )


def test_int_checkpoint_deletes_old_processed_noise(agent):
    """The bug: this must NOT raise, and must actually delete eligible files."""
    events_dir = node_agent.EVENTS_DIR
    state_dir = node_agent.STATE_DIR
    now = int(time.time())
    checkpoint = now - 100

    old_healthy = _write_event(events_dir, "piha", now - 4 * 86_400, "service_healthy")
    old_node_health = _write_event(events_dir, "piha", now - 4 * 86_400, "node_health")
    _write_checkpoint(state_dir, {"piha": checkpoint})

    agent._cleanup_control_plane_fs()

    assert not old_healthy.exists()
    assert not old_node_health.exists()


def test_keeps_real_signals_regardless_of_age(agent):
    """healthcheck_failed / incidents must survive this sweep indefinitely."""
    events_dir = node_agent.EVENTS_DIR
    state_dir = node_agent.STATE_DIR
    now = int(time.time())
    checkpoint = now - 100

    healthcheck = _write_event(events_dir, "piha", now - 30 * 86_400, "healthcheck_failed")
    ha_event = _write_event(events_dir, "piha", now - 30 * 86_400, "ha_entity_unavailable_long")
    _write_checkpoint(state_dir, {"piha": checkpoint})

    agent._cleanup_control_plane_fs()

    assert healthcheck.exists()
    assert ha_event.exists()


def test_keeps_unprocessed_noise_even_if_old(agent):
    """An event newer than the checkpoint is not guaranteed processed — keep it."""
    events_dir = node_agent.EVENTS_DIR
    state_dir = node_agent.STATE_DIR
    now = int(time.time())
    checkpoint = now - 100

    unprocessed = _write_event(events_dir, "piha", now - 4 * 86_400, "service_healthy", svc="future")
    # Force it to sort after the checkpoint despite an old mtime by using a
    # timestamp greater than checkpoint in the filename itself.
    unprocessed.rename(events_dir / "piha" / f"evt-piha-{checkpoint + 50}-service_healthy-future.json")
    _write_checkpoint(state_dir, {"piha": checkpoint})

    agent._cleanup_control_plane_fs()

    assert (events_dir / "piha" / f"evt-piha-{checkpoint + 50}-service_healthy-future.json").exists()


def test_keeps_recent_processed_noise_within_three_day_buffer(agent):
    """Past checkpoint but within the 3-day safety buffer: not yet eligible."""
    events_dir = node_agent.EVENTS_DIR
    state_dir = node_agent.STATE_DIR
    now = int(time.time())
    checkpoint = now - 100

    recent = _write_event(events_dir, "piha", now - 3600, "service_healthy")
    _write_checkpoint(state_dir, {"piha": checkpoint})

    agent._cleanup_control_plane_fs()

    assert recent.exists()


def test_legacy_path_string_checkpoint_still_migrates_correctly(agent):
    """Old observer builds stored a lexical path string per node; the embedded
    timestamp must still be extracted so retention works during a rolling
    upgrade instead of silently no-op'ing again."""
    events_dir = node_agent.EVENTS_DIR
    state_dir = node_agent.STATE_DIR
    now = int(time.time())
    checkpoint_ts = now - 100
    legacy_path_value = f"/opt/homelab/events/piha/evt-piha-{checkpoint_ts}-node_health-node.json"

    old_healthy = _write_event(events_dir, "piha", now - 4 * 86_400, "service_healthy")
    _write_checkpoint(state_dir, {"piha": legacy_path_value})

    agent._cleanup_control_plane_fs()

    assert not old_healthy.exists()


def test_no_checkpoint_file_is_noop_not_crash(agent):
    events_dir = node_agent.EVENTS_DIR
    _write_event(events_dir, "piha", int(time.time()) - 999999, "service_healthy")
    # No checkpoint file written at all.
    agent._cleanup_control_plane_fs()  # must not raise