Move futures-util tests into futures/tests

Author: taiki-e

futures-util/src/sink/fanout.rs

@@ -104,34 +104,3 @@ impl<Si1, Si2, Item> Sink<Item> for Fanout<Si1, Si2>
         if ready { Poll::Ready(Ok(())) } else { Poll::Pending }
     }
 }
-
-#[cfg(test)]
-#[cfg(feature = "std")]
-mod tests {
-    use crate::future::join3;
-    use crate::sink::SinkExt;
-    use crate::stream::{self, StreamExt};
-    use futures_executor::block_on;
-    use futures_channel::mpsc;
-    use std::iter::Iterator;
-    use std::vec::Vec;
-
-    #[test]
-    fn it_works() {
-        let (tx1, rx1) = mpsc::channel(1);
-        let (tx2, rx2) = mpsc::channel(2);
-        let tx = tx1.fanout(tx2).sink_map_err(|_| ());
-
-        let src = stream::iter((0..10).map(Ok));
-        let fwd = src.forward(tx);
-
-        let collect_fut1 = rx1.collect::<Vec<_>>();
-        let collect_fut2 = rx2.collect::<Vec<_>>();
-        let (_, vec1, vec2) = block_on(join3(fwd, collect_fut1, collect_fut2));
-
-        let expected = (0..10).collect::<Vec<_>>();
-
-        assert_eq!(vec1, expected);
-        assert_eq!(vec2, expected);
-    }
-}

futures-util/src/task/arc_wake.rs

@@ -82,53 +82,3 @@ pub(super) unsafe fn wake_by_ref_arc_raw<T: ArcWake>(data: *const ()) {
     ArcWake::wake_by_ref(&arc);
     mem::forget(arc);
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::sync::Mutex;
-
-    struct CountingWaker {
-        nr_wake: Mutex<i32>,
-    }
-
-    impl CountingWaker {
-        fn new() -> CountingWaker {
-            CountingWaker {
-                nr_wake: Mutex::new(0),
-            }
-        }
-
-        pub fn wakes(&self) -> i32 {
-            *self.nr_wake.lock().unwrap()
-        }
-    }
-
-    impl ArcWake for CountingWaker {
-        fn wake_by_ref(arc_self: &Arc<Self>) {
-            let mut lock = arc_self.nr_wake.lock().unwrap();
-            *lock += 1;
-        }
-    }
-
-    #[test]
-    fn create_waker_from_arc() {
-        let some_w = Arc::new(CountingWaker::new());
-
-        let w1: Waker = ArcWake::into_waker(some_w.clone());
-        assert_eq!(2, Arc::strong_count(&some_w));
-        w1.wake_by_ref();
-        assert_eq!(1, some_w.wakes());
-
-        let w2 = w1.clone();
-        assert_eq!(3, Arc::strong_count(&some_w));
-
-        w2.wake_by_ref();
-        assert_eq!(2, some_w.wakes());
-
-        drop(w2);
-        assert_eq!(2, Arc::strong_count(&some_w));
-        drop(w1);
-        assert_eq!(1, Arc::strong_count(&some_w));
-    }
-}

futures-util/src/try_stream/into_async_read.rs

@@ -155,102 +155,3 @@ where
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use futures::stream::{self, TryStreamExt};
-    use futures_io::AsyncRead;
-    use futures_test::{task::noop_context, stream::StreamTestExt};
-
-    macro_rules! assert_read {
-        ($reader:expr, $buf:expr, $item:expr) => {
-            let mut cx = noop_context();
-            loop {
-                match Pin::new(&mut $reader).poll_read(&mut cx, $buf) {
-                    Poll::Ready(Ok(x)) => {
-                        assert_eq!(x, $item);
-                        break;
-                    }
-                    Poll::Ready(Err(err)) => {
-                        panic!("assertion failed: expected value but got {}", err);
-                    }
-                    Poll::Pending => {
-                        continue;
-                    }
-                }
-            }
-        };
-    }
-
-    macro_rules! assert_fill_buf {
-        ($reader:expr, $buf:expr) => {
-            let mut cx = noop_context();
-            loop {
-                match Pin::new(&mut $reader).poll_fill_buf(&mut cx) {
-                    Poll::Ready(Ok(x)) => {
-                        assert_eq!(x, $buf);
-                        break;
-                    }
-                    Poll::Ready(Err(err)) => {
-                        panic!("assertion failed: expected value but got {}", err);
-                    }
-                    Poll::Pending => {
-                        continue;
-                    }
-                }
-            }
-        };
-    }
-
-    #[test]
-    fn test_into_async_read() {
-        let stream = stream::iter((1..=3).flat_map(|_| vec![Ok(vec![]), Ok(vec![1, 2, 3, 4, 5])]));
-        let mut reader = stream.interleave_pending().into_async_read();
-        let mut buf = vec![0; 3];
-
-        assert_read!(reader, &mut buf, 3);
-        assert_eq!(&buf, &[1, 2, 3]);
-
-        assert_read!(reader, &mut buf, 2);
-        assert_eq!(&buf[..2], &[4, 5]);
-
-        assert_read!(reader, &mut buf, 3);
-        assert_eq!(&buf, &[1, 2, 3]);
-
-        assert_read!(reader, &mut buf, 2);
-        assert_eq!(&buf[..2], &[4, 5]);
-
-        assert_read!(reader, &mut buf, 3);
-        assert_eq!(&buf, &[1, 2, 3]);
-
-        assert_read!(reader, &mut buf, 2);
-        assert_eq!(&buf[..2], &[4, 5]);
-
-        assert_read!(reader, &mut buf, 0);
-    }
-
-    #[test]
-    fn test_into_async_bufread() -> std::io::Result<()> {
-        let stream = stream::iter((1..=2).flat_map(|_| vec![Ok(vec![]), Ok(vec![1, 2, 3, 4, 5])]));
-        let mut reader = stream.interleave_pending().into_async_read();
-
-        let mut reader = Pin::new(&mut reader);
-
-        assert_fill_buf!(reader, &[1, 2, 3, 4, 5][..]);
-        reader.as_mut().consume(3);
-
-        assert_fill_buf!(reader, &[4, 5][..]);
-        reader.as_mut().consume(2);
-
-        assert_fill_buf!(reader, &[1, 2, 3, 4, 5][..]);
-        reader.as_mut().consume(2);
-
-        assert_fill_buf!(reader, &[3, 4, 5][..]);
-        reader.as_mut().consume(3);
-
-        assert_fill_buf!(reader, &[][..]);
-
-        Ok(())
-    }
-}

