Nexus Agent
adee25deb6
Use chr(34) inside the bash-wrapped python -c probe so monitor slots work again after hardware detail rollout.
228 lines
6.7 KiB
Python
228 lines
6.7 KiB
Python
"""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_parse_ssh_watch_payload_includes_hardware_specs():
|
|
sample = parse_ssh_watch_payload(
|
|
{
|
|
"cpu_usage": 12.5,
|
|
"mem_usage": 40,
|
|
"disk_usage": 55,
|
|
"cpu_cores": 8,
|
|
"cpu_cores_physical": 4,
|
|
"cpu_model": "Intel Xeon",
|
|
"memory_total_gb": 16,
|
|
"memory_used_gb": 6.4,
|
|
"memory_available_gb": 9.6,
|
|
"disk_total_gb": 500,
|
|
"disk_used_gb": 275,
|
|
"disk_free_gb": 225,
|
|
"swap_total_gb": 4,
|
|
"swap_used_gb": 0.5,
|
|
"swap_pct": 12.5,
|
|
"os_release": "Ubuntu 22.04 LTS",
|
|
"uptime_seconds": 86400,
|
|
}
|
|
)
|
|
assert sample.cpu_cores == 8
|
|
assert sample.cpu_cores_physical == 4
|
|
assert sample.cpu_model == "Intel Xeon"
|
|
assert sample.memory_used_gb == 6.4
|
|
assert sample.disk_free_gb == 225.0
|
|
assert sample.os_release == "Ubuntu 22.04 LTS"
|
|
assert sample.uptime_seconds == 86400
|
|
|
|
|
|
def test_fill_hardware_gaps_from_percent():
|
|
from server.utils.watch_metrics import fill_hardware_gaps
|
|
|
|
sample = WatchProbeSample(
|
|
probe_status=PROBE_OK,
|
|
source="redis",
|
|
mem_pct=50,
|
|
disk_pct=25,
|
|
memory_total_gb=16.0,
|
|
disk_total_gb=400.0,
|
|
)
|
|
fill_hardware_gaps(sample)
|
|
assert sample.memory_used_gb == 8.0
|
|
assert sample.memory_available_gb == 8.0
|
|
assert sample.disk_used_gb == 100.0
|
|
assert sample.disk_free_gb == 300.0
|
|
|
|
|
|
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,
|
|
cpu_cores=4,
|
|
memory_total_gb=8.0,
|
|
disk_total_gb=120.0,
|
|
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.cpu_cores == 4
|
|
assert merged.memory_total_gb == 8.0
|
|
assert merged.disk_total_gb == 120.0
|
|
assert merged.net_bytes_sent == 99
|
|
|
|
|
|
def test_ssh_watch_metrics_cmd_runs_locally():
|
|
import subprocess
|
|
|
|
from server.infrastructure.ssh.remote_probe import SSH_WATCH_METRICS_CMD, _parse_json_stdout
|
|
|
|
result = subprocess.run(
|
|
["bash", "-c", SSH_WATCH_METRICS_CMD],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=20,
|
|
)
|
|
parsed = _parse_json_stdout(result.stdout)
|
|
assert parsed is not None
|
|
assert parsed.get("status") == "ok"
|
|
assert parsed.get("cpu_cores") is not None
|
|
|
|
|
|
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
|