mirror of
https://github.com/nicolargo/glances.git
synced 2026-06-03 03:15:09 -04:00
293 lines
11 KiB
Python
293 lines
11 KiB
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 ``processlist`` plugin (collection)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from glances.config_v5 import GlancesConfigV5
|
|
from glances.plugins.processlist.model_v5 import PluginModel
|
|
from glances.stats_store_v5 import StatsStoreV5
|
|
|
|
|
|
@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 _proc(**overrides):
|
|
"""Build a representative engine process dict."""
|
|
base = {
|
|
"pid": 1234,
|
|
"name": "python3",
|
|
"username": "alice",
|
|
"status": "S",
|
|
"nice": 0,
|
|
"num_threads": 4,
|
|
"cpu_percent": 12.5,
|
|
"memory_percent": 3.1,
|
|
"cmdline": ["python3", "myscript.py"],
|
|
"cpu_num": 2,
|
|
"memory_info": (1024, 2048, 0, 0, 0, 0, 0),
|
|
"cpu_times": (1.0, 0.5),
|
|
"io_counters": [0, 0, 0, 0, 0],
|
|
"time_since_update": 1.0,
|
|
"key": "pid",
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
# ---------------------------------------------------------- contract
|
|
|
|
|
|
def test_plugin_identity(store, config):
|
|
plugin = PluginModel(store, config)
|
|
assert plugin.plugin_name == "processlist"
|
|
assert plugin.IS_COLLECTION is True
|
|
assert plugin._primary_key == "pid"
|
|
|
|
|
|
def test_processlist_opts_out_of_alerts(store, config):
|
|
"""processlist colours cells but must NOT emit alerts events / actions.
|
|
|
|
v4 never paged on individual processes; v5 mirrors this — `_levels` is
|
|
computed for the renderer only. See base ``EMITS_ALERTS`` doc."""
|
|
assert PluginModel.EMITS_ALERTS is False
|
|
|
|
|
|
def test_pid_is_primary_key(store, config):
|
|
fields = PluginModel(store, config)._fields
|
|
assert fields["pid"].get("primary_key") is True
|
|
|
|
|
|
def test_cpu_and_mem_percent_are_watched_not_prominent(store, config):
|
|
fields = PluginModel(store, config)._fields
|
|
for name in ("cpu_percent", "memory_percent"):
|
|
assert fields[name].get("watched") is True, name
|
|
assert fields[name].get("prominent") is False, name
|
|
assert fields[name].get("default_thresholds") == {
|
|
"careful": 50.0,
|
|
"warning": 70.0,
|
|
"critical": 90.0,
|
|
}, name
|
|
|
|
|
|
def test_internal_fields_flagged(store, config):
|
|
fields = PluginModel(store, config)._fields
|
|
for name in ("memory_info", "cpu_times", "io_counters", "gids", "time_since_update", "key"):
|
|
assert fields[name].get("internal") is True, name
|
|
|
|
|
|
# ---------------------------------------------------------- update pipeline
|
|
|
|
|
|
async def test_update_surfaces_engine_list(store, config):
|
|
plugin = PluginModel(store, config)
|
|
procs = [_proc(pid=1, name="systemd"), _proc(pid=42, name="bash")]
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
data = store.get("processlist")["data"]
|
|
pids = sorted(p["pid"] for p in data)
|
|
assert pids == [1, 42]
|
|
|
|
|
|
async def test_update_filters_undeclared_fields(store, config):
|
|
plugin = PluginModel(store, config)
|
|
procs = [_proc(pid=1, name="x", fancy_extra="surprise")]
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
item = store.get("processlist")["data"][0]
|
|
assert "fancy_extra" not in item
|
|
|
|
|
|
async def test_update_returns_copy_so_engine_state_isolated(store, config):
|
|
plugin = PluginModel(store, config)
|
|
engine_list = [_proc(pid=1, name="x")]
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=engine_list):
|
|
await plugin.update()
|
|
stored = store.get("processlist")["data"][0]
|
|
stored["name"] = "MUTATED"
|
|
# Engine's source dict must be untouched.
|
|
assert engine_list[0]["name"] == "x"
|
|
|
|
|
|
async def test_update_handles_engine_failure(store, config):
|
|
plugin = PluginModel(store, config)
|
|
with patch(
|
|
"glances.plugins.processlist.model_v5.glances_processes.get_list",
|
|
side_effect=RuntimeError("engine boom"),
|
|
):
|
|
await plugin.update()
|
|
assert store.get("processlist")["data"] == []
|
|
|
|
|
|
async def test_update_handles_non_list_return(store, config):
|
|
plugin = PluginModel(store, config)
|
|
with patch(
|
|
"glances.plugins.processlist.model_v5.glances_processes.get_list",
|
|
return_value=None,
|
|
):
|
|
await plugin.update()
|
|
assert store.get("processlist")["data"] == []
|
|
|
|
|
|
async def test_update_skips_non_dict_entries(store, config):
|
|
plugin = PluginModel(store, config)
|
|
procs = [_proc(pid=1, name="x"), "garbage", None, _proc(pid=2, name="y")]
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
pids = sorted(p["pid"] for p in store.get("processlist")["data"])
|
|
assert pids == [1, 2]
|
|
|
|
|
|
# ---------------------------------------------------------- thresholds
|
|
|
|
|
|
async def test_cpu_percent_default_thresholds_trigger_level(store, config):
|
|
plugin = PluginModel(store, config)
|
|
procs = [_proc(pid=1, cpu_percent=75.0)] # > warning (70)
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
levels = store.get("processlist")["_levels"]
|
|
assert levels[1]["cpu_percent"]["level"] == "warning"
|
|
assert levels[1]["cpu_percent"]["prominent"] is False
|
|
|
|
|
|
async def test_memory_percent_default_thresholds_trigger_level(store, config):
|
|
plugin = PluginModel(store, config)
|
|
procs = [_proc(pid=1, memory_percent=95.0)] # > critical (90)
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
levels = store.get("processlist")["_levels"]
|
|
assert levels[1]["memory_percent"]["level"] == "critical"
|
|
|
|
|
|
async def test_user_can_override_cpu_percent_threshold(tmp_path, monkeypatch, store):
|
|
config = _config_with(
|
|
tmp_path,
|
|
monkeypatch,
|
|
"[processlist]\ncpu_percent_careful=20\ncpu_percent_warning=40\ncpu_percent_critical=60\n",
|
|
)
|
|
plugin = PluginModel(store, config)
|
|
procs = [_proc(pid=1, cpu_percent=45.0)] # > 40 → warning under override
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
levels = store.get("processlist")["_levels"]
|
|
assert levels[1]["cpu_percent"]["level"] == "warning"
|
|
|
|
|
|
# ---------------------------------------------------------- categorical thresholds
|
|
|
|
|
|
async def test_status_categorical_threshold_from_config(tmp_path, monkeypatch, store):
|
|
"""`status_critical=Z,D` should mark zombies as critical.
|
|
|
|
Values explicitly listed in ``status_ok`` get a level=ok entry. Values
|
|
in NO bucket (here: 'S') get **no** entry — the renderer keeps the
|
|
default colour. Mirrors v4 ``get_alert`` returning ``'DEFAULT'`` for
|
|
unmatched categorical values.
|
|
"""
|
|
config = _config_with(
|
|
tmp_path,
|
|
monkeypatch,
|
|
"[processlist]\nstatus_ok=R,W,P,I\nstatus_critical=Z,D\n",
|
|
)
|
|
plugin = PluginModel(store, config)
|
|
procs = [
|
|
_proc(pid=1, status="R"),
|
|
_proc(pid=2, status="Z"),
|
|
_proc(pid=3, status="S"), # not in any bucket → no level entry
|
|
]
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
levels = store.get("processlist")["_levels"]
|
|
assert levels[1]["status"]["level"] == "ok"
|
|
assert levels[2]["status"]["level"] == "critical"
|
|
# pid 3 has no status entry (or no entry at all if no other watched
|
|
# field fired) — the renderer falls back to DEFAULT colour.
|
|
assert "status" not in levels.get(3, {})
|
|
|
|
|
|
async def test_status_without_config_emits_no_level_entry(store, config):
|
|
"""No `status_*=` in conf → status is watched but no level computed."""
|
|
plugin = PluginModel(store, config)
|
|
procs = [_proc(pid=1, status="Z")]
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
levels = store.get("processlist")["_levels"]
|
|
# status not configured → no entry under levels[1].
|
|
assert "status" not in levels.get(1, {})
|
|
|
|
|
|
async def test_nice_categorical_threshold_from_config(tmp_path, monkeypatch, store):
|
|
"""`nice_warning=-1,1,...` flags non-zero nice values as warning.
|
|
|
|
nice=0 stays default-coloured (no level entry) — listing it in
|
|
``nice_ok=0`` would be the way to force an explicit OK.
|
|
"""
|
|
config = _config_with(
|
|
tmp_path,
|
|
monkeypatch,
|
|
"[processlist]\nnice_warning=-1,1,2,3,4,5\n",
|
|
)
|
|
plugin = PluginModel(store, config)
|
|
procs = [
|
|
_proc(pid=1, nice=0), # not in any list → no level entry
|
|
_proc(pid=2, nice=3), # in list → warning
|
|
_proc(pid=3, nice=-1), # in list → warning
|
|
]
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
levels = store.get("processlist")["_levels"]
|
|
assert "nice" not in levels.get(1, {})
|
|
assert levels[2]["nice"]["level"] == "warning"
|
|
assert levels[3]["nice"]["level"] == "warning"
|
|
|
|
|
|
async def test_nice_escalating_categorical_thresholds(tmp_path, monkeypatch, store):
|
|
"""Buckets careful / warning / critical escalate by membership."""
|
|
config = _config_with(
|
|
tmp_path,
|
|
monkeypatch,
|
|
"[processlist]\nnice_careful=1,2,3,4,5,6,7,8,9\nnice_warning=10,11,12,13,14\nnice_critical=15,16,17,18,19\n",
|
|
)
|
|
plugin = PluginModel(store, config)
|
|
procs = [
|
|
_proc(pid=1, nice=5),
|
|
_proc(pid=2, nice=12),
|
|
_proc(pid=3, nice=18),
|
|
]
|
|
with patch("glances.plugins.processlist.model_v5.glances_processes.get_list", return_value=procs):
|
|
await plugin.update()
|
|
levels = store.get("processlist")["_levels"]
|
|
assert levels[1]["nice"]["level"] == "careful"
|
|
assert levels[2]["nice"]["level"] == "warning"
|
|
assert levels[3]["nice"]["level"] == "critical"
|