From 0bb17bd01640492db2685e67bacac12dd54a9f59 Mon Sep 17 00:00:00 2001
From: Owen Jacobson <owen@grimoire.ca>
Date: Thu, 24 Oct 2024 22:37:22 -0400
Subject: Tests for channel, invite, setup, and message deletion events.

This also found a bug! No live event was being emitted during invite accept. The only way to find out about invites was to reconnect.
---
 src/event/routes/test/message.rs | 316 +++++++++++++++++++++++++++++++++++++++
 1 file changed, 316 insertions(+)
 create mode 100644 src/event/routes/test/message.rs

(limited to 'src/event/routes/test/message.rs')

diff --git a/src/event/routes/test/message.rs b/src/event/routes/test/message.rs
new file mode 100644
index 0000000..9bbbc7d
--- /dev/null
+++ b/src/event/routes/test/message.rs
@@ -0,0 +1,316 @@
+use axum::extract::State;
+use axum_extra::extract::Query;
+use futures::{
+    future,
+    stream::{self, StreamExt as _},
+};
+
+use crate::{
+    event::routes::get,
+    test::fixtures::{self, future::Immediately as _},
+};
+
+#[tokio::test]
+async fn sending() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+
+    // Call the endpoint
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) =
+        get::handler(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
+
+    // Send a message
+
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
+    let message = app
+        .messages()
+        .send(
+            &channel.id,
+            &sender,
+            &fixtures::now(),
+            &fixtures::message::propose(),
+        )
+        .await
+        .expect("sending a message succeeds");
+
+    // Verify that an event is delivered
+
+    let _ = events
+        .filter_map(fixtures::message::events)
+        .filter_map(fixtures::message::sent)
+        .filter(|event| future::ready(event.message == message))
+        .next()
+        .immediately()
+        .await
+        .expect("delivered message sent event");
+}
+
+#[tokio::test]
+async fn previously_sent() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+
+    // Send a message
+
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
+    let message = app
+        .messages()
+        .send(
+            &channel.id,
+            &sender,
+            &fixtures::now(),
+            &fixtures::message::propose(),
+        )
+        .await
+        .expect("sending a message succeeds");
+
+    // Call the endpoint
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) =
+        get::handler(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
+
+    // Verify that an event is delivered
+
+    let _ = events
+        .filter_map(fixtures::message::events)
+        .filter_map(fixtures::message::sent)
+        .filter(|event| future::ready(event.message == message))
+        .next()
+        .immediately()
+        .await
+        .expect("delivered message sent event");
+}
+
+#[tokio::test]
+async fn sent_in_multiple_channels() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
+
+    let channels = [
+        fixtures::channel::create(&app, &fixtures::now()).await,
+        fixtures::channel::create(&app, &fixtures::now()).await,
+    ];
+
+    let messages = stream::iter(channels)
+        .then(|channel| {
+            let app = app.clone();
+            let sender = sender.clone();
+            let channel = channel.clone();
+            async move { fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await }
+        })
+        .collect::<Vec<_>>()
+        .await;
+
+    // Call the endpoint
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) = get::handler(State(app), subscriber, None, Query::default())
+        .await
+        .expect("subscribe never fails");
+
+    // Verify the structure of the response.
+
+    let events = events
+        .filter_map(fixtures::message::events)
+        .filter_map(fixtures::message::sent)
+        .take(messages.len())
+        .collect::<Vec<_>>()
+        .immediately()
+        .await;
+
+    for message in &messages {
+        assert!(events.iter().any(|event| &event.message == message));
+    }
+}
+
+#[tokio::test]
+async fn sent_sequentially() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
+
+    let messages = vec![
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+    ];
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) = get::handler(State(app), subscriber, None, Query::default())
+        .await
+        .expect("subscribe never fails");
+
+    // Verify the expected events in the expected order
+
+    let mut events = events
+        .filter_map(fixtures::message::events)
+        .filter_map(fixtures::message::sent)
+        .filter(|event| future::ready(messages.iter().any(|message| &event.message == message)));
+
+    for message in &messages {
+        let event = events
+            .next()
+            .immediately()
+            .await
+            .expect("undelivered messages remaining");
+
+        assert_eq!(message, &event.message);
+    }
+}
+
+#[tokio::test]
+async fn expiring() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
+    let sender = fixtures::login::create(&app, &fixtures::ancient()).await;
+    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::ancient()).await;
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) =
+        get::handler(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
+
+    // Expire messages
+
+    app.messages()
+        .expire(&fixtures::now())
+        .await
+        .expect("expiring messages always succeeds");
+
+    // Check for expiry event
+    let _ = events
+        .filter_map(fixtures::message::events)
+        .filter_map(fixtures::message::deleted)
+        .filter(|event| future::ready(event.id == message.id))
+        .next()
+        .immediately()
+        .await
+        .expect("a deleted message event will be delivered");
+}
+
+#[tokio::test]
+async fn previously_expired() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
+    let sender = fixtures::login::create(&app, &fixtures::ancient()).await;
+    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::ancient()).await;
+
+    // Expire messages
+
+    app.messages()
+        .expire(&fixtures::now())
+        .await
+        .expect("expiring messages always succeeds");
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) =
+        get::handler(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
+
+    // Check for expiry event
+    let _ = events
+        .filter_map(fixtures::message::events)
+        .filter_map(fixtures::message::deleted)
+        .filter(|event| future::ready(event.id == message.id))
+        .next()
+        .immediately()
+        .await
+        .expect("a deleted message event will be delivered");
+}
+
+#[tokio::test]
+async fn deleting() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
+    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) =
+        get::handler(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
+
+    // Delete the message
+
+    app.messages()
+        .delete(&message.id, &fixtures::now())
+        .await
+        .expect("deleting a valid message succeeds");
+
+    // Check for delete event
+    let _ = events
+        .filter_map(fixtures::message::events)
+        .filter_map(fixtures::message::deleted)
+        .filter(|event| future::ready(event.id == message.id))
+        .next()
+        .immediately()
+        .await
+        .expect("a deleted message event will be delivered");
+}
+
+#[tokio::test]
+async fn previously_deleted() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let sender = fixtures::login::create(&app, &fixtures::now()).await;
+    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
+
+    // Delete the message
+
+    app.messages()
+        .delete(&message.id, &fixtures::now())
+        .await
+        .expect("deleting a valid message succeeds");
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) =
+        get::handler(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
+
+    // Check for delete event
+    let _ = events
+        .filter_map(fixtures::message::events)
+        .filter_map(fixtures::message::deleted)
+        .filter(|event| future::ready(event.id == message.id))
+        .next()
+        .immediately()
+        .await
+        .expect("a deleted message event will be delivered");
+}
-- 
cgit v1.2.3


