crates/signald/tests/hardware_ingest.rs

 1//! Hardware ingest: frames written by the out-of-process collector are read by
 2//! the same `wire::read_frame` the socket uses and published into the hub, so
 3//! a subscriber's snapshot carries them like any other collector's signals.
 4
 5use std::io::Cursor;
 6
 7use signal_schema::{wire, Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
 8use signald::collectors::hardware;
 9use signald::hub::Hub;
10
11fn sig(source: Source, name: SignalName, value: f64) -> Signal {
12    Signal {
13        schema_version: SCHEMA_VERSION,
14        ts: 1_723_100_000_000,
15        source,
16        name,
17        value: Value(value),
18        tag: None,
19    }
20}
21
22/// The five signals `macos-collector` emits per tick, as it emits them.
23fn one_tick() -> Vec<Signal> {
24    vec![
25        sig(Source::Hardware, SignalName::CpuLoad, 0.25),
26        sig(Source::Macos, SignalName::BatteryPct, 80.0),
27        sig(Source::Macos, SignalName::Charging, 1.0),
28        sig(Source::Hardware, SignalName::BatteryDrawW, 12.5),
29        sig(Source::Macos, SignalName::ThermalState, 0.0),
30    ]
31}
32
33fn frames(signals: &[Signal]) -> Vec<u8> {
34    let mut buf = Vec::new();
35    for s in signals {
36        wire::write_frame(&mut buf, s).unwrap();
37    }
38    buf
39}
40
41#[test]
42fn canned_frames_reach_a_subscriber_snapshot() {
43    let hub = Hub::new();
44    let mut first = one_tick();
45    let mut second = one_tick();
46    second[0].value = Value(0.75); // cpu_load changes on the second tick
47    first.append(&mut second);
48
49    let n = hardware::ingest(&mut Cursor::new(frames(&first)), &hub).unwrap();
50    assert_eq!(n, 10, "every frame is published");
51
52    let (snapshot, _rx) = hub.subscribe();
53    assert_eq!(snapshot.len(), 5, "one cached value per metric");
54    let cpu = snapshot.iter().find(|s| s.name == SignalName::CpuLoad).unwrap();
55    assert_eq!(cpu.value, Value(0.75), "keep-latest");
56    assert_eq!(cpu.source, Source::Hardware);
57    for name in [
58        SignalName::BatteryPct,
59        SignalName::Charging,
60        SignalName::BatteryDrawW,
61        SignalName::ThermalState,
62    ] {
63        assert!(snapshot.iter().any(|s| s.name == name), "{name:?} missing");
64    }
65}
66
67#[test]
68fn frame_that_breaks_the_tag_rule_is_dropped_and_ingest_continues() {
69    let hub = Hub::new();
70    let good = sig(Source::Hardware, SignalName::CpuLoad, 0.5);
71    // CpuLoad never allows a tag. Encode one anyway, bypassing the schema's
72    // well-formedness check, as a misbehaving collector could.
73    let mut bad = good.clone();
74    bad.tag = Some(Tag::repo_path("/not/allowed").unwrap());
75    let after = sig(Source::Macos, SignalName::BatteryPct, 42.0);
76
77    let mut buf = frames(&[good]);
78    buf.extend(wire::encode(&bad));
79    buf.extend(frames(&[after]));
80
81    let n = hardware::ingest(&mut Cursor::new(buf), &hub).unwrap();
82    assert_eq!(n, 2, "the tagged frame is dropped, the other two published");
83
84    let snapshot = hub.snapshot();
85    assert_eq!(snapshot.len(), 2);
86    assert!(
87        snapshot.iter().all(|s| s.tag.is_none()),
88        "no tagged frame reaches the hub"
89    );
90    assert!(
91        snapshot.iter().any(|s| s.name == SignalName::BatteryPct),
92        "ingest kept reading past the rejected frame"
93    );
94}