futures-util/tests/futures_unordered.rs

@@ -0,0 +1,46 @@
+use futures::task::{ArcWake, Waker};
+use std::sync::{Arc, Mutex};
+
+struct CountingWaker {
+    nr_wake: Mutex<i32>,
+}
+
+impl CountingWaker {
+    fn new() -> CountingWaker {
+        CountingWaker {
+            nr_wake: Mutex::new(0),
+        }
+    }
+
+    fn wakes(&self) -> i32 {
+        *self.nr_wake.lock().unwrap()
+    }
+}
+
+impl ArcWake for CountingWaker {
+    fn wake_by_ref(arc_self: &Arc<Self>) {
+        let mut lock = arc_self.nr_wake.lock().unwrap();
+        *lock += 1;
+    }
+}
+
+#[test]
+fn create_waker_from_arc() {
+    let some_w = Arc::new(CountingWaker::new());
+
+    let w1: Waker = ArcWake::into_waker(some_w.clone());
+    assert_eq!(2, Arc::strong_count(&some_w));
+    w1.wake_by_ref();
+    assert_eq!(1, some_w.wakes());
+
+    let w2 = w1.clone();
+    assert_eq!(3, Arc::strong_count(&some_w));
+
+    w2.wake_by_ref();
+    assert_eq!(2, some_w.wakes());
+
+    drop(w2);
+    assert_eq!(2, Arc::strong_count(&some_w));
+    drop(w1);
+    assert_eq!(1, Arc::strong_count(&some_w));
+}

futures/tests/arc_wake.rs

@@ -1,12 +1,39 @@
 use futures::channel::oneshot;
 use futures::executor::{block_on, block_on_stream};
 use futures::future::{self, join, Future, FutureExt};
-use futures::stream::{StreamExt, FuturesUnordered};
+use futures::stream::{FusedStream, FuturesUnordered, StreamExt};
 use futures::task::Poll;
 use futures_test::{assert_stream_done, assert_stream_next};
 use futures_test::future::FutureTestExt;
 use futures_test::task::noop_context;
 
+#[test]
+fn is_terminated() {
+    let mut cx = noop_context();
+    let mut tasks = FuturesUnordered::new();
+
+    assert_eq!(tasks.is_terminated(), false);
+    assert_eq!(tasks.poll_next_unpin(&mut cx), Poll::Ready(None));
+    assert_eq!(tasks.is_terminated(), true);
+
+    // Test that the sentinel value doesn't leak
+    assert_eq!(tasks.is_empty(), true);
+    assert_eq!(tasks.len(), 0);
+    assert_eq!(tasks.iter_mut().len(), 0);
+
+    tasks.push(future::ready(1));
+
+    assert_eq!(tasks.is_empty(), false);
+    assert_eq!(tasks.len(), 1);
+    assert_eq!(tasks.iter_mut().len(), 1);
+
+    assert_eq!(tasks.is_terminated(), false);
+    assert_eq!(tasks.poll_next_unpin(&mut cx), Poll::Ready(Some(1)));
+    assert_eq!(tasks.is_terminated(), false);
+    assert_eq!(tasks.poll_next_unpin(&mut cx), Poll::Ready(None));
+    assert_eq!(tasks.is_terminated(), true);
+}
+
 #[test]
 fn works_1() {
     let (a_tx, a_rx) = oneshot::channel::<i32>();

futures-util/tests/atomic_waker.rs renamed to futures/tests/atomic_waker.rs

@@ -0,0 +1,24 @@
+use futures::channel::mpsc;
+use futures::executor::block_on;
+use futures::future::join3;
+use futures::sink::SinkExt;
+use futures::stream::{self, StreamExt};
+
+#[test]
+fn it_works() {
+    let (tx1, rx1) = mpsc::channel(1);
+    let (tx2, rx2) = mpsc::channel(2);
+    let tx = tx1.fanout(tx2).sink_map_err(|_| ());
+
+    let src = stream::iter((0..10).map(Ok));
+    let fwd = src.forward(tx);
+
+    let collect_fut1 = rx1.collect::<Vec<_>>();
+    let collect_fut2 = rx2.collect::<Vec<_>>();
+    let (_, vec1, vec2) = block_on(join3(fwd, collect_fut1, collect_fut2));
+
+    let expected = (0..10).collect::<Vec<_>>();
+
+    assert_eq!(vec1, expected);
+    assert_eq!(vec2, expected);
+}