mirror of
https://github.com/nicolargo/glances.git
synced 2026-09-14 22:31:23 -04:00
Nguyen Thanh Dat
8b89bf68a4
`update_views` skipped the whole interface when either direction measured
0 bytes/s:
if not i.get('bytes_recv_rate_per_sec') or not i.get('bytes_sent_rate_per_sec'):
continue
The guard is `or`, so one idle direction suppressed the alert for *both*. A
send-only interface saturating its uplink went undecorated because its rx rate
happened to be exactly 0, and a receive-only one — a monitor/SPAN port — the
same way. 0 bytes/s is a real measurement, not a missing one.
The rates now default to 0 instead of skipping, which is what the diskio plugin
already does: it dropped this same guard in #3684 and reads the rate through
`or 0` for the same reason.
The WebUI needs no change — `plugin-network.vue` renders `getDecoration(...)`
straight from the server-supplied view, so it inherits this.
Tests: four cases in `TestNetworkPluginZeroRateAlerts`. Three fail before this
change (saturated tx with idle rx, saturated rx with idle tx, and a fully idle
interface reading DEFAULT instead of OK); the fourth — both directions busy —
passes before and after, which is the path that already worked.
`tests/test_plugin_network.py` 39 passed.
387 lines
15 KiB
Python
Executable File
387 lines
15 KiB
Python
Executable File
#!/usr/bin/env python
|
|
#
|
|
# Glances - An eye on your system
|
|
#
|
|
# SPDX-FileCopyrightText: 2024 Nicolas Hennion <nicolas@nicolargo.com>
|
|
#
|
|
# SPDX-License-Identifier: LGPL-3.0-only
|
|
#
|
|
|
|
"""Tests for the Network plugin."""
|
|
|
|
import json
|
|
import time
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def network_plugin(glances_stats):
|
|
"""Return the Network plugin instance from glances_stats."""
|
|
return glances_stats.get_plugin('network')
|
|
|
|
|
|
class TestNetworkPluginBasics:
|
|
"""Test basic Network plugin functionality."""
|
|
|
|
def test_plugin_name(self, network_plugin):
|
|
"""Test plugin name is correctly set."""
|
|
assert network_plugin.plugin_name == 'network'
|
|
|
|
def test_plugin_is_enabled(self, network_plugin):
|
|
"""Test that the plugin is enabled by default."""
|
|
assert network_plugin.is_enabled() is True
|
|
|
|
def test_display_curse_enabled(self, network_plugin):
|
|
"""Test that curse display is enabled."""
|
|
assert network_plugin.display_curse is True
|
|
|
|
def test_get_key_returns_interface_name(self, network_plugin):
|
|
"""Test that get_key returns interface_name."""
|
|
assert network_plugin.get_key() == 'interface_name'
|
|
|
|
def test_history_items_defined(self, network_plugin):
|
|
"""Test that history items are properly defined."""
|
|
items = network_plugin.get_items_history_list()
|
|
assert items is not None
|
|
item_names = [item['name'] for item in items]
|
|
assert 'bytes_recv_rate_per_sec' in item_names
|
|
assert 'bytes_sent_rate_per_sec' in item_names
|
|
|
|
|
|
class TestNetworkPluginUpdate:
|
|
"""Test Network plugin update functionality."""
|
|
|
|
def test_update_returns_list(self, network_plugin):
|
|
"""Test that update returns a list."""
|
|
network_plugin.update()
|
|
stats = network_plugin.get_raw()
|
|
assert isinstance(stats, list)
|
|
|
|
def test_each_interface_has_name(self, network_plugin):
|
|
"""Test that each interface entry has a name."""
|
|
network_plugin.update()
|
|
stats = network_plugin.get_raw()
|
|
for iface in stats:
|
|
assert 'interface_name' in iface
|
|
|
|
def test_bytes_recv_and_sent_present(self, network_plugin):
|
|
"""Test that bytes_recv and bytes_sent are present."""
|
|
network_plugin.update()
|
|
time.sleep(0.1)
|
|
network_plugin.update()
|
|
stats = network_plugin.get_raw()
|
|
for iface in stats:
|
|
assert 'bytes_recv' in iface
|
|
assert 'bytes_sent' in iface
|
|
|
|
def test_bytes_values_non_negative(self, network_plugin):
|
|
"""Test that byte values are non-negative."""
|
|
network_plugin.update()
|
|
stats = network_plugin.get_raw()
|
|
for iface in stats:
|
|
if 'bytes_recv' in iface and iface['bytes_recv'] is not None:
|
|
assert iface['bytes_recv'] >= 0
|
|
if 'bytes_sent' in iface and iface['bytes_sent'] is not None:
|
|
assert iface['bytes_sent'] >= 0
|
|
|
|
|
|
class TestNetworkPluginRateCalculation:
|
|
"""Test Network plugin rate calculation."""
|
|
|
|
def test_rate_fields_after_two_updates(self, network_plugin):
|
|
"""Test that rate fields are populated after two updates."""
|
|
network_plugin.update()
|
|
time.sleep(0.2)
|
|
network_plugin.update()
|
|
stats = network_plugin.get_raw()
|
|
for iface in stats:
|
|
# Rate fields should be present after second update
|
|
assert 'bytes_recv_rate_per_sec' in iface or 'bytes_recv' in iface
|
|
assert 'bytes_sent_rate_per_sec' in iface or 'bytes_sent' in iface
|
|
|
|
def test_bytes_all_calculated(self, network_plugin):
|
|
"""Test that bytes_all is calculated as sum of recv and sent."""
|
|
network_plugin.update()
|
|
stats = network_plugin.get_raw()
|
|
for iface in stats:
|
|
if all(k in iface for k in ['bytes_recv', 'bytes_sent', 'bytes_all']):
|
|
if iface['bytes_recv'] is not None and iface['bytes_sent'] is not None:
|
|
expected = iface['bytes_recv'] + iface['bytes_sent']
|
|
assert iface['bytes_all'] == expected
|
|
|
|
|
|
class TestNetworkPluginInterfaceInfo:
|
|
"""Test Network plugin interface information."""
|
|
|
|
def test_is_up_field_present(self, network_plugin):
|
|
"""Test that is_up field is present for interfaces when available."""
|
|
network_plugin.update()
|
|
stats = network_plugin.get_raw()
|
|
for iface in stats:
|
|
# is_up may not be present in all stats entries
|
|
if 'is_up' in iface:
|
|
assert isinstance(iface['is_up'], bool)
|
|
|
|
def test_speed_field_present(self, network_plugin):
|
|
"""Test that speed field is present for interfaces when available."""
|
|
network_plugin.update()
|
|
stats = network_plugin.get_raw()
|
|
for iface in stats:
|
|
if 'speed' in iface:
|
|
assert isinstance(iface['speed'], (int, float))
|
|
|
|
def test_alias_field_present(self, network_plugin):
|
|
"""Test that alias field is present for interfaces."""
|
|
network_plugin.update()
|
|
stats = network_plugin.get_raw()
|
|
for iface in stats:
|
|
assert 'alias' in iface
|
|
|
|
|
|
class TestNetworkPluginViews:
|
|
"""Test Network plugin views functionality."""
|
|
|
|
def test_update_views_creates_views(self, network_plugin):
|
|
"""Test that update_views creates views dictionary."""
|
|
network_plugin.update()
|
|
network_plugin.update_views()
|
|
views = network_plugin.get_views()
|
|
assert isinstance(views, dict)
|
|
|
|
def test_views_keyed_by_interface(self, network_plugin):
|
|
"""Test that views are keyed by interface name."""
|
|
network_plugin.update()
|
|
time.sleep(0.1)
|
|
network_plugin.update()
|
|
network_plugin.update_views()
|
|
views = network_plugin.get_views()
|
|
stats = network_plugin.get_raw()
|
|
for iface in stats:
|
|
if iface['interface_name'] in views:
|
|
assert isinstance(views[iface['interface_name']], dict)
|
|
|
|
|
|
class TestNetworkPluginJSON:
|
|
"""Test Network plugin JSON serialization."""
|
|
|
|
def test_get_stats_returns_json(self, network_plugin):
|
|
"""Test that get_stats returns valid JSON."""
|
|
network_plugin.update()
|
|
stats_json = network_plugin.get_stats()
|
|
parsed = json.loads(stats_json)
|
|
assert isinstance(parsed, list)
|
|
|
|
def test_json_preserves_interface_data(self, network_plugin):
|
|
"""Test that JSON output preserves interface data."""
|
|
network_plugin.update()
|
|
stats_json = network_plugin.get_stats()
|
|
parsed = json.loads(stats_json)
|
|
for iface in parsed:
|
|
assert 'interface_name' in iface
|
|
|
|
|
|
class TestNetworkPluginHistory:
|
|
"""Test Network plugin history functionality."""
|
|
|
|
def test_history_enable(self, network_plugin):
|
|
"""Test that history can be enabled."""
|
|
assert network_plugin.history_enable() is not None
|
|
|
|
|
|
class TestNetworkPluginReset:
|
|
"""Test Network plugin reset functionality."""
|
|
|
|
def test_reset_clears_stats(self, network_plugin):
|
|
"""Test that reset clears stats."""
|
|
network_plugin.update()
|
|
network_plugin.reset()
|
|
stats = network_plugin.get_raw()
|
|
assert stats == network_plugin.get_init_value()
|
|
|
|
def test_reset_views(self, network_plugin):
|
|
"""Test that reset_views clears views."""
|
|
network_plugin.update()
|
|
network_plugin.update_views()
|
|
network_plugin.reset_views()
|
|
assert network_plugin.get_views() == {}
|
|
|
|
|
|
class TestNetworkPluginFieldsDescription:
|
|
"""Test Network plugin fields description."""
|
|
|
|
def test_fields_description_exists(self, network_plugin):
|
|
"""Test that fields_description is defined."""
|
|
assert network_plugin.fields_description is not None
|
|
|
|
def test_mandatory_fields_described(self, network_plugin):
|
|
"""Test that mandatory fields have descriptions."""
|
|
mandatory_fields = ['interface_name', 'bytes_recv', 'bytes_sent']
|
|
for field in mandatory_fields:
|
|
assert field in network_plugin.fields_description
|
|
|
|
def test_rate_fields_described(self, network_plugin):
|
|
"""Test that rate-enabled fields have rate flag."""
|
|
rate_fields = ['bytes_recv', 'bytes_sent', 'bytes_all']
|
|
for field in rate_fields:
|
|
if field in network_plugin.fields_description:
|
|
assert network_plugin.fields_description[field].get('rate') is True
|
|
|
|
|
|
class TestNetworkPluginConfiguration:
|
|
"""Test Network plugin configuration options."""
|
|
|
|
def test_hide_zero_attribute(self, network_plugin):
|
|
"""Test that hide_zero attribute exists."""
|
|
assert hasattr(network_plugin, 'hide_zero')
|
|
|
|
def test_hide_zero_fields_defined(self, network_plugin):
|
|
"""Test that hide_zero_fields is defined."""
|
|
assert hasattr(network_plugin, 'hide_zero_fields')
|
|
assert isinstance(network_plugin.hide_zero_fields, list)
|
|
|
|
def test_hide_no_up_attribute(self, network_plugin):
|
|
"""Test that hide_no_up attribute exists."""
|
|
assert hasattr(network_plugin, 'hide_no_up')
|
|
|
|
def test_hide_no_ip_attribute(self, network_plugin):
|
|
"""Test that hide_no_ip attribute exists."""
|
|
assert hasattr(network_plugin, 'hide_no_ip')
|
|
|
|
|
|
class TestNetworkPluginMsgCurse:
|
|
"""Test Network plugin curse message generation."""
|
|
|
|
def test_msg_curse_empty_without_max_width(self, network_plugin):
|
|
"""Test that msg_curse returns empty without max_width."""
|
|
network_plugin.update()
|
|
msg = network_plugin.msg_curse()
|
|
# Without max_width, should return empty list
|
|
assert isinstance(msg, list)
|
|
assert len(msg) == 0
|
|
|
|
def test_msg_curse_with_args_and_max_width(self, network_plugin):
|
|
"""Test that msg_curse works with args and max_width after views update."""
|
|
network_plugin.update()
|
|
network_plugin.update_views()
|
|
# msg_curse requires args with network_cumul and network_sum
|
|
if hasattr(network_plugin, 'args') and network_plugin.args:
|
|
try:
|
|
msg = network_plugin.msg_curse(args=network_plugin.args, max_width=80)
|
|
assert isinstance(msg, list)
|
|
except KeyError:
|
|
# Views might not be synced with stats in some cases
|
|
pass
|
|
|
|
|
|
class TestNetworkPluginSorting:
|
|
"""Test Network plugin sorting functionality."""
|
|
|
|
def test_sorted_stats_returns_list(self, network_plugin):
|
|
"""Test that sorted_stats returns a list."""
|
|
network_plugin.update()
|
|
sorted_stats = network_plugin.sorted_stats()
|
|
assert isinstance(sorted_stats, list)
|
|
|
|
def test_sorted_stats_preserves_count(self, network_plugin):
|
|
"""Test that sorted_stats preserves interface count."""
|
|
network_plugin.update()
|
|
raw_count = len(network_plugin.get_raw())
|
|
sorted_count = len(network_plugin.sorted_stats())
|
|
assert raw_count == sorted_count
|
|
|
|
|
|
class TestNetworkPluginExport:
|
|
"""Test Network plugin export functionality."""
|
|
|
|
def test_get_export_returns_list(self, network_plugin):
|
|
"""Test that get_export returns a list."""
|
|
network_plugin.update()
|
|
export = network_plugin.get_export()
|
|
assert isinstance(export, list)
|
|
|
|
def test_export_equals_raw(self, network_plugin):
|
|
"""Test that export equals raw stats by default."""
|
|
network_plugin.update()
|
|
assert network_plugin.get_export() == network_plugin.get_raw()
|
|
|
|
|
|
class TestNetworkPluginAlerts:
|
|
"""Test Network plugin alert functionality."""
|
|
|
|
def test_views_have_decoration(self, network_plugin):
|
|
"""Test that interface views have decoration after update."""
|
|
network_plugin.update()
|
|
time.sleep(0.1)
|
|
network_plugin.update()
|
|
network_plugin.update_views()
|
|
views = network_plugin.get_views()
|
|
stats = network_plugin.get_raw()
|
|
|
|
for iface in stats:
|
|
iface_name = iface['interface_name']
|
|
if iface_name in views and 'bytes_recv' in views[iface_name]:
|
|
# Check that decoration exists for rate fields
|
|
assert 'decoration' in views[iface_name]['bytes_recv']
|
|
|
|
|
|
class TestNetworkPluginZeroRateAlerts:
|
|
"""A rate of exactly 0 in one direction must not suppress the other.
|
|
|
|
`update_views` skipped the whole interface when either direction measured
|
|
0 bytes/s, so a saturated uplink went undecorated whenever the downlink
|
|
happened to be idle -- and vice versa for a receive-only interface.
|
|
"""
|
|
|
|
#: 1 Gbps, in bits, which is what the `speed` field carries.
|
|
LINK_SPEED_BPS = 1_000_000_000
|
|
|
|
#: 120 MB/s == 960 Mbps == 96% of the link above.
|
|
SATURATED_BYTES_PER_SEC = 120_000_000
|
|
|
|
def _views_for(self, plugin, rx_rate, tx_rate):
|
|
plugin.stats = [
|
|
{
|
|
'interface_name': 'eth0',
|
|
'alias': None,
|
|
'bytes_recv': 0,
|
|
'bytes_sent': 0,
|
|
'bytes_all': 0,
|
|
'bytes_recv_rate_per_sec': rx_rate,
|
|
'bytes_sent_rate_per_sec': tx_rate,
|
|
'bytes_all_rate_per_sec': rx_rate + tx_rate,
|
|
'speed': self.LINK_SPEED_BPS,
|
|
'is_up': True,
|
|
}
|
|
]
|
|
plugin.update_views()
|
|
return plugin.get_views()['eth0']
|
|
|
|
def test_saturated_tx_is_flagged_while_rx_is_idle(self, network_plugin):
|
|
"""A send-only interface at 96% of link speed must still alert."""
|
|
views = self._views_for(network_plugin, rx_rate=0, tx_rate=self.SATURATED_BYTES_PER_SEC)
|
|
assert views['bytes_sent_rate_per_sec']['decoration'] == 'CRITICAL'
|
|
assert views['bytes_sent']['decoration'] == 'CRITICAL'
|
|
|
|
def test_saturated_rx_is_flagged_while_tx_is_idle(self, network_plugin):
|
|
"""The mirror case: a receive-only interface, e.g. a monitor port."""
|
|
views = self._views_for(network_plugin, rx_rate=self.SATURATED_BYTES_PER_SEC, tx_rate=0)
|
|
assert views['bytes_recv_rate_per_sec']['decoration'] == 'CRITICAL'
|
|
assert views['bytes_recv']['decoration'] == 'CRITICAL'
|
|
|
|
def test_a_fully_idle_interface_is_ok_not_undecorated(self, network_plugin):
|
|
"""0 in both directions is a real measurement, not a missing one."""
|
|
views = self._views_for(network_plugin, rx_rate=0, tx_rate=0)
|
|
assert views['bytes_recv_rate_per_sec']['decoration'] == 'OK'
|
|
assert views['bytes_sent_rate_per_sec']['decoration'] == 'OK'
|
|
|
|
def test_both_directions_busy_still_alert(self, network_plugin):
|
|
"""The path that already worked must keep working."""
|
|
views = self._views_for(
|
|
network_plugin,
|
|
rx_rate=self.SATURATED_BYTES_PER_SEC,
|
|
tx_rate=self.SATURATED_BYTES_PER_SEC,
|
|
)
|
|
assert views['bytes_recv_rate_per_sec']['decoration'] == 'CRITICAL'
|
|
assert views['bytes_sent_rate_per_sec']['decoration'] == 'CRITICAL'
|