[test](trx-server): add Tier 1 unit tests for pure helpers

Cover the side-effect-free helpers in audio.rs and main.rs with table-driven tests. New: split_history_chunks (boundary cases, oversize records, truncated tails), classify_stream_error, enforce_capacity, downmix_if_needed/downmix_mono, frame_rms, apply_decode_audio_gate, resample_to_12k, opus_channels, retain_ft2_window, should_emit_ft2_decode, parse_serial_addr, parse_rig_mode, default_audio_bandwidth_for_mode.

33 new tests; trx-server suite now reports 71 passed (was 38).

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Signed-off-by: Stan Grams <sjg@haxx.space>

src/trx-server/src/audio.rs

@@ -200,6 +200,41 @@ impl StreamErrorLogger {
     }
 }
 
+/// Walk a length-prefixed history blob and return the byte ranges of each
+/// chunk to send. Each record is `[u8 type][u32 BE len][payload]`. A chunk is
+/// flushed when adding the next record would push it past `threshold`. A
+/// single record larger than `threshold` becomes its own chunk — the wire-side
+/// `MAX_HISTORY_PAYLOAD_SIZE` cap is generous enough for any realistic decode.
+/// A malformed tail (truncated header or payload) is dropped silently; the
+/// blob is self-built so this is defence in depth, not error reporting.
+fn split_history_chunks(blob: &[u8], threshold: usize) -> Vec<std::ops::Range<usize>> {
+    let mut chunks = Vec::new();
+    let mut chunk_start = 0usize;
+    let mut pos = 0usize;
+    while pos < blob.len() {
+        let len_end = pos.saturating_add(5);
+        if len_end > blob.len() {
+            break;
+        }
+        let payload_len =
+            u32::from_be_bytes([blob[pos + 1], blob[pos + 2], blob[pos + 3], blob[pos + 4]])
+                as usize;
+        let msg_end = len_end.saturating_add(payload_len);
+        if msg_end > blob.len() {
+            break;
+        }
+        if pos > chunk_start && (msg_end - chunk_start) > threshold {
+            chunks.push(chunk_start..pos);
+            chunk_start = pos;
+        }
+        pos = msg_end;
+    }
+    if chunk_start < pos {
+        chunks.push(chunk_start..pos);
+    }
+    chunks
+}
+
 fn classify_stream_error(err: &str) -> &'static str {
     if err.contains("snd_pcm_poll_descriptors") || err.contains("alsa::poll() returned POLLERR") {
         "alsa_poll_failure"
@@ -3392,9 +3427,11 @@ async fn handle_audio_client(
 
     let history_replay_started_at = Instant::now();
 
-    // Serialize the entire history into a single in-memory blob so we can
+    // Build one in-memory blob of all history records, then split it into
-    // send it with one write_all + one flush, avoiding repeated partial
+    // chunks at message boundaries before gzipping. A single oversized blob
-    // flushes that dominate replay time for large histories.
+    // can exceed the 16 MiB MAX_HISTORY_PAYLOAD_SIZE the client enforces;
+    // chunking keeps each frame well under the limit while still amortising
+    // the gzip cost across many records per chunk.
     let history_blob = {
         // Estimate ~256 bytes per message; pre-allocate to avoid repeated
         // reallocation for large histories.
@@ -3478,24 +3515,34 @@ async fn handle_audio_client(
     };
     let (blob, replayed_history_count) = history_blob;
     if !blob.is_empty() {
-        // Gzip-compress the blob before sending. JSON history compresses very
+        // Target uncompressed bytes per chunk. JSON compresses ~10-20x so
-        // well (~10-20x) so this dramatically reduces both transfer size and
+        // 8 MiB uncompressed lands well under the 16 MiB compressed cap even
-        // the time the client spends waiting for data.
+        // in adversarial cases (image-heavy LRPT/WEFAX history, low ratio).
-        let compressed = {
+        const CHUNK_THRESHOLD_BYTES: usize = 8 * 1024 * 1024;
-            let mut enc = GzEncoder::new(Vec::with_capacity(blob.len() / 8), Compression::fast());
+
-            enc.write_all(&blob)
+        let mut total_compressed = 0usize;
+        let mut chunks_sent = 0usize;
+        for range in split_history_chunks(&blob, CHUNK_THRESHOLD_BYTES) {
+            let slice = &blob[range];
+            let mut enc = GzEncoder::new(Vec::with_capacity(slice.len() / 8), Compression::fast());
+            let compressed = enc
+                .write_all(slice)
                 .and_then(|_| enc.finish())
-                .unwrap_or(blob.clone())
+                .unwrap_or_else(|_| slice.to_vec());
-        };
+            write_audio_msg(&mut writer, AUDIO_MSG_HISTORY_COMPRESSED, &compressed).await?;
-        write_audio_msg(&mut writer, AUDIO_MSG_HISTORY_COMPRESSED, &compressed).await?;
+            total_compressed += compressed.len();
+            chunks_sent += 1;
+        }
+
         info!(
-            "Audio client {} replayed {} history messages in {:?} ({} → {} bytes, {:.1}x)",
+            "Audio client {} replayed {} history messages in {} chunk(s) in {:?} ({} → {} bytes, {:.1}x)",
             peer,
             replayed_history_count,
+            chunks_sent,
             history_replay_started_at.elapsed(),
             blob.len(),
-            compressed.len(),
+            total_compressed,
-            blob.len() as f64 / compressed.len().max(1) as f64,
+            blob.len() as f64 / total_compressed.max(1) as f64,
         );
     }
 
@@ -4014,3 +4061,282 @@ async fn handle_audio_client(
     rx_handle.abort();
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn frame(msg_type: u8, payload: &[u8]) -> Vec<u8> {
+        let mut out = Vec::with_capacity(5 + payload.len());
+        out.push(msg_type);
+        out.extend_from_slice(&(payload.len() as u32).to_be_bytes());
+        out.extend_from_slice(payload);
+        out
+    }
+
+    #[test]
+    fn split_history_chunks_empty_blob_yields_no_chunks() {
+        assert!(split_history_chunks(&[], 1024).is_empty());
+    }
+
+    #[test]
+    fn split_history_chunks_single_record_under_threshold_one_chunk() {
+        let blob = frame(AUDIO_MSG_FT8_DECODE, &[0u8; 100]);
+        let chunks = split_history_chunks(&blob, 1024);
+        assert_eq!(chunks, vec![0..blob.len()]);
+    }
+
+    #[test]
+    fn split_history_chunks_multiple_records_one_chunk_when_under() {
+        let mut blob = Vec::new();
+        blob.extend(frame(AUDIO_MSG_FT8_DECODE, &[0u8; 100]));
+        blob.extend(frame(AUDIO_MSG_FT4_DECODE, &[0u8; 100]));
+        blob.extend(frame(AUDIO_MSG_AIS_DECODE, &[0u8; 100]));
+        let chunks = split_history_chunks(&blob, 1024);
+        assert_eq!(chunks, vec![0..blob.len()]);
+    }
+
+    #[test]
+    fn split_history_chunks_splits_at_record_boundaries() {
+        // Three 100-byte payloads (105 bytes per record). Threshold of 200
+        // means: record 1 fits alone; adding record 2 stays at 210 (over) so
+        // chunk 1 contains only record 1; record 2 starts chunk 2; record 3
+        // pushes chunk 2 to 210 (over) so chunk 2 contains only record 2; etc.
+        let r1 = frame(AUDIO_MSG_FT8_DECODE, &[0u8; 100]);
+        let r2 = frame(AUDIO_MSG_FT4_DECODE, &[0u8; 100]);
+        let r3 = frame(AUDIO_MSG_AIS_DECODE, &[0u8; 100]);
+        let mut blob = Vec::new();
+        blob.extend_from_slice(&r1);
+        blob.extend_from_slice(&r2);
+        blob.extend_from_slice(&r3);
+        let chunks = split_history_chunks(&blob, 200);
+        assert_eq!(
+            chunks,
+            vec![
+                0..r1.len(),
+                r1.len()..r1.len() + r2.len(),
+                r1.len() + r2.len()..blob.len(),
+            ]
+        );
+    }
+
+    #[test]
+    fn split_history_chunks_oversize_record_alone() {
+        // Record exceeds threshold; it still appears as its own chunk because
+        // we never split a single record across frames.
+        let blob = frame(AUDIO_MSG_LRPT_IMAGE, &[0u8; 4096]);
+        let chunks = split_history_chunks(&blob, 1024);
+        assert_eq!(chunks, vec![0..blob.len()]);
+    }
+
+    #[test]
+    fn split_history_chunks_drops_truncated_header() {
+        // Tail has only 3 bytes of a 5-byte header — must be dropped.
+        let mut blob = frame(AUDIO_MSG_FT8_DECODE, &[1u8; 50]);
+        let valid_end = blob.len();
+        blob.extend_from_slice(&[0xAA, 0xBB, 0xCC]);
+        let chunks = split_history_chunks(&blob, 1024);
+        assert_eq!(chunks, vec![0..valid_end]);
+    }
+
+    #[test]
+    fn split_history_chunks_drops_truncated_payload() {
+        // Header claims 200 bytes but only 50 follow — drop it.
+        let mut blob = frame(AUDIO_MSG_FT8_DECODE, &[1u8; 50]);
+        let valid_end = blob.len();
+        blob.push(AUDIO_MSG_AIS_DECODE);
+        blob.extend_from_slice(&200u32.to_be_bytes());
+        blob.extend_from_slice(&[0u8; 50]);
+        let chunks = split_history_chunks(&blob, 1024);
+        assert_eq!(chunks, vec![0..valid_end]);
+    }
+
+    #[test]
+    fn split_history_chunks_threshold_exact_keeps_record_in_chunk() {
+        // Two records of exactly 100 bytes each (total 105 framed). Threshold
+        // == record-end => `>` is false, so the second record stays in chunk 1.
+        let r1 = frame(AUDIO_MSG_FT8_DECODE, &[0u8; 95]);
+        assert_eq!(r1.len(), 100);
+        let mut blob = Vec::new();
+        blob.extend_from_slice(&r1);
+        blob.extend_from_slice(&r1);
+        let chunks = split_history_chunks(&blob, 200);
+        assert_eq!(chunks, vec![0..200]);
+    }
+
+    #[test]
+    fn classify_stream_error_categorises_known_strings() {
+        assert_eq!(
+            classify_stream_error("snd_pcm_poll_descriptors revents error"),
+            "alsa_poll_failure"
+        );
+        assert_eq!(
+            classify_stream_error("alsa::poll() returned POLLERR"),
+            "alsa_poll_failure"
+        );
+        assert_eq!(
+            classify_stream_error("Input stream closed"),
+            "input_stream_error"
+        );
+        assert_eq!(
+            classify_stream_error("Output stream xrun"),
+            "output_stream_error"
+        );
+        assert_eq!(
+            classify_stream_error("device disappeared"),
+            "other_stream_error"
+        );
+        assert_eq!(classify_stream_error(""), "other_stream_error");
+    }
+
+    #[test]
+    fn enforce_capacity_keeps_newest_when_over() {
+        let mut q: VecDeque<i32> = (0..20).collect();
+        enforce_capacity(&mut q, 5);
+        assert_eq!(q.len(), 5);
+        assert_eq!(q.front().copied(), Some(15));
+        assert_eq!(q.back().copied(), Some(19));
+    }
+
+    #[test]
+    fn enforce_capacity_no_op_when_under_or_equal() {
+        let mut q: VecDeque<i32> = (0..3).collect();
+        enforce_capacity(&mut q, 5);
+        assert_eq!(q.len(), 3);
+        let mut q: VecDeque<i32> = (0..5).collect();
+        enforce_capacity(&mut q, 5);
+        assert_eq!(q.len(), 5);
+    }
+
+    #[test]
+    fn enforce_capacity_handles_zero_max() {
+        let mut q: VecDeque<i32> = (0..5).collect();
+        enforce_capacity(&mut q, 0);
+        assert!(q.is_empty());
+    }
+
+    #[test]
+    fn downmix_if_needed_passthrough_for_mono() {
+        let v = vec![1.0, 2.0, 3.0];
+        assert_eq!(downmix_if_needed(v.clone(), 1), v);
+        // channels == 0 also passes through (defensive)
+        assert_eq!(downmix_if_needed(v.clone(), 0), v);
+    }
+
+    #[test]
+    fn downmix_if_needed_takes_first_channel_for_stereo() {
+        // Interleaved L/R: result keeps L only.
+        let v = vec![1.0, 9.0, 2.0, 9.0, 3.0, 9.0];
+        assert_eq!(downmix_if_needed(v, 2), vec![1.0, 2.0, 3.0]);
+    }
+
+    #[test]
+    fn downmix_mono_matches_downmix_if_needed() {
+        let v = vec![1.0, 9.0, 2.0, 9.0, 3.0, 9.0];
+        assert_eq!(downmix_mono(v.clone(), 2), downmix_if_needed(v, 2));
+    }
+
+    #[test]
+    fn frame_rms_zero_for_empty_or_silent() {
+        assert_eq!(frame_rms(&[]), 0.0);
+        assert_eq!(frame_rms(&[0.0; 16]), 0.0);
+    }
+
+    #[test]
+    fn frame_rms_dc_signal_equals_amplitude() {
+        let r = frame_rms(&[0.5; 32]);
+        assert!((r - 0.5).abs() < 1e-6, "rms={}", r);
+    }
+
+    #[test]
+    fn frame_rms_sine_equals_amp_over_sqrt2() {
+        // 1024-sample full-cycle sine at amplitude 1.0 — expected RMS ≈ 0.707
+        let n = 1024;
+        let samples: Vec<f32> = (0..n)
+            .map(|i| (i as f32 * std::f32::consts::TAU / n as f32).sin())
+            .collect();
+        let r = frame_rms(&samples);
+        assert!(
+            (r - std::f32::consts::FRAC_1_SQRT_2).abs() < 1e-3,
+            "rms={}",
+            r
+        );
+    }
+
+    #[test]
+    fn apply_decode_audio_gate_passes_loud_audio() {
+        let mut s = vec![0.5; 64];
+        let muted = apply_decode_audio_gate(&mut s);
+        assert!(!muted);
+        assert_eq!(s, vec![0.5; 64]);
+    }
+
+    #[test]
+    fn apply_decode_audio_gate_mutes_quiet_audio() {
+        let mut s = vec![1e-6; 64];
+        let muted = apply_decode_audio_gate(&mut s);
+        assert!(muted);
+        assert!(s.iter().all(|&x| x == 0.0));
+    }
+
+    #[test]
+    fn resample_to_12k_passthrough_when_already_12k() {
+        let s = vec![1.0, 2.0, 3.0];
+        assert_eq!(resample_to_12k(&s, FT8_SAMPLE_RATE), Some(s));
+    }
+
+    #[test]
+    fn resample_to_12k_decimates_48k_to_12k_with_averaging() {
+        // 4:1 decimation should average each group of 4 samples.
+        let input = vec![1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0];
+        let out = resample_to_12k(&input, 48_000).expect("supported rate");
+        assert_eq!(out, vec![2.5, 6.5]);
+    }
+
+    #[test]
+    fn resample_to_12k_rejects_non_multiple_rate() {
+        // 11_025 Hz is not a multiple of 12_000.
+        assert!(resample_to_12k(&[0.0; 1024], 11_025).is_none());
+    }
+
+    #[test]
+    fn opus_channels_maps_supported_counts() {
+        assert!(matches!(opus_channels(1), Ok(opus::Channels::Mono)));
+        assert!(matches!(opus_channels(2), Ok(opus::Channels::Stereo)));
+        assert!(opus_channels(0).is_err());
+        assert!(opus_channels(3).is_err());
+    }
+
+    #[cfg(feature = "ft2")]
+    #[test]
+    fn retain_ft2_window_drops_oldest_excess() {
+        let mut buf: Vec<f32> = (0..(FT2_ASYNC_BUFFER_SAMPLES + 100) as i32)
+            .map(|i| i as f32)
+            .collect();
+        retain_ft2_window(&mut buf);
+        assert_eq!(buf.len(), FT2_ASYNC_BUFFER_SAMPLES);
+        // Oldest 100 should have been drained.
+        assert!((buf[0] - 100.0).abs() < f32::EPSILON);
+    }
+
+    #[cfg(feature = "ft2")]
+    #[test]
+    fn retain_ft2_window_no_op_when_within_size() {
+        let mut buf: Vec<f32> = vec![1.0; FT2_ASYNC_BUFFER_SAMPLES];
+        retain_ft2_window(&mut buf);
+        assert_eq!(buf.len(), FT2_ASYNC_BUFFER_SAMPLES);
+    }
+
+    #[cfg(feature = "ft2")]
+    #[test]
+    fn should_emit_ft2_decode_dedupes_and_freq_round() {
+        let mut recent = HashMap::new();
+        assert!(should_emit_ft2_decode(&mut recent, "CQ DX", 14_074_000.4));
+        // Same text + same rounded freq => suppressed.
+        assert!(!should_emit_ft2_decode(&mut recent, "CQ DX", 14_074_000.4));
+        // Different freq (rounded) => emitted.
+        assert!(should_emit_ft2_decode(&mut recent, "CQ DX", 14_074_001.4));
+        // Different text => emitted.
+        assert!(should_emit_ft2_decode(&mut recent, "CQ ZZ", 14_074_000.4));
+    }
+}

src/trx-server/src/main.rs

@@ -1228,3 +1228,96 @@ async fn main() -> DynResult<()> {
     // (e.g. SoapySDR/USB transfers in D-state) cannot prevent shutdown.
     std::process::exit(0);
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use trx_core::rig::state::RigMode;
+
+    #[test]
+    fn parse_serial_addr_accepts_path_and_baud() {
+        let (path, baud) = parse_serial_addr("/dev/ttyUSB0 9600").unwrap();
+        assert_eq!(path, "/dev/ttyUSB0");
+        assert_eq!(baud, 9600);
+    }
+
+    #[test]
+    fn parse_serial_addr_collapses_whitespace() {
+        let (path, baud) = parse_serial_addr("/dev/ttyUSB0    38400").unwrap();
+        assert_eq!(path, "/dev/ttyUSB0");
+        assert_eq!(baud, 38400);
+    }
+
+    #[test]
+    fn parse_serial_addr_rejects_missing_baud() {
+        assert!(parse_serial_addr("/dev/ttyUSB0").is_err());
+    }
+
+    #[test]
+    fn parse_serial_addr_rejects_extra_tokens() {
+        assert!(parse_serial_addr("/dev/ttyUSB0 9600 extra").is_err());
+    }
+
+    #[test]
+    fn parse_serial_addr_rejects_non_numeric_baud() {
+        assert!(parse_serial_addr("/dev/ttyUSB0 fast").is_err());
+    }
+
+    #[test]
+    fn parse_serial_addr_rejects_empty_input() {
+        assert!(parse_serial_addr("").is_err());
+        assert!(parse_serial_addr("   ").is_err());
+    }
+
+    #[test]
+    fn default_audio_bandwidth_for_mode_table() {
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::USB), 3_000);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::LSB), 3_000);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::DIG), 3_000);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::PKT), 25_000);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::CW), 500);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::CWR), 500);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::AM), 9_000);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::SAM), 9_000);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::FM), 12_500);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::WFM), 180_000);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::AIS), 25_000);
+        assert_eq!(default_audio_bandwidth_for_mode(&RigMode::VDES), 100_000);
+        assert_eq!(
+            default_audio_bandwidth_for_mode(&RigMode::Other("X".into())),
+            3_000
+        );
+    }
+
+    #[cfg(feature = "soapysdr")]
+    #[test]
+    fn parse_rig_mode_recognises_known_tokens() {
+        let initial = RigMode::USB;
+        for (s, expected) in [
+            ("LSB", RigMode::LSB),
+            ("USB", RigMode::USB),
+            ("CW", RigMode::CW),
+            ("CWR", RigMode::CWR),
+            ("AM", RigMode::AM),
+            ("FM", RigMode::FM),
+            ("WFM", RigMode::WFM),
+            ("AIS", RigMode::AIS),
+            ("VDES", RigMode::VDES),
+            ("DIG", RigMode::DIG),
+            ("PKT", RigMode::PKT),
+        ] {
+            assert_eq!(parse_rig_mode(s, &initial), expected, "token={}", s);
+        }
+    }
+
+    #[cfg(feature = "soapysdr")]
+    #[test]
+    fn parse_rig_mode_falls_back_on_unknown_or_auto() {
+        let initial = RigMode::FM;
+        assert_eq!(parse_rig_mode("auto", &initial), RigMode::FM);
+        assert_eq!(parse_rig_mode("nonsense", &initial), RigMode::FM);
+        assert_eq!(parse_rig_mode("", &initial), RigMode::FM);
+        // Case-sensitive: lowercase "usb" doesn't match — falls back.
+        assert_eq!(parse_rig_mode("usb", &initial), RigMode::FM);
+    }
+}