From 062192bc1289419a0dd6b9ef775a083a5d8f3dfa Mon Sep 17 00:00:00 2001
From: Matthew Hounslow <hounslowmatthew@gmail.com>
Date: Tue, 23 Dec 2025 16:30:12 -0700
Subject: [PATCH 1/6] Add comprehensive tests for buswatch-types module

- Add extensive tests for Microseconds (conversions, ordering, edge cases)
- Add tests for ReadMetrics and WriteMetrics (builders, health checks, edge cases)
- Add tests for ModuleMetrics (multiple topics, unicode, special chars)
- Add tests for SchemaVersion (compatibility, const functions)
- Add tests for Snapshot (builders, iteration, serialization roundtrips)
- Test coverage for serde JSON and minicbor CBOR serialization
---
 buswatch-types/src/duration.rs | 100 +++++++++
 buswatch-types/src/metrics.rs  | 362 +++++++++++++++++++++++++++++++++
 buswatch-types/src/snapshot.rs | 220 ++++++++++++++++++++
 buswatch-types/src/version.rs  |  99 +++++++++
 4 files changed, 781 insertions(+)

diff --git a/buswatch-types/src/duration.rs b/buswatch-types/src/duration.rs
index 068f5d7..b748bd2 100644
--- a/buswatch-types/src/duration.rs
+++ b/buswatch-types/src/duration.rs
@@ -81,4 +81,104 @@ mod tests {
         let d2: Duration = m.into();
         assert_eq!(d, d2);
     }
+
+    #[test]
+    fn from_micros() {
+        let m = Microseconds::from_micros(1_500_000);
+        assert_eq!(m.as_micros(), 1_500_000);
+        assert_eq!(m.as_millis(), 1500);
+        assert_eq!(m.as_secs(), 1);
+    }
+
+    #[test]
+    fn from_millis() {
+        let m = Microseconds::from_millis(2500);
+        assert_eq!(m.as_micros(), 2_500_000);
+        assert_eq!(m.as_millis(), 2500);
+        assert_eq!(m.as_secs(), 2);
+    }
+
+    #[test]
+    fn from_secs() {
+        let m = Microseconds::from_secs(5);
+        assert_eq!(m.as_micros(), 5_000_000);
+        assert_eq!(m.as_millis(), 5000);
+        assert_eq!(m.as_secs(), 5);
+    }
+
+    #[test]
+    fn to_duration() {
+        let m = Microseconds::from_millis(567);
+        let d = m.to_duration();
+        assert_eq!(d, Duration::from_millis(567));
+    }
+
+    #[test]
+    fn default_is_zero() {
+        let m = Microseconds::default();
+        assert_eq!(m.as_micros(), 0);
+        assert_eq!(m.to_duration(), Duration::ZERO);
+    }
+
+    #[test]
+    fn ordering() {
+        let a = Microseconds::from_millis(100);
+        let b = Microseconds::from_millis(200);
+        let c = Microseconds::from_millis(100);
+
+        assert!(a < b);
+        assert!(b > a);
+        assert_eq!(a, c);
+        assert!(a <= c);
+        assert!(a >= c);
+    }
+
+    #[test]
+    fn truncation_behavior() {
+        // 1,500,999 microseconds should truncate to 1500 millis and 1 sec
+        let m = Microseconds::from_micros(1_500_999);
+        assert_eq!(m.as_millis(), 1500); // truncated, not rounded
+        assert_eq!(m.as_secs(), 1); // truncated
+    }
+
+    #[test]
+    fn large_values() {
+        // Test with very large values (days worth of microseconds)
+        let days_in_micros = 86_400_000_000u64 * 30; // 30 days
+        let m = Microseconds::from_micros(days_in_micros);
+        assert_eq!(m.as_secs(), 86_400 * 30);
+    }
+
+    #[test]
+    fn zero_values() {
+        let m = Microseconds::from_micros(0);
+        assert_eq!(m.as_micros(), 0);
+        assert_eq!(m.as_millis(), 0);
+        assert_eq!(m.as_secs(), 0);
+    }
+
+    #[test]
+    fn hash_impl() {
+        use std::collections::HashSet;
+        let mut set = HashSet::new();
+        set.insert(Microseconds::from_secs(1));
+        set.insert(Microseconds::from_secs(2));
+        set.insert(Microseconds::from_secs(1)); // duplicate
+        assert_eq!(set.len(), 2);
+    }
+
+    #[test]
+    fn copy_semantics() {
+        let m1 = Microseconds::from_secs(5);
+        let m2 = m1; // Copy
+        assert_eq!(m1, m2);
+        assert_eq!(m1.as_secs(), 5); // m1 still usable
+    }
+
+    #[test]
+    fn debug_format() {
+        let m = Microseconds::from_millis(1500);
+        let debug = format!("{:?}", m);
+        assert!(debug.contains("1500000"));
+    }
 }
