heartbeat/tests/test_http_host_info.py

"""Tests for _build_host_info helper in http.py."""
import pytest
from unittest.mock import MagicMock
from hbd.server.http import _build_host_info


class _FakeConn:
    def __init__(self, lastbeat):
        self.lastbeat = lastbeat


class _FakeHost:
    def __init__(self, name="myhost", owner=None, managers=None,
                 connections=None, os_data=None, plugin_data=None):
        self.name = name
        self.owner = owner
        self.managers = managers or []
        self.connections = connections or {}
        self._os_data = os_data
        self.plugin_data = plugin_data or {}

    def get_latest_plugin_data(self, plugin_name):
        if plugin_name == "os_info" and self._os_data is not None:
            return (1234567890.0, self._os_data)
        return None


def test_build_host_info_basic_fields():
    host = _FakeHost(owner="alice", managers=["bob", "carol"])
    result = _build_host_info(host)
    assert result["owner"] == "alice"
    assert result["managers"] == ["bob", "carol"]
    assert result["hbc_version"] is None
    assert result["hbc_type"] is None
    assert result["last_packet"] is None
    assert result["thresholds"] is None


def test_build_host_info_no_owner():
    host = _FakeHost()
    result = _build_host_info(host)
    assert result["owner"] is None
    assert result["managers"] == []


def test_build_host_info_reads_hbc_from_os_info():
    host = _FakeHost(os_data={"hbc_version": "5.3.0", "hbc_type": "full"})
    result = _build_host_info(host)
    assert result["hbc_version"] == "5.3.0"
    assert result["hbc_type"] == "full"


def test_build_host_info_hbc_none_when_no_os_info():
    host = _FakeHost(os_data=None)
    result = _build_host_info(host)
    assert result["hbc_version"] is None
    assert result["hbc_type"] is None


def test_build_host_info_last_packet_is_max_lastbeat():
    host = _FakeHost(connections={
        "IPv4": _FakeConn(1000.0),
        "IPv6": _FakeConn(2000.0),
    })
    result = _build_host_info(host)
    assert result["last_packet"] == 2000.0


def test_build_host_info_last_packet_none_when_no_connections():
    host = _FakeHost(connections={})
    result = _build_host_info(host)
    assert result["last_packet"] is None


def test_build_host_info_thresholds_none_without_checker():
    host = _FakeHost()
    result = _build_host_info(host, threshold_checker=None)
    assert result["thresholds"] is None


def test_build_host_info_thresholds_sorted_by_metric():
    from hbd.server.threshold import ThresholdConfig
    tc_cpu = ThresholdConfig("cpu_monitor.cpu_percent", warning=80.0, critical=95.0)
    tc_mem = ThresholdConfig("memory_monitor.memory_percent", warning=85.0, critical=98.0)

    checker = MagicMock()
    checker.get_thresholds_for_host.return_value = {
        "memory_monitor.memory_percent": tc_mem,
        "cpu_monitor.cpu_percent": tc_cpu,
    }

    host = _FakeHost()
    result = _build_host_info(host, threshold_checker=checker)

    assert result["thresholds"] is not None
    assert len(result["thresholds"]) == 2
    assert result["thresholds"][0]["metric"] == "cpu_monitor.cpu_percent"
    assert result["thresholds"][0]["warning"] == 80.0
    assert result["thresholds"][0]["critical"] == 95.0
    assert result["thresholds"][0]["operator"] == ">"
    assert result["thresholds"][1]["metric"] == "memory_monitor.memory_percent"


def test_build_host_info_thresholds_empty_list_when_no_thresholds():
    checker = MagicMock()
    checker.get_thresholds_for_host.return_value = {}
    host = _FakeHost()
    result = _build_host_info(host, threshold_checker=checker)
    assert result["thresholds"] == []


def test_build_host_info_threshold_null_warning_critical():
    from hbd.server.threshold import ThresholdConfig
    tc = ThresholdConfig("rtt.myhost", warning=None, critical=500.0)
    checker = MagicMock()
    checker.get_thresholds_for_host.return_value = {"rtt.myhost": tc}
    host = _FakeHost()
    result = _build_host_info(host, threshold_checker=checker)
    assert result["thresholds"][0]["warning"] is None
    assert result["thresholds"][0]["critical"] == 500.0


def test_build_host_info_nagios_operator_serialized():
    from hbd.server.threshold import ThresholdConfig
    tc = ThresholdConfig("nagios_runner.check_http", operator="nagios")
    checker = MagicMock()
    checker.get_thresholds_for_host.return_value = {"nagios_runner.check_http": tc}
    host = _FakeHost()
    result = _build_host_info(host, threshold_checker=checker)
    assert result["thresholds"][0]["operator"] == "nagios"


def test_build_host_info_covers_suffix_matched_metrics():
    """memory_monitor.percent threshold covers swap_percent via suffix match."""
    from hbd.server.threshold import ThresholdConfig
    tc_pct = ThresholdConfig("memory_monitor.percent", warning=85.0, critical=95.0)
    checker = MagicMock()
    checker.get_thresholds_for_host.return_value = {"memory_monitor.percent": tc_pct}

    host = _FakeHost(
        connections={},
        os_data=None,
    )
    # Simulate plugin_data with both percent and swap_percent fields
    host.plugin_data = {
        "memory_monitor": [(1234567890.0, {
            "percent": 80.0,
            "swap_percent": 25.0,
            "available_mb": 2000,
        })]
    }

    result = _build_host_info(host, threshold_checker=checker)
    assert result["thresholds"] is not None
    t = result["thresholds"][0]
    assert t["metric"] == "memory_monitor.percent"
    assert t["covers"] == ["memory_monitor.swap_percent"]


def test_build_host_info_covers_empty_when_exact_matches_only():
    """No covers when all plugin fields match their threshold exactly."""
    from hbd.server.threshold import ThresholdConfig
    tc_pct = ThresholdConfig("memory_monitor.percent", warning=85.0, critical=95.0)
    checker = MagicMock()
    checker.get_thresholds_for_host.return_value = {"memory_monitor.percent": tc_pct}

    host = _FakeHost()
    host.plugin_data = {
        "memory_monitor": [(1234567890.0, {"percent": 80.0})]
    }

    result = _build_host_info(host, threshold_checker=checker)
    t = result["thresholds"][0]
    assert t["covers"] == []


class _FakeConnFull:
    def __init__(self, afam, addr, state, rtt, statetime, lastbeat):
        self.afam = afam
        self.addr = addr
        self.state = state
        self.rtts = [rtt]
        self.statetime = statetime
        self.lastbeat = lastbeat


def test_build_host_info_includes_connections():
    host = _FakeHost(connections={
        "IPv4": _FakeConnFull("IPv4", "10.0.0.5", "up", 14.2, 1000.0, 2000.0),
        "IPv6": _FakeConnFull("IPv6", "fd00::5", "overdue", 0, 1500.0, 1900.0),
    })
    result = _build_host_info(host)
    conns = {c["family"]: c for c in result["connections"]}
    assert conns["IPv4"] == {"family": "IPv4", "addr": "10.0.0.5", "state": "up",
                             "rtt": 14.2, "statetime": 1000.0, "lastbeat": 2000.0}
    assert conns["IPv6"]["state"] == "overdue"


def test_build_host_info_connections_empty():
    host = _FakeHost()
    result = _build_host_info(host)
    assert result["connections"] == []