Skip to content

GitLab

Switch branch/tag
Find file Normal viewHistoryPermalink
test_net_collector.py 2.49 KB
Newer Older
one's avatar
[hytop] Add hytop net  
one committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80 
"""Tests for hytop.net.collector helpers."""

from __future__ import annotations

from hytop.net.collector import (
    _normalize_ib_port_state,
    _parse_ibdev2netdev_output,
    parse_counter_payload,
    parse_kind_filter,
    parse_name_filter,
    should_keep_iface,
)


class TestParseKindFilter:
    def test_all(self):
        assert parse_kind_filter("all") == {"eth", "ib"}

    def test_eth_only(self):
        assert parse_kind_filter("eth") == {"eth"}

    def test_ib_only(self):
        assert parse_kind_filter("ib") == {"ib"}


class TestParseNameFilter:
    def test_empty(self):
        assert parse_name_filter("") == set()

    def test_trim_and_split(self):
        assert parse_name_filter(" eth0, ib0 ,eth1 ") == {"eth0", "ib0", "eth1"}


class TestShouldKeepIface:
    def test_include_filter(self):
        assert should_keep_iface("eth0", include={"eth0"})
        assert not should_keep_iface("eth1", include={"eth0"})

    def test_virtual_excluded_by_default(self):
        assert not should_keep_iface("lo", include=set())
        assert not should_keep_iface("docker0", include=set())


class TestParseCounterPayload:
    def test_parse_valid_payload(self):
        payload = (
            '{"counters":['
            '{"kind":"eth","name":"eth0","rx_bytes":10,"tx_bytes":20,"link_state":"up"},'
            '{"kind":"ib","name":"mlx5_0/p1","rx_bytes":30,"tx_bytes":40,"link_state":"4: ACTIVE"}'
            "]}"
        )
        result = parse_counter_payload(payload)
        assert result["eth:eth0"].rx_bytes == 10
        assert result["ib:mlx5_0/p1"].tx_bytes == 40
        assert result["eth:eth0"].link_state == "up"
        assert result["ib:mlx5_0/p1"].link_state == "active"

    def test_invalid_json_raises(self):
        try:
            parse_counter_payload("{")
        except ValueError as exc:
            assert "invalid collector payload" in str(exc)
        else:
            raise AssertionError("expected ValueError")


class TestNormalizeIbState:
    def test_state_with_numeric_prefix(self):
        assert _normalize_ib_port_state("1: DOWN") == "down"

    def test_state_without_prefix(self):
        assert _normalize_ib_port_state("ACTIVE") == "active"


class TestParseIbdev2netdevOutput:
    def test_parse_mapping_lines(self):
        stdout = "mlx5_0 port 1 ==> p6p1 (Down)\nmlx5_3 port 1 ==> ibs61f0 (Up)\n"
        mapping = _parse_ibdev2netdev_output(stdout)
        assert mapping[("mlx5_0", "1")] == ("p6p1", "down")
        assert mapping[("mlx5_3", "1")] == ("ibs61f0", "up")