test(nodes): add tests for ogg, pixel_ops, colorbars, and moq push

crates/nodes/src/containers/ogg.rs

@@ -875,3 +875,180 @@ pub fn register_ogg_nodes(registry: &mut NodeRegistry) {
         );
     }
 }
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used, clippy::expect_used)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn ogg_muxer_config_defaults() {
+        let config = OggMuxerConfig::default();
+        assert_eq!(config.stream_serial, 0);
+        assert!(matches!(config.codec, OggMuxerCodec::Opus));
+        assert_eq!(config.channels, 1);
+        assert_eq!(config.chunk_size, 65536);
+    }
+
+    #[test]
+    fn ogg_muxer_config_deserialization() {
+        let json = r#"{"stream_serial": 42, "channels": 2, "chunk_size": 4096}"#;
+        let config: OggMuxerConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(config.stream_serial, 42);
+        assert_eq!(config.channels, 2);
+        assert_eq!(config.chunk_size, 4096);
+    }
+
+    #[test]
+    fn ogg_demuxer_config_defaults() {
+        let config: OggDemuxerConfig = serde_json::from_str("{}").unwrap();
+        let _ = config;
+    }
+
+    #[test]
+    fn ogg_muxer_content_type() {
+        let node = OggMuxerNode::new(OggMuxerConfig::default());
+        assert_eq!(node.content_type(), Some("audio/ogg".to_string()));
+    }
+
+    #[test]
+    fn ogg_muxer_pins() {
+        let node = OggMuxerNode::new(OggMuxerConfig::default());
+        let inputs = node.input_pins();
+        assert_eq!(inputs.len(), 1);
+        assert_eq!(inputs[0].name, "in");
+
+        let outputs = node.output_pins();
+        assert_eq!(outputs.len(), 1);
+        assert_eq!(outputs[0].name, "out");
+        assert!(matches!(outputs[0].produces_type, PacketType::Binary));
+    }
+
+    #[test]
+    fn ogg_demuxer_pins() {
+        let node = OggDemuxerNode::new(OggDemuxerConfig::default());
+        let inputs = node.input_pins();
+        assert_eq!(inputs.len(), 1);
+        assert_eq!(inputs[0].name, "in");
+
+        let outputs = node.output_pins();
+        assert_eq!(outputs.len(), 1);
+        assert_eq!(outputs[0].name, "out");
+        assert!(matches!(
+            outputs[0].produces_type,
+            PacketType::EncodedAudio(EncodedAudioFormat { codec: AudioCodec::Opus, .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn ogg_mux_produces_valid_ogg_output() {
+        use crate::test_utils::{create_test_binary_packet, create_test_context};
+
+        let (input_tx, input_rx) = tokio::sync::mpsc::channel(10);
+        let mut inputs = std::collections::HashMap::new();
+        inputs.insert("in".to_string(), input_rx);
+
+        let (context, mock_sender, _state_rx) = create_test_context(inputs, 1);
+
+        let config = OggMuxerConfig { chunk_size: 128, ..OggMuxerConfig::default() };
+        let node = Box::new(OggMuxerNode::new(config));
+
+        let handle = tokio::spawn(async move { node.run(context).await });
+
+        for _ in 0..5 {
+            input_tx.send(create_test_binary_packet(vec![0xAB; 160])).await.unwrap();
+        }
+        drop(input_tx);
+
+        handle.await.unwrap().unwrap();
+
+        let packets = mock_sender.collect_packets().await;
+        assert!(!packets.is_empty(), "muxer should produce output");
+
+        let mut ogg_data = Vec::new();
+        for (_, _, pkt) in &packets {
+            if let Packet::Binary { data, content_type, .. } = pkt {
+                assert_eq!(content_type.as_deref(), Some("audio/ogg"));
+                ogg_data.extend_from_slice(data);
+            }
+        }
+
+        assert!(ogg_data.len() >= 4);
+        assert_eq!(&ogg_data[..4], b"OggS");
+    }
+
+    #[tokio::test]
+    async fn ogg_mux_demux_round_trip() {
+        use crate::test_utils::{create_test_binary_packet, create_test_context};
+
+        let (input_tx, input_rx) = tokio::sync::mpsc::channel(10);
+        let mut inputs = std::collections::HashMap::new();
+        inputs.insert("in".to_string(), input_rx);
+
+        let (context, mock_sender, _state_rx) = create_test_context(inputs, 1);
+
+        let config = OggMuxerConfig { chunk_size: 128, ..OggMuxerConfig::default() };
+        let node = Box::new(OggMuxerNode::new(config));
+
+        let handle = tokio::spawn(async move { node.run(context).await });
+
+        let original_payloads: Vec<Vec<u8>> = (0..3).map(|i| vec![0x10 + i; 160]).collect();
+
+        for payload in &original_payloads {
+            input_tx.send(create_test_binary_packet(payload.clone())).await.unwrap();
+        }
+        drop(input_tx);
+
+        handle.await.unwrap().unwrap();
+
+        let muxed_packets = mock_sender.collect_packets().await;
+        let mut ogg_data = Vec::new();
+        for (_, _, pkt) in &muxed_packets {
+            if let Packet::Binary { data, .. } = pkt {
+                ogg_data.extend_from_slice(data);
+            }
+        }
+        assert!(!ogg_data.is_empty());
+
+        let (demux_input_tx, demux_input_rx) = tokio::sync::mpsc::channel(10);
+        let mut demux_inputs = std::collections::HashMap::new();
+        demux_inputs.insert("in".to_string(), demux_input_rx);
+
+        let (demux_context, demux_sender, _demux_state_rx) = create_test_context(demux_inputs, 1);
+
+        let demux_node = Box::new(OggDemuxerNode::new(OggDemuxerConfig::default()));
+
+        let demux_handle = tokio::spawn(async move { demux_node.run(demux_context).await });
+
+        demux_input_tx.send(create_test_binary_packet(ogg_data)).await.unwrap();
+        drop(demux_input_tx);
+
+        demux_handle.await.unwrap().unwrap();
+
+        let demuxed = demux_sender.collect_packets().await;
+        let data_packets: Vec<_> = demuxed
+            .iter()
+            .filter_map(|(_, _, pkt)| {
+                if let Packet::Binary { data, .. } = pkt {
+                    if !data.is_empty()
+                        && !data.starts_with(b"OpusHead")
+                        && !data.starts_with(b"OpusTags")
+                    {
+                        return Some(data.clone());
+                    }
+                }
+                None
+            })
+            .collect();
+
+        assert_eq!(
+            data_packets.len(),
+            original_payloads.len(),
+            "demuxer should output one data packet per muxed payload"
+        );
+
+        for (i, data) in data_packets.iter().enumerate() {
+            assert_eq!(data.as_ref(), &original_payloads[i], "round-trip payload {i} should match");
+        }
+    }
+}

crates/nodes/src/transport/moq/push.rs

@@ -930,4 +930,71 @@ mod tests {
         let vp9: MoqPushConfig = serde_json::from_str(r#"{"video_codec": "vp9"}"#).unwrap();
         assert_eq!(vp9.video_codec, Some(VideoCodec::Vp9));
     }
+
+    #[test]
+    fn moq_push_config_all_defaults() {
+        let config: super::MoqPushConfig = serde_json::from_str("{}").unwrap();
+        assert!(config.url.is_empty());
+        assert!(config.jwt.is_none());
+        assert!(config.broadcast.is_empty());
+        assert_eq!(config.channels, 2);
+        assert!(config.audio.is_none());
+        assert!(config.video.is_none());
+        assert!(config.video_codec.is_none());
+        assert!(config.audio_codec.is_none());
+        assert_eq!(config.group_duration_ms, 40);
+        assert_eq!(config.initial_delay_ms, 0);
+    }
+
+    #[test]
+    fn is_video_pin_empty_string() {
+        assert!(!is_video_pin(""));
+    }
+
+    #[test]
+    fn is_video_pin_slashes_only() {
+        assert!(!is_video_pin("/"));
+        assert!(!is_video_pin("///"));
+    }
+
+    #[test]
+    fn is_video_pin_prefix_is_case_sensitive() {
+        assert!(!is_video_pin("Video/hd"));
+        assert!(!is_video_pin("VIDEO/hd"));
+    }
+
+    #[test]
+    fn track_name_from_pin_empty_string() {
+        assert_eq!(track_name_from_pin(""), "audio/");
+    }
+
+    #[test]
+    fn track_name_from_pin_slashes_only() {
+        assert_eq!(track_name_from_pin("/"), "audio//");
+        assert_eq!(track_name_from_pin("video/"), "video/");
+        assert_eq!(track_name_from_pin("audio/"), "audio/");
+    }
+
+    #[test]
+    fn moq_push_config_full_deserialization() {
+        let json = r#"{
+            "url": "https://relay.example.com",
+            "jwt": "tok123",
+            "broadcast": "my-stream",
+            "channels": 1,
+            "audio": true,
+            "video": false,
+            "group_duration_ms": 100,
+            "initial_delay_ms": 50
+        }"#;
+        let config: super::MoqPushConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(config.url, "https://relay.example.com");
+        assert_eq!(config.jwt.as_deref(), Some("tok123"));
+        assert_eq!(config.broadcast, "my-stream");
+        assert_eq!(config.channels, 1);
+        assert_eq!(config.audio, Some(true));
+        assert_eq!(config.video, Some(false));
+        assert_eq!(config.group_duration_ms, 100);
+        assert_eq!(config.initial_delay_ms, 50);
+    }
 }

crates/nodes/src/video/colorbars.rs

@@ -810,5 +810,112 @@ mod tests {
         assert_eq!(config.height, 480);
         assert_eq!(config.fps, 30);
         assert_eq!(config.frame_count, 0);
+        assert_eq!(config.pixel_format, "nv12");
+        assert!(!config.draw_time);
+        assert!(!config.animate);
+    }
+
+    #[test]
+    fn test_colorbars_config_custom_deserialization() {
+        let json = r#"{
+            "width": 1280,
+            "height": 720,
+            "fps": 60,
+            "frame_count": 10,
+            "pixel_format": "rgba8"
+        }"#;
+        let config: ColorBarsConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(config.width, 1280);
+        assert_eq!(config.height, 720);
+        assert_eq!(config.fps, 60);
+        assert_eq!(config.frame_count, 10);
+        assert_eq!(config.pixel_format, "rgba8");
+    }
+
+    #[test]
+    fn test_smpte_colorbars_nv12() {
+        let width = 640u32;
+        let height = 480u32;
+        let layout = streamkit_core::types::VideoLayout::packed(width, height, PixelFormat::Nv12);
+        let total = layout.total_bytes();
+        let mut data = vec![0u8; total];
+        generate_smpte_colorbars_nv12(width, height, &mut data, &layout);
+
+        // Y plane: first pixel should be white (Y=180).
+        assert_eq!(data[0], 180);
+        // Last bar (rightmost column) should be blue (Y=35).
+        let last_y_col = (width - 1) as usize;
+        assert_eq!(data[last_y_col], 35);
+
+        // UV plane should be non-zero (chroma data present).
+        let planes = layout.planes();
+        let uv_plane = planes[1];
+        let uv_start = uv_plane.offset;
+        let uv_len = uv_plane.stride * uv_plane.height as usize;
+        let uv_data = &data[uv_start..uv_start + uv_len];
+        assert!(uv_data.iter().any(|&b| b != 0), "UV plane should contain chroma data");
+    }
+
+    #[test]
+    fn test_smpte_colorbars_rgba8() {
+        let width = 640u32;
+        let height = 480u32;
+        let total = (width * height * 4) as usize;
+        let mut data = vec![0u8; total];
+        generate_smpte_colorbars_rgba8(width, height, &mut data);
+
+        // First pixel should be 75% white (191, 191, 191, 255).
+        assert_eq!(&data[0..4], &[191, 191, 191, 255]);
+        // Last column should be blue (0, 0, 191, 255).
+        let last_px = ((width - 1) as usize) * 4;
+        assert_eq!(&data[last_px..last_px + 4], &[0, 0, 191, 255]);
+
+        // All alpha values should be 255.
+        for px in data.chunks_exact(4) {
+            assert_eq!(px[3], 255, "alpha should always be 255");
+        }
+    }
+
+    #[tokio::test]
+    async fn test_colorbars_frame_count_limit() {
+        use crate::test_utils::create_oneshot_test_context;
+
+        let inputs = std::collections::HashMap::new();
+        let (mut context, mock_sender, mut state_rx) = create_oneshot_test_context(inputs, 1);
+
+        let (control_tx, control_rx) = tokio::sync::mpsc::channel(10);
+        context.control_rx = control_rx;
+
+        let config = ColorBarsConfig {
+            width: 32,
+            height: 32,
+            fps: 30,
+            frame_count: 5,
+            pixel_format: "i420".to_string(),
+            ..ColorBarsConfig::default()
+        };
+        let pixel_format = parse_pixel_format(&config.pixel_format).unwrap();
+        let node = Box::new(ColorBarsNode { config, pixel_format });
+
+        let handle = tokio::spawn(async move { node.run(context).await });
+
+        // Drain Initializing + Ready states, then send Start.
+        crate::test_utils::assert_state_initializing(&mut state_rx).await;
+        crate::test_utils::assert_state_update(
+            &mut state_rx,
+            |s| matches!(s, streamkit_core::NodeState::Ready),
+            "Ready",
+        )
+        .await;
+        control_tx.send(streamkit_core::control::NodeControlMessage::Start).await.unwrap();
+
+        handle.await.unwrap().unwrap();
+
+        let packets = mock_sender.collect_packets().await;
+        assert_eq!(packets.len(), 5, "should produce exactly 5 frames");
+
+        for (_, _, pkt) in &packets {
+            assert!(matches!(pkt, streamkit_core::types::Packet::Video(_)));
+        }
     }
 }