glances/tests/test_plugin_cpu_v5.py

#!/usr/bin/env python
#
# Glances - An eye on your system
#
# SPDX-FileCopyrightText: 2026 Nicolas Hennion <nicolas@nicolargo.com>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""Glances v5 — unit tests for the `cpu` plugin (scalar)."""

from __future__ import annotations

from collections import namedtuple
from unittest.mock import AsyncMock, patch

import pytest

from glances.config_v5 import GlancesConfigV5
from glances.plugins.cpu.model_v5 import PluginModel
from glances.stats_store_v5 import StatsStoreV5

# psutil result stubs ------------------------------------------------------

CpuTimesPercent = namedtuple(
    "scputimes_percent",
    ["user", "system", "idle", "nice", "iowait", "irq", "softirq", "steal", "guest", "guest_nice"],
)

CpuStats = namedtuple("scpustats", ["ctx_switches", "interrupts", "soft_interrupts", "syscalls"])


def _agg(idle: float = 70.0, system: float = 15.0, user: float = 10.0, steal: float = 0.0) -> CpuTimesPercent:
    return CpuTimesPercent(
        user=user,
        system=system,
        idle=idle,
        nice=0.5,
        iowait=2.0,
        irq=0.1,
        softirq=0.1,
        steal=steal,
        guest=0.0,
        guest_nice=0.0,
    )


def _stats(ctx: int = 12_000) -> CpuStats:
    return CpuStats(ctx_switches=ctx, interrupts=2_000, soft_interrupts=1_000, syscalls=0)


# ---------------------------------------------------------- fixtures


@pytest.fixture
def store() -> StatsStoreV5:
    return StatsStoreV5()


@pytest.fixture
def config(tmp_path, monkeypatch) -> GlancesConfigV5:
    monkeypatch.setattr(GlancesConfigV5, "SYSTEM_CONFIG_PATH", tmp_path / "etc" / "glances.conf")
    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg"))
    return GlancesConfigV5()


def _config_with(tmp_path, monkeypatch, body: str) -> GlancesConfigV5:
    monkeypatch.setattr(GlancesConfigV5, "SYSTEM_CONFIG_PATH", tmp_path / "etc" / "glances.conf")
    xdg = tmp_path / "xdg"
    cfg_dir = xdg / "glances"
    cfg_dir.mkdir(parents=True)
    (cfg_dir / "glances.conf").write_text(body)
    monkeypatch.setenv("XDG_CONFIG_HOME", str(xdg))
    return GlancesConfigV5()


def _patch_sampler(agg=None, stats=None, cpu_count: int = 4):
    """Patch the module-level sampler used by cpu/model_v5.py.

    Returns an ExitStack — usable as a `with` block — that wires
    `get_aggregate`, `get_stats` and `cpu_count` to deterministic values
    for the duration of the test.
    """
    from contextlib import ExitStack

    # Reset the cached cpu_count on the module-level singleton so the
    # patched psutil.cpu_count is honoured for this test.
    from glances.cpu_sampler_v5 import sampler

    sampler._cpu_count = None

    stack = ExitStack()
    stack.enter_context(
        patch(
            "glances.plugins.cpu.model_v5.sampler.get_aggregate",
            new_callable=AsyncMock,
            return_value=agg if agg is not None else _agg(),
        )
    )
    stack.enter_context(
        patch(
            "glances.plugins.cpu.model_v5.sampler.get_stats",
            new_callable=AsyncMock,
            return_value=stats if stats is not None else _stats(),
        )
    )
    # cpu_count is a property on the singleton; patch the underlying
    # psutil call so the sampler computes the desired value lazily.
    stack.enter_context(patch("glances.cpu_sampler_v5.psutil.cpu_count", return_value=cpu_count))
    return stack


# ---------------------------------------------------------- contract


def test_plugin_identity(store, config):
    plugin = PluginModel(store, config)
    assert plugin.plugin_name == "cpu"
    assert plugin.IS_COLLECTION is False


def test_total_is_watched_prominent(store, config):
    schema = PluginModel(store, config)._fields["total"]
    assert schema["watched"] is True
    assert schema["prominent"] is True


def test_system_user_dpc_are_watched_non_prominent(store, config):
    fields = PluginModel(store, config)._fields
    for name in ("system", "user", "dpc"):
        assert fields[name]["watched"] is True, name
        assert fields[name]["prominent"] is False, name


def test_iowait_is_watched_non_prominent(store, config):
    """iowait is watched (level-coloured) but not prominent — the maintainer
    rejected the filled/reverse-video badge on iowait after smoke-testing the
    v5 Web UI (G9-8 smoke fix 5; see `test_iowait_is_not_prominent_by_default`
    for the pinned schema value)."""
    schema = PluginModel(store, config)._fields["iowait"]
    assert schema["watched"] is True
    assert schema["prominent"] is False