diff --git a/buswatch-types/src/metrics.rs b/buswatch-types/src/metrics.rs
index 1a608f3..06652c1 100644
--- a/buswatch-types/src/metrics.rs
+++ b/buswatch-types/src/metrics.rs
@@ -289,6 +289,215 @@ mod tests {
     use super::*;
     use core::time::Duration;
 
+    // ========================================================================
+    // ReadMetrics Tests
+    // ========================================================================
+
+    #[test]
+    fn read_metrics_new_with_count() {
+        let r = ReadMetrics::new(100);
+        assert_eq!(r.count, 100);
+        assert!(r.backlog.is_none());
+        assert!(r.pending.is_none());
+        assert!(r.rate.is_none());
+    }
+
+    #[test]
+    fn read_metrics_builder_all_fields() {
+        let r = ReadMetrics::builder()
+            .count(500)
+            .backlog(50)
+            .pending(Duration::from_secs(2))
+            .rate(25.5)
+            .build();
+
+        assert_eq!(r.count, 500);
+        assert_eq!(r.backlog, Some(50));
+        assert_eq!(r.pending, Some(Microseconds::from_secs(2)));
+        assert_eq!(r.rate, Some(25.5));
+    }
+
+    #[test]
+    fn read_metrics_builder_partial_fields() {
+        let r = ReadMetrics::builder().count(100).backlog(10).build();
+
+        assert_eq!(r.count, 100);
+        assert_eq!(r.backlog, Some(10));
+        assert!(r.pending.is_none());
+        assert!(r.rate.is_none());
+    }
+
+    #[test]
+    fn read_metrics_builder_default_count() {
+        let r = ReadMetrics::builder().build();
+        assert_eq!(r.count, 0);
+    }
+
+    #[test]
+    fn read_metrics_is_healthy_with_no_issues() {
+        let r = ReadMetrics::new(100);
+        assert!(r.is_healthy(10, Microseconds::from_secs(5)));
+    }
+
+    #[test]
+    fn read_metrics_is_healthy_backlog_at_threshold() {
+        let r = ReadMetrics::builder().count(100).backlog(10).build();
+        assert!(r.is_healthy(10, Microseconds::from_secs(5))); // equal is healthy
+    }
+
+    #[test]
+    fn read_metrics_is_unhealthy_backlog_exceeds_threshold() {
+        let r = ReadMetrics::builder().count(100).backlog(11).build();
+        assert!(!r.is_healthy(10, Microseconds::from_secs(5)));
+    }
+
+    #[test]
+    fn read_metrics_is_healthy_pending_at_threshold() {
+        let r = ReadMetrics::builder()
+            .count(100)
+            .pending(Duration::from_secs(5))
+            .build();
+        assert!(r.is_healthy(10, Microseconds::from_secs(5))); // equal is healthy
+    }
+
+    #[test]
+    fn read_metrics_is_unhealthy_pending_exceeds_threshold() {
+        let r = ReadMetrics::builder()
+            .count(100)
+            .pending(Duration::from_secs(6))
+            .build();
+        assert!(!r.is_healthy(10, Microseconds::from_secs(5)));
+    }
+
+    #[test]
+    fn read_metrics_is_unhealthy_both_exceed() {
+        let r = ReadMetrics::builder()
+            .count(100)
+            .backlog(20)
+            .pending(Duration::from_secs(10))
+            .build();
+        assert!(!r.is_healthy(10, Microseconds::from_secs(5)));
+    }
+
+    #[test]
+    fn read_metrics_default_is_empty() {
+        let r = ReadMetrics::default();
+        assert_eq!(r.count, 0);
+        assert!(r.backlog.is_none());
+        assert!(r.pending.is_none());
+        assert!(r.rate.is_none());
+    }
+
+    #[test]
+    fn read_metrics_clone_and_equality() {
+        let r1 = ReadMetrics::builder().count(100).backlog(5).build();
+        let r2 = r1.clone();
+        assert_eq!(r1, r2);
+    }
+
+    #[test]
+    fn read_metrics_max_values() {
+        let r = ReadMetrics::builder()
+            .count(u64::MAX)
+            .backlog(u64::MAX)
+            .build();
+        assert_eq!(r.count, u64::MAX);
+        assert_eq!(r.backlog, Some(u64::MAX));
+    }
+
+    #[test]
+    fn read_metrics_special_rate_values() {
+        // Negative rate (counts could theoretically decrease)
+        let r = ReadMetrics::builder().rate(-10.0).build();
+        assert_eq!(r.rate, Some(-10.0));
+
+        // NaN rate
+        let r = ReadMetrics::builder().rate(f64::NAN).build();
+        assert!(r.rate.unwrap().is_nan());
+
+        // Infinity rate
+        let r = ReadMetrics::builder().rate(f64::INFINITY).build();
+        assert!(r.rate.unwrap().is_infinite());
+    }
+
+    // ========================================================================
+    // WriteMetrics Tests
+    // ========================================================================
+
+    #[test]
+    fn write_metrics_new_with_count() {
+        let w = WriteMetrics::new(200);
+        assert_eq!(w.count, 200);
+        assert!(w.pending.is_none());
+        assert!(w.rate.is_none());
+    }
+
+    #[test]
+    fn write_metrics_builder_all_fields() {
+        let w = WriteMetrics::builder()
+            .count(1000)
+            .pending(Duration::from_millis(500))
+            .rate(100.0)
+            .build();
+
+        assert_eq!(w.count, 1000);
+        assert_eq!(w.pending, Some(Microseconds::from_millis(500)));
+        assert_eq!(w.rate, Some(100.0));
+    }
+
+    #[test]
+    fn write_metrics_builder_partial_fields() {
+        let w = WriteMetrics::builder().count(50).build();
+
+        assert_eq!(w.count, 50);
+        assert!(w.pending.is_none());
+        assert!(w.rate.is_none());
+    }
+
+    #[test]
+    fn write_metrics_is_healthy_with_no_pending() {
+        let w = WriteMetrics::new(100);
+        assert!(w.is_healthy(Microseconds::from_secs(5)));
+    }
+
+    #[test]
+    fn write_metrics_is_healthy_pending_at_threshold() {
+        let w = WriteMetrics::builder()
+            .count(100)
+            .pending(Duration::from_secs(5))
+            .build();
+        assert!(w.is_healthy(Microseconds::from_secs(5)));
+    }
+
+    #[test]
+    fn write_metrics_is_unhealthy_pending_exceeds_threshold() {
+        let w = WriteMetrics::builder()
+            .count(100)
+            .pending(Duration::from_secs(6))
+            .build();
+        assert!(!w.is_healthy(Microseconds::from_secs(5)));
+    }
+
+    #[test]
+    fn write_metrics_default_is_empty() {
+        let w = WriteMetrics::default();
+        assert_eq!(w.count, 0);
+        assert!(w.pending.is_none());
+        assert!(w.rate.is_none());
+    }
+
+    // ========================================================================
+    // ModuleMetrics Tests
+    // ========================================================================
+
+    #[test]
+    fn module_metrics_new_is_empty() {
+        let m = ModuleMetrics::new();
+        assert!(m.is_empty());
+        assert_eq!(m.total_reads(), 0);
+        assert_eq!(m.total_writes(), 0);
+    }
+
     #[test]
     fn test_module_metrics_builder() {
         let metrics = ModuleMetrics::builder()
@@ -301,6 +510,134 @@ mod tests {
         assert_eq!(metrics.reads.get("input").unwrap().backlog, Some(5));
     }
 
+    #[test]
+    fn module_metrics_builder_with_multiple_topics() {
+        let m = ModuleMetrics::builder()
+            .read("input-a", |r| r.count(100).backlog(5))
+            .read("input-b", |r| r.count(200))
+            .write("output-a", |w| w.count(150))
+            .write("output-b", |w| w.count(140))
+            .build();
+
+        assert!(!m.is_empty());
+        assert_eq!(m.total_reads(), 300);
+        assert_eq!(m.total_writes(), 290);
+        assert_eq!(m.reads.len(), 2);
+        assert_eq!(m.writes.len(), 2);
+    }
+
+    #[test]
+    fn module_metrics_builder_only_reads() {
+        let m = ModuleMetrics::builder()
+            .read("events", |r| r.count(500))
+            .build();
+
+        assert!(!m.is_empty());
+        assert_eq!(m.total_reads(), 500);
+        assert_eq!(m.total_writes(), 0);
+    }
+
+    #[test]
+    fn module_metrics_builder_only_writes() {
+        let m = ModuleMetrics::builder()
+            .write("events", |w| w.count(1000))
+            .build();
+
+        assert!(!m.is_empty());
+        assert_eq!(m.total_reads(), 0);
+        assert_eq!(m.total_writes(), 1000);
+    }
+
+    #[test]
+    fn module_metrics_is_empty_with_empty_collections() {
+        let m = ModuleMetrics::builder().build();
+        assert!(m.is_empty());
+    }
+
+    #[test]
+    fn module_metrics_access_individual_topics() {
+        let m = ModuleMetrics::builder()
+            .read("orders", |r| r.count(42).backlog(3))
+            .write("notifications", |w| w.count(40).rate(5.0))
+            .build();
+
+        let orders = m.reads.get("orders").unwrap();
+        assert_eq!(orders.count, 42);
+        assert_eq!(orders.backlog, Some(3));
+
+        let notifications = m.writes.get("notifications").unwrap();
+        assert_eq!(notifications.count, 40);
+        assert_eq!(notifications.rate, Some(5.0));
+    }
+
+    #[test]
+    fn module_metrics_default_is_empty() {
+        let m = ModuleMetrics::default();
+        assert!(m.is_empty());
+    }
+
+    #[test]
+    fn module_metrics_duplicate_topic_overwrites() {
+        let m = ModuleMetrics::builder()
+            .read("topic", |r| r.count(100))
+            .read("topic", |r| r.count(200)) // Same topic, should overwrite
+            .build();
+
+        assert_eq!(m.reads.len(), 1);
+        assert_eq!(m.reads.get("topic").unwrap().count, 200);
+    }
+
+    #[test]
+    fn module_metrics_unicode_topic_names() {
+        let m = ModuleMetrics::builder()
+            .read("订单", |r| r.count(100))
+            .read("événements", |r| r.count(200))
+            .read("イベント", |r| r.count(300))
+            .build();
+
+        assert_eq!(m.reads.get("订单").unwrap().count, 100);
+        assert_eq!(m.reads.get("événements").unwrap().count, 200);
+        assert_eq!(m.reads.get("イベント").unwrap().count, 300);
+        assert_eq!(m.total_reads(), 600);
+    }
+
+    #[test]
+    fn module_metrics_special_characters_in_names() {
+        let m = ModuleMetrics::builder()
+            .read("topic/with/slashes", |r| r.count(1))
+            .read("topic.with.dots", |r| r.count(2))
+            .read("topic:with:colons", |r| r.count(3))
+            .read("topic-with-dashes", |r| r.count(4))
+            .read("topic_with_underscores", |r| r.count(5))
+            .build();
+
+        assert_eq!(m.total_reads(), 15);
+    }
+
+    #[test]
+    fn module_metrics_empty_topic_name() {
+        let m = ModuleMetrics::builder()
+            .read("", |r| r.count(100))
+            .write("", |w| w.count(50))
+            .build();
+
+        assert_eq!(m.reads.get("").unwrap().count, 100);
+        assert_eq!(m.writes.get("").unwrap().count, 50);
+    }
+
+    #[test]
+    fn module_metrics_many_topics() {
+        let mut m_builder = ModuleMetrics::builder();
+        for i in 0..100 {
+            m_builder = m_builder.read(alloc::format!("read-topic-{}", i), |r| r.count(i as u64));
+            m_builder = m_builder.write(alloc::format!("write-topic-{}", i), |w| w.count(i as u64));
+        }
+        let m = m_builder.build();
+
+        assert_eq!(m.reads.len(), 100);
+        assert_eq!(m.writes.len(), 100);
+    }
+
     #[test]
     fn test_read_metrics_health() {
         let healthy = ReadMetrics::new(100);
@@ -315,4 +652,29 @@ mod tests {
             .build();
         assert!(!with_pending.is_healthy(10, Microseconds::from_secs(5)));
     }
+
+    // ========================================================================
+    // Builder Default Tests
+    // ========================================================================
+
+    #[test]
+    fn read_metrics_builder_default() {
+        let b = ReadMetricsBuilder::default();
+        let r = b.build();
+        assert_eq!(r.count, 0);
+    }
+
+    #[test]
+    fn write_metrics_builder_default() {
+        let b = WriteMetricsBuilder::default();
+        let w = b.build();
+        assert_eq!(w.count, 0);
+    }
+
+    #[test]
+    fn module_metrics_builder_default() {
+        let b = ModuleMetricsBuilder::default();
+        let m = b.build();
+        assert!(m.is_empty());
+    }
 }
diff --git a/buswatch-types/src/snapshot.rs b/buswatch-types/src/snapshot.rs
index 2b3787e..4d60c5d 100644
--- a/buswatch-types/src/snapshot.rs
+++ b/buswatch-types/src/snapshot.rs
@@ -210,6 +210,168 @@ mod tests {
         assert!(snapshot.version.is_compatible());
     }
 
+    #[test]
+    fn with_timestamp() {
+        let s = Snapshot::with_timestamp(1703160000000);
+        assert_eq!(s.timestamp_ms, 1703160000000);
+        assert!(s.is_empty());
+        assert!(s.version.is_compatible());
+    }
+
+    #[test]
+    fn builder_with_prebuilt_metrics() {
+        let metrics = ModuleMetrics::builder()
+            .read("input", |r| r.count(100))
+            .build();
+
+        let s = Snapshot::builder()
+            .timestamp_ms(1000)
+            .module_metrics("my-service", metrics)
+            .build();
+
+        assert_eq!(s.len(), 1);
+        assert!(s.get("my-service").is_some());
+    }
+
+    #[test]
+    fn get_module() {
+        let s = Snapshot::builder()
+            .module("test", |m| m.read("topic", |r| r.count(42)))
+            .build();
+
+        assert!(s.get("test").is_some());
+        assert!(s.get("nonexistent").is_none());
+
+        let m = s.get("test").unwrap();
+        assert_eq!(m.total_reads(), 42);
+    }
+
+    #[test]
+    fn iterate_modules() {
+        let s = Snapshot::builder()
+            .module("a", |m| m.read("t", |r| r.count(1)))
+            .module("b", |m| m.read("t", |r| r.count(2)))
+            .module("c", |m| m.read("t", |r| r.count(3)))
+            .build();
+
+        let names: Vec<_> = s.iter().map(|(name, _)| name.as_str()).collect();
+        // BTreeMap iterates in sorted order
+        assert_eq!(names, vec!["a", "b", "c"]);
+
+        let total: u64 = s.iter().map(|(_, m)| m.total_reads()).sum();
+        assert_eq!(total, 6);
+    }
+
+    #[test]
+    fn empty_snapshot() {
+        let s = Snapshot::builder().timestamp_ms(0).build();
+        assert!(s.is_empty());
+        assert_eq!(s.len(), 0);
+        assert_eq!(s.total_reads(), 0);
+        assert_eq!(s.total_writes(), 0);
+    }
+
+    #[test]
+    fn version_is_set() {
+        let s = Snapshot::builder().build();
+        assert_eq!(s.version.major, crate::SCHEMA_VERSION);
+    }
+
+    #[cfg(feature = "std")]
+    #[test]
+    fn new_has_current_timestamp() {
+        let before = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+
+        let s = Snapshot::new();
+
+        let after = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+
+        assert!(s.timestamp_ms >= before);
+        assert!(s.timestamp_ms <= after);
+    }
+
+    #[cfg(feature = "std")]
+    #[test]
+    fn default_uses_new() {
+        let before = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+
+        let s = Snapshot::default();
+
+        let after = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+
+        assert!(s.timestamp_ms >= before);
+        assert!(s.timestamp_ms <= after);
+    }
+
+    #[test]
+    fn zero_timestamp() {
+        let s = Snapshot::with_timestamp(0);
+        assert_eq!(s.timestamp_ms, 0);
+    }
+
+    #[test]
+    fn max_timestamp() {
+        let s = Snapshot::with_timestamp(u64::MAX);
+        assert_eq!(s.timestamp_ms, u64::MAX);
+    }
+
+    #[test]
+    fn many_modules() {
+        let mut builder = Snapshot::builder().timestamp_ms(0);
+        for i in 0..100 {
+            builder = builder.module(alloc::format!("module-{}", i), |m| {
+                m.read("topic", |r| r.count(i as u64))
+            });
+        }
+        let s = builder.build();
+
+        assert_eq!(s.len(), 100);
+        assert_eq!(s.total_reads(), (0..100u64).sum::<u64>());
+    }
+
+    #[test]
+    fn snapshot_builder_default() {
+        let b = SnapshotBuilder::default();
+        let s = b.build();
+        assert!(s.is_empty());
+    }
+
+    #[test]
+    fn clone_and_equality() {
+        let s1 = Snapshot::builder()
+            .timestamp_ms(1000)
+            .module("test", |m| m.read("t", |r| r.count(42)))
+            .build();
+        let s2 = s1.clone();
+        assert_eq!(s1, s2);
+    }
+
+    #[test]
+    fn debug_format() {
+        let s = Snapshot::builder()
+            .module("test", |m| m.read("t", |r| r.count(1)))
+            .build();
+        let debug = alloc::format!("{:?}", s);
+        assert!(debug.contains("Snapshot"));
+        assert!(debug.contains("test"));
+    }
+
+    // ========================================================================
+    // Serialization Tests
+    // ========================================================================
+
     #[cfg(feature = "serde")]
     #[test]
     fn test_serde_roundtrip() {
@@ -224,6 +386,64 @@ mod tests {
         assert_eq!(snapshot, parsed);
     }
 
+    #[cfg(feature = "serde")]
+    #[test]
+    fn serde_json_structure() {
+        let s = Snapshot::builder()
+            .timestamp_ms(1703160000000)
+            .module("test", |m| m.read("topic", |r| r.count(42)))
+            .build();
+
+        let json: serde_json::Value = serde_json::to_value(&s).unwrap();
+
+        assert!(json.get("version").is_some());
+        assert!(json.get("timestamp_ms").is_some());
+        assert!(json.get("modules").is_some());
+
+        let version = json.get("version").unwrap();
+        assert_eq!(version.get("major").unwrap(), crate::SCHEMA_VERSION);
+    }
+
+    #[cfg(feature = "serde")]
+    #[test]
+    fn serde_deserialize_from_external_json() {
+        let json = r#"{
+            "version": { "major": 1, "minor": 0 },
+            "timestamp_ms": 1703160000000,
+            "modules": {
+                "my-service": {
+                    "reads": {
+                        "input": { "count": 100, "backlog": 5 }
+                    },
+                    "writes": {
+                        "output": { "count": 95 }
+                    }
+                }
+            }
+        }"#;
+
+        let s: Snapshot = serde_json::from_str(json).unwrap();
+        assert_eq!(s.timestamp_ms, 1703160000000);
+        assert_eq!(s.len(), 1);
+
+        let service = s.get("my-service").unwrap();
+        assert_eq!(service.reads.get("input").unwrap().count, 100);
+        assert_eq!(service.reads.get("input").unwrap().backlog, Some(5));
+    }
+
+    #[cfg(feature = "serde")]
+    #[test]
+    fn serde_pretty_print() {
+        let s = Snapshot::builder()
+            .timestamp_ms(1000)
+            .module("test", |m| m.read("t", |r| r.count(1)))
+            .build();
+
+        let pretty = serde_json::to_string_pretty(&s).unwrap();
+        assert!(pretty.contains('\n'));
+        assert!(pretty.contains("  ")); // indentation
+    }
+
     #[cfg(feature = "minicbor")]
     #[test]
     fn test_minicbor_roundtrip() {
diff --git a/buswatch-types/src/version.rs b/buswatch-types/src/version.rs
index 6a69fbc..df30de8 100644
--- a/buswatch-types/src/version.rs
+++ b/buswatch-types/src/version.rs
@@ -45,3 +45,102 @@ impl Default for SchemaVersion {
         Self::current()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::SCHEMA_VERSION;
+
+    #[test]
+    fn current_version() {
+        let v = SchemaVersion::current();
+        assert_eq!(v.major, SCHEMA_VERSION);
+        assert_eq!(v.minor, 0);
+    }
+
+    #[test]
+    fn new_version() {
+        let v = SchemaVersion::new(2, 5);
+        assert_eq!(v.major, 2);
+        assert_eq!(v.minor, 5);
+    }
+
+    #[test]
+    fn is_compatible_same_major() {
+        let v = SchemaVersion::new(SCHEMA_VERSION, 10);
+        assert!(v.is_compatible());
+    }
+
+    #[test]
+    fn is_incompatible_different_major() {
+        let v = SchemaVersion::new(SCHEMA_VERSION + 1, 0);
+        assert!(!v.is_compatible());
+
+        // Test with major version 0 if current is > 0
+        if SCHEMA_VERSION > 0 {
+            let v2 = SchemaVersion::new(0, 0);
+            assert!(!v2.is_compatible());
+        }
+    }
+
+    #[test]
+    fn default_is_current() {
+        let v = SchemaVersion::default();
+        assert_eq!(v, SchemaVersion::current());
+        assert!(v.is_compatible());
+    }
+
+    #[test]
+    fn equality() {
+        let v1 = SchemaVersion::new(1, 0);
+        let v2 = SchemaVersion::new(1, 0);
+        let v3 = SchemaVersion::new(1, 1);
+        let v4 = SchemaVersion::new(2, 0);
+
+        assert_eq!(v1, v2);
+        assert_ne!(v1, v3);
+        assert_ne!(v1, v4);
+    }
+
+    #[test]
+    fn hash_impl() {
+        use std::collections::HashSet;
+        let mut set = HashSet::new();
+        set.insert(SchemaVersion::new(1, 0));
+        set.insert(SchemaVersion::new(1, 1));
+        set.insert(SchemaVersion::new(1, 0)); // duplicate
+        assert_eq!(set.len(), 2);
+    }
+
+    #[test]
+    fn copy_semantics() {
+        let v1 = SchemaVersion::new(1, 0);
+        let v2 = v1; // Copy
+        assert_eq!(v1, v2);
+        assert_eq!(v1.major, 1); // v1 still usable
+    }
+
+    #[test]
+    fn debug_format() {
+        let v = SchemaVersion::new(1, 5);
+        let debug = format!("{:?}", v);
+        assert!(debug.contains("SchemaVersion"));
+        assert!(debug.contains("1"));
+        assert!(debug.contains("5"));
+    }
+
+    #[test]
+    fn const_new() {
+        // Verify new is const
+        const V: SchemaVersion = SchemaVersion::new(1, 0);
+        assert_eq!(V.major, 1);
+        assert_eq!(V.minor, 0);
+    }
+
+    #[test]
+    fn const_current() {
+        // Verify current is const
+        const V: SchemaVersion = SchemaVersion::current();
+        assert_eq!(V.major, SCHEMA_VERSION);
+    }
+}

From 676407fda587d69c4c684efd88eb76a5e3d37baf Mon Sep 17 00:00:00 2001
From: Matthew Hounslow <hounslowmatthew@gmail.com>
Date: Tue, 23 Dec 2025 16:34:40 -0700
Subject: [PATCH 2/6] Add comprehensive tests for buswatch-sdk module

- Add tests for state management (Arc sharing, backlog computation)
- Add tests for ModuleHandle (pending guards, cloning, global counters)
- Add tests for Instrumentor (multi-module scenarios, builder patterns)
- Add thread safety test for concurrent metric increments
- Add tests verifying pending time measurement
---
 buswatch-sdk/src/handle.rs       | 111 +++++++++++++++++++
 buswatch-sdk/src/instrumentor.rs | 116 ++++++++++++++++++++
 buswatch-sdk/src/state.rs        | 181 +++++++++++++++++++++++++++++++
 3 files changed, 408 insertions(+)

diff --git a/buswatch-sdk/src/handle.rs b/buswatch-sdk/src/handle.rs
index d5ac3fd..b279dfd 100644
--- a/buswatch-sdk/src/handle.rs
+++ b/buswatch-sdk/src/handle.rs
@@ -202,4 +202,115 @@ mod tests {
         let state = handle.state.get_or_create_read("topic");
         assert!(state.pending_since.read().is_none());
     }
+
+    #[test]
+    fn record_write_updates_global_topic_counter() {
+        let global = Arc::new(GlobalState::default());
+        let state = global.register_module("producer");
+        let handle = ModuleHandle {
+            state,
+            global: global.clone(),
+            name: "producer".to_string(),
+        };
+
+        handle.record_write("events", 50);
+        handle.record_write("events", 25);
+
+        let counter = global.get_topic_write_counter("events");
+        assert_eq!(counter.load(std::sync::atomic::Ordering::Relaxed), 75);
+    }
+
+    #[test]
+    fn start_write_sets_pending_and_guard_clears_it() {
+        let handle = create_handle();
+
+        {
+            let _guard = handle.start_write("output");
+            let state = handle.state.get_or_create_write("output");
+            assert!(state.pending_since.read().is_some());
+        }
+
+        let state = handle.state.get_or_create_write("output");
+        assert!(state.pending_since.read().is_none());
+    }
+
+    #[test]
+    fn set_read_pending_directly() {
+        let handle = create_handle();
+
+        handle.set_read_pending("topic", Some(Instant::now()));
+        let state = handle.state.get_or_create_read("topic");
+        assert!(state.pending_since.read().is_some());
+
+        handle.set_read_pending("topic", None);
+        assert!(state.pending_since.read().is_none());
+    }
+
+    #[test]
+    fn set_write_pending_directly() {
+        let handle = create_handle();
+
+        handle.set_write_pending("topic", Some(Instant::now()));
+        let state = handle.state.get_or_create_write("topic");
+        assert!(state.pending_since.read().is_some());
+
+        handle.set_write_pending("topic", None);
+        assert!(state.pending_since.read().is_none());
+    }
+
+    #[test]
+    fn module_handle_is_clone() {
+        let handle1 = create_handle();
+        let handle2 = handle1.clone();
+
+        handle1.record_read("topic", 10);
+        handle2.record_read("topic", 5);
+
+        // Both should contribute to same state
+        let metrics = handle1.state.collect();
+        assert_eq!(metrics.reads.get("topic").unwrap().count, 15);
+    }
+
+    #[test]
+    fn name_returns_module_name() {
+        let handle = create_handle();
+        assert_eq!(handle.name(), "test");
+    }
+
+    #[test]
+    fn multiple_topics_independent() {
+        let handle = create_handle();
+
+        handle.record_read("topic-a", 10);
+        handle.record_read("topic-b", 20);
+        handle.record_write("topic-c", 30);
+
+        let metrics = handle.state.collect();
+        assert_eq!(metrics.reads.get("topic-a").unwrap().count, 10);
+        assert_eq!(metrics.reads.get("topic-b").unwrap().count, 20);
+        assert_eq!(metrics.writes.get("topic-c").unwrap().count, 30);
+    }
+
+    #[test]
+    fn pending_guard_measures_elapsed_time() {
+        let handle = create_handle();
+
+        let _guard = handle.start_read("topic");
+        std::thread::sleep(std::time::Duration::from_millis(10));
+
+        let metrics = handle.state.collect();
+        let pending = metrics.reads.get("topic").unwrap().pending;
+
+        assert!(pending.is_some());
+        // Should be at least 10ms (10000 microseconds)
+        assert!(pending.unwrap().as_micros() >= 10000);
+    }
+
+    #[test]
+    fn debug_format_shows_name() {
+        let handle = create_handle();
+        let debug = format!("{:?}", handle);
+        assert!(debug.contains("test"));
+        assert!(debug.contains("ModuleHandle"));
+    }
 }
diff --git a/buswatch-sdk/src/instrumentor.rs b/buswatch-sdk/src/instrumentor.rs
index 5a924d2..0475ee0 100644
--- a/buswatch-sdk/src/instrumentor.rs
+++ b/buswatch-sdk/src/instrumentor.rs
@@ -254,4 +254,120 @@ mod tests {
         assert_eq!(instrumentor.interval, Duration::from_millis(500));
         assert_eq!(instrumentor.outputs.len(), 1);
     }
+
+    #[test]
+    fn default_interval_is_one_second() {
+        let instrumentor = Instrumentor::new();
+        assert_eq!(instrumentor.interval, Duration::from_secs(1));
+    }
+
+    #[test]
+    fn default_has_no_outputs() {
+        let instrumentor = Instrumentor::new();
+        assert!(instrumentor.outputs.is_empty());
+    }
+
+    #[test]
+    fn builder_can_add_multiple_outputs() {
+        let instrumentor = Instrumentor::builder()
+            .output(Output::file("metrics1.json"))
+            .output(Output::file("metrics2.json"))
+            .output(Output::tcp("localhost:9090"))
+            .build();
+
+        assert_eq!(instrumentor.outputs.len(), 3);
+    }
+
+    #[test]
+    fn register_same_module_twice_returns_same_state() {
+        let instrumentor = Instrumentor::new();
+
+        let handle1 = instrumentor.register("service");
+        let handle2 = instrumentor.register("service");
+
+        handle1.record_read("topic", 10);
+        handle2.record_read("topic", 5);
+
+        let snapshot = instrumentor.collect();
+        let metrics = snapshot.modules.get("service").unwrap();
+        assert_eq!(metrics.reads.get("topic").unwrap().count, 15);
+    }
+
+    #[test]
+    fn collect_returns_snapshot_with_timestamp() {
+        let instrumentor = Instrumentor::new();
+        let _ = instrumentor.register("test");
+
+        let before = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+
+        let snapshot = instrumentor.collect();
+
+        let after = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+
+        assert!(snapshot.timestamp_ms >= before);
+        assert!(snapshot.timestamp_ms <= after);
+    }
+
+    #[test]
+    fn collect_includes_schema_version() {
+        let instrumentor = Instrumentor::new();
+        let snapshot = instrumentor.collect();
+        assert!(snapshot.version.is_compatible());
+    }
+
+    #[test]
+    fn instrumentor_default_same_as_new() {
+        let i1 = Instrumentor::new();
+        let i2 = Instrumentor::default();
+
+        assert_eq!(i1.interval, i2.interval);
+        assert_eq!(i1.outputs.len(), i2.outputs.len());
+    }
+
+    #[test]
+    fn builder_default_interval_when_not_specified() {
+        let instrumentor = Instrumentor::builder()
+            .output(Output::file("test.json"))
+            .build();
+
+        assert_eq!(instrumentor.interval, Duration::from_secs(1));
+    }
+
+    #[test]
+    fn complex_multi_module_scenario() {
+        let instrumentor = Instrumentor::new();
+
+        // Simulate a pipeline: API -> Processor -> Notifier
+        let api = instrumentor.register("api");
+        let processor = instrumentor.register("processor");
+        let notifier = instrumentor.register("notifier");
+
+        // API receives requests and writes to orders topic
+        api.record_write("orders", 1000);
+
+        // Processor reads orders, writes to notifications
+        processor.record_read("orders", 950);
+        processor.record_write("notifications", 950);
+
+        // Notifier reads notifications
+        notifier.record_read("notifications", 900);
+
+        let snapshot = instrumentor.collect();
+
+        // Verify backlogs
+        let proc_metrics = snapshot.modules.get("processor").unwrap();
+        assert_eq!(proc_metrics.reads.get("orders").unwrap().backlog, Some(50)); // 1000 - 950
+
+        let notif_metrics = snapshot.modules.get("notifier").unwrap();
+        assert_eq!(
+            notif_metrics.reads.get("notifications").unwrap().backlog,
+            Some(50)
+        ); // 950 - 900
+    }
 }
diff --git a/buswatch-sdk/src/state.rs b/buswatch-sdk/src/state.rs
index 427a1b0..a5842c8 100644
--- a/buswatch-sdk/src/state.rs
+++ b/buswatch-sdk/src/state.rs
@@ -235,4 +235,185 @@ mod tests {
         let snapshot = global.collect();
         assert_eq!(snapshot.modules.len(), 2);
     }
+
+    #[test]
+    fn get_or_create_read_returns_same_arc_on_second_call() {
+        let state = ModuleState::default();
+
+        let read1 = state.get_or_create_read("topic");
+        let read2 = state.get_or_create_read("topic");
+
+        // Should be the same Arc (pointer equality)
+        assert!(Arc::ptr_eq(&read1, &read2));
+
+        // Mutations should be visible through both handles
+        read1.count.fetch_add(10, Ordering::Relaxed);
+        assert_eq!(read2.count.load(Ordering::Relaxed), 10);
+    }
+
+    #[test]
+    fn get_or_create_write_returns_same_arc_on_second_call() {
+        let state = ModuleState::default();
+
+        let write1 = state.get_or_create_write("topic");
+        let write2 = state.get_or_create_write("topic");
+
+        assert!(Arc::ptr_eq(&write1, &write2));
+
+        write1.count.fetch_add(5, Ordering::Relaxed);
+        assert_eq!(write2.count.load(Ordering::Relaxed), 5);
+    }
+
+    #[test]
+    fn register_module_returns_same_arc_on_second_call() {
+        let global = GlobalState::default();
+
+        let m1 = global.register_module("service");
+        let m2 = global.register_module("service");
+
+        assert!(Arc::ptr_eq(&m1, &m2));
+    }
+
+    #[test]
+    fn backlog_computed_correctly_when_writes_exceed_reads() {
+        let global = GlobalState::default();
+
+        let producer = global.register_module("producer");
+        let consumer = global.register_module("consumer");
+
+        // Producer writes 100 messages
+        producer
+            .get_or_create_write("events")
+            .count
+            .fetch_add(100, Ordering::Relaxed);
+        global
+            .get_topic_write_counter("events")
+            .fetch_add(100, Ordering::Relaxed);
+
+        // Consumer reads 70 messages
+        consumer
+            .get_or_create_read("events")
+            .count
+            .fetch_add(70, Ordering::Relaxed);
+
+        let snapshot = global.collect();
+        let consumer_metrics = snapshot.modules.get("consumer").unwrap();
+        let events_read = consumer_metrics.reads.get("events").unwrap();
+
+        assert_eq!(events_read.count, 70);
+        assert_eq!(events_read.backlog, Some(30)); // 100 - 70 = 30 unread
+    }
+
+    #[test]
+    fn backlog_is_none_when_no_global_writes_tracked() {
+        let global = GlobalState::default();
+
+        let consumer = global.register_module("consumer");
+        consumer
+            .get_or_create_read("events")
+            .count
+            .fetch_add(50, Ordering::Relaxed);
+
+        // No writes tracked globally
+        let snapshot = global.collect();
+        let consumer_metrics = snapshot.modules.get("consumer").unwrap();
+        let events_read = consumer_metrics.reads.get("events").unwrap();
+
+        assert_eq!(events_read.backlog, None);
+    }
+
+    #[test]
+    fn backlog_is_none_when_reads_equal_writes() {
+        let global = GlobalState::default();
+
+        let producer = global.register_module("producer");
+        let consumer = global.register_module("consumer");
+
+        producer
+            .get_or_create_write("events")
+            .count
+            .fetch_add(100, Ordering::Relaxed);
+        global
+            .get_topic_write_counter("events")
+            .fetch_add(100, Ordering::Relaxed);
+
+        consumer
+            .get_or_create_read("events")
+            .count
+            .fetch_add(100, Ordering::Relaxed);
+
+        let snapshot = global.collect();
+        let consumer_metrics = snapshot.modules.get("consumer").unwrap();
+        let events_read = consumer_metrics.reads.get("events").unwrap();
+
+        // No backlog when fully caught up
+        assert_eq!(events_read.backlog, None);
+    }
+
+    #[test]
+    fn pending_time_captured_in_collect() {
+        let state = ModuleState::default();
+
+        let read = state.get_or_create_read("topic");
+        *read.pending_since.write() = Some(Instant::now());
+
+        // Small sleep to ensure measurable pending time
+        std::thread::sleep(std::time::Duration::from_millis(5));
+
+        let metrics = state.collect();
+        let pending = metrics.reads.get("topic").unwrap().pending;
+
+        assert!(pending.is_some());
+        assert!(pending.unwrap().as_micros() >= 5000); // At least 5ms
+    }
+
+    #[test]
+    fn concurrent_increments_are_thread_safe() {
+        use std::thread;
+
+        let global = Arc::new(GlobalState::default());
+        let module = global.register_module("test");
+
+        let mut handles = vec![];
+        for _ in 0..10 {
+            let m = module.clone();
+            handles.push(thread::spawn(move || {
+                for _ in 0..100 {
+                    m.get_or_create_read("topic")
+                        .count
+                        .fetch_add(1, Ordering::Relaxed);
+                }
+            }));
+        }
+
+        for h in handles {
+            h.join().unwrap();
+        }
+
+        let metrics = module.collect();
+        assert_eq!(metrics.reads.get("topic").unwrap().count, 1000);
+    }
+
+    #[test]
+    fn multiple_topics_tracked_independently() {
+        let state = ModuleState::default();
+
+        state
+            .get_or_create_read("topic-a")
+            .count
+            .fetch_add(10, Ordering::Relaxed);
+        state
+            .get_or_create_read("topic-b")
+            .count
+            .fetch_add(20, Ordering::Relaxed);
+        state
+            .get_or_create_write("topic-c")
+            .count
+            .fetch_add(30, Ordering::Relaxed);
+
+        let metrics = state.collect();
+        assert_eq!(metrics.reads.get("topic-a").unwrap().count, 10);
+        assert_eq!(metrics.reads.get("topic-b").unwrap().count, 20);
+        assert_eq!(metrics.writes.get("topic-c").unwrap().count, 30);
+    }
 }

