mirror of
https://github.com/nicolargo/glances.git
synced 2026-07-31 07:16:17 -04:00
225 lines
7.5 KiB
Python
Executable File
225 lines
7.5 KiB
Python
Executable File
#!/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
|
||
#
|
||
|
||
"""Tests for the Glances v5 containers plugin model."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from glances.plugins.containers.model_v5 import PluginModel
|
||
|
||
|
||
def _mk(store_with, config_with, section=None):
|
||
return PluginModel(store_with(), config_with({"containers": section or {}}))
|
||
|
||
|
||
def test_identity(store_with, config_with):
|
||
p = _mk(store_with, config_with)
|
||
assert p.plugin_name == "containers"
|
||
assert p.IS_COLLECTION is True
|
||
assert p.EMITS_ALERTS is True
|
||
assert p._primary_key == "name"
|
||
|
||
|
||
def test_fields_present(store_with, config_with):
|
||
p = _mk(store_with, config_with)
|
||
fd = p.fields_description
|
||
for key in (
|
||
"name",
|
||
"id",
|
||
"image",
|
||
"status",
|
||
"created",
|
||
"command",
|
||
"cpu_percent",
|
||
"cpu_limit",
|
||
"memory_usage",
|
||
"memory_usage_no_cache",
|
||
"memory_limit",
|
||
"memory_percent",
|
||
"io_rx",
|
||
"io_wx",
|
||
"network_rx",
|
||
"network_tx",
|
||
"ports",
|
||
"uptime",
|
||
"engine",
|
||
"pod_name",
|
||
"pod_id",
|
||
):
|
||
assert key in fd, key
|
||
assert fd["name"].get("primary_key") is True
|
||
# Threshold aliases (design §5.2).
|
||
assert fd["cpu_percent"]["threshold_field"] == "cpu"
|
||
assert fd["memory_percent"]["threshold_field"] == "mem"
|
||
|
||
|
||
def test_cpu_level_uses_cpu_prefix_thresholds(store_with, config_with):
|
||
p = _mk(store_with, config_with, {"cpu_warning": "70", "cpu_critical": "90"})
|
||
p._stats = [{"name": "web", "cpu_percent": 95.0, "memory_percent": None}]
|
||
p._derived_parameters()
|
||
assert p._levels["web"]["cpu_percent"]["level"] == "critical"
|
||
|
||
|
||
def test_mem_level_uses_mem_prefix_thresholds(store_with, config_with):
|
||
p = _mk(store_with, config_with, {"mem_careful": "20", "mem_warning": "50"})
|
||
p._stats = [{"name": "web", "cpu_percent": None, "memory_percent": 60.0}]
|
||
p._derived_parameters()
|
||
assert p._levels["web"]["memory_percent"]["level"] == "warning"
|
||
|
||
|
||
def test_per_container_cpu_override(store_with, config_with):
|
||
p = _mk(store_with, config_with, {"cpu_warning": "70", "web_cpu_warning": "10"})
|
||
p._stats = [{"name": "web", "cpu_percent": 15.0}]
|
||
p._derived_parameters()
|
||
assert p._levels["web"]["cpu_percent"]["level"] == "warning"
|
||
|
||
|
||
class _FakeWatcher:
|
||
def __init__(self, containers, raises=False):
|
||
self._containers = containers
|
||
self._raises = raises
|
||
self.stopped = False
|
||
|
||
def update(self, all_tag):
|
||
if self._raises:
|
||
raise RuntimeError("engine down")
|
||
return {}, [dict(c) for c in self._containers]
|
||
|
||
def stop(self):
|
||
self.stopped = True
|
||
|
||
|
||
def _model_with_watchers(store_with, config_with, watchers, section=None):
|
||
p = PluginModel(store_with(), config_with({"containers": section or {}}))
|
||
p.watchers = watchers
|
||
return p
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grab_merges_engines_and_injects_engine_field(store_with, config_with):
|
||
# memory_usage=250 simulates the engine's v4 export value (usage−cache);
|
||
# the nested memory dict drives the no-cache + percent surfaces.
|
||
d = {
|
||
"name": "web",
|
||
"key": "name",
|
||
"memory_usage": 250,
|
||
"memory": {"usage": 300, "inactive_file": 100, "limit": 1000},
|
||
}
|
||
p = _model_with_watchers(store_with, config_with, {"docker": _FakeWatcher([d])})
|
||
out = await p._grab_stats()
|
||
assert len(out) == 1
|
||
assert out[0]["engine"] == "docker"
|
||
# Three memory surfaces:
|
||
assert out[0]["memory_usage"] == 250 # export (v4 value, untouched)
|
||
assert out[0]["memory_usage_no_cache"] == 200 # display (usage − inactive_file)
|
||
assert out[0]["memory_percent"] == 20.0 # alert (200 / 1000 * 100)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grab_partial_failure_keeps_other_engine(store_with, config_with):
|
||
ok = {"name": "web", "memory": {}}
|
||
p = _model_with_watchers(
|
||
store_with,
|
||
config_with,
|
||
{"bad": _FakeWatcher([], raises=True), "docker": _FakeWatcher([ok])},
|
||
)
|
||
out = await p._grab_stats()
|
||
assert [c["name"] for c in out] == ["web"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grab_empty_when_no_watcher(store_with, config_with):
|
||
p = _model_with_watchers(store_with, config_with, {})
|
||
assert await p._grab_stats() == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grab_memory_percent_none_when_limit_zero(store_with, config_with):
|
||
# limit=0 → no meaningful percent, no divide-by-zero: memory_percent is None.
|
||
d = {"name": "web", "memory": {"usage": 300, "inactive_file": 100, "limit": 0}}
|
||
p = _model_with_watchers(store_with, config_with, {"docker": _FakeWatcher([d])})
|
||
out = await p._grab_stats()
|
||
assert out[0]["memory_usage_no_cache"] == 200
|
||
assert out[0]["memory_percent"] is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grab_memory_percent_none_when_limit_missing(store_with, config_with):
|
||
# No limit key → memory_percent is None (guarded), no exception.
|
||
d = {"name": "web", "memory": {"usage": 300, "inactive_file": 100}}
|
||
p = _model_with_watchers(store_with, config_with, {"docker": _FakeWatcher([d])})
|
||
out = await p._grab_stats()
|
||
assert out[0]["memory_usage_no_cache"] == 200
|
||
assert out[0]["memory_percent"] is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grab_all_tag_forwarded_to_watcher(store_with, config_with):
|
||
seen = {}
|
||
|
||
class _Capturing(_FakeWatcher):
|
||
def update(self, all_tag):
|
||
seen["all_tag"] = all_tag
|
||
return {}, []
|
||
|
||
p = _model_with_watchers(store_with, config_with, {"docker": _Capturing([])}, section={"all": "True"})
|
||
await p._grab_stats()
|
||
assert seen["all_tag"] is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_grab_sort_follows_glances_processes_sort_key(store_with, config_with):
|
||
from glances.processes import glances_processes
|
||
|
||
cs = [
|
||
{"name": "a", "cpu_percent": 1.0, "memory": {}},
|
||
{"name": "b", "cpu_percent": 9.0, "memory": {}},
|
||
]
|
||
p = _model_with_watchers(store_with, config_with, {"docker": _FakeWatcher(cs)})
|
||
saved = glances_processes.sort_key
|
||
try:
|
||
glances_processes.set_sort_key("cpu_percent", auto=False)
|
||
out = await p._grab_stats()
|
||
# cpu_percent sort is reverse (highest first).
|
||
assert [c["name"] for c in out] == ["b", "a"]
|
||
|
||
glances_processes.set_sort_key("name", auto=False)
|
||
out = await p._grab_stats()
|
||
# name sort is ascending.
|
||
assert [c["name"] for c in out] == ["a", "b"]
|
||
finally:
|
||
glances_processes.set_sort_key(saved, auto=False)
|
||
|
||
|
||
def test_stop_calls_each_watcher(store_with, config_with):
|
||
w1, w2 = _FakeWatcher([]), _FakeWatcher([])
|
||
p = _model_with_watchers(store_with, config_with, {"docker": w1, "podman": w2})
|
||
p.stop()
|
||
assert w1.stopped and w2.stopped
|
||
|
||
|
||
def test_stop_one_raising_watcher_does_not_block_others(store_with, config_with):
|
||
class _Boom(_FakeWatcher):
|
||
def stop(self):
|
||
raise RuntimeError("boom")
|
||
|
||
good = _FakeWatcher([])
|
||
p = _model_with_watchers(store_with, config_with, {"bad": _Boom([]), "docker": good})
|
||
p.stop() # must not raise
|
||
assert good.stopped
|
||
|
||
|
||
def test_metadata_carries_disable_stats_and_max_name_size(store_with, config_with):
|
||
p = PluginModel(store_with(), config_with({"containers": {"disable_stats": "command", "max_name_size": "12"}}))
|
||
p._add_metadata()
|
||
assert "command" in p._metadata["disable_stats"]
|
||
assert p._metadata["max_name_size"] == 12
|