Test pipe and bulk_pipe

fuzz/Cargo.toml

@@ -103,6 +103,34 @@ test = false
 doc = false
 bench = false
 
+[[bin]]
+name = "sync_pipe"
+path = "fuzz_targets/sync/pipe.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "local_nb_pipe"
+path = "fuzz_targets/local_nb/pipe.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "sync_bulk_pipe"
+path = "fuzz_targets/sync/bulk_pipe.rs"
+test = false
+doc = false
+bench = false
+
+[[bin]]
+name = "local_nb_bulk_pipe"
+path = "fuzz_targets/local_nb/bulk_pipe.rs"
+test = false
+doc = false
+bench = false
+
 ####################
 ## ufotofu_queues ##
 ####################

fuzz/fuzz_targets/local_nb/bulk_pipe.rs

@@ -0,0 +1,21 @@
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+use ufotofu::local_nb::{bulk_pipe, consumer::TestConsumer, pipe, producer::TestProducer};
+
+fuzz_target!(
+    |data: (TestConsumer<u16, u16, u16>, TestProducer<u16, u16, u16>)| {
+        smol::block_on(async {
+            let (mut con, mut pro) = data;
+            let mut control_con = con.clone();
+            let mut control_pro = pro.clone();
+
+            assert_eq!(
+                bulk_pipe(&mut pro, &mut con).await,
+                pipe(&mut control_pro, &mut control_con).await,
+            );
+            assert_eq!(con, control_con);
+            assert_eq!(pro, control_pro);
+        });
+    }
+);

fuzz/fuzz_targets/local_nb/pipe.rs

@@ -0,0 +1,44 @@
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+use ufotofu::local_nb::{consumer::TestConsumer, pipe, producer::TestProducer, PipeError};
+
+fuzz_target!(
+    |data: (TestProducer<u16, u16, u16>, TestConsumer<u16, u16, u16>)| {
+        smol::block_on(async {
+            let (mut pro, mut con) = data;
+
+            let items = pro.remaining().to_vec();
+            let last = pro.peek_last().unwrap().clone();
+            let consumer_err = con.peek_error().unwrap().clone();
+
+            match pipe(&mut pro, &mut con).await {
+                Ok(()) => {
+                    assert!(pro.did_already_emit_last());
+                    assert_eq!(con.consumed(), &items[..]);
+                    assert_eq!(con.final_consumed(), Some(&last.unwrap()));
+                }
+                Err(PipeError::Consumer(err)) => {
+                    if pro.peek_last().is_none() && last.is_err() {
+                        panic!();
+                    }
+                    assert!(con.did_error());
+                    assert_eq!(err, consumer_err);
+                    assert_eq!(con.consumed(), &items[..con.consumed().len()]);
+                    assert_eq!(
+                        pro.remaining(),
+                        &items[con.consumed().len()
+                            + if pro.did_already_emit_last() { 0 } else { 1 }..]
+                    ); // TODO
+                }
+                Err(PipeError::Producer(err)) => {
+                    assert!(pro.did_already_emit_last());
+                    assert!(!con.did_error());
+                    assert_eq!(err, last.unwrap_err());
+                    assert_eq!(con.consumed(), &items[..con.consumed().len()]);
+                    assert_eq!(pro.remaining(), &items[con.consumed().len()..]);
+                }
+            }
+        });
+    }
+);

fuzz/fuzz_targets/sync/bulk_pipe.rs

@@ -0,0 +1,17 @@
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+use ufotofu::sync::{consumer::TestConsumer, producer::TestProducer, pipe, bulk_pipe};
+
+fuzz_target!(|data: (TestConsumer<u16, u16, u16>, TestProducer<u16, u16, u16>)| {
+    let (mut con, mut pro) = data;
+    let mut control_con = con.clone();
+    let mut control_pro = pro.clone();
+
+    assert_eq!(
+        bulk_pipe(&mut pro, &mut con),
+        pipe(&mut control_pro, &mut control_con),
+    );
+    assert_eq!(con, control_con);
+    assert_eq!(pro, control_pro);
+});

fuzz/fuzz_targets/sync/pipe.rs

@@ -0,0 +1,38 @@
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+use ufotofu::sync::{consumer::TestConsumer, pipe, producer::TestProducer, PipeError};
+
+fuzz_target!(
+    |data: (TestProducer<u16, u16, u16>, TestConsumer<u16, u16, u16>)| {
+        let (mut pro, mut con) = data;
+
+        let items = pro.remaining().to_vec();
+        let last = pro.peek_last().unwrap().clone();
+        let consumer_err = con.peek_error().unwrap().clone();
+
+        match pipe(&mut pro, &mut con) {
+            Ok(()) => {
+                assert!(pro.did_already_emit_last());
+                assert_eq!(con.consumed(), &items[..]);
+                assert_eq!(con.final_consumed(), Some(&last.unwrap()));
+            }
+            Err(PipeError::Consumer(err)) => {
+                if pro.peek_last().is_none() && last.is_err() {
+                    panic!();
+                }
+                assert!(con.did_error());
+                assert_eq!(err, consumer_err);
+                assert_eq!(con.consumed(), &items[..con.consumed().len()]);
+                assert_eq!(pro.remaining(), &items[con.consumed().len() + if pro.did_already_emit_last() { 0 } else { 1 } ..]); // TODO
+            }
+            Err(PipeError::Producer(err)) => {
+                assert!(pro.did_already_emit_last());
+                assert!(!con.did_error());
+                assert_eq!(err, last.unwrap_err());
+                assert_eq!(con.consumed(), &items[..con.consumed().len()]);
+                assert_eq!(pro.remaining(), &items[con.consumed().len()..]);
+            }
+        }
+    }
+);