From fe8a3cef9f4bf63c4e3419d38a6731050e608ed4 Mon Sep 17 00:00:00 2001
From: Matthew Hounslow <hounslowmatthew@gmail.com>
Date: Tue, 23 Dec 2025 16:36:34 -0700
Subject: [PATCH 3/6] Add comprehensive tests for buswatch-adapters module

- Add tests for AdapterError display formatting
- Add tests for RabbitMQ queue metrics conversion edge cases
- Add tests for RabbitMQ JSON deserialization of queue info
- Add tests for Kafka builder options
- Add tests for NATS builder options
---
 buswatch-adapters/src/error.rs    |  48 ++++++++++++
 buswatch-adapters/src/kafka.rs    |  48 +++++++++++-
 buswatch-adapters/src/nats.rs     |  29 ++++++++
 buswatch-adapters/src/rabbitmq.rs | 118 ++++++++++++++++++++++++++++++
 4 files changed, 242 insertions(+), 1 deletion(-)

diff --git a/buswatch-adapters/src/error.rs b/buswatch-adapters/src/error.rs
index bc3f4ae..db975e7 100644
--- a/buswatch-adapters/src/error.rs
+++ b/buswatch-adapters/src/error.rs
@@ -42,3 +42,51 @@ impl From<reqwest::Error> for AdapterError {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn error_display_http() {
+        let err = AdapterError::Http("404 Not Found".to_string());
+        assert_eq!(err.to_string(), "HTTP request failed: 404 Not Found");
+    }
+
+    #[test]
+    fn error_display_parse() {
+        let err = AdapterError::Parse("invalid JSON".to_string());
+        assert_eq!(err.to_string(), "Failed to parse response: invalid JSON");
+    }
+
+    #[test]
+    fn error_display_auth() {
+        let err = AdapterError::Auth("bad credentials".to_string());
+        assert_eq!(err.to_string(), "Authentication failed: bad credentials");
+    }
+
+    #[test]
+    fn error_display_connection() {
+        let err = AdapterError::Connection("connection refused".to_string());
+        assert_eq!(err.to_string(), "Connection failed: connection refused");
+    }
+
+    #[test]
+    fn error_display_timeout() {
+        let err = AdapterError::Timeout;
+        assert_eq!(err.to_string(), "Request timed out");
+    }
+
+    #[test]
+    fn error_display_unsupported() {
+        let err = AdapterError::Unsupported("feature X".to_string());
+        assert_eq!(err.to_string(), "Feature not supported: feature X");
+    }
+
+    #[test]
+    fn error_is_debug() {
+        let err = AdapterError::Timeout;
+        let debug = format!("{:?}", err);
+        assert!(debug.contains("Timeout"));
+    }
+}
diff --git a/buswatch-adapters/src/kafka.rs b/buswatch-adapters/src/kafka.rs
index f4ba0b5..763d0a0 100644
--- a/buswatch-adapters/src/kafka.rs
+++ b/buswatch-adapters/src/kafka.rs
@@ -251,7 +251,6 @@ mod tests {
 
     #[test]
     fn test_builder_defaults() {
-        // Just test that builder compiles correctly
         let builder = KafkaAdapter::builder()
             .brokers("localhost:9092")
             .group_id("test-group")
@@ -260,4 +259,51 @@ mod tests {
         assert!(builder.brokers.is_some());
         assert!(builder.group_id.is_some());
     }
+
+    #[test]
+    fn builder_stores_brokers() {
+        let builder = KafkaAdapter::builder().brokers("broker1:9092,broker2:9092");
+        assert_eq!(builder.brokers.unwrap(), "broker1:9092,broker2:9092");
+    }
+
+    #[test]
+    fn builder_stores_group_id() {
+        let builder = KafkaAdapter::builder().group_id("my-consumer-group");
+        assert_eq!(builder.group_id.unwrap(), "my-consumer-group");
+    }
+
+    #[test]
+    fn builder_stores_group_filter() {
+        let builder = KafkaAdapter::builder().group_filter("prefix-");
+        assert_eq!(builder.group_filter.unwrap(), "prefix-");
+    }
+
+    #[test]
+    fn builder_stores_timeout() {
+        let builder = KafkaAdapter::builder().timeout(Duration::from_secs(30));
+        assert_eq!(builder.timeout.unwrap(), Duration::from_secs(30));
+    }
+
+    #[test]
+    fn builder_chains_all_options() {
+        let builder = KafkaAdapter::builder()
+            .brokers("localhost:9092")
+            .group_id("test")
+            .group_filter("app-")
+            .timeout(Duration::from_secs(15));
+
+        assert_eq!(builder.brokers.unwrap(), "localhost:9092");
+        assert_eq!(builder.group_id.unwrap(), "test");
+        assert_eq!(builder.group_filter.unwrap(), "app-");
+        assert_eq!(builder.timeout.unwrap(), Duration::from_secs(15));
+    }
+
+    #[test]
+    fn builder_default_is_empty() {
+        let builder = KafkaAdapterBuilder::default();
+        assert!(builder.brokers.is_none());
+        assert!(builder.group_id.is_none());
+        assert!(builder.group_filter.is_none());
+        assert!(builder.timeout.is_none());
+    }
 }
diff --git a/buswatch-adapters/src/nats.rs b/buswatch-adapters/src/nats.rs
index 15a4717..35ee678 100644
--- a/buswatch-adapters/src/nats.rs
+++ b/buswatch-adapters/src/nats.rs
@@ -200,4 +200,33 @@ mod tests {
         assert!(builder.url.is_some());
         assert_eq!(builder.url.unwrap(), "nats://localhost:4222");
     }
+
+    #[test]
+    fn builder_stores_url() {
+        let builder = NatsAdapter::builder().url("nats://nats.example.com:4222");
+        assert_eq!(builder.url.unwrap(), "nats://nats.example.com:4222");
+    }
+
+    #[test]
+    fn builder_stores_credentials_file() {
+        let builder = NatsAdapter::builder().credentials_file("/path/to/creds.txt");
+        assert_eq!(builder.credentials.unwrap(), "/path/to/creds.txt");
+    }
+
+    #[test]
+    fn builder_chains_options() {
+        let builder = NatsAdapter::builder()
+            .url("nats://localhost:4222")
+            .credentials_file("/etc/nats/creds");
+
+        assert_eq!(builder.url.unwrap(), "nats://localhost:4222");
+        assert_eq!(builder.credentials.unwrap(), "/etc/nats/creds");
+    }
+
+    #[test]
+    fn builder_default_is_empty() {
+        let builder = NatsAdapterBuilder::default();
+        assert!(builder.url.is_none());
+        assert!(builder.credentials.is_none());
+    }
 }
