1 files changed, 61 insertions, 5 deletions

diff --git a/src/event/routes/test/token.rs b/src/event/routes/test/token.rs
index 2039d9b..fa76865 100644
--- a/src/event/routes/test/token.rs
+++ b/src/event/routes/test/token.rs
@@ -14,6 +14,7 @@ async fn terminates_on_token_expiry() {
     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 resume_point = fixtures::boot::resume_point(&app).await;
 
     // Subscribe via the endpoint
 
@@ -21,10 +22,14 @@ async fn terminates_on_token_expiry() {
     let subscriber =
         fixtures::identity::logged_in(&app, &subscriber_creds, &fixtures::ancient()).await;
 
-    let get::Response(events) =
-        get::handler(State(app.clone()), subscriber, None, Query::default())
-            .await
-            .expect("subscribe never fails");
+    let get::Response(events) = get::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Verify the resulting stream's behaviour
 
@@ -56,6 +61,7 @@ async fn terminates_on_logout() {
     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 resume_point = fixtures::boot::resume_point(&app).await;
 
     // Subscribe via the endpoint
 
@@ -65,7 +71,7 @@ async fn terminates_on_logout() {
         State(app.clone()),
         subscriber.clone(),
         None,
-        Query::default(),
+        Query(get::QueryParams { resume_point }),
     )
     .await
     .expect("subscribe never fails");
@@ -93,3 +99,53 @@ async fn terminates_on_logout() {
         .expect_none("end of stream")
         .await;
 }
+
+#[tokio::test]
+async fn terminates_on_password_change() {
+    // 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 resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Subscribe via the endpoint
+
+    let creds = fixtures::login::create_with_password(&app, &fixtures::now()).await;
+    let cookie = fixtures::cookie::logged_in(&app, &creds, &fixtures::now()).await;
+    let subscriber = fixtures::identity::from_cookie(&app, &cookie, &fixtures::now()).await;
+
+    let get::Response(events) = get::handler(
+        State(app.clone()),
+        subscriber.clone(),
+        None,
+        Query(get::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Verify the resulting stream's behaviour
+
+    let (_, password) = creds;
+    let to = fixtures::login::propose_password();
+    app.tokens()
+        .change_password(&subscriber.login, &password, &to, &fixtures::now())
+        .await
+        .expect("expiring tokens succeeds");
+
+    // These should not be delivered.
+
+    let messages = [
+        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,
+    ];
+
+    events
+        .filter_map(fixtures::event::message)
+        .filter_map(fixtures::event::message::sent)
+        .filter(|event| future::ready(messages.iter().any(|message| &event.message == message)))
+        .next()
+        .expect_none("end of stream")
+        .await;
+}