1 files changed, 55 insertions, 22 deletions

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;
 }