refactor(value-source): MetricSpec registry for SystemMetricsValueStream

SystemMetricsValueStream used to dispatch on its ``self._metric`` string
across three independent if/elif chains (audit finding M5):

  * priming in ``start()`` (cpu_percent seed, initial network counter);
  * raw reading in ``_read_metric_psutil`` plus ``_read_metric_fallback``;
  * normalisation in ``_normalize`` (percent / min-max range / max-rate).

Adding a new metric meant editing all three chains plus the Android
fallback — and forgetting one branch made the metric silently return 0.

Lift each per-metric concern into a free function and register them as a
``MetricSpec(name, read_psutil, read_fallback, normalize, prime)`` in a
new ``core.processing.metric_readers`` module. Shared normalisers
(``_norm_percent`` / ``_norm_range`` / ``_norm_rate`` / ``_zero``) live
once. The stream's ``start()`` / ``_read_metric()`` / ``_normalize()``
collapse to a single registry lookup + delegation.

The stream still owns its mutable state (``_disk_path``,
``_sensor_label``, ``_gpu_unavailable``, ``_prev_net_bytes``,
``_prev_net_time``, etc.) — readers operate on the stream by
parameter, not by inheritance, so the kitchen-sink class shrinks by
~140 lines without losing the per-stream cadence bookkeeping. Each
spec function's docstring documents which fields it reads or mutates.

Tests: 16 new tests cover the 10-metric coverage set, callable shape
of every spec field, the three normaliser primitives' clamping +
divide-by-zero behaviour, prime-hook presence (only the three metrics
that need a baseline: cpu_load + network_rx + network_tx), and
fallback-path expectations (desktop-only sensors -> _zero, cpu/ram ->
real MetricsProvider).

754 existing core / storage / api tests stay green; ruff clean.
This commit is contained in:
2026-05-22 23:29:33 +03:00
parent 98fb61d932
commit 9f3f346543
3 changed files with 443 additions and 152 deletions
@@ -0,0 +1,152 @@
"""Tests for the SystemMetricsValueStream metric reader registry.
Locks in the spec-driven dispatch that replaced the three independent
if/elif chains (priming, raw read, normalise) inside
``SystemMetricsValueStream``.
"""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from ledgrab.core.processing.metric_readers import (
METRIC_SPECS,
MetricSpec,
_norm_percent,
_norm_range,
_norm_rate,
_zero,
get_spec,
)
EXPECTED_METRICS = {
"cpu_load",
"ram_usage",
"disk_usage",
"battery_level",
"cpu_temp",
"fan_speed",
"gpu_load",
"gpu_temp",
"network_rx",
"network_tx",
}
# ---------------------------------------------------------------------------
# Registry shape
# ---------------------------------------------------------------------------
def test_registry_covers_known_metrics():
assert set(METRIC_SPECS.keys()) == EXPECTED_METRICS
def test_every_spec_has_callable_readers_and_normalize():
for name, spec in METRIC_SPECS.items():
assert isinstance(spec, MetricSpec), f"{name} is not a MetricSpec"
assert callable(spec.read_psutil), f"{name}.read_psutil not callable"
assert callable(spec.read_fallback), f"{name}.read_fallback not callable"
assert callable(spec.normalize), f"{name}.normalize not callable"
# prime is Optional[PrimeFn]
if spec.prime is not None:
assert callable(spec.prime), f"{name}.prime present but not callable"
def test_get_spec_returns_none_for_unknown_metric():
assert get_spec("totally_unregistered") is None
# ---------------------------------------------------------------------------
# Normaliser primitives
# ---------------------------------------------------------------------------
def test_norm_percent_clamps_to_unit_interval():
s = SimpleNamespace() # normalise primitives ignore the stream
assert _norm_percent(s, 50.0) == 0.5
assert _norm_percent(s, 0.0) == 0.0
assert _norm_percent(s, 100.0) == 1.0
# Clamps
assert _norm_percent(s, -10.0) == 0.0
assert _norm_percent(s, 200.0) == 1.0
def test_norm_range_handles_min_max_window():
s = SimpleNamespace(_min_val=20.0, _max_val=70.0)
assert _norm_range(s, 20.0) == 0.0
assert _norm_range(s, 70.0) == 1.0
assert _norm_range(s, 45.0) == 0.5
# Clamps
assert _norm_range(s, 10.0) == 0.0
assert _norm_range(s, 100.0) == 1.0
def test_norm_range_returns_half_on_zero_window():
"""Avoid division by zero when min==max — pick the middle of [0,1]."""
s = SimpleNamespace(_min_val=50.0, _max_val=50.0)
assert _norm_range(s, 50.0) == 0.5
assert _norm_range(s, 999.0) == 0.5
def test_norm_rate_uses_max_rate():
s = SimpleNamespace(_max_rate=100.0)
assert _norm_rate(s, 0.0) == 0.0
assert _norm_rate(s, 50.0) == 0.5
assert _norm_rate(s, 100.0) == 1.0
assert _norm_rate(s, 200.0) == 1.0
def test_norm_rate_returns_half_on_zero_max():
s = SimpleNamespace(_max_rate=0.0)
assert _norm_rate(s, 12345.0) == 0.5
def test_zero_reader_always_returns_zero():
assert _zero(None) == 0.0
assert _zero(SimpleNamespace()) == 0.0
# ---------------------------------------------------------------------------
# Spec wiring sanity
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"name, expected_normalize",
[
("cpu_load", _norm_percent),
("ram_usage", _norm_percent),
("disk_usage", _norm_percent),
("battery_level", _norm_percent),
("gpu_load", _norm_percent),
("cpu_temp", _norm_range),
("gpu_temp", _norm_range),
("fan_speed", _norm_range),
("network_rx", _norm_rate),
("network_tx", _norm_rate),
],
)
def test_metric_uses_expected_normaliser(name, expected_normalize):
assert METRIC_SPECS[name].normalize is expected_normalize
def test_metrics_without_psutil_use_zero_fallback_for_desktop_only():
"""Desktop-only sensors fall back to a constant 0.0 when psutil is absent."""
for name in ("disk_usage", "battery_level", "cpu_temp", "fan_speed", "gpu_load", "gpu_temp"):
assert METRIC_SPECS[name].read_fallback is _zero, name
def test_cpu_and_ram_have_meaningful_fallbacks():
"""cpu_load and ram_usage go through the platform-aware MetricsProvider."""
for name in ("cpu_load", "ram_usage"):
assert METRIC_SPECS[name].read_fallback is not _zero, name
def test_only_cpu_load_and_network_have_prime_hooks():
"""Priming is only meaningful for metrics that need an initial baseline."""
primed = {name for name, spec in METRIC_SPECS.items() if spec.prime is not None}
assert primed == {"cpu_load", "network_rx", "network_tx"}