Nexus/tests/test_watch_metrics.py

"""Watch metrics unit tests — rates, redis IO skip, failure samples."""

import pytest

from server.utils.watch_metrics import (
    WatchProbeSample,
    WATCH_PIN_TTL_DEFAULT_MINUTES,
    apply_rates_to_sample,
    compute_rates,
    failure_sample,
    merge_redis_and_ssh,
    normalize_watch_ttl_minutes,
    parse_ssh_watch_payload,
    redis_sample_has_io,
    redis_sample_has_basic_metrics,
    sanitize_processes,
    PROBE_OK,
)


def test_compute_rates_basic():
    from server.utils.watch_metrics import _RateState

    prev = _RateState(
        net_bytes_sent=1000,
        net_bytes_recv=2000,
        disk_read_bytes=500,
        disk_write_bytes=800,
        recorded_at_ts=100.0,
    )
    up, down, dr, dw = compute_rates(
        prev,
        net_bytes_sent=1500,
        net_bytes_recv=2600,
        disk_read_bytes=700,
        disk_write_bytes=900,
        now_ts=105.0,
    )
    assert up == 100
    assert down == 120
    assert dr == 40
    assert dw == 20


def test_compute_rates_no_prev():
    up, down, dr, dw = compute_rates(None, net_bytes_sent=1, net_bytes_recv=1, disk_read_bytes=1, disk_write_bytes=1, now_ts=10)
    assert up is None and down is None and dr is None and dw is None


def test_redis_sample_has_basic_metrics():
    basic = WatchProbeSample(probe_status=PROBE_OK, source="redis", cpu_pct=10)
    assert redis_sample_has_basic_metrics(basic) is True
    assert redis_sample_has_io(basic) is False
    assert redis_sample_has_basic_metrics(None) is False


def test_parse_json_stdout_with_prefix_noise():
    from server.infrastructure.ssh.remote_probe import _parse_json_stdout

    stdout = "Warning: something\nOK {\"status\": \"ok\", \"cpu_usage\": 1.0}\n"
    parsed = _parse_json_stdout(stdout)
    assert parsed is not None
    assert parsed.get("status") == "ok"


def test_redis_sample_has_io():
    complete = WatchProbeSample(
        probe_status=PROBE_OK,
        source="redis",
        cpu_pct=10,
        mem_pct=20,
        net_bytes_sent=1,
        net_bytes_recv=2,
        disk_read_bytes=3,
        disk_write_bytes=4,
    )
    incomplete = WatchProbeSample(probe_status=PROBE_OK, source="redis", cpu_pct=10, mem_pct=20)
    assert redis_sample_has_io(complete) is True
    assert redis_sample_has_io(incomplete) is False


def test_failure_sample_null_metrics():
    s = failure_sample(status="ssh_timeout", source="ssh", error="timeout")
    assert s.cpu_pct is None
    assert s.probe_status == "ssh_timeout"


def test_parse_ssh_watch_payload():
    data = {
        "cpu_usage": 12.4,
        "mem_usage": 45.2,
        "disk_usage": 61.0,
        "load_avg": [0.8, 0.9, 1.0],
        "net_io": {"bytes_sent": 100, "bytes_recv": 200},
        "disk_io": {"read_bytes": 300, "write_bytes": 400},
        "top_processes": [{"pid": 1, "name": "nginx", "cpu_pct": 2.0, "mem_pct": 1.0}],
    }
    sample = parse_ssh_watch_payload(data)
    assert sample.cpu_pct == 12
    assert sample.net_bytes_sent == 100
    assert sample.processes_json and sample.processes_json[0]["name"] == "nginx"


def test_sanitize_processes_truncates_name():
    raw = [{"pid": 1, "name": "x" * 100, "cpu_pct": 1}]
    out = sanitize_processes(raw)
    assert out and len(out[0]["name"]) == 80


def test_apply_rates_to_sample():
    from server.utils.watch_metrics import _RateState

    sample = WatchProbeSample(
        probe_status=PROBE_OK,
        source="ssh",
        net_bytes_sent=2000,
        net_bytes_recv=3000,
        disk_read_bytes=4000,
        disk_write_bytes=5000,
    )
    prev = _RateState(
        net_bytes_sent=1000,
        net_bytes_recv=2000,
        disk_read_bytes=3000,
        disk_write_bytes=4000,
        recorded_at_ts=0.0,
    )
    apply_rates_to_sample(sample, prev, 10.0)
    assert sample.net_up_bps == 100
    assert sample.net_down_bps == 100


def test_merge_redis_and_ssh_prefers_ssh_io():
    redis_s = WatchProbeSample(probe_status=PROBE_OK, source="redis", cpu_pct=10, mem_pct=20)
    ssh_s = WatchProbeSample(
        probe_status=PROBE_OK,
        source="ssh",
        cpu_pct=11,
        net_bytes_sent=99,
        net_bytes_recv=88,
        disk_read_bytes=77,
        disk_write_bytes=66,
    )
    merged = merge_redis_and_ssh(redis_s, ssh_s)
    assert merged.source == "mixed"
    assert merged.cpu_pct == 10
    assert merged.net_bytes_sent == 99


def test_normalize_watch_ttl_minutes():
    assert normalize_watch_ttl_minutes(30) == 30
    assert normalize_watch_ttl_minutes(60) == 60
    assert normalize_watch_ttl_minutes(1440) == 1440
    assert normalize_watch_ttl_minutes(8) == 480
    assert normalize_watch_ttl_minutes(99) == WATCH_PIN_TTL_DEFAULT_MINUTES