Tests for purged channels and messages.

This required a re-think of the `.immediately()` combinator, to generalize it to cases where a message is _not_ expected. That (more or less immediately) suggested some mixed combinators, particularly for stream futures (futures of `Option<T>`).

2 files changed, 15 insertions, 23 deletions

diff --git a/src/channel/routes/channel/test/post.rs b/src/channel/routes/channel/test/post.rs
index d81715f..111a703 100644
--- a/src/channel/routes/channel/test/post.rs
+++ b/src/channel/routes/channel/test/post.rs
@@ -1,11 +1,11 @@
 use axum::extract::{Json, Path, State};
-use futures::stream::StreamExt;
+use futures::stream::{self, StreamExt as _};
 
 use crate::{
     channel::{self, routes::channel::post},
     event::Sequenced,
-    message::{self, app::SendError},
-    test::fixtures::{self, future::Immediately as _},
+    message::app::SendError,
+    test::fixtures::{self, future::Expect as _},
 };
 
 #[tokio::test]
@@ -39,24 +39,23 @@ async fn messages_in_order() {
 
     // Verify the semantics
 
-    let events = app
+    let mut events = app
         .events()
         .subscribe(None)
         .await
         .expect("subscribing to a valid channel succeeds")
         .filter_map(fixtures::event::message)
-        .take(requests.len());
+        .filter_map(fixtures::event::message::sent)
+        .zip(stream::iter(requests));
 
-    let events = events.collect::<Vec<_>>().immediately().await;
-
-    for ((sent_at, message), event) in requests.into_iter().zip(events) {
+    while let Some((event, (sent_at, body))) = events
+        .next()
+        .expect_ready("an event should be ready for each message")
+        .await
+    {
         assert_eq!(*sent_at, event.at());
-        assert!(matches!(
-            event,
-            message::Event::Sent(event)
-                if event.message.sender == sender.login.id
-                    && event.message.body == message
-        ));
+        assert_eq!(sender.login.id, event.message.sender);
+        assert_eq!(body, event.message.body);
     }
 }
 
diff --git a/src/channel/routes/test.rs b/src/channel/routes/test.rs
index 46c58b0..10b1e8d 100644
--- a/src/channel/routes/test.rs
+++ b/src/channel/routes/test.rs
@@ -7,7 +7,7 @@ use super::post;
 use crate::{
     channel::app,
     name::Name,
-    test::fixtures::{self, future::Immediately as _},
+    test::fixtures::{self, future::Expect as _},
 };
 
 #[tokio::test]
@@ -39,7 +39,6 @@ async fn new_channel() {
         .channels()
         .get(&response.id)
         .await
-        .expect("searching for channels by ID never fails")
         .expect("the newly-created channel exists");
     assert_eq!(response, channel);
 
@@ -52,11 +51,7 @@ async fn new_channel() {
         .filter_map(fixtures::event::channel::created)
         .filter(|event| future::ready(event.channel == response));
 
-    let event = events
-        .next()
-        .immediately()
-        .await
-        .expect("creation event published");
+    let event = events.next().expect_some("creation event published").await;
 
     assert_eq!(event.channel, response);
 }
@@ -165,7 +160,6 @@ async fn name_reusable_after_delete() {
         .channels()
         .get(&response.id)
         .await
-        .expect("searching for channels by ID never fails")
         .expect("the newly-created channel exists");
     assert_eq!(response, channel);
 }
@@ -215,7 +209,6 @@ async fn name_reusable_after_expiry() {
         .channels()
         .get(&response.id)
         .await
-        .expect("searching for channels by ID never fails")
         .expect("the newly-created channel exists");
     assert_eq!(response, channel);
 }