1 files changed, 203 insertions, 37 deletions
diff --git a/src/test/fixtures/future.rs b/src/test/fixtures/future.rs
index bbdc9f8..2f810a3 100644
--- a/src/test/fixtures/future.rs
+++ b/src/test/fixtures/future.rs
@@ -1,55 +1,221 @@
-use std::{future::IntoFuture, time::Duration};
+use std::{future::Future, pin::Pin, task};
 
-use futures::{stream, Stream};
-use tokio::time::timeout;
+use futures::stream;
 
-async fn immediately<F>(fut: F) -> F::Output
+// Combinators for futures that prevent waits, even when the underlying future
+// would block.
+//
+// These are only useful for futures with no bound on how long they may wait,
+// and this trait is only implemented on futures that are likely to have that
+// characteristic. Trying to apply this to futures that already have some
+// bounded wait time may make tests fail inappropriately and can hide other
+// logic errors.
+pub trait Expect: Sized {
+    // The returned future expects the underlying future to be ready immediately,
+    // and panics with the provided message if it is not.
+    //
+    // For stream operations, can be used to assert immediate completion.
+    fn expect_ready(self, message: &str) -> Ready<Self>
+    where
+        Self: Future;
+
+    // The returned future expects the underlying future _not_ to be ready, and
+    // panics if it is. This is usually a useful proxy for "I expect this to never
+    // arrive" or "to not be here yet." The future is transformed to return `()`,
+    // since the underlying future can never provide a value.
+    //
+    // For stream operations, can be used to assert that completion hasn't happened
+    // yet.
+    fn expect_wait(self, message: &str) -> Wait<Self>
+    where
+        Self: Future;
+
+    // The returned future expects the underlying future to resolve immediately, to
+    // a `Some` value. If it resolves to `None` or is not ready, it panics. The
+    // future is transformed to return the inner value from the `Some` case, like
+    // [`Option::expect`].
+    //
+    // For stream operations, can be used to assert that the stream has at least one
+    // message.
+    fn expect_some<T>(self, message: &str) -> Some<Self>
+    where
+        Self: Future<Output = Option<T>>;
+
+    // The returned future expects the underlying future to resolve immediately, to
+    // a `None` value. If it resolves to `Some(_)`, or is not ready, it panics. The
+    // future is transformed to return `()`, since the underlying future's value is
+    // fixed.
+    //
+    // For stream operations, can be used to assert that the stream has ended.
+    fn expect_none<T>(self, message: &str) -> None<Self>
+    where
+        Self: Future<Output = Option<T>>;
+}
+
+impl<'a, St> Expect for stream::Next<'a, St> {
+    fn expect_ready(self, message: &str) -> Ready<Self> {
+        Ready {
+            future: self,
+            message,
+        }
+    }
+
+    fn expect_wait(self, message: &str) -> Wait<Self> {
+        Wait {
+            future: self,
+            message,
+        }
+    }
+
+    fn expect_some<T>(self, message: &str) -> Some<Self>
+    where
+        Self: Future<Output = Option<T>>,
+    {
+        Some {
+            future: self,
+            message,
+        }
+    }
+
+    fn expect_none<T>(self, message: &str) -> None<Self>
+    where
+        Self: Future<Output = Option<T>>,
+    {
+        None {
+            future: self,
+            message,
+        }
+    }
+}
+
+impl<St, C> Expect for stream::Collect<St, C> {
+    fn expect_ready(self, message: &str) -> Ready<Self> {
+        Ready {
+            future: self,
+            message,
+        }
+    }
+
+    fn expect_wait(self, message: &str) -> Wait<Self> {
+        Wait {
+            future: self,
+            message,
+        }
+    }
+
+    fn expect_some<T>(self, message: &str) -> Some<Self>
+    where
+        Self: Future<Output = Option<T>>,
+    {
+        Some {
+            future: self,
+            message,
+        }
+    }
+
+    fn expect_none<T>(self, message: &str) -> None<Self>
+    where
+        Self: Future<Output = Option<T>>,
+    {
+        None {
+            future: self,
+            message,
+        }
+    }
+}
+
+#[pin_project::pin_project]
+pub struct Ready<'m, F> {
+    #[pin]
+    future: F,
+    message: &'m str,
+}
+
+impl<'m, F> Future for Ready<'m, F>
 where
-    F: IntoFuture,
+    F: Future + std::fmt::Debug,
 {
-    // I haven't been particularly rigorous here. Zero delay _seems to work_,
-    // but this can be set higher; it makes tests that fail to meet the
-    // "immediate" expectation take longer, but gives slow tests time to
-    // succeed, as well.
-    let duration = Duration::from_nanos(0);
-    timeout(duration, fut)
-        .await
-        .expect("expected result immediately")
-}
-
-// This is only intended for streams, since their `next()`, `collect()`, and
-// so on can all block indefinitely on an empty stream. There's no need to
-// force immediacy on futures that "can't" block forever, and it can hide logic
-// errors if you do that.
-//
-// The impls below _could_ be replaced with a blanket impl for all future
-// types, otherwise. The choice to restrict impls to stream futures is
-// deliberate.
-pub trait Immediately {
-    type Output;
+    type Output = F::Output;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
+        let this = self.project();
+
+        if let task::Poll::Ready(value) = this.future.poll(cx) {
+            task::Poll::Ready(value)
+        } else {
+            panic!("{}", this.message);
+        }
+    }
+}
+
+#[pin_project::pin_project]
+pub struct Wait<'m, F> {
+    #[pin]
+    future: F,
+    message: &'m str,
+}
 
-    async fn immediately(self) -> Self::Output;
+impl<'m, F> Future for Wait<'m, F>
+where
+    F: Future + std::fmt::Debug,
+{
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
+        let this = self.project();
+
+        if this.future.poll(cx).is_pending() {
+            task::Poll::Ready(())
+        } else {
+            panic!("{}", this.message);
+        }
+    }
 }
 
-impl<'a, St> Immediately for stream::Next<'a, St>
+#[pin_project::pin_project]
+pub struct Some<'m, F> {
+    #[pin]
+    future: F,
+    message: &'m str,
+}
+
+impl<'m, F, T> Future for Some<'m, F>
 where
-    St: Stream + Unpin + ?Sized,
+    F: Future<Output = Option<T>> + std::fmt::Debug,
 {
-    type Output = Option<<St as Stream>::Item>;
+    type Output = T;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
+        let this = self.project();
 
-    async fn immediately(self) -> Self::Output {
-        immediately(self).await
+        if let task::Poll::Ready(Option::Some(value)) = this.future.poll(cx) {
+            task::Poll::Ready(value)
+        } else {
+            panic!("{}", this.message)
+        }
     }
 }
 
-impl<St, C> Immediately for stream::Collect<St, C>
+#[pin_project::pin_project]
+pub struct None<'m, F> {
+    #[pin]
+    future: F,
+    message: &'m str,
+}
+
+impl<'m, F, T> Future for None<'m, F>
 where
-    St: Stream,
-    C: Default + Extend<<St as Stream>::Item>,
+    F: Future<Output = Option<T>> + std::fmt::Debug,
 {
-    type Output = C;
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
+        let this = self.project();
 
-    async fn immediately(self) -> Self::Output {
-        immediately(self).await
+        if let task::Poll::Ready(Option::None) = this.future.poll(cx) {
+            task::Poll::Ready(())
+        } else {
+            panic!("{}", this.message)
+        }
     }
 }