def test_iowait_is_not_prominent_by_default():
    """Pin `prominent: False` on `iowait`.

    A `True` here makes both surfaces paint a filled badge on an alerting
    `iowait` (the TUI's reverse-video cell and the Web UI's filled badge,
    `levelClass()` in levels.js honours the `entry.prominent` flag verbatim)
    -- the maintainer rejected that look after smoke-testing the v5 Web UI
    (G9-8 smoke fix 5, the same call already made for quicklook's bar fields
    in `test_bar_fields_are_not_prominent_by_default`, G9-8 smoke fix 2).
    Only the tier colour remains, like every other cpu field except `total`.
    Flipping this back to True re-introduces the badge; that intentional a
    change should touch this test, not slip in silently.
    """
    fd = PluginModel.fields_description
    assert fd["iowait"].get("prominent") is False, f"got {fd['iowait']!r}"


def test_steal_is_watched_non_prominent_with_strict_thresholds(store, config):
    schema = PluginModel(store, config)._fields["steal"]
    assert schema["watched"] is True
    assert schema["prominent"] is False
    # Strict — any non-trivial steal is worth surfacing (level only, not prominent).
    assert schema["default_thresholds"]["critical"] == 30.0


def test_ctx_switches_is_rate_watched_with_per_core_thresholds(store, config):
    """``ctx_switches`` is watched with **per-core** thresholds.

    V4-compatible: v4 calls ``get_alert(..., maximum=100 * cpucore)``,
    which compares ``rate / cpucore`` against the configured value.
    Defaults follow the ``50000 * cpucore`` reference of
    ``conf/glances.conf``, scaled by 0.5 / 0.75 / 1.
    """
    schema = PluginModel(store, config)._fields["ctx_switches"]
    assert schema["rate"] is True
    assert schema["watched"] is True
    assert schema["prominent"] is False
    assert schema["normalize_by"] == "cpucore"
    assert schema["default_thresholds"] == {
        "careful": 50000.0 * 0.5,
        "warning": 50000.0 * 0.75,
        "critical": 50000.0,
    }


def test_interrupts_and_soft_interrupts_and_syscalls_are_rate_only(store, config):
    fields = PluginModel(store, config)._fields
    for name in ("interrupts", "soft_interrupts", "syscalls"):
        assert fields[name].get("rate") is True, name
        assert fields[name].get("watched", False) is False, name


# ---------------------------------------------------------- update pipeline


async def test_update_writes_aggregate_fields(store, config):
    plugin = PluginModel(store, config)
    with _patch_sampler():
        await plugin.update()
    payload = store.get("cpu")
    assert payload["user"] == 10.0
    assert payload["system"] == 15.0
    assert payload["idle"] == 70.0
    # total = 100 - idle (single source of truth, no extra psutil call)
    assert payload["total"] == 30.0
    assert payload["cpucore"] == 4


async def test_update_drops_undeclared_psutil_fields(store, config):
    """psutil fields not declared in fields_description are stripped."""

    class WeirdNamedTuple(tuple):
        # Simulate a future psutil attribute we don't know about.
        @property
        def some_future_field(self) -> float:
            return 99.9

    plugin = PluginModel(store, config)
    with _patch_sampler():
        await plugin.update()
    payload = store.get("cpu")
    assert "some_future_field" not in payload


async def test_first_cycle_rate_fields_are_none(store, config):
    """Counter fields (ctx_switches, interrupts, ...) are present but None on the first cycle."""
    plugin = PluginModel(store, config)
    with _patch_sampler():
        await plugin.update()
    payload = store.get("cpu")
    # `rate: True` fields kept, value None — no previous sample to diff against.
    assert "ctx_switches" in payload
    assert payload["ctx_switches"] is None
    assert "interrupts" in payload
    assert payload["interrupts"] is None
    assert "soft_interrupts" in payload
    assert payload["soft_interrupts"] is None
    assert "syscalls" in payload
    assert payload["syscalls"] is None


async def test_second_cycle_computes_ctx_switches_rate(store, config, monkeypatch):
    """Second cycle: ctx_switches becomes a per-second rate."""
    plugin = PluginModel(store, config)

    fake_now = [100.0]
    import glances.plugins.plugin.base_v5 as base_module

    monkeypatch.setattr(base_module.time, "monotonic", lambda: fake_now[0])

    with _patch_sampler(stats=_stats(ctx=10_000)):
        await plugin.update()  # cycle 1 — ctx_switches is None

    fake_now[0] = 102.0  # +2 s elapsed
    with _patch_sampler(stats=_stats(ctx=10_500)):
        await plugin.update()  # cycle 2

    payload = store.get("cpu")
    # delta = 500 over 2 s = 250 events/s
    assert payload["ctx_switches"] == 250.0


