"""Watch metrics unit tests — rates, failure samples, merge.""" 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_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_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_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], "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.load_1 == 0.8 assert isinstance(sample.processes_json, dict) assert sample.processes_json["top_cpu"][0]["name"] == "nginx" def test_parse_ssh_watch_payload_baota_detail_fields(): sample = parse_ssh_watch_payload( { "cpu_usage": 30, "mem_usage": 13, "disk_usage": 50, "load_avg": [0.31, 0.38, 0.42], "process_running": 1, "process_total": 146, "per_cpu_pct": [332.4, 29.5, 30.0, 27.2], "mem_free_mb": 329, "mem_shared_mb": 2, "mem_buffers_mb": 3046, "mem_cached_mb": 77, "top_cpu": [{"pid": 1, "name": "a", "cpu_pct": 10, "mem_pct": 1}], "top_mem": [{"pid": 2, "name": "b", "cpu_pct": 1, "mem_pct": 20}], }, ) assert sample.load_1 == 0.31 assert sample.process_total == 146 assert sample.per_cpu_pct == [332.4, 29.5, 30.0, 27.2] assert sample.mem_free_mb == 329 bundle = sample.processes_json assert bundle["top_mem"][0]["name"] == "b" 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_metrics(): 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, mem_pct=22, load_1=1.2, cpu_cores=4, memory_total_gb=8.0, disk_total_gb=120.0, ) merged = merge_redis_and_ssh(redis_s, ssh_s) assert merged.source == "mixed" assert merged.cpu_pct == 11 assert merged.mem_pct == 22 assert merged.load_1 == 1.2 assert merged.cpu_cores == 4 assert merged.memory_total_gb == 8.0 assert merged.disk_total_gb == 120.0 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