diff --git a/buswatch-adapters/src/rabbitmq.rs b/buswatch-adapters/src/rabbitmq.rs
index f039cad..d0ce036 100644
--- a/buswatch-adapters/src/rabbitmq.rs
+++ b/buswatch-adapters/src/rabbitmq.rs
@@ -359,4 +359,122 @@ mod tests {
         assert_eq!(write.count, 600);
         assert_eq!(write.rate, Some(10.5));
     }
+
+    #[test]
+    fn queue_with_no_consumers_shows_backlog_only() {
+        let adapter = RabbitMqAdapter::builder().build();
+
+        let queue = QueueInfo {
+            name: "idle-queue".to_string(),
+            messages_ready: 50,
+            messages_unacknowledged: 0,
+            consumers: 0, // No consumers
+            messages_delivered: None,
+            messages_published: Some(50),
+            message_stats: None,
+        };
+
+        let metrics = adapter.queue_to_metrics(&queue);
+
+        let read = metrics.reads.get("messages").unwrap();
+        assert_eq!(read.count, 0); // No delivered messages
+        assert_eq!(read.backlog, Some(50)); // But backlog is shown
+    }
+
+    #[test]
+    fn queue_with_no_message_stats_has_no_rates() {
+        let adapter = RabbitMqAdapter::builder().build();
+
+        let queue = QueueInfo {
+            name: "new-queue".to_string(),
+            messages_ready: 0,
+            messages_unacknowledged: 0,
+            consumers: 1,
+            messages_delivered: Some(10),
+            messages_published: Some(10),
+            message_stats: None,
+        };
+
+        let metrics = adapter.queue_to_metrics(&queue);
+
+        assert!(metrics.reads.get("messages").unwrap().rate.is_none());
+        assert!(metrics.writes.get("messages").unwrap().rate.is_none());
+    }
+
+    #[test]
+    fn builder_with_timeout() {
+        let adapter = RabbitMqAdapter::builder()
+            .timeout(Duration::from_secs(30))
+            .build();
+
+        // Can't directly test timeout, but verify it doesn't panic
+        assert_eq!(adapter.endpoint, "http://localhost:15672");
+    }
+
+    #[test]
+    fn urlencoded_handles_multiple_slashes() {
+        assert_eq!(urlencoded("/a/b/c"), "%2Fa%2Fb%2Fc");
+    }
+
+    #[test]
+    fn urlencoded_preserves_other_special_chars() {
+        // Only slashes are encoded by this function
+        assert_eq!(urlencoded("name:with:colons"), "name:with:colons");
+        assert_eq!(urlencoded("name.with.dots"), "name.with.dots");
+    }
+
+    #[test]
+    fn message_stats_rate_extraction() {
+        let stats = MessageStats {
+            publish_details: Some(RateDetails { rate: 100.5 }),
+            deliver_get_details: Some(RateDetails { rate: 99.2 }),
+        };
+
+        assert_eq!(stats.publish_rate(), Some(100.5));
+        assert_eq!(stats.deliver_get_rate(), Some(99.2));
+    }
+
+    #[test]
+    fn message_stats_missing_details_returns_none() {
+        let stats = MessageStats {
+            publish_details: None,
+            deliver_get_details: None,
+        };
+
+        assert_eq!(stats.publish_rate(), None);
+        assert_eq!(stats.deliver_get_rate(), None);
+    }
+
+    #[test]
+    fn queue_info_deserializes_from_json() {
+        let json = r#"{
+            "name": "orders",
+            "messages_ready": 42,
+            "messages_unacknowledged": 3,
+            "consumers": 2,
+            "message_stats": {
+                "publish_details": { "rate": 10.0 },
+                "deliver_get_details": { "rate": 9.5 }
+            }
+        }"#;
+
+        let queue: QueueInfo = serde_json::from_str(json).unwrap();
+        assert_eq!(queue.name, "orders");
+        assert_eq!(queue.messages_ready, 42);
+        assert_eq!(queue.consumers, 2);
+        assert_eq!(queue.message_stats.unwrap().publish_rate(), Some(10.0));
+    }
+
+    #[test]
+    fn queue_info_deserializes_with_missing_optional_fields() {
+        let json = r#"{
+            "name": "simple-queue"
+        }"#;
+
+        let queue: QueueInfo = serde_json::from_str(json).unwrap();
+        assert_eq!(queue.name, "simple-queue");
+        assert_eq!(queue.messages_ready, 0);
+        assert_eq!(queue.consumers, 0);
+        assert!(queue.message_stats.is_none());
+    }
 }