From eae0edb57e9ade7c73affb78baf2ae267b6290b8 Mon Sep 17 00:00:00 2001
From: Owen Jacobson <owen@grimoire.ca>
Date: Thu, 24 Oct 2024 22:50:23 -0400
Subject: Consolidate test helper event functions

---
 src/channel/routes/channel/test/post.rs |  2 +-
 src/channel/routes/test.rs              |  4 +-
 src/event/routes/test/channel.rs        | 30 ++++++-------
 src/event/routes/test/invite.rs         |  8 ++--
 src/event/routes/test/message.rs        | 32 ++++++-------
 src/event/routes/test/resume.rs         | 20 ++++-----
 src/event/routes/test/setup.rs          |  4 +-
 src/event/routes/test/token.rs          |  8 ++--
 src/test/fixtures/channel.rs            | 24 ----------
 src/test/fixtures/event.rs              | 79 +++++++++++++++++++++++++++++++++
 src/test/fixtures/login.rs              | 16 -------
 src/test/fixtures/message.rs            | 26 +----------
 src/test/fixtures/mod.rs                |  1 +
 13 files changed, 134 insertions(+), 120 deletions(-)
 create mode 100644 src/test/fixtures/event.rs

(limited to 'src/event/routes/test/message.rs')

diff --git a/src/channel/routes/channel/test/post.rs b/src/channel/routes/channel/test/post.rs
index 67e7d36..d81715f 100644
--- a/src/channel/routes/channel/test/post.rs
+++ b/src/channel/routes/channel/test/post.rs
@@ -44,7 +44,7 @@ async fn messages_in_order() {
         .subscribe(None)
         .await
         .expect("subscribing to a valid channel succeeds")
-        .filter_map(fixtures::message::events)
+        .filter_map(fixtures::event::message)
         .take(requests.len());
 
     let events = events.collect::<Vec<_>>().immediately().await;
diff --git a/src/channel/routes/test.rs b/src/channel/routes/test.rs
index 216eba1..46c58b0 100644
--- a/src/channel/routes/test.rs
+++ b/src/channel/routes/test.rs
@@ -48,8 +48,8 @@ async fn new_channel() {
         .subscribe(None)
         .await
         .expect("subscribing never fails")
-        .filter_map(fixtures::channel::events)
-        .filter_map(fixtures::channel::created)
+        .filter_map(fixtures::event::channel)
+        .filter_map(fixtures::event::channel::created)
         .filter(|event| future::ready(event.channel == response));
 
     let event = events
diff --git a/src/event/routes/test/channel.rs b/src/event/routes/test/channel.rs
index ac45bfc..0ab28c4 100644
--- a/src/event/routes/test/channel.rs
+++ b/src/event/routes/test/channel.rs
@@ -33,8 +33,8 @@ async fn creating() {
     // Verify channel created event
 
     let _ = events
-        .filter_map(fixtures::channel::events)
-        .filter_map(fixtures::channel::created)
+        .filter_map(fixtures::event::channel)
+        .filter_map(fixtures::event::channel::created)
         .filter(|event| future::ready(event.channel == channel))
         .next()
         .immediately()
@@ -68,8 +68,8 @@ async fn previously_created() {
     // Verify channel created event
 
     let _ = events
-        .filter_map(fixtures::channel::events)
-        .filter_map(fixtures::channel::created)
+        .filter_map(fixtures::event::channel)
+        .filter_map(fixtures::event::channel::created)
         .filter(|event| future::ready(event.channel == channel))
         .next()
         .immediately()
@@ -101,8 +101,8 @@ async fn expiring() {
 
     // Check for expiry event
     let _ = events
-        .filter_map(fixtures::channel::events)
-        .filter_map(fixtures::channel::deleted)
+        .filter_map(fixtures::event::channel)
+        .filter_map(fixtures::event::channel::deleted)
         .filter(|event| future::ready(event.id == channel.id))
         .next()
         .immediately()
@@ -134,8 +134,8 @@ async fn previously_expired() {
 
     // Check for expiry event
     let _ = events
-        .filter_map(fixtures::channel::events)
-        .filter_map(fixtures::channel::deleted)
+        .filter_map(fixtures::event::channel)
+        .filter_map(fixtures::event::channel::deleted)
         .filter(|event| future::ready(event.id == channel.id))
         .next()
         .immediately()
@@ -167,8 +167,8 @@ async fn deleting() {
 
     // Check for delete event
     let _ = events
-        .filter_map(fixtures::channel::events)
-        .filter_map(fixtures::channel::deleted)
+        .filter_map(fixtures::event::channel)
+        .filter_map(fixtures::event::channel::deleted)
         .filter(|event| future::ready(event.id == channel.id))
         .next()
         .immediately()
@@ -182,8 +182,6 @@ async fn previously_deleted() {
 
     let app = fixtures::scratch_app().await;
     let channel = fixtures::channel::create(&app, &fixtures::now()).await;
-    let sender = fixtures::login::create(&app, &fixtures::now()).await;
-    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
 
     // Delete the channel
 
@@ -202,11 +200,11 @@ async fn previously_deleted() {
 
     // Check for expiry event
     let _ = events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::deleted)
-        .filter(|event| future::ready(event.id == message.id))
+        .filter_map(fixtures::event::channel)
+        .filter_map(fixtures::event::channel::deleted)
+        .filter(|event| future::ready(event.id == channel.id))
         .next()
         .immediately()
         .await
-        .expect("a deleted message will be delivered");
+        .expect("a deleted channel event will be delivered");
 }
diff --git a/src/event/routes/test/invite.rs b/src/event/routes/test/invite.rs
index 10e4521..afd3aeb 100644
--- a/src/event/routes/test/invite.rs
+++ b/src/event/routes/test/invite.rs
@@ -35,8 +35,8 @@ async fn accepting_invite() {
     // Expect a login created event
 
     let _ = events
-        .filter_map(fixtures::login::events)
-        .filter_map(fixtures::login::created)
+        .filter_map(fixtures::event::login)
+        .filter_map(fixtures::event::login::created)
         .filter(|event| future::ready(event.login == joiner))
         .next()
         .immediately()
@@ -72,8 +72,8 @@ async fn previously_accepted_invite() {
     // Expect a login created event
 
     let _ = events
-        .filter_map(fixtures::login::events)
-        .filter_map(fixtures::login::created)
+        .filter_map(fixtures::event::login)
+        .filter_map(fixtures::event::login::created)
         .filter(|event| future::ready(event.login == joiner))
         .next()
         .immediately()
diff --git a/src/event/routes/test/message.rs b/src/event/routes/test/message.rs
index 9bbbc7d..df42a89 100644
--- a/src/event/routes/test/message.rs
+++ b/src/event/routes/test/message.rs
@@ -42,8 +42,8 @@ async fn sending() {
     // Verify that an event is delivered
 
     let _ = events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::sent)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::sent)
         .filter(|event| future::ready(event.message == message))
         .next()
         .immediately()
@@ -83,8 +83,8 @@ async fn previously_sent() {
     // Verify that an event is delivered
 
     let _ = events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::sent)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::sent)
         .filter(|event| future::ready(event.message == message))
         .next()
         .immediately()
@@ -124,8 +124,8 @@ async fn sent_in_multiple_channels() {
     // Verify the structure of the response.
 
     let events = events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::sent)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::sent)
         .take(messages.len())
         .collect::<Vec<_>>()
         .immediately()
@@ -160,8 +160,8 @@ async fn sent_sequentially() {
     // Verify the expected events in the expected order
 
     let mut events = events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::sent)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::sent)
         .filter(|event| future::ready(messages.iter().any(|message| &event.message == message)));
 
     for message in &messages {
@@ -201,8 +201,8 @@ async fn expiring() {
 
     // Check for expiry event
     let _ = events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::deleted)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::deleted)
         .filter(|event| future::ready(event.id == message.id))
         .next()
         .immediately()
@@ -236,8 +236,8 @@ async fn previously_expired() {
 
     // Check for expiry event
     let _ = events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::deleted)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::deleted)
         .filter(|event| future::ready(event.id == message.id))
         .next()
         .immediately()
@@ -271,8 +271,8 @@ async fn deleting() {
 
     // Check for delete event
     let _ = events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::deleted)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::deleted)
         .filter(|event| future::ready(event.id == message.id))
         .next()
         .immediately()
@@ -306,8 +306,8 @@ async fn previously_deleted() {
 
     // Check for delete event
     let _ = events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::deleted)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::deleted)
         .filter(|event| future::ready(event.id == message.id))
         .next()
         .immediately()
diff --git a/src/event/routes/test/resume.rs b/src/event/routes/test/resume.rs
index c393d38..e4751bb 100644
--- a/src/event/routes/test/resume.rs
+++ b/src/event/routes/test/resume.rs
@@ -40,8 +40,8 @@ async fn resume() {
         .expect("subscribe never fails");
 
         let event = events
-            .filter_map(fixtures::message::events)
-            .filter_map(fixtures::message::sent)
+            .filter_map(fixtures::event::message)
+            .filter_map(fixtures::event::message::sent)
             .filter(|event| future::ready(event.message == initial_message))
             .next()
             .immediately()
@@ -64,8 +64,8 @@ async fn resume() {
     // Verify final events
 
     let mut events = resumed
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::sent)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::sent)
         .zip(stream::iter(later_messages));
 
     while let Some((event, message)) = events.next().immediately().await {
@@ -124,8 +124,8 @@ async fn serial_resume() {
         // Check for expected events
 
         let events = events
-            .filter_map(fixtures::message::events)
-            .filter_map(fixtures::message::sent)
+            .filter_map(fixtures::event::message)
+            .filter_map(fixtures::event::message::sent)
             .zip(stream::iter(initial_messages))
             .collect::<Vec<_>>()
             .immediately()
@@ -165,8 +165,8 @@ async fn serial_resume() {
         // Check for expected events
 
         let events = events
-            .filter_map(fixtures::message::events)
-            .filter_map(fixtures::message::sent)
+            .filter_map(fixtures::event::message)
+            .filter_map(fixtures::event::message::sent)
             .zip(stream::iter(resume_messages))
             .collect::<Vec<_>>()
             .immediately()
@@ -206,8 +206,8 @@ async fn serial_resume() {
         // Check for expected events
 
         let events = events
-            .filter_map(fixtures::message::events)
-            .filter_map(fixtures::message::sent)
+            .filter_map(fixtures::event::message)
+            .filter_map(fixtures::event::message::sent)
             .zip(stream::iter(final_messages))
             .collect::<Vec<_>>()
             .immediately()
diff --git a/src/event/routes/test/setup.rs b/src/event/routes/test/setup.rs
index 234c2d9..a54b65b 100644
--- a/src/event/routes/test/setup.rs
+++ b/src/event/routes/test/setup.rs
@@ -36,8 +36,8 @@ async fn previously_completed() {
     // Expect a login created event
 
     let _ = events
-        .filter_map(fixtures::login::events)
-        .filter_map(fixtures::login::created)
+        .filter_map(fixtures::event::login)
+        .filter_map(fixtures::event::login::created)
         .filter(|event| future::ready(event.login == owner))
         .next()
         .immediately()
diff --git a/src/event/routes/test/token.rs b/src/event/routes/test/token.rs
index a545988..577fabd 100644
--- a/src/event/routes/test/token.rs
+++ b/src/event/routes/test/token.rs
@@ -41,8 +41,8 @@ async fn terminates_on_token_expiry() {
     ];
 
     assert!(events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::sent)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::sent)
         .filter(|event| future::ready(messages.iter().any(|message| &event.message == message)))
         .next()
         .immediately()
@@ -87,8 +87,8 @@ async fn terminates_on_logout() {
     ];
 
     assert!(events
-        .filter_map(fixtures::message::events)
-        .filter_map(fixtures::message::sent)
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::sent)
         .filter(|event| future::ready(messages.iter().any(|message| &event.message == message)))
         .next()
         .immediately()
diff --git a/src/test/fixtures/channel.rs b/src/test/fixtures/channel.rs
index 1fd8d23..0c6480b 100644
--- a/src/test/fixtures/channel.rs
+++ b/src/test/fixtures/channel.rs
@@ -1,5 +1,3 @@
-use std::future;
-
 use faker_rand::{
     en_us::{addresses::CityName, names::FullName},
     faker_impl_from_templates,
@@ -10,7 +8,6 @@ use crate::{
     app::App,
     channel::{self, Channel},
     clock::RequestedAt,
-    event::Event,
     name::Name,
 };
 
@@ -31,27 +28,6 @@ faker_impl_from_templates! {
     NameTemplate; "{} {}", CityName, FullName;
 }
 
-pub fn events(event: Event) -> future::Ready<Option<channel::Event>> {
-    future::ready(match event {
-        Event::Channel(channel) => Some(channel),
-        _ => None,
-    })
-}
-
-pub fn created(event: channel::Event) -> future::Ready<Option<channel::event::Created>> {
-    future::ready(match event {
-        channel::Event::Created(event) => Some(event),
-        channel::Event::Deleted(_) => None,
-    })
-}
-
-pub fn deleted(event: channel::Event) -> future::Ready<Option<channel::event::Deleted>> {
-    future::ready(match event {
-        channel::Event::Deleted(event) => Some(event),
-        channel::Event::Created(_) => None,
-    })
-}
-
 pub fn fictitious() -> channel::Id {
     channel::Id::generate()
 }
diff --git a/src/test/fixtures/event.rs b/src/test/fixtures/event.rs
new file mode 100644
index 0000000..de02d4d
--- /dev/null
+++ b/src/test/fixtures/event.rs
@@ -0,0 +1,79 @@
+use std::future::{self, Ready};
+
+use crate::event::Event;
+
+pub fn channel(event: Event) -> Ready<Option<channel::Event>> {
+    future::ready(match event {
+        Event::Channel(channel) => Some(channel),
+        _ => None,
+    })
+}
+
+pub fn message(event: Event) -> Ready<Option<message::Event>> {
+    future::ready(match event {
+        Event::Message(event) => Some(event),
+        _ => None,
+    })
+}
+
+pub fn login(event: Event) -> Ready<Option<login::Event>> {
+    future::ready(match event {
+        Event::Login(event) => Some(event),
+        _ => None,
+    })
+}
+
+pub mod channel {
+    use std::future::{self, Ready};
+
+    use crate::channel::event;
+    pub use crate::channel::Event;
+
+    pub fn created(event: Event) -> Ready<Option<event::Created>> {
+        future::ready(match event {
+            Event::Created(event) => Some(event),
+            Event::Deleted(_) => None,
+        })
+    }
+
+    pub fn deleted(event: Event) -> Ready<Option<event::Deleted>> {
+        future::ready(match event {
+            Event::Deleted(event) => Some(event),
+            Event::Created(_) => None,
+        })
+    }
+}
+
+pub mod message {
+    use std::future::{self, Ready};
+
+    use crate::message::event;
+    pub use crate::message::Event;
+
+    pub fn sent(event: Event) -> Ready<Option<event::Sent>> {
+        future::ready(match event {
+            Event::Sent(event) => Some(event),
+            Event::Deleted(_) => None,
+        })
+    }
+
+    pub fn deleted(event: Event) -> future::Ready<Option<event::Deleted>> {
+        future::ready(match event {
+            Event::Deleted(event) => Some(event),
+            Event::Sent(_) => None,
+        })
+    }
+}
+
+pub mod login {
+    use std::future::{self, Ready};
+
+    use crate::login::event;
+    pub use crate::login::Event;
+
+    pub fn created(event: Event) -> Ready<Option<event::Created>> {
+        future::ready(match event {
+            Event::Created(event) => Some(event),
+        })
+    }
+}
diff --git a/src/test/fixtures/login.rs b/src/test/fixtures/login.rs
index cbcbdd4..e308289 100644
--- a/src/test/fixtures/login.rs
+++ b/src/test/fixtures/login.rs
@@ -1,12 +1,9 @@
-use std::future::{self, Ready};
-
 use faker_rand::en_us::internet;
 use uuid::Uuid;
 
 use crate::{
     app::App,
     clock::RequestedAt,
-    event::Event,
     login::{self, Login, Password},
     name::Name,
 };
@@ -48,16 +45,3 @@ fn propose_name() -> Name {
 pub fn propose_password() -> Password {
     Uuid::new_v4().to_string().into()
 }
-
-pub fn events(event: Event) -> Ready<Option<login::Event>> {
-    future::ready(match event {
-        Event::Login(event) => Some(event),
-        _ => None,
-    })
-}
-
-pub fn created(event: login::Event) -> Ready<Option<login::event::Created>> {
-    future::ready(match event {
-        login::Event::Created(event) => Some(event),
-    })
-}
diff --git a/src/test/fixtures/message.rs b/src/test/fixtures/message.rs
index 8cb50c1..d3b4719 100644
--- a/src/test/fixtures/message.rs
+++ b/src/test/fixtures/message.rs
@@ -1,14 +1,11 @@
-use std::future;
-
 use faker_rand::lorem::Paragraphs;
 
 use crate::{
     app::App,
     channel::Channel,
     clock::RequestedAt,
-    event::Event,
     login::Login,
-    message::{self, event, Body, Message},
+    message::{self, Body, Message},
 };
 
 pub async fn send(app: &App, channel: &Channel, sender: &Login, sent_at: &RequestedAt) -> Message {
@@ -24,27 +21,6 @@ pub fn propose() -> Body {
     rand::random::<Paragraphs>().to_string().into()
 }
 
-pub fn events(event: Event) -> future::Ready<Option<message::Event>> {
-    future::ready(match event {
-        Event::Message(event) => Some(event),
-        _ => None,
-    })
-}
-
-pub fn sent(event: message::Event) -> future::Ready<Option<event::Sent>> {
-    future::ready(match event {
-        message::Event::Sent(event) => Some(event),
-        message::Event::Deleted(_) => None,
-    })
-}
-
-pub fn deleted(event: message::Event) -> future::Ready<Option<event::Deleted>> {
-    future::ready(match event {
-        message::Event::Deleted(event) => Some(event),
-        message::Event::Sent(_) => None,
-    })
-}
-
 pub fn fictitious() -> message::Id {
     message::Id::generate()
 }
diff --git a/src/test/fixtures/mod.rs b/src/test/fixtures/mod.rs
index cf30e02..2b7b6af 100644
--- a/src/test/fixtures/mod.rs
+++ b/src/test/fixtures/mod.rs
@@ -4,6 +4,7 @@ use crate::{app::App, clock::RequestedAt, db};
 
 pub mod channel;
 pub mod cookie;
+pub mod event;
 pub mod future;
 pub mod identity;
 pub mod invite;
-- 
cgit v1.2.3


From 5423ec3937a4e28f3958a71b3db7498a4c427dc1 Mon Sep 17 00:00:00 2001
From: Owen Jacobson <owen@grimoire.ca>
Date: Fri, 25 Oct 2024 00:33:16 -0400
Subject: 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>`).
---
 Cargo.lock                              |  21 +++
 Cargo.toml                              |   1 +
 src/channel/app.rs                      |  10 +-
 src/channel/routes/channel/test/post.rs |  27 ++--
 src/channel/routes/test.rs              |  11 +-
 src/event/routes/test/channel.rs        |  69 ++++++---
 src/event/routes/test/invite.rs         |  12 +-
 src/event/routes/test/message.rs        |  77 +++++++---
 src/event/routes/test/resume.rs         |  15 +-
 src/event/routes/test/setup.rs          |   7 +-
 src/event/routes/test/token.rs          |  16 +--
 src/test/fixtures/future.rs             | 240 +++++++++++++++++++++++++++-----
 src/ui/mime.rs                          |   5 +-
 src/ui/routes/ch/channel.rs             |  19 ++-
 14 files changed, 386 insertions(+), 144 deletions(-)

(limited to 'src/event/routes/test/message.rs')

diff --git a/Cargo.lock b/Cargo.lock
index 2b8eb5e..f5ba5ff 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -815,6 +815,7 @@ dependencies = [
  "itertools",
  "mime",
  "password-hash",
+ "pin-project",
  "rand",
  "rand_core",
  "rusqlite",
@@ -1260,6 +1261,26 @@ version = "2.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
 
+[[package]]
+name = "pin-project"
+version = "1.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "be57f64e946e500c8ee36ef6331845d40a93055567ec57e8fae13efd33759b95"
+dependencies = [
+ "pin-project-internal",
+]
+
+[[package]]
+name = "pin-project-internal"
+version = "1.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3c0f5fad0874fc7abcd4d750e76917eaebbecaa2c20bde22e1dbeeba8beb758c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "pin-project-lite"
 version = "0.2.14"
diff --git a/Cargo.toml b/Cargo.toml
index e9c9616..989f8cc 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -39,4 +39,5 @@ uuid = { version = "1.11.0", features = ["v4"] }
 
 [dev-dependencies]
 faker_rand = "0.1.1"
+pin-project = "1.1.7"
 rand = "0.8.5"
diff --git a/src/channel/app.rs b/src/channel/app.rs
index 7bfa0f7..8359277 100644
--- a/src/channel/app.rs
+++ b/src/channel/app.rs
@@ -4,7 +4,7 @@ use sqlx::sqlite::SqlitePool;
 
 use super::{
     repo::{LoadError, Provider as _},
-    Channel, History, Id,
+    Channel, Id,
 };
 use crate::{
     clock::DateTime,
@@ -42,12 +42,14 @@ impl<'a> Channels<'a> {
 
     // This function is careless with respect to time, and gets you the channel as
     // it exists in the specific moment when you call it.
-    pub async fn get(&self, channel: &Id) -> Result<Option<Channel>, Error> {
+    pub async fn get(&self, channel: &Id) -> Result<Channel, Error> {
+        let not_found = || Error::NotFound(channel.clone());
+
         let mut tx = self.db.begin().await?;
-        let channel = tx.channels().by_id(channel).await.optional()?;
+        let channel = tx.channels().by_id(channel).await.not_found(not_found)?;
         tx.commit().await?;
 
-        Ok(channel.as_ref().and_then(History::as_snapshot))
+        channel.as_snapshot().ok_or_else(not_found)
     }
 
     pub async fn delete(&self, channel: &Id, deleted_at: &DateTime) -> Result<(), Error> {
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);
 }
diff --git a/src/event/routes/test/channel.rs b/src/event/routes/test/channel.rs
index 0ab28c4..6a0a803 100644
--- a/src/event/routes/test/channel.rs
+++ b/src/event/routes/test/channel.rs
@@ -4,7 +4,7 @@ use futures::{future, stream::StreamExt as _};
 
 use crate::{
     event::routes::get,
-    test::fixtures::{self, future::Immediately as _},
+    test::fixtures::{self, future::Expect as _},
 };
 
 #[tokio::test]
@@ -32,14 +32,13 @@ async fn creating() {
 
     // Verify channel created event
 
-    let _ = events
+    events
         .filter_map(fixtures::event::channel)
         .filter_map(fixtures::event::channel::created)
         .filter(|event| future::ready(event.channel == channel))
         .next()
-        .immediately()
-        .await
-        .expect("channel created event is delivered");
+        .expect_some("channel created event is delivered")
+        .await;
 }
 
 #[tokio::test]
@@ -72,9 +71,8 @@ async fn previously_created() {
         .filter_map(fixtures::event::channel::created)
         .filter(|event| future::ready(event.channel == channel))
         .next()
-        .immediately()
-        .await
-        .expect("channel created event is delivered");
+        .expect_some("channel created event is delivered")
+        .await;
 }
 
 #[tokio::test]
@@ -105,9 +103,8 @@ async fn expiring() {
         .filter_map(fixtures::event::channel::deleted)
         .filter(|event| future::ready(event.id == channel.id))
         .next()
-        .immediately()
-        .await
-        .expect("a deleted channel event will be delivered");
+        .expect_some("a deleted channel event will be delivered")
+        .await;
 }
 
 #[tokio::test]
@@ -138,9 +135,8 @@ async fn previously_expired() {
         .filter_map(fixtures::event::channel::deleted)
         .filter(|event| future::ready(event.id == channel.id))
         .next()
-        .immediately()
-        .await
-        .expect("a deleted channel event will be delivered");
+        .expect_some("a deleted channel event will be delivered")
+        .await;
 }
 
 #[tokio::test]
@@ -171,9 +167,8 @@ async fn deleting() {
         .filter_map(fixtures::event::channel::deleted)
         .filter(|event| future::ready(event.id == channel.id))
         .next()
-        .immediately()
-        .await
-        .expect("a deleted channel event will be delivered");
+        .expect_some("a deleted channel event will be delivered")
+        .await;
 }
 
 #[tokio::test]
@@ -204,7 +199,43 @@ async fn previously_deleted() {
         .filter_map(fixtures::event::channel::deleted)
         .filter(|event| future::ready(event.id == channel.id))
         .next()
-        .immediately()
+        .expect_some("a deleted channel event will be delivered")
+        .await;
+}
+
+#[tokio::test]
+async fn previously_purged() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
+
+    // Delete and purge the channel
+
+    app.channels()
+        .delete(&channel.id, &fixtures::ancient())
+        .await
+        .expect("deleting a valid channel succeeds");
+
+    app.channels()
+        .purge(&fixtures::now())
         .await
-        .expect("a deleted channel event will be delivered");
+        .expect("purging channels always succeeds");
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) =
+        get::handler(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
+
+    // Check for expiry event
+    events
+        .filter_map(fixtures::event::channel)
+        .filter_map(fixtures::event::channel::deleted)
+        .filter(|event| future::ready(event.id == channel.id))
+        .next()
+        .expect_wait("deleted channel events not delivered")
+        .await;
 }
diff --git a/src/event/routes/test/invite.rs b/src/event/routes/test/invite.rs
index afd3aeb..d24f474 100644
--- a/src/event/routes/test/invite.rs
+++ b/src/event/routes/test/invite.rs
@@ -4,7 +4,7 @@ use futures::{future, stream::StreamExt as _};
 
 use crate::{
     event::routes::get,
-    test::fixtures::{self, future::Immediately as _},
+    test::fixtures::{self, future::Expect as _},
 };
 
 #[tokio::test]
@@ -39,9 +39,8 @@ async fn accepting_invite() {
         .filter_map(fixtures::event::login::created)
         .filter(|event| future::ready(event.login == joiner))
         .next()
-        .immediately()
-        .await
-        .expect("a login created event is sent");
+        .expect_some("a login created event is sent")
+        .await;
 }
 
 #[tokio::test]
@@ -76,7 +75,6 @@ async fn previously_accepted_invite() {
         .filter_map(fixtures::event::login::created)
         .filter(|event| future::ready(event.login == joiner))
         .next()
-        .immediately()
-        .await
-        .expect("a login created event is sent");
+        .expect_some("a login created event is sent")
+        .await;
 }
diff --git a/src/event/routes/test/message.rs b/src/event/routes/test/message.rs
index df42a89..63a3f43 100644
--- a/src/event/routes/test/message.rs
+++ b/src/event/routes/test/message.rs
@@ -7,7 +7,7 @@ use futures::{
 
 use crate::{
     event::routes::get,
-    test::fixtures::{self, future::Immediately as _},
+    test::fixtures::{self, future::Expect as _},
 };
 
 #[tokio::test]
@@ -46,9 +46,8 @@ async fn sending() {
         .filter_map(fixtures::event::message::sent)
         .filter(|event| future::ready(event.message == message))
         .next()
-        .immediately()
-        .await
-        .expect("delivered message sent event");
+        .expect_some("delivered message sent event")
+        .await;
 }
 
 #[tokio::test]
@@ -87,9 +86,8 @@ async fn previously_sent() {
         .filter_map(fixtures::event::message::sent)
         .filter(|event| future::ready(event.message == message))
         .next()
-        .immediately()
-        .await
-        .expect("delivered message sent event");
+        .expect_some("delivered message sent event")
+        .await;
 }
 
 #[tokio::test]
@@ -128,7 +126,7 @@ async fn sent_in_multiple_channels() {
         .filter_map(fixtures::event::message::sent)
         .take(messages.len())
         .collect::<Vec<_>>()
-        .immediately()
+        .expect_ready("events ready")
         .await;
 
     for message in &messages {
@@ -167,9 +165,8 @@ async fn sent_sequentially() {
     for message in &messages {
         let event = events
             .next()
-            .immediately()
-            .await
-            .expect("undelivered messages remaining");
+            .expect_some("undelivered messages remaining")
+            .await;
 
         assert_eq!(message, &event.message);
     }
@@ -205,9 +202,8 @@ async fn expiring() {
         .filter_map(fixtures::event::message::deleted)
         .filter(|event| future::ready(event.id == message.id))
         .next()
-        .immediately()
-        .await
-        .expect("a deleted message event will be delivered");
+        .expect_some("a deleted message event will be delivered")
+        .await;
 }
 
 #[tokio::test]
@@ -240,9 +236,8 @@ async fn previously_expired() {
         .filter_map(fixtures::event::message::deleted)
         .filter(|event| future::ready(event.id == message.id))
         .next()
-        .immediately()
-        .await
-        .expect("a deleted message event will be delivered");
+        .expect_some("a deleted message event will be delivered")
+        .await;
 }
 
 #[tokio::test]
@@ -275,9 +270,8 @@ async fn deleting() {
         .filter_map(fixtures::event::message::deleted)
         .filter(|event| future::ready(event.id == message.id))
         .next()
-        .immediately()
-        .await
-        .expect("a deleted message event will be delivered");
+        .expect_some("a deleted message event will be delivered")
+        .await;
 }
 
 #[tokio::test]
@@ -310,7 +304,46 @@ async fn previously_deleted() {
         .filter_map(fixtures::event::message::deleted)
         .filter(|event| future::ready(event.id == message.id))
         .next()
-        .immediately()
+        .expect_some("a deleted message event will be delivered")
+        .await;
+}
+
+#[tokio::test]
+async fn previously_purged() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
+    let sender = fixtures::login::create(&app, &fixtures::ancient()).await;
+    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::ancient()).await;
+
+    // Purge the message
+
+    app.messages()
+        .delete(&message.id, &fixtures::ancient())
+        .await
+        .expect("deleting a valid message succeeds");
+
+    app.messages()
+        .purge(&fixtures::now())
         .await
-        .expect("a deleted message event will be delivered");
+        .expect("purge always succeeds");
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let get::Response(events) =
+        get::handler(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
+
+    // Check for delete event
+
+    events
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::deleted)
+        .filter(|event| future::ready(event.id == message.id))
+        .next()
+        .expect_wait("no deleted message will be delivered")
+        .await;
 }
diff --git a/src/event/routes/test/resume.rs b/src/event/routes/test/resume.rs
index e4751bb..62b9bad 100644
--- a/src/event/routes/test/resume.rs
+++ b/src/event/routes/test/resume.rs
@@ -6,7 +6,7 @@ use futures::stream::{self, StreamExt as _};
 
 use crate::{
     event::{routes::get, Sequenced as _},
-    test::fixtures::{self, future::Immediately as _},
+    test::fixtures::{self, future::Expect as _},
 };
 
 #[tokio::test]
@@ -44,9 +44,8 @@ async fn resume() {
             .filter_map(fixtures::event::message::sent)
             .filter(|event| future::ready(event.message == initial_message))
             .next()
-            .immediately()
-            .await
-            .expect("delivered events");
+            .expect_some("delivered event for initial message")
+            .await;
 
         event.sequence()
     };
@@ -68,7 +67,7 @@ async fn resume() {
         .filter_map(fixtures::event::message::sent)
         .zip(stream::iter(later_messages));
 
-    while let Some((event, message)) = events.next().immediately().await {
+    while let Some((event, message)) = events.next().expect_ready("event ready").await {
         assert_eq!(message, event.message);
     }
 }
@@ -128,7 +127,7 @@ async fn serial_resume() {
             .filter_map(fixtures::event::message::sent)
             .zip(stream::iter(initial_messages))
             .collect::<Vec<_>>()
-            .immediately()
+            .expect_ready("zipping a finite list of events is ready immediately")
             .await;
 
         assert!(events
@@ -169,7 +168,7 @@ async fn serial_resume() {
             .filter_map(fixtures::event::message::sent)
             .zip(stream::iter(resume_messages))
             .collect::<Vec<_>>()
-            .immediately()
+            .expect_ready("zipping a finite list of events is ready immediately")
             .await;
 
         assert!(events
@@ -210,7 +209,7 @@ async fn serial_resume() {
             .filter_map(fixtures::event::message::sent)
             .zip(stream::iter(final_messages))
             .collect::<Vec<_>>()
-            .immediately()
+            .expect_ready("zipping a finite list of events is ready immediately")
             .await;
 
         assert!(events
diff --git a/src/event/routes/test/setup.rs b/src/event/routes/test/setup.rs
index a54b65b..007b03d 100644
--- a/src/event/routes/test/setup.rs
+++ b/src/event/routes/test/setup.rs
@@ -4,7 +4,7 @@ use futures::{future, stream::StreamExt as _};
 
 use crate::{
     event::routes::get,
-    test::fixtures::{self, future::Immediately as _},
+    test::fixtures::{self, future::Expect as _},
 };
 
 // There's no test for this in subscribe-then-setup order because creating an
@@ -40,7 +40,6 @@ async fn previously_completed() {
         .filter_map(fixtures::event::login::created)
         .filter(|event| future::ready(event.login == owner))
         .next()
-        .immediately()
-        .await
-        .expect("a login created event is sent");
+        .expect_some("a login created event is sent")
+        .await;
 }
diff --git a/src/event/routes/test/token.rs b/src/event/routes/test/token.rs
index 577fabd..2039d9b 100644
--- a/src/event/routes/test/token.rs
+++ b/src/event/routes/test/token.rs
@@ -4,7 +4,7 @@ use futures::{future, stream::StreamExt as _};
 
 use crate::{
     event::routes::get,
-    test::fixtures::{self, future::Immediately as _},
+    test::fixtures::{self, future::Expect as _},
 };
 
 #[tokio::test]
@@ -40,14 +40,13 @@ async fn terminates_on_token_expiry() {
         fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
     ];
 
-    assert!(events
+    events
         .filter_map(fixtures::event::message)
         .filter_map(fixtures::event::message::sent)
         .filter(|event| future::ready(messages.iter().any(|message| &event.message == message)))
         .next()
-        .immediately()
-        .await
-        .is_none());
+        .expect_none("end of stream")
+        .await;
 }
 
 #[tokio::test]
@@ -86,12 +85,11 @@ async fn terminates_on_logout() {
         fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
     ];
 
-    assert!(events
+    events
         .filter_map(fixtures::event::message)
         .filter_map(fixtures::event::message::sent)
         .filter(|event| future::ready(messages.iter().any(|message| &event.message == message)))
         .next()
-        .immediately()
-        .await
-        .is_none());
+        .expect_none("end of stream")
+        .await;
 }
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)
+        }
     }
 }
diff --git a/src/ui/mime.rs b/src/ui/mime.rs
index 9c724f0..7818ac1 100644
--- a/src/ui/mime.rs
+++ b/src/ui/mime.rs
@@ -1,7 +1,10 @@
 use mime::Mime;
 use unix_path::Path;
 
-// Extremely manual; using `std::path` here would result in platform-dependent behaviour when it's not appropriate (the URLs passed here always use `/` and are parsed like URLs). Using `unix_path` might be an option, but it's not clearly
+// Extremely manual; using `std::path` here would result in platform-dependent
+// behaviour when it's not appropriate (the URLs passed here always use `/` and
+// are parsed like URLs). Using `unix_path` might be an option, but it's not
+// clearly
 pub fn from_path<P>(path: P) -> Result<Mime, mime::FromStrError>
 where
     P: AsRef<Path>,
diff --git a/src/ui/routes/ch/channel.rs b/src/ui/routes/ch/channel.rs
index a338f1f..a854f14 100644
--- a/src/ui/routes/ch/channel.rs
+++ b/src/ui/routes/ch/channel.rs
@@ -6,7 +6,7 @@ pub mod get {
 
     use crate::{
         app::App,
-        channel,
+        channel::{self, app},
         error::Internal,
         token::extract::Identity,
         ui::{
@@ -21,18 +21,14 @@ pub mod get {
         Path(channel): Path<channel::Id>,
     ) -> Result<Asset, Error> {
         let _ = identity.ok_or(Error::NotLoggedIn)?;
-        app.channels()
-            .get(&channel)
-            .await
-            .map_err(Error::internal)?
-            .ok_or(Error::NotFound)?;
+        app.channels().get(&channel).await.map_err(Error::from)?;
 
         Assets::index().map_err(Error::Internal)
     }
 
     #[derive(Debug, thiserror::Error)]
     pub enum Error {
-        #[error("requested channel not found")]
+        #[error("channel not found")]
         NotFound,
         #[error("not logged in")]
         NotLoggedIn,
@@ -40,9 +36,12 @@ pub mod get {
         Internal(Internal),
     }
 
-    impl Error {
-        fn internal(err: impl Into<Internal>) -> Self {
-            Self::Internal(err.into())
+    impl From<app::Error> for Error {
+        fn from(error: app::Error) -> Self {
+            match error {
+                app::Error::NotFound(_) | app::Error::Deleted(_) => Self::NotFound,
+                other => Self::Internal(other.into()),
+            }
         }
     }
 
-- 
cgit v1.2.3