Skip to content

GitLab

Switch branch/tag
Find file Normal viewHistoryPermalink
test_net_service.py 1.78 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 
"""Tests for hytop.net.service core logic."""

from __future__ import annotations

from hytop.net.models import NetCounter, NodeCounterSnapshot
from hytop.net.service import apply_node_results, init_monitor_state


class TestApplyNodeResults:
    def test_second_sample_produces_rate(self):
        state = init_monitor_state(hosts=["localhost"], max_window=10.0)
        first = NodeCounterSnapshot(
            host="localhost",
            counters={"eth:eth0": NetCounter(kind="eth", name="eth0", rx_bytes=100, tx_bytes=200)},
            sample_ts=1.0,
        )
        second = NodeCounterSnapshot(
            host="localhost",
            counters={"eth:eth0": NetCounter(kind="eth", name="eth0", rx_bytes=300, tx_bytes=500)},
            sample_ts=2.0,
        )
        apply_node_results([first], interval=1.0, state=state)
        apply_node_results([second], interval=1.0, state=state)

        history = state.histories[("localhost", "eth:eth0")]
        latest = history.latest()
        assert latest is not None
        assert latest.rx_bps == 200.0
        assert latest.tx_bps == 300.0

    def test_counter_reset_is_skipped(self):
        state = init_monitor_state(hosts=["localhost"], max_window=10.0)
        first = NodeCounterSnapshot(
            host="localhost",
            counters={"eth:eth0": NetCounter(kind="eth", name="eth0", rx_bytes=100, tx_bytes=100)},
            sample_ts=1.0,
        )
        reset = NodeCounterSnapshot(
            host="localhost",
            counters={"eth:eth0": NetCounter(kind="eth", name="eth0", rx_bytes=50, tx_bytes=50)},
            sample_ts=2.0,
        )
        apply_node_results([first], interval=1.0, state=state)
        apply_node_results([reset], interval=1.0, state=state)
        assert ("localhost", "eth:eth0") not in state.histories