LEDMatrix/test/test_resource_monitor.py

"""
Tests for src/plugin_system/resource_monitor.py

Focus areas:
- Execution-time metrics are captured regardless of psutil availability.
- CPU sampling is non-blocking (regression guard for the previous
  ``cpu_percent(interval=0.1)`` call that blocked 100 ms per monitored call).
- Resource limits are enforced.
"""

import time

import pytest
from unittest.mock import MagicMock, patch

from src.plugin_system.resource_monitor import (
    PluginResourceMonitor,
    ResourceLimits,
    ResourceLimitExceeded,
    PSUTIL_AVAILABLE,
)


def _cache():
    cache = MagicMock()
    cache.get.return_value = None
    return cache


class TestExecutionTimeMetrics:
    def test_monitor_call_returns_value_and_records_call(self):
        mon = PluginResourceMonitor(_cache(), enable_monitoring=False)
        result = mon.monitor_call("p", lambda: 42)
        assert result == 42
        metrics = mon.get_metrics("p")
        assert metrics.call_count == 1
        assert metrics.total_execution_time >= 0.0

    def test_avg_and_max_execution_time(self):
        mon = PluginResourceMonitor(_cache(), enable_monitoring=False)
        mon.monitor_call("p", lambda: time.sleep(0.01))
        mon.monitor_call("p", lambda: None)
        summary = mon.get_metrics_summary("p")
        assert summary["call_count"] == 2
        assert summary["max_execution_time"] >= summary["avg_execution_time"] >= 0.0

    def test_exception_propagates_but_is_still_timed(self):
        mon = PluginResourceMonitor(_cache(), enable_monitoring=False)

        def boom():
            raise ValueError("nope")

        with pytest.raises(ValueError):
            mon.monitor_call("p", boom)
        # Execution time is still recorded even when the call raised.
        assert mon.get_metrics("p").execution_time >= 0.0


class TestNonBlockingCpu:
    def test_cpu_sampling_is_fast_when_disabled(self):
        mon = PluginResourceMonitor(_cache(), enable_monitoring=False)
        start = time.time()
        for _ in range(50):
            mon._get_process_cpu_percent()
        # The old implementation blocked ~0.1s/call (~5s for 50). Non-blocking
        # must complete near-instantly.
        assert time.time() - start < 0.5
        assert mon._get_process_cpu_percent() == 0.0

    @pytest.mark.skipif(not PSUTIL_AVAILABLE, reason="psutil not installed")
    def test_cpu_sampling_is_fast_with_psutil(self):
        mon = PluginResourceMonitor(_cache(), enable_monitoring=True)
        assert mon._process is not None
        start = time.time()
        for _ in range(30):
            mon._get_process_cpu_percent()
        # 30 blocking 0.1s samples would be ~3s; non-blocking must be well under.
        assert time.time() - start < 0.5

    def test_monitor_call_does_not_block_on_cpu_sampling(self):
        mon = PluginResourceMonitor(_cache())  # enable depends on psutil
        start = time.time()
        for _ in range(25):
            mon.monitor_call("p", lambda: None)
        # 25 * 0.1s = 2.5s under the old blocking bug; must be far faster now.
        assert time.time() - start < 1.0


class TestResourceLimits:
    def test_execution_time_limit_raises(self):
        mon = PluginResourceMonitor(_cache(), enable_monitoring=False)
        mon.set_limits("p", ResourceLimits(max_execution_time=0.001))
        with pytest.raises(ResourceLimitExceeded):
            mon.monitor_call("p", lambda: time.sleep(0.02))

    def test_memory_limit_judges_each_call_on_its_own_growth(self):
        """One expensive call must not fail every call after it.

        The check used to compare the stored high-water mark, which never
        decreases, so after one call grew memory past the limit every later
        call raised too and the plugin never updated again.
        """
        mon = PluginResourceMonitor(_cache(), enable_monitoring=False)
        mon.enable_monitoring = True  # measure without needing psutil
        readings = iter([100.0, 200.0,   # first call grows RSS by 100 MB
                         200.0, 201.0])  # second call grows it by 1 MB
        mon._get_process_memory_mb = lambda: next(readings)
        mon._get_process_cpu_percent = lambda: 0.0
        mon.set_limits("p", ResourceLimits(max_memory_mb=50))

        with pytest.raises(ResourceLimitExceeded):
            mon.monitor_call("p", lambda: None)
        assert mon.monitor_call("p", lambda: "ok") == "ok"
        # The high-water mark is still reported.
        assert mon.get_metrics("p").memory_mb == 100.0

    def test_reset_metrics_clears_counts(self):
        cache = _cache()
        mon = PluginResourceMonitor(cache, enable_monitoring=False)
        mon.monitor_call("p", lambda: None)
        assert mon.get_metrics("p").call_count == 1
        mon.reset_metrics("p")
        assert mon.get_metrics("p").call_count == 0