# ---------------------------------------------------------- _levels


async def test_total_level_uses_default_thresholds(store, config):
    plugin = PluginModel(store, config)
    with _patch_sampler(agg=_agg(idle=20.0, user=50.0, system=30.0)):
        await plugin.update()
    payload = store.get("cpu")
    # total = 100 - 20 = 80 → between warning (70) and critical (90) → warning
    assert payload["total"] == 80.0
    assert payload["_levels"]["total"] == {"level": "warning", "prominent": True}


async def test_steal_strict_thresholds(store, config):
    """`steal=20` exceeds warning (15) but not critical (30)."""
    plugin = PluginModel(store, config)
    with _patch_sampler(agg=_agg(steal=20.0)):
        await plugin.update()
    assert store.get("cpu")["_levels"]["steal"] == {"level": "warning", "prominent": False}


async def test_ctx_switches_level_scales_with_core_count(store, config, monkeypatch):
    """Thresholds scale with the logical core count (v4 semantics).

    With 4 cores the defaults are 100_000 / 150_000 / 200_000 in absolute
    rate; 160_000/s lands between warning and critical.
    """
    plugin = PluginModel(store, config)

    fake_now = [100.0]
    import glances.plugins.plugin.base_v5 as base_module

    monkeypatch.setattr(base_module.time, "monotonic", lambda: fake_now[0])

    with _patch_sampler(stats=_stats(ctx=0), cpu_count=4):
        await plugin.update()

    fake_now[0] = 101.0  # +1 s — 160_000/s over 4 cores = 40_000/core → warning
    with _patch_sampler(stats=_stats(ctx=160_000), cpu_count=4):
        await plugin.update()

    payload = store.get("cpu")
    assert payload["ctx_switches"] == 160_000.0
    assert payload["_levels"]["ctx_switches"] == {"level": "warning", "prominent": False}


async def test_ctx_switches_same_rate_is_ok_on_a_bigger_machine(store, config, monkeypatch):
    """Same absolute rate, twice the cores → half the per-core pressure.

    160_000/s over 8 cores = 20_000/core, below careful (25_000).
    """
    plugin = PluginModel(store, config)

    fake_now = [100.0]
    import glances.plugins.plugin.base_v5 as base_module

    monkeypatch.setattr(base_module.time, "monotonic", lambda: fake_now[0])

    with _patch_sampler(stats=_stats(ctx=0), cpu_count=8):
        await plugin.update()

    fake_now[0] = 101.0
    with _patch_sampler(stats=_stats(ctx=160_000), cpu_count=8):
        await plugin.update()

    assert store.get("cpu")["_levels"]["ctx_switches"]["level"] == "ok"


async def test_ctx_switches_level_ok_at_typical_desktop_rate(store, config, monkeypatch):
    """Typical desktop ctx rate (~4k/s) must stay at level ``ok`` — regression
    guard against defaults low enough to alert on an idle desktop.
    """
    plugin = PluginModel(store, config)

    fake_now = [100.0]
    import glances.plugins.plugin.base_v5 as base_module

    monkeypatch.setattr(base_module.time, "monotonic", lambda: fake_now[0])

    with _patch_sampler(stats=_stats(ctx=0), cpu_count=8):
        await plugin.update()

    fake_now[0] = 101.0  # +1 s — ctx rate = 4_000/s = 500/core, well below careful=25_000
    with _patch_sampler(stats=_stats(ctx=4_000), cpu_count=8):
        await plugin.update()

    payload = store.get("cpu")
    assert payload["_levels"]["ctx_switches"]["level"] == "ok"


async def test_user_config_overrides_total_threshold(tmp_path, monkeypatch, store):
    config = _config_with(tmp_path, monkeypatch, "[cpu]\ntotal_warning=85\n")
    plugin = PluginModel(store, config)
    with _patch_sampler(agg=_agg(idle=20.0)):
        await plugin.update()
    # total = 80, override warning to 85 → still careful (default 50)
    assert store.get("cpu")["_levels"]["total"]["level"] == "careful"


# ---------------------------------------------------------- export


async def test_get_export_strips_internals_and_levels(store, config):
    plugin = PluginModel(store, config)
    with _patch_sampler():
        await plugin.update()
    exported = plugin.get_export()
    assert "_levels" not in exported
    assert "time_since_update" not in exported
    assert exported["total"] == 30.0