//! Hardware ingest: frames written by the out-of-process collector are read by //! the same `wire::read_frame` the socket uses and published into the hub, so //! a subscriber's snapshot carries them like any other collector's signals. use std::io::Cursor; use signal_schema::{wire, Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION}; use signald::collectors::hardware; use signald::hub::Hub; fn sig(source: Source, name: SignalName, value: f64) -> Signal { Signal { schema_version: SCHEMA_VERSION, ts: 1_723_100_000_000, source, name, value: Value(value), tag: None, } } /// The five signals `macos-collector` emits per tick, as it emits them. fn one_tick() -> Vec { vec![ sig(Source::Hardware, SignalName::CpuLoad, 0.25), sig(Source::Macos, SignalName::BatteryPct, 80.0), sig(Source::Macos, SignalName::Charging, 1.0), sig(Source::Hardware, SignalName::BatteryDrawW, 12.5), sig(Source::Macos, SignalName::ThermalState, 0.0), ] } fn frames(signals: &[Signal]) -> Vec { let mut buf = Vec::new(); for s in signals { wire::write_frame(&mut buf, s).unwrap(); } buf } #[test] fn canned_frames_reach_a_subscriber_snapshot() { let hub = Hub::new(); let mut first = one_tick(); let mut second = one_tick(); second[0].value = Value(0.75); // cpu_load changes on the second tick first.append(&mut second); let n = hardware::ingest(&mut Cursor::new(frames(&first)), &hub).unwrap(); assert_eq!(n, 10, "every frame is published"); let (snapshot, _rx) = hub.subscribe(); assert_eq!(snapshot.len(), 5, "one cached value per metric"); let cpu = snapshot.iter().find(|s| s.name == SignalName::CpuLoad).unwrap(); assert_eq!(cpu.value, Value(0.75), "keep-latest"); assert_eq!(cpu.source, Source::Hardware); for name in [ SignalName::BatteryPct, SignalName::Charging, SignalName::BatteryDrawW, SignalName::ThermalState, ] { assert!(snapshot.iter().any(|s| s.name == name), "{name:?} missing"); } } #[test] fn frame_that_breaks_the_tag_rule_is_rejected_at_the_boundary() { let hub = Hub::new(); let good = sig(Source::Hardware, SignalName::CpuLoad, 0.5); // CpuLoad never allows a tag. Encode one anyway, bypassing the schema's // well-formedness check, as a misbehaving collector could. let mut bad = good.clone(); bad.tag = Some(Tag::repo_path("/not/allowed").unwrap()); let mut buf = frames(&[good]); buf.extend(wire::encode(&bad)); let err = hardware::ingest(&mut Cursor::new(buf), &hub).unwrap_err(); assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); let snapshot = hub.snapshot(); assert_eq!(snapshot.len(), 1, "the good frame before it was published"); assert!(snapshot[0].tag.is_none()); }