class TestForceReload:
    def test_force_reload_refreshes_stale_snapshot(self):
        """A read-only consumer must see the writer process's latest persisted
        metrics rather than a pinned first snapshot."""
        cache = MagicMock()
        persisted = {"value": None}  # only the metrics key returns data

        def cache_get(key, max_age=None, memory_ttl=None):
            return persisted["value"] if key.startswith("plugin_metrics:") else None

        cache.get.side_effect = cache_get
        mon = PluginResourceMonitor(cache, enable_monitoring=False)

        # First read snapshots empty metrics.
        assert mon.get_metrics_summary("p")["call_count"] == 0

        # The display service later persists real metrics.
        persisted["value"] = {"call_count": 7, "total_execution_time": 1.4}

        # Plain read stays stale...
        assert mon.get_metrics_summary("p")["call_count"] == 0
        # ...force_reload picks up the persisted values and bypasses memory.
        fresh = mon.get_metrics_summary("p", force_reload=True)
        assert fresh["call_count"] == 7
        assert any(c.kwargs.get("memory_ttl") == 0 for c in cache.get.call_args_list)


class TestMetricsPersistenceChurn:
    """Metrics are telemetry; writing them on every call wore the SD card.

    Each write is a ~350-byte file, which on ext4 costs a 4KB block plus a
    journal entry. At roughly nine calls a minute per plugin across fourteen
    plugins it dominated the device's write volume.
    """

    def test_the_first_snapshot_is_written_even_seconds_after_boot(self):
        """The throttle must key off "have we written?", not process uptime.

        time.monotonic() is time since boot on Linux, and systemd starts this
        service at boot. With 0.0 as the missing-timestamp default,
        `now - 0.0 < 30` was true for the first half-minute of every run, so
        the very first metrics write -- the one that matters most after a
        restart -- was silently skipped.
        """
        import src.plugin_system.resource_monitor as rm
        cache = _cache()
        mon = PluginResourceMonitor(cache, enable_monitoring=False)
        # 12 seconds after boot: inside the interval, but nothing written yet.
        with patch.object(rm.time, "monotonic", return_value=12.0):
            mon.monitor_call("p", lambda: None)
        writes = [c for c in cache.set.call_args_list
                  if "plugin_metrics:" in str(c)]
        assert writes, \
            "the first snapshot was dropped because the process was young"

    def test_repeated_calls_persist_once_per_interval(self):
        cache = _cache()
        mon = PluginResourceMonitor(cache, enable_monitoring=False)
        for _ in range(50):
            mon.monitor_call("p", lambda: None)
        writes = [c for c in cache.set.call_args_list
                  if c.args and str(c.args[0]).startswith("plugin_metrics:")]
        assert len(writes) == 1, (
            f"50 calls produced {len(writes)} metric writes; expected 1")

    def test_the_interval_elapsing_allows_the_next_write(self, monkeypatch):
        import src.plugin_system.resource_monitor as rm
        cache = _cache()
        mon = PluginResourceMonitor(cache, enable_monitoring=False)
        mon.monitor_call("p", lambda: None)
        # pretend the interval has passed
        mon._metrics_persisted_at["p"] -= rm._METRICS_PERSIST_INTERVAL + 1
        mon.monitor_call("p", lambda: None)
        writes = [c for c in cache.set.call_args_list
                  if c.args and str(c.args[0]).startswith("plugin_metrics:")]
        assert len(writes) == 2

    def test_in_memory_metrics_stay_exact_while_writes_are_skipped(self):
        mon = PluginResourceMonitor(_cache(), enable_monitoring=False)
        for _ in range(20):
            mon.monitor_call("p", lambda: None)
        assert mon.get_metrics("p").call_count == 20

    def test_reset_lets_the_next_call_persist_immediately(self):
        cache = _cache()
        mon = PluginResourceMonitor(cache, enable_monitoring=False)
        mon.monitor_call("p", lambda: None)
        mon.reset_metrics("p")
        mon.monitor_call("p", lambda: None)
        writes = [c for c in cache.set.call_args_list
                  if c.args and str(c.args[0]).startswith("plugin_metrics:")]
        assert len(writes) == 2, "reset should clear the throttle timestamp"

    def test_a_failed_write_does_not_buy_the_next_interval_of_silence(self):
        """A set() that raises must not count as having persisted.

        Marking the timestamp before the write would leave no snapshot in the
        cache and still suppress the next 30 seconds of attempts.
        """
        cache = _cache()
        cache.set.side_effect = [OSError("disk full"), None]
        mon = PluginResourceMonitor(cache, enable_monitoring=False)
        with pytest.raises(OSError):
            mon.monitor_call("p", lambda: None)
        # the very next call must try again rather than skip the interval
        mon.monitor_call("p", lambda: None)
        writes = [c for c in cache.set.call_args_list
                  if c.args and str(c.args[0]).startswith("plugin_metrics:")]
        assert len(writes) == 2, "a failed write should be retried, not skipped"