glances/tests/test_plugin_ports.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
#

"""Tests for the Ports plugin."""

import pytest

import glances.plugins.ports as ports_mod
from glances.plugins.ports import PortsPlugin, ThreadScanner


@pytest.fixture
def ports_plugin():
    """Return a Ports plugin instance without running its full init."""
    return PortsPlugin.__new__(PortsPlugin)


def web_scan(status, elapsed=0, rtt_warning=1):
    """Return a web scan result as stored by the ports plugin."""
    return {'status': status, 'elapsed': elapsed, 'rtt_warning': rtt_warning}


class TestPortsPluginAlertLevel:
    """Test that the alert level is resolved by severity, not by dict ordering."""

    @pytest.mark.parametrize(
        ('conds', 'expected'),
        [
            ({'CAREFUL': True, 'CRITICAL': True, 'WARNING': True}, 'CRITICAL'),
            ({'CAREFUL': True, 'CRITICAL': False, 'WARNING': True}, 'WARNING'),
            ({'CAREFUL': True, 'CRITICAL': False, 'WARNING': False}, 'CAREFUL'),
            ({'CAREFUL': False, 'CRITICAL': False, 'WARNING': False}, 'OK'),
        ],
    )
    def test_most_severe_condition_wins(self, ports_plugin, conds, expected):
        """Test that the most severe matching condition is returned."""
        assert ports_plugin.get_default_ret_value(conds) == expected

    def test_web_not_scanned_yet_is_careful(self, ports_plugin):
        """Test that a URL whose first scan did not complete is not CRITICAL."""
        conds = ports_plugin.get_conds_if_url(web_scan(None))
        assert ports_plugin.get_default_ret_value(conds) == 'CAREFUL'

    def test_web_failing_and_slow_is_critical(self, ports_plugin):
        """Test that a failing URL stays CRITICAL even when it is also slow."""
        conds = ports_plugin.get_conds_if_url(web_scan(404, elapsed=5))
        assert ports_plugin.get_default_ret_value(conds) == 'CRITICAL'

    def test_web_failing_and_fast_is_critical(self, ports_plugin):
        """Test that a failing URL is CRITICAL."""
        conds = ports_plugin.get_conds_if_url(web_scan(404))
        assert ports_plugin.get_default_ret_value(conds) == 'CRITICAL'

    def test_web_ok_but_slow_is_warning(self, ports_plugin):
        """Test that a reachable but slow URL is WARNING."""
        conds = ports_plugin.get_conds_if_url(web_scan(200, elapsed=5))
        assert ports_plugin.get_default_ret_value(conds) == 'WARNING'

    def test_web_ok_and_fast_is_ok(self, ports_plugin):
        """Test that a reachable and fast URL is OK."""
        conds = ports_plugin.get_conds_if_url(web_scan(200))
        assert ports_plugin.get_default_ret_value(conds) == 'OK'


@pytest.fixture
def scanner():
    """Return a ThreadScanner instance without running its full init."""
    return ThreadScanner.__new__(ThreadScanner)


class TestIcmpPingCommand:
    """Test the ping command line built for an ICMP (port 0) check."""

    @pytest.fixture
    def ping_cmd(self, scanner, monkeypatch):
        """Return the ping command line built on the given platform."""

        def build(platform, timeout=3):
            recorded = {}

            def fake_check_call(cmd, **kwargs):
                recorded['cmd'] = cmd
                return 0

            for name in ('WINDOWS', 'MACOS', 'BSD'):
                monkeypatch.setattr(ports_mod, name, name == platform)
            monkeypatch.setattr(ports_mod.subprocess, 'check_call', fake_check_call)
            monkeypatch.setattr(ThreadScanner, '_resolv_name', lambda self, host: host)
            scanner._port_scan_icmp({'host': 'example.net', 'port': 0, 'timeout': timeout})
            return recorded['cmd']

        return build

    def test_windows_timeout_is_expressed_in_milliseconds(self, ping_cmd):
        """Windows ping -w is a per-reply timeout in ms, so 3s must be sent as 3000."""
        assert ping_cmd('WINDOWS') == ['ping', '-n', '1', '-w', '3000', 'example.net']

    def test_linux_timeout_stays_in_seconds(self, ping_cmd):
        """Linux ping -W is in seconds, so the value is passed through."""
        assert ping_cmd('LINUX') == ['ping', '-c', '1', '-W', '3', 'example.net']

    def test_macos_timeout_stays_in_seconds(self, ping_cmd):
        """macOS and BSD ping -t is in seconds, so the value is passed through."""
        assert ping_cmd('MACOS') == ['ping', '-c', '1', '-t', '3', 'example.net']

    def test_timeout_is_not_sent_through_the_name_resolver(self, scanner, monkeypatch):
        """The timeout is a number of seconds, not a hostname to look up."""
        resolved = []

        for name in ('WINDOWS', 'MACOS', 'BSD'):
            monkeypatch.setattr(ports_mod, name, False)
        monkeypatch.setattr(ports_mod.subprocess, 'check_call', lambda cmd, **kwargs: 0)
        monkeypatch.setattr(ThreadScanner, '_resolv_name', lambda self, host: resolved.append(host) or host)
        scanner._port_scan_icmp({'host': 'example.net', 'port': 0, 'timeout': 3})
        assert resolved == ['example.net']