From 0fe172d9740ae93b798193fbdf4ce692745326a0 Mon Sep 17 00:00:00 2001
From: Matthew Hounslow <hounslowmatthew@gmail.com>
Date: Tue, 23 Dec 2025 16:40:13 -0700
Subject: [PATCH 4/6] Add comprehensive tests for buswatch-tui module

- Add tests for duration parsing and formatting
- Add tests for MonitorData health status computation and sorting
- Add tests for History sparkline generation and rate calculation
- Add tests for DataFlowGraph producer/consumer tracking
---
 buswatch-tui/src/data/duration.rs |  69 +++++++++++
 buswatch-tui/src/data/flow.rs     | 150 +++++++++++++++++++++++
 buswatch-tui/src/data/history.rs  | 151 +++++++++++++++++++++++
 buswatch-tui/src/data/monitor.rs  | 197 ++++++++++++++++++++++++++++++
 4 files changed, 567 insertions(+)

diff --git a/buswatch-tui/src/data/duration.rs b/buswatch-tui/src/data/duration.rs
index 427e405..0990cd7 100644
--- a/buswatch-tui/src/data/duration.rs
+++ b/buswatch-tui/src/data/duration.rs
@@ -68,4 +68,73 @@ mod tests {
         let d = parse_duration("0ns").unwrap();
         assert_eq!(d.as_nanos(), 0);
     }
+
+    #[test]
+    fn parse_microseconds_with_us_suffix() {
+        let d = parse_duration("500us").unwrap();
+        assert_eq!(d.as_nanos(), 500_000);
+    }
+
+    #[test]
+    fn parse_whole_seconds() {
+        let d = parse_duration("5s").unwrap();
+        assert_eq!(d.as_secs(), 5);
+    }
+
+    #[test]
+    fn parse_with_whitespace() {
+        let d = parse_duration("  100ms  ").unwrap();
+        assert_eq!(d.as_millis(), 100);
+    }
+
+    #[test]
+    fn parse_unknown_format_fails() {
+        let result = parse_duration("100x");
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn parse_invalid_number_fails() {
+        let result = parse_duration("abcms");
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn format_zero() {
+        assert_eq!(format_duration(Duration::ZERO), "0ns");
+    }
+
+    #[test]
+    fn format_nanoseconds() {
+        assert_eq!(format_duration(Duration::from_nanos(500)), "500ns");
+    }
+
+    #[test]
+    fn format_microseconds() {
+        let d = Duration::from_nanos(1500);
+        let s = format_duration(d);
+        assert!(s.contains("µs"));
+    }
+
+    #[test]
+    fn format_milliseconds() {
+        let d = Duration::from_micros(1500);
+        let s = format_duration(d);
+        assert!(s.contains("ms"));
+    }
+
+    #[test]
+    fn format_seconds() {
+        let d = Duration::from_millis(1500);
+        let s = format_duration(d);
+        assert!(s.contains("s"));
+        assert!(!s.contains("ms")); // Should be "1.50s", not "1500.00ms"
+    }
+
+    #[test]
+    fn format_large_duration() {
+        let d = Duration::from_secs(3600); // 1 hour
+        let s = format_duration(d);
+        assert!(s.contains("3600"));
+    }
 }
diff --git a/buswatch-tui/src/data/flow.rs b/buswatch-tui/src/data/flow.rs
index d2d707e..20027ae 100644
--- a/buswatch-tui/src/data/flow.rs
+++ b/buswatch-tui/src/data/flow.rs
@@ -55,3 +55,153 @@ impl DataFlowGraph {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::data::monitor::{HealthStatus, ModuleData, TopicRead, TopicWrite};
+    use std::time::Instant;
+
+    fn make_module(name: &str, reads: Vec<&str>, writes: Vec<&str>) -> ModuleData {
+        ModuleData {
+            name: name.to_string(),
+            reads: reads
+                .into_iter()
+                .map(|t| TopicRead {
+                    topic: t.to_string(),
+                    read: 0,
+                    pending_for: None,
+                    unread: None,
+                    status: HealthStatus::Healthy,
+                })
+                .collect(),
+            writes: writes
+                .into_iter()
+                .map(|t| TopicWrite {
+                    topic: t.to_string(),
+                    written: 0,
+                    pending_for: None,
+                    status: HealthStatus::Healthy,
+                })
+                .collect(),
+            total_read: 0,
+            total_written: 0,
+            health: HealthStatus::Healthy,
+        }
+    }
+
+    fn make_monitor_data(modules: Vec<ModuleData>) -> MonitorData {
+        MonitorData {
+            modules,
+            last_updated: Instant::now(),
+        }
+    }
+
+    #[test]
+    fn empty_data_produces_empty_graph() {
+        let data = make_monitor_data(vec![]);
+        let graph = DataFlowGraph::from_monitor_data(&data);
+
+        assert!(graph.producers.is_empty());
+        assert!(graph.consumers.is_empty());
+        assert!(graph.topics.is_empty());
+    }
+
+    #[test]
+    fn single_producer_tracked() {
+        let data = make_monitor_data(vec![make_module("api", vec![], vec!["events"])]);
+        let graph = DataFlowGraph::from_monitor_data(&data);
+
+        assert_eq!(graph.topics, vec!["events"]);
+        assert_eq!(graph.producers.get("events").unwrap(), &vec!["api"]);
+        assert!(graph.consumers.get("events").is_none());
+    }
+
+    #[test]
+    fn single_consumer_tracked() {
+        let data = make_monitor_data(vec![make_module("worker", vec!["events"], vec![])]);
+        let graph = DataFlowGraph::from_monitor_data(&data);
+
+        assert_eq!(graph.topics, vec!["events"]);
+        assert!(graph.producers.get("events").is_none());
+        assert_eq!(graph.consumers.get("events").unwrap(), &vec!["worker"]);
+    }
+
+    #[test]
+    fn producer_consumer_pipeline() {
+        let data = make_monitor_data(vec![
+            make_module("api", vec![], vec!["orders"]),
+            make_module("processor", vec!["orders"], vec!["notifications"]),
+            make_module("notifier", vec!["notifications"], vec![]),
+        ]);
+        let graph = DataFlowGraph::from_monitor_data(&data);
+
+        assert_eq!(graph.topics, vec!["notifications", "orders"]);
+
+        assert_eq!(graph.producers.get("orders").unwrap(), &vec!["api"]);
+        assert_eq!(graph.consumers.get("orders").unwrap(), &vec!["processor"]);
+
+        assert_eq!(
+            graph.producers.get("notifications").unwrap(),
+            &vec!["processor"]
+        );
+        assert_eq!(
+            graph.consumers.get("notifications").unwrap(),
+            &vec!["notifier"]
+        );
+    }
+
+    #[test]
+    fn multiple_producers_same_topic() {
+        let data = make_monitor_data(vec![
+            make_module("api-1", vec![], vec!["events"]),
+            make_module("api-2", vec![], vec!["events"]),
+        ]);
+        let graph = DataFlowGraph::from_monitor_data(&data);
+
+        let producers = graph.producers.get("events").unwrap();
+        assert_eq!(producers.len(), 2);
+        assert!(producers.contains(&"api-1".to_string()));
+        assert!(producers.contains(&"api-2".to_string()));
+    }
+
+    #[test]
+    fn multiple_consumers_same_topic() {
+        let data = make_monitor_data(vec![
+            make_module("worker-1", vec!["events"], vec![]),
+            make_module("worker-2", vec!["events"], vec![]),
+        ]);
+        let graph = DataFlowGraph::from_monitor_data(&data);
+
+        let consumers = graph.consumers.get("events").unwrap();
+        assert_eq!(consumers.len(), 2);
+        assert!(consumers.contains(&"worker-1".to_string()));
+        assert!(consumers.contains(&"worker-2".to_string()));
+    }
+
+    #[test]
+    fn topics_are_sorted_alphabetically() {
+        let data = make_monitor_data(vec![make_module(
+            "m",
+            vec!["zebra", "alpha", "middle"],
+            vec![],
+        )]);
+        let graph = DataFlowGraph::from_monitor_data(&data);
+
+        assert_eq!(graph.topics, vec!["alpha", "middle", "zebra"]);
+    }
+
+    #[test]
+    fn module_can_be_both_producer_and_consumer() {
+        let data = make_monitor_data(vec![make_module(
+            "transformer",
+            vec!["input"],
+            vec!["output"],
+        )]);
+        let graph = DataFlowGraph::from_monitor_data(&data);
+
+        assert_eq!(graph.topics.len(), 2);
+        assert_eq!(graph.consumers.get("input").unwrap(), &vec!["transformer"]);
+        assert_eq!(graph.producers.get("output").unwrap(), &vec!["transformer"]);
+    }
+}
diff --git a/buswatch-tui/src/data/history.rs b/buswatch-tui/src/data/history.rs
index 1af0dbb..9e140bd 100644
--- a/buswatch-tui/src/data/history.rs
+++ b/buswatch-tui/src/data/history.rs
@@ -126,3 +126,154 @@ impl History {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::data::monitor::{HealthStatus, ModuleData, Thresholds};
+    use buswatch_types::Snapshot;
+
+    fn make_monitor_data(modules: Vec<(&str, u64, u64)>) -> MonitorData {
+        let module_data: Vec<ModuleData> = modules
+            .into_iter()
+            .map(|(name, reads, writes)| ModuleData {
+                name: name.to_string(),
+                reads: vec![],
+                writes: vec![],
+                total_read: reads,
+                total_written: writes,
+                health: HealthStatus::Healthy,
+            })
+            .collect();
+
+        MonitorData {
+            modules: module_data,
+            last_updated: Instant::now(),
+        }
+    }
+
+    #[test]
+    fn new_history_is_empty() {
+        let h = History::new();
+        assert!(h.module_reads.is_empty());
+        assert!(h.module_writes.is_empty());
+        assert!(h.timestamps.is_empty());
+    }
+
+    #[test]
+    fn record_stores_module_data() {
+        let mut h = History::new();
+        let data = make_monitor_data(vec![("service", 100, 50)]);
+
+        h.record(&data);
+
+        assert!(h.module_reads.contains_key("service"));
+        assert!(h.module_writes.contains_key("service"));
+        assert_eq!(h.module_reads.get("service").unwrap().len(), 1);
+    }
+
+    #[test]
+    fn record_accumulates_history() {
+        let mut h = History::new();
+
+        for i in 0..5 {
+            let data = make_monitor_data(vec![("service", i * 10, i * 5)]);
+            h.record(&data);
+        }
+
+        assert_eq!(h.module_reads.get("service").unwrap().len(), 5);
+        assert_eq!(h.timestamps.len(), 5);
+    }
+
+    #[test]
+    fn history_caps_at_max_size() {
+        let mut h = History::new();
+
+        // Record more than MAX_HISTORY_SIZE (60) entries
+        for i in 0..70 {
+            let data = make_monitor_data(vec![("service", i, 0)]);
+            h.record(&data);
+        }
+
+        // Should be capped at 60
+        assert_eq!(h.module_reads.get("service").unwrap().len(), 60);
+        assert_eq!(h.timestamps.len(), 60);
+    }
+
+    #[test]
+    fn sparkline_empty_for_unknown_module() {
+        let h = History::new();
+        let sparkline = h.get_reads_sparkline("unknown");
+        assert!(sparkline.is_empty());
+    }
+
+    #[test]
+    fn sparkline_empty_with_single_reading() {
+        let mut h = History::new();
+        let data = make_monitor_data(vec![("service", 100, 0)]);
+        h.record(&data);
+
+        let sparkline = h.get_reads_sparkline("service");
+        assert!(sparkline.is_empty()); // Need at least 2 readings
+    }
+
+    #[test]
+    fn sparkline_returns_normalized_deltas() {
+        let mut h = History::new();
+
+        // Record increasing values
+        for i in 0..5 {
+            let data = make_monitor_data(vec![("service", i * 100, 0)]);
+            h.record(&data);
+        }
+
+        let sparkline = h.get_reads_sparkline("service");
+        assert_eq!(sparkline.len(), 4); // 5 readings = 4 deltas
+
+        // All deltas are equal (100), so all should normalize to same value
+        assert!(sparkline.iter().all(|&v| v == sparkline[0]));
+    }
+
+    #[test]
+    fn read_rate_none_for_unknown_module() {
+        let h = History::new();
+        assert!(h.get_read_rate("unknown").is_none());
+    }
+
+    #[test]
+    fn read_rate_none_with_single_reading() {
+        let mut h = History::new();
+        let data = make_monitor_data(vec![("service", 100, 0)]);
+        h.record(&data);
+
+        assert!(h.get_read_rate("service").is_none());
+    }
+
+    #[test]
+    fn read_rate_calculated_from_last_two_readings() {
+        let mut h = History::new();
+
+        let data1 = make_monitor_data(vec![("service", 100, 0)]);
+        h.record(&data1);
+
+        // Small delay to ensure non-zero elapsed time
+        std::thread::sleep(std::time::Duration::from_millis(10));
+
+        let data2 = make_monitor_data(vec![("service", 200, 0)]);
+        h.record(&data2);
+
+        let rate = h.get_read_rate("service");
+        assert!(rate.is_some());
+        // Rate should be positive (100 messages over ~10ms)
+        assert!(rate.unwrap() > 0.0);
+    }
+
+    #[test]
+    fn default_is_same_as_new() {
+        let h1 = History::new();
+        let h2 = History::default();
+
+        assert_eq!(h1.module_reads.len(), h2.module_reads.len());
+        assert_eq!(h1.timestamps.len(), h2.timestamps.len());
+    }
+}
diff --git a/buswatch-tui/src/data/monitor.rs b/buswatch-tui/src/data/monitor.rs
index e577f98..e0f2816 100644
--- a/buswatch-tui/src/data/monitor.rs
+++ b/buswatch-tui/src/data/monitor.rs
@@ -309,3 +309,200 @@ impl UnhealthyTopic {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use buswatch_types::Snapshot;
+
+    fn default_thresholds() -> Thresholds {
+        Thresholds::default()
+    }
+
+    fn make_snapshot_json(modules_json: &str) -> String {
+        format!(
+            r#"{{"version":{{"major":1,"minor":0}},"timestamp_ms":1000,"modules":{{{}}}}}"#,
+            modules_json
+        )
+    }
+
+    #[test]
+    fn thresholds_default_values() {
+        let t = Thresholds::default();
+        assert_eq!(t.pending_warning, Duration::from_secs(1));
+        assert_eq!(t.pending_critical, Duration::from_secs(10));
+        assert_eq!(t.unread_warning, 1000);
+        assert_eq!(t.unread_critical, 5000);
+    }
+
+    #[test]
+    fn health_status_ordering() {
+        assert!(HealthStatus::Healthy < HealthStatus::Warning);
+        assert!(HealthStatus::Warning < HealthStatus::Critical);
+        assert!(HealthStatus::Healthy < HealthStatus::Critical);
+    }
+
+    #[test]
+    fn health_status_symbols() {
+        assert_eq!(HealthStatus::Healthy.symbol(), "OK");
+        assert_eq!(HealthStatus::Warning.symbol(), "WARN");
+        assert_eq!(HealthStatus::Critical.symbol(), "CRIT");
+    }
+
+    #[test]
+    fn parse_empty_snapshot() {
+        let json = make_snapshot_json("");
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+        assert!(data.modules.is_empty());
+    }
+
+    #[test]
+    fn parse_single_module() {
+        let json = make_snapshot_json(
+            r#""service":{"reads":{"input":{"count":100}},"writes":{"output":{"count":50}}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        assert_eq!(data.modules.len(), 1);
+        assert_eq!(data.modules[0].name, "service");
+        assert_eq!(data.modules[0].total_read, 100);
+        assert_eq!(data.modules[0].total_written, 50);
+    }
+
+    #[test]
+    fn module_health_is_worst_of_topics() {
+        // Create a module with one healthy read and one critical read
+        let json = make_snapshot_json(
+            r#""service":{"reads":{"healthy":{"count":100},"critical":{"count":50,"backlog":10000}},"writes":{}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        assert_eq!(data.modules[0].health, HealthStatus::Critical);
+    }
+
+    #[test]
+    fn read_status_healthy_when_under_thresholds() {
+        let json = make_snapshot_json(
+            r#""service":{"reads":{"topic":{"count":100,"backlog":500}},"writes":{}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        assert_eq!(data.modules[0].reads[0].status, HealthStatus::Healthy);
+    }
+
+    #[test]
+    fn read_status_warning_when_backlog_exceeds_warning_threshold() {
+        let json = make_snapshot_json(
+            r#""service":{"reads":{"topic":{"count":100,"backlog":1500}},"writes":{}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        assert_eq!(data.modules[0].reads[0].status, HealthStatus::Warning);
+    }
+
+    #[test]
+    fn read_status_critical_when_backlog_exceeds_critical_threshold() {
+        let json = make_snapshot_json(
+            r#""service":{"reads":{"topic":{"count":100,"backlog":6000}},"writes":{}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        assert_eq!(data.modules[0].reads[0].status, HealthStatus::Critical);
+    }
+
+    #[test]
+    fn modules_sorted_by_health_critical_first() {
+        let json = make_snapshot_json(
+            r#""healthy":{"reads":{},"writes":{}},
+               "critical":{"reads":{"t":{"count":0,"backlog":10000}},"writes":{}},
+               "warning":{"reads":{"t":{"count":0,"backlog":2000}},"writes":{}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        assert_eq!(data.modules[0].name, "critical");
+        assert_eq!(data.modules[1].name, "warning");
+        assert_eq!(data.modules[2].name, "healthy");
+    }
+
+    #[test]
+    fn topics_sorted_by_status_within_module() {
+        let json = make_snapshot_json(
+            r#""service":{"reads":{
+                "healthy":{"count":100},
+                "critical":{"count":0,"backlog":10000},
+                "warning":{"count":0,"backlog":2000}
+            },"writes":{}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        let reads = &data.modules[0].reads;
+        assert_eq!(reads[0].topic, "critical");
+        assert_eq!(reads[1].topic, "warning");
+        assert_eq!(reads[2].topic, "healthy");
+    }
+
+    #[test]
+    fn unhealthy_topics_returns_only_non_healthy() {
+        let json = make_snapshot_json(
+            r#""service":{"reads":{
+                "healthy":{"count":100},
+                "warning":{"count":0,"backlog":2000}
+            },"writes":{}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        let unhealthy = data.unhealthy_topics();
+        assert_eq!(unhealthy.len(), 1);
+        assert_eq!(unhealthy[0].1.topic(), "warning");
+    }
+
+    #[test]
+    fn unhealthy_topic_accessors() {
+        let read = TopicRead {
+            topic: "orders".to_string(),
+            read: 100,
+            pending_for: Some(Duration::from_secs(5)),
+            unread: Some(1500),
+            status: HealthStatus::Warning,
+        };
+        let unhealthy = UnhealthyTopic::Read(read);
+
+        assert_eq!(unhealthy.topic(), "orders");
+        assert_eq!(unhealthy.status(), HealthStatus::Warning);
+        assert_eq!(unhealthy.pending_for(), Some(Duration::from_secs(5)));
+    }
+
+    #[test]
+    fn from_snapshot_works_directly() {
+        let snapshot = Snapshot::builder()
+            .timestamp_ms(1000)
+            .module("test", |m| m.read("topic", |r| r.count(42)))
+            .build();
+
+        let data = MonitorData::from_snapshot(snapshot, &default_thresholds());
+        assert_eq!(data.modules.len(), 1);
+        assert_eq!(data.modules[0].total_read, 42);
+    }
+
+    #[test]
+    fn pending_duration_converted_correctly() {
+        // pending is in microseconds in the JSON
+        let json = make_snapshot_json(
+            r#""service":{"reads":{"topic":{"count":100,"pending":15000000}},"writes":{}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        let pending = data.modules[0].reads[0].pending_for.unwrap();
+        assert_eq!(pending, Duration::from_secs(15));
+    }
+
+    #[test]
+    fn write_pending_triggers_warning() {
+        let json = make_snapshot_json(
+            r#""service":{"reads":{},"writes":{"topic":{"count":100,"pending":5000000}}}"#,
+        );
+        let data = MonitorData::parse(&json, &default_thresholds()).unwrap();
+
+        assert_eq!(data.modules[0].writes[0].status, HealthStatus::Warning);
+    }
+}

From 2df9a3c3064395e2ffc4639ef885e49e72958f79 Mon Sep 17 00:00:00 2001
From: Matthew Hounslow <hounslowmatthew@gmail.com>
Date: Tue, 23 Dec 2025 16:42:02 -0700
Subject: [PATCH 5/6] Add code coverage workflow with cargo-tarpaulin

- Run coverage on push to main and on PRs
- Generate XML and HTML coverage reports
- Upload reports as artifacts for 14 days
---
 .github/workflows/coverage.yml | 38 ++++++++++++++++++++++++++++++++++
 1 file changed, 38 insertions(+)
 create mode 100644 .github/workflows/coverage.yml

diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml
new file mode 100644
index 0000000..b2bab5e
--- /dev/null
+++ b/.github/workflows/coverage.yml
@@ -0,0 +1,38 @@
+name: Code Coverage
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  coverage:
+    name: Coverage
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - uses: Swatinem/rust-cache@v2
+
+      - name: Install cargo-tarpaulin
+        run: cargo install cargo-tarpaulin
+
+      - name: Run tests with coverage
+        run: cargo tarpaulin --all-features --workspace --out xml --out html --skip-clean
+
+      - name: Upload coverage report
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-report
+          path: |
+            cobertura.xml
+            tarpaulin-report.html
+          retention-days: 14

From fe7a7a8cc25ac4c2f07b9f9347d6b59f6d0aa828 Mon Sep 17 00:00:00 2001
From: Matthew Hounslow <hounslowmatthew@gmail.com>
Date: Tue, 23 Dec 2025 16:46:54 -0700
Subject: [PATCH 6/6] Fix clippy warnings and run tests across full workspace

- Replace .is_none() with !contains_key() in flow.rs tests
- Add --workspace flag to CI test job to run all crate tests
---
 .github/workflows/ci.yml      | 4 ++--
 buswatch-tui/src/data/flow.rs | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 2e75953..7b2106f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -40,9 +40,9 @@ jobs:
       - uses: Swatinem/rust-cache@v2
       - run: cargo fetch
       - name: Build tests
-        run: cargo build --tests --all-features
+        run: cargo build --tests --all-features --workspace
       - name: Run tests
-        run: cargo test --all-features
+        run: cargo test --all-features --workspace
 
   deny:
     name: Security & Licenses
diff --git a/buswatch-tui/src/data/flow.rs b/buswatch-tui/src/data/flow.rs
index 20027ae..87eb148 100644
--- a/buswatch-tui/src/data/flow.rs
+++ b/buswatch-tui/src/data/flow.rs
@@ -114,7 +114,7 @@ mod tests {
 
         assert_eq!(graph.topics, vec!["events"]);
         assert_eq!(graph.producers.get("events").unwrap(), &vec!["api"]);
-        assert!(graph.consumers.get("events").is_none());
+        assert!(!graph.consumers.contains_key("events"));
     }
 
     #[test]
@@ -123,7 +123,7 @@ mod tests {
         let graph = DataFlowGraph::from_monitor_data(&data);
 
         assert_eq!(graph.topics, vec!["events"]);
-        assert!(graph.producers.get("events").is_none());
+        assert!(!graph.producers.contains_key("events"));
         assert_eq!(graph.consumers.get("events").unwrap(), &vec!["worker"]);
     }