ledgrab/server/tests/test_metrics_provider.py

"""Tests for the metrics provider abstraction."""

from __future__ import annotations

import sys
from unittest.mock import MagicMock, mock_open, patch

import pytest

from ledgrab.utils.metrics import (
    AndroidMetricsProvider,
    MemorySnapshot,
    NullMetricsProvider,
    ProcessSnapshot,
    PsutilMetricsProvider,
    ThermalSnapshot,
    get_metrics_provider,
    reset_metrics_provider,
)
from ledgrab.utils.metrics import android_provider as android_mod


@pytest.fixture(autouse=True)
def _reset_provider_cache(monkeypatch):
    # Disable the Windows CPU-temp background reader so tests don't spawn
    # PowerShell when run on a Windows host.
    monkeypatch.setenv("LEDGRAB_DISABLE_WIN_TEMP", "1")
    reset_metrics_provider()
    yield
    reset_metrics_provider()


def test_null_provider_returns_zero_metrics() -> None:
    p = NullMetricsProvider()
    assert p.available is False
    assert p.cpu_percent() == 0.0
    assert p.cpu_count() == 1
    assert p.virtual_memory() == MemorySnapshot(0, 0, 0.0)
    assert p.process_snapshot() == ProcessSnapshot(0.0, 0)


def test_psutil_provider_normalizes_process_cpu() -> None:
    psutil_mock = MagicMock()
    psutil_mock.cpu_percent.return_value = 42.5
    psutil_mock.cpu_count.return_value = 8
    mem = MagicMock(used=2_000_000, total=8_000_000, percent=25.0)
    psutil_mock.virtual_memory.return_value = mem
    proc_mock = MagicMock()
    # Per-core 0–N*100% — 800% means all 8 cores fully busy → 100% normalized.
    proc_mock.cpu_percent.return_value = 800.0
    proc_mock.memory_info.return_value = MagicMock(rss=1_500_000)
    psutil_mock.Process.return_value = proc_mock

    provider = PsutilMetricsProvider(psutil_mock)

    # Two priming calls expected at construction (host + process counters).
    assert psutil_mock.cpu_percent.call_count == 1
    assert proc_mock.cpu_percent.call_count == 1

    assert provider.available is True
    assert provider.cpu_percent() == 42.5
    assert provider.cpu_count() == 8

    snap = provider.virtual_memory()
    assert snap.used_bytes == 2_000_000
    assert snap.total_bytes == 8_000_000
    assert snap.percent == 25.0

    proc = provider.process_snapshot()
    assert proc.cpu_percent == 100.0  # 800% / 8 cores
    assert proc.rss_bytes == 1_500_000


def test_psutil_provider_handles_unknown_cpu_count() -> None:
    psutil_mock = MagicMock()
    psutil_mock.cpu_count.return_value = None  # psutil sometimes returns None
    psutil_mock.Process.return_value = MagicMock()

    provider = PsutilMetricsProvider(psutil_mock)

    assert provider.cpu_count() == 1  # falls back to 1 to avoid div-by-zero


def test_factory_returns_psutil_provider_when_available() -> None:
    pytest.importorskip("psutil")
    provider = get_metrics_provider()
    assert isinstance(provider, PsutilMetricsProvider)
    assert provider.available is True
    # Same instance on subsequent calls — provider is cached.
    assert get_metrics_provider() is provider


def test_factory_falls_back_to_null_when_psutil_missing(monkeypatch) -> None:
    # Hide psutil from the import system for this test.
    monkeypatch.setitem(sys.modules, "psutil", None)
    provider = get_metrics_provider()
    assert isinstance(provider, NullMetricsProvider)
    assert provider.available is False


# ── Android provider ────────────────────────────────────────────────


def test_android_meminfo_parses_kb_values(monkeypatch) -> None:
    sample = (
        "MemTotal:        2000000 kB\n"
        "MemFree:          500000 kB\n"
        "MemAvailable:    1500000 kB\n"
    )
    with patch("builtins.open", mock_open(read_data=sample)):
        snap = android_mod._read_meminfo()
    assert snap.total_bytes == 2000000 * 1024
    # used = total - available
    assert snap.used_bytes == 500000 * 1024
    assert snap.percent == 25.0


def test_android_meminfo_returns_zero_on_missing_file(monkeypatch) -> None:
    def _raise(*args, **kwargs):
        raise OSError("simulated")

    monkeypatch.setattr("builtins.open", _raise)
    snap = android_mod._read_meminfo()
    assert snap == MemorySnapshot(0, 0, 0.0)


def test_android_proc_self_stat_parses_with_paren_in_comm() -> None:
    # Process name "(weird) name" — embedded parens are the classic /proc trap.
    fields = ["S", "1", "1", "1", "0", "-1", "0", "0", "0", "0", "0", "150", "75"]
    raw = b"42 ((weird) name) " + " ".join(fields).encode() + b"\n"
    m = mock_open(read_data=raw)
    with patch("builtins.open", m):
        jiffies = android_mod._read_proc_self_stat_jiffies()
    assert jiffies == 150 + 75


def test_android_provider_cpu_percent_uses_delta() -> None:
    # First sample: total=1000, busy=200. Second sample: total=2000, busy=900.
    # Delta busy/total = 700/1000 = 70%.
    samples = iter(
        [
            android_mod._CpuSample(total=1000, busy=200),
            android_mod._CpuSample(total=2000, busy=900),
        ]
    )
    with patch.object(android_mod, "_read_proc_stat", lambda: next(samples)):
        with patch.object(android_mod, "_read_proc_self_stat_jiffies", lambda: 0):
            provider = AndroidMetricsProvider()
            assert provider.cpu_percent() == 70.0


def test_android_provider_process_cpu_normalized_across_cores() -> None:
    # Process consumed 400 jiffies while host clock advanced 1000 jiffies
    # across all cores → 40% of one CPU's worth of work.
    host_samples = iter(
        [
            android_mod._CpuSample(total=1000, busy=500),
            android_mod._CpuSample(total=2000, busy=1500),
        ]
    )
    proc_samples = iter([100, 500])
    with patch.object(android_mod, "_read_proc_stat", lambda: next(host_samples)):
        with patch.object(android_mod, "_read_proc_self_stat_jiffies", lambda: next(proc_samples)):
            with patch.object(android_mod, "_read_self_rss_bytes", lambda: 12345):
                with patch.object(android_mod.os, "cpu_count", lambda: 4):
                    provider = AndroidMetricsProvider()
                    snap = provider.process_snapshot()
    assert snap.cpu_percent == 40.0
    assert snap.rss_bytes == 12345


def test_android_provider_handles_missing_proc_files() -> None:
    with patch.object(android_mod, "_read_proc_stat", lambda: None):
        with patch.object(android_mod, "_read_proc_self_stat_jiffies", lambda: None):
            with patch.object(android_mod, "_read_self_rss_bytes", lambda: 0):
                provider = AndroidMetricsProvider()
                # No samples available → 0.0, not an exception.
                assert provider.cpu_percent() == 0.0
                snap = provider.process_snapshot()
                assert snap == ProcessSnapshot(0.0, 0)


def test_factory_prefers_android_when_running_on_android(monkeypatch) -> None:
    monkeypatch.setattr("ledgrab.utils.metrics.is_android", lambda: True)
    monkeypatch.setattr("ledgrab.utils.metrics._android_supported", lambda: True)
    provider = get_metrics_provider()
    assert isinstance(provider, AndroidMetricsProvider)


# ── Thermals ────────────────────────────────────────────────────────


def test_null_provider_thermals_are_all_none() -> None:
    snap = NullMetricsProvider().thermals()
    assert snap == ThermalSnapshot()


def test_psutil_provider_thermals_picks_hottest_sensor() -> None:
    psutil_mock = MagicMock()
    psutil_mock.Process.return_value = MagicMock()
    psutil_mock.cpu_count.return_value = 4
    bat = MagicMock(percent=78.0)
    psutil_mock.sensors_battery.return_value = bat
    psutil_mock.sensors_temperatures.return_value = {
        "coretemp": [
            MagicMock(current=55.0),
            MagicMock(current=72.5),
        ],
        "acpi": [MagicMock(current=40.0)],
    }
    provider = PsutilMetricsProvider(psutil_mock)
    snap = provider.thermals()
    assert snap.battery_percent == 78.0
    assert snap.cpu_temp_c == 72.5  # hottest across all sensors
    assert snap.battery_temp_c is None  # psutil doesn't expose battery temp


def test_psutil_provider_thermals_handles_missing_sensors() -> None:
    psutil_mock = MagicMock()
    psutil_mock.Process.return_value = MagicMock()
    psutil_mock.cpu_count.return_value = 1
    # Strip the optional sensor methods entirely (e.g. Windows psutil).
    del psutil_mock.sensors_battery
    del psutil_mock.sensors_temperatures
    provider = PsutilMetricsProvider(psutil_mock)
    assert provider.thermals() == ThermalSnapshot()


def test_android_battery_parses_tenths_of_celsius(monkeypatch) -> None:
    def _fake_int(path: str):
        return {
            "/sys/class/power_supply/battery/capacity": 78,
            "/sys/class/power_supply/battery/temp": 312,  # tenths of °C → 31.2°C
        }.get(path)

    monkeypatch.setattr(android_mod, "_read_int_file", _fake_int)
    pct, temp = android_mod._read_battery()
    assert pct == 78.0
    assert temp == 31.2


def test_android_cpu_temp_filters_non_cpu_zones_and_picks_hottest(monkeypatch) -> None:
    monkeypatch.setattr(
        "glob.glob",
        lambda _: [
            "/sys/class/thermal/thermal_zone0",
            "/sys/class/thermal/thermal_zone1",
            "/sys/class/thermal/thermal_zone2",
            "/sys/class/thermal/thermal_zone3",
        ],
    )

    def _fake_text(path: str):
        return {
            "/sys/class/thermal/thermal_zone0/type": "battery",
            "/sys/class/thermal/thermal_zone1/type": "cpu-thermal",
            "/sys/class/thermal/thermal_zone2/type": "soc-max",
            "/sys/class/thermal/thermal_zone3/type": "skin-therm",
        }.get(path)

    def _fake_int(path: str):
        return {
            # Battery & skin should be filtered out by zone type
            "/sys/class/thermal/thermal_zone0/temp": 99000,
            "/sys/class/thermal/thermal_zone1/temp": 52000,  # 52°C
            "/sys/class/thermal/thermal_zone2/temp": 67500,  # 67.5°C ← hottest
            "/sys/class/thermal/thermal_zone3/temp": 99000,
        }.get(path)

    monkeypatch.setattr(android_mod, "_read_text_file", _fake_text)
    monkeypatch.setattr(android_mod, "_read_int_file", _fake_int)

    assert android_mod._read_cpu_temp_c() == 67.5


def test_android_cpu_temp_rejects_nonsense_values(monkeypatch) -> None:
    monkeypatch.setattr("glob.glob", lambda _: ["/sys/class/thermal/thermal_zone0"])
    monkeypatch.setattr(android_mod, "_read_text_file", lambda _: "cpu-thermal")
    # Some buggy zones report INT_MAX
    monkeypatch.setattr(android_mod, "_read_int_file", lambda _: 2147483647)
    assert android_mod._read_cpu_temp_c() is None