3 files changed, 371 insertions, 105 deletions

diff --git a/src/push/handlers/ping/mod.rs b/src/push/handlers/ping/mod.rs
index db828fa..2a86984 100644
--- a/src/push/handlers/ping/mod.rs
+++ b/src/push/handlers/ping/mod.rs
@@ -1,7 +1,10 @@
 use axum::{Json, extract::State, http::StatusCode};
-use web_push::WebPushClient;
 
-use crate::{error::Internal, push::app::Push, token::extract::Identity};
+use crate::{
+    error::Internal,
+    push::{Publish, app::Push},
+    token::extract::Identity,
+};
 
 #[cfg(test)]
 mod test;
@@ -15,7 +18,7 @@ pub async fn handler<P>(
     Json(_): Json<Request>,
 ) -> Result<StatusCode, Internal>
 where
-    P: WebPushClient,
+    P: Publish,
 {
     push.ping(&identity.login).await?;
 
diff --git a/src/push/handlers/ping/test.rs b/src/push/handlers/ping/test.rs
index 5725131..cc07ef0 100644
--- a/src/push/handlers/ping/test.rs
+++ b/src/push/handlers/ping/test.rs
@@ -1,9 +1,13 @@
+use std::{collections::HashSet, io};
+
 use axum::{
     extract::{Json, State},
     http::StatusCode,
 };
+use itertools::Itertools;
+use web_push::{SubscriptionInfo, WebPushError};
 
-use crate::test::fixtures;
+use crate::{event::Heartbeat, test::fixtures};
 
 #[tokio::test]
 async fn ping_without_subscriptions() {
@@ -11,30 +15,360 @@ async fn ping_without_subscriptions() {
 
     let recipient = fixtures::identity::create(&app, &fixtures::now()).await;
 
-    app.vapid()
-        .refresh_key(&fixtures::now())
+    let response = super::handler(State(app.push()), recipient, Json(super::Request {}))
+        .await
+        .expect("sending a ping with no subscriptions always succeeds");
+
+    assert_eq!(StatusCode::ACCEPTED, response);
+
+    assert!(
+        app.publisher()
+            .sent()
+            .into_iter()
+            .filter(|publish| publish.message_eq(&Heartbeat::Heartbeat)
+                && publish.recipients.is_empty())
+            .exactly_one()
+            .is_ok()
+    );
+}
+
+#[tokio::test]
+async fn ping() {
+    let app = fixtures::scratch_app().await;
+
+    let recipient = fixtures::identity::create(&app, &fixtures::now()).await;
+    let vapid = fixtures::vapid::key(&app).await;
+
+    // Create a subscription
+    let subscription = SubscriptionInfo::new(
+        "https://push.example.com/endpoint",
+        "testing-p256dh-key",
+        "testing-auth",
+    );
+    app.push()
+        .subscribe(&recipient, &subscription, &vapid)
+        .await
+        .expect("creating a subscription succeeds");
+
+    // Send a ping
+    let response = super::handler(State(app.push()), recipient, Json(super::Request {}))
+        .await
+        .expect("sending a ping succeeds");
+
+    assert_eq!(StatusCode::ACCEPTED, response);
+
+    // Confirm that it was actually sent
+    let subscriptions = HashSet::from([subscription]);
+    assert!(
+        app.publisher()
+            .sent()
+            .into_iter()
+            .filter(|publish| publish.message_eq(&Heartbeat::Heartbeat)
+                && publish.recipients == subscriptions)
+            .exactly_one()
+            .is_ok()
+    );
+}
+
+#[tokio::test]
+async fn ping_multiple_subscriptions() {
+    let app = fixtures::scratch_app().await;
+
+    let recipient = fixtures::identity::create(&app, &fixtures::now()).await;
+    let vapid = fixtures::vapid::key(&app).await;
+
+    // Create a subscription
+    let subscriptions = HashSet::from([
+        SubscriptionInfo::new(
+            "https://push.example.com/endpoint-1",
+            "testing-p256dh-key-1",
+            "testing-auth-1",
+        ),
+        SubscriptionInfo::new(
+            "https://push.example.com/endpoint-2",
+            "testing-p256dh-key-2",
+            "testing-auth-2",
+        ),
+    ]);
+    for subscription in &subscriptions {
+        app.push()
+            .subscribe(&recipient, subscription, &vapid)
+            .await
+            .expect("creating a subscription succeeds");
+    }
+
+    // Send a ping
+    let response = super::handler(State(app.push()), recipient, Json(super::Request {}))
         .await
-        .expect("refreshing the VAPID key always succeeds");
+        .expect("sending a ping succeeds");
+
+    assert_eq!(StatusCode::ACCEPTED, response);
 
+    // Confirm that it was actually sent
+    assert!(
+        app.publisher()
+            .sent()
+            .into_iter()
+            .filter(|publish| publish.message_eq(&Heartbeat::Heartbeat)
+                && publish.recipients == subscriptions)
+            .exactly_one()
+            .is_ok()
+    );
+}
+
+#[tokio::test]
+async fn ping_recipient_only() {
+    let app = fixtures::scratch_app().await;
+
+    let recipient = fixtures::identity::create(&app, &fixtures::now()).await;
+    let spectator = fixtures::identity::create(&app, &fixtures::now()).await;
+
+    let vapid = fixtures::vapid::key(&app).await;
+
+    // Create subscriptions for each user
+    let recipient_subscription = SubscriptionInfo::new(
+        "https://push.example.com/recipient/endpoint",
+        "recipient-p256dh-key",
+        "recipient-auth",
+    );
+    app.push()
+        .subscribe(&recipient, &recipient_subscription, &vapid)
+        .await
+        .expect("creating a subscription succeeds");
+
+    let spectator_subscription = SubscriptionInfo::new(
+        "https://push.example.com/spectator/endpoint",
+        "spectator-p256dh-key",
+        "spectator-auth",
+    );
+    app.push()
+        .subscribe(&spectator, &spectator_subscription, &vapid)
+        .await
+        .expect("creating a subscription succeeds");
+
+    // Send a ping
+    let response = super::handler(State(app.push()), recipient, Json(super::Request {}))
+        .await
+        .expect("sending a ping succeeds");
+
+    assert_eq!(StatusCode::ACCEPTED, response);
+
+    // Confirm that it was actually sent to the recipient
+    let sent = app.publisher().sent();
+
+    let recipient_subscriptions = HashSet::from([recipient_subscription]);
+    assert!(
+        sent.iter()
+            .filter(|publish| publish.message_eq(&Heartbeat::Heartbeat)
+                && publish.recipients == recipient_subscriptions)
+            .exactly_one()
+            .is_ok()
+    );
+
+    // Confirm that it was not sent to the spectator
+    assert!(
+        !sent
+            .iter()
+            .any(|publish| publish.recipients.contains(&spectator_subscription))
+    );
+}
+
+#[tokio::test]
+async fn ping_permanent_error() {
+    let app = fixtures::scratch_app().await;
+
+    let recipient = fixtures::identity::create(&app, &fixtures::now()).await;
+
+    let vapid = fixtures::vapid::key(&app).await;
+
+    // Create subscriptions
+    let subscription = SubscriptionInfo::new(
+        "https://push.example.com/recipient/endpoint",
+        "recipient-p256dh-key",
+        "recipient-auth",
+    );
+    app.push()
+        .subscribe(&recipient, &subscription, &vapid)
+        .await
+        .expect("creating a subscription succeeds");
+
+    // Prepare the next ping attempt to fail
+    app.publisher()
+        .fail_next(&subscription, WebPushError::InvalidUri);
+
+    // Send a ping
+    super::handler(
+        State(app.push()),
+        recipient.clone(),
+        Json(super::Request {}),
+    )
+    .await
+    .expect_err("sending a ping with a permanently-failing subscription fails");
+
+    // Confirm that it was actually sent
+    let sent = app.publisher().sent();
+
+    let subscriptions = HashSet::from([subscription.clone()]);
+    assert!(
+        sent.iter()
+            .filter(|publish| publish.message_eq(&Heartbeat::Heartbeat)
+                && publish.recipients == subscriptions)
+            .exactly_one()
+            .is_ok()
+    );
+
+    // Send a second ping
     let response = super::handler(State(app.push()), recipient, Json(super::Request {}))
         .await
         .expect("sending a ping with no subscriptions always succeeds");
 
     assert_eq!(StatusCode::ACCEPTED, response);
 
-    assert!(app.webpush().sent().is_empty());
+    // Confirm that it was _not_ sent this time, since it failed permanently last time
+    let sent = app.publisher().sent();
+
+    assert!(
+        !sent
+            .iter()
+            .any(|publish| publish.recipients.contains(&subscription))
+    );
 }
 
-// More complete testing requires that we figure out how to generate working p256 ECDH keys for
-// testing _with_, as `web_push` will actually parse and use those keys even if push messages are
-// ultimately never serialized or sent over HTTP.
-//
-// Tests that are missing:
-//
-// * Verify that subscribing and sending a ping causes a ping to be delivered to that subscription.
-// * Verify that two subscriptions both get pings.
-// * Verify that other users' subscriptions are not pinged.
-// * Verify that a ping that causes a permanent error causes the subscription to be deleted.
-// * Verify that a ping that causes a non-permanent error does not cause the subscription to be
-//   deleted.
-// * Verify that a failure on one subscription doesn't affect delivery on other subscriptions.
+#[tokio::test]
+async fn ping_temporary_error() {
+    let app = fixtures::scratch_app().await;
+
+    let recipient = fixtures::identity::create(&app, &fixtures::now()).await;
+
+    let vapid = fixtures::vapid::key(&app).await;
+
+    // Create subscriptions
+    let subscription = SubscriptionInfo::new(
+        "https://push.example.com/recipient/endpoint",
+        "recipient-p256dh-key",
+        "recipient-auth",
+    );
+    app.push()
+        .subscribe(&recipient, &subscription, &vapid)
+        .await
+        .expect("creating a subscription succeeds");
+
+    // Prepare the next ping attempt to fail
+    app.publisher().fail_next(
+        &subscription,
+        WebPushError::from(io::Error::other("transient IO error")),
+    );
+
+    // Send a ping
+    super::handler(
+        State(app.push()),
+        recipient.clone(),
+        Json(super::Request {}),
+    )
+    .await
+    .expect_err("sending a ping with a temporarily-failing subscription fails");
+
+    // Confirm that it was actually sent
+    let sent = app.publisher().sent();
+
+    let subscriptions = HashSet::from([subscription.clone()]);
+    assert!(
+        sent.iter()
+            .filter(|publish| publish.message_eq(&Heartbeat::Heartbeat)
+                && publish.recipients == subscriptions)
+            .exactly_one()
+            .is_ok()
+    );
+
+    // Send a second ping
+    let response = super::handler(State(app.push()), recipient, Json(super::Request {}))
+        .await
+        .expect("sending a ping subscription succeeds now that the transient error has cleared");
+    assert_eq!(StatusCode::ACCEPTED, response);
+
+    // Confirm that it was _not_ sent this time, since it failed permanently last time
+    let sent = app.publisher().sent();
+
+    let subscriptions = HashSet::from([subscription]);
+    assert!(
+        sent.iter()
+            .filter(|publish| publish.message_eq(&Heartbeat::Heartbeat)
+                && publish.recipients == subscriptions)
+            .exactly_one()
+            .is_ok()
+    );
+}
+
+#[tokio::test]
+async fn ping_multiple_subscriptions_with_failure() {
+    let app = fixtures::scratch_app().await;
+
+    let recipient = fixtures::identity::create(&app, &fixtures::now()).await;
+    let vapid = fixtures::vapid::key(&app).await;
+
+    // Create subscriptions
+    let failing = SubscriptionInfo::new(
+        "https://push.example.com/endpoint-failing",
+        "testing-p256dh-key-failing",
+        "testing-auth-failing",
+    );
+    let succeeding = SubscriptionInfo::new(
+        "https://push.example.com/endpoint-succeeding",
+        "testing-p256dh-key-succeeding",
+        "testing-auth-succeeding",
+    );
+    let subscriptions = HashSet::from([failing.clone(), succeeding.clone()]);
+    for subscription in &subscriptions {
+        app.push()
+            .subscribe(&recipient, subscription, &vapid)
+            .await
+            .expect("creating a subscription succeeds");
+    }
+
+    // Rig one of them to fail permanently
+    app.publisher()
+        .fail_next(&failing, WebPushError::InvalidUri);
+
+    // Send a ping
+    super::handler(
+        State(app.push()),
+        recipient.clone(),
+        Json(super::Request {}),
+    )
+    .await
+    .expect_err("sending a ping with a failing subscription fails");
+
+    // Confirm that it was actually sent to both subs the first time
+    let subscriptions = HashSet::from([failing.clone(), succeeding.clone()]);
+    assert!(
+        app.publisher()
+            .sent()
+            .iter()
+            .filter(|publish| publish.message_eq(&Heartbeat::Heartbeat)
+                && publish.recipients == subscriptions)
+            .exactly_one()
+            .is_ok()
+    );
+
+    // Send a second ping
+    let response = super::handler(State(app.push()), recipient, Json(super::Request {}))
+        .await
+        .expect("sending a second ping succeeds after the failing subscription is removed");
+    assert_eq!(StatusCode::ACCEPTED, response);
+
+    // Confirm that it was only sent to the succeeding subscription
+    let subscriptions = HashSet::from([succeeding]);
+    let sent = app.publisher().sent();
+    assert!(
+        sent.iter()
+            .filter(|publish| publish.message_eq(&Heartbeat::Heartbeat)
+                && publish.recipients == subscriptions)
+            .exactly_one()
+            .is_ok()
+    );
+    assert!(
+        !sent
+            .iter()
+            .any(|publish| publish.recipients.contains(&failing))
+    );
+}
diff --git a/src/push/handlers/subscribe/test.rs b/src/push/handlers/subscribe/test.rs
index b72624d..793bcef 100644
--- a/src/push/handlers/subscribe/test.rs
+++ b/src/push/handlers/subscribe/test.rs
@@ -3,33 +3,16 @@ use axum::{
     http::StatusCode,
 };
 
-use crate::{
-    push::app::SubscribeError,
-    test::{fixtures, fixtures::event},
-};
+use crate::{push::app::SubscribeError, test::fixtures};
 
 #[tokio::test]
 async fn accepts_new_subscription() {
     let app = fixtures::scratch_app().await;
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
 
-    // Issue a VAPID key.
-
-    app.vapid()
-        .refresh_key(&fixtures::now())
-        .await
-        .expect("refreshing the VAPID key always succeeds");
-
     // Find out what that VAPID key is.
 
-    let boot = app.boot().snapshot().await.expect("boot always succeeds");
-    let vapid = boot
-        .events
-        .into_iter()
-        .filter_map(event::vapid)
-        .filter_map(event::vapid::changed)
-        .next_back()
-        .expect("the application will have a vapid key after a refresh");
+    let vapid = fixtures::vapid::key(&app).await;
 
     // Create a dummy subscription with that key.
 
@@ -41,7 +24,7 @@ async fn accepts_new_subscription() {
                 auth: String::from("test-auth-value"),
             },
         },
-        vapid: vapid.key,
+        vapid,
     };
     let response = super::handler(State(app.push()), subscriber, Json(request))
         .await
@@ -57,23 +40,9 @@ async fn accepts_repeat_subscription() {
     let app = fixtures::scratch_app().await;
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
 
-    // Issue a VAPID key.
-
-    app.vapid()
-        .refresh_key(&fixtures::now())
-        .await
-        .expect("refreshing the VAPID key always succeeds");
-
     // Find out what that VAPID key is.
 
-    let boot = app.boot().snapshot().await.expect("boot always succeeds");
-    let vapid = boot
-        .events
-        .into_iter()
-        .filter_map(event::vapid)
-        .filter_map(event::vapid::changed)
-        .next_back()
-        .expect("the application will have a vapid key after a refresh");
+    let vapid = fixtures::vapid::key(&app).await;
 
     // Create a dummy subscription with that key.
 
@@ -85,7 +54,7 @@ async fn accepts_repeat_subscription() {
                 auth: String::from("test-auth-value"),
             },
         },
-        vapid: vapid.key,
+        vapid,
     };
     let response = super::handler(State(app.push()), subscriber.clone(), Json(request.clone()))
         .await
@@ -111,23 +80,9 @@ async fn rejects_duplicate_subscription() {
     let app = fixtures::scratch_app().await;
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
 
-    // Issue a VAPID key.
-
-    app.vapid()
-        .refresh_key(&fixtures::now())
-        .await
-        .expect("refreshing the VAPID key always succeeds");
-
     // Find out what that VAPID key is.
 
-    let boot = app.boot().snapshot().await.expect("boot always succeeds");
-    let vapid = boot
-        .events
-        .into_iter()
-        .filter_map(event::vapid)
-        .filter_map(event::vapid::changed)
-        .next_back()
-        .expect("the application will have a vapid key after a refresh");
+    let vapid = fixtures::vapid::key(&app).await;
 
     // Create a dummy subscription with that key.
 
@@ -139,7 +94,7 @@ async fn rejects_duplicate_subscription() {
                 auth: String::from("test-auth-value"),
             },
         },
-        vapid: vapid.key,
+        vapid,
     };
     super::handler(State(app.push()), subscriber.clone(), Json(request))
         .await
@@ -155,7 +110,7 @@ async fn rejects_duplicate_subscription() {
                 auth: String::from("different-test-auth-value"),
             },
         },
-        vapid: vapid.key,
+        vapid,
     };
     let response = super::handler(State(app.push()), subscriber, Json(request))
         .await
@@ -170,29 +125,12 @@ async fn rejects_duplicate_subscription() {
 async fn rejects_stale_vapid_key() {
     let app = fixtures::scratch_app().await;
     let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-
-    // Issue a VAPID key.
-
-    app.vapid()
-        .refresh_key(&fixtures::now())
-        .await
-        .expect("refreshing the VAPID key always succeeds");
-
-    // Find out what that VAPID key is.
-
-    let boot = app.boot().snapshot().await.expect("boot always succeeds");
-    let vapid = boot
-        .events
-        .into_iter()
-        .filter_map(event::vapid)
-        .filter_map(event::vapid::changed)
-        .next_back()
-        .expect("the application will have a vapid key after a refresh");
+    let stale_vapid = fixtures::vapid::key(&app).await;
 
     // Change the VAPID key.
 
     app.vapid()
-        .rotate_key()
+        .revoke_key()
         .await
         .expect("key rotation always succeeds");
     app.vapid()
@@ -200,16 +138,7 @@ async fn rejects_stale_vapid_key() {
         .await
         .expect("refreshing the VAPID key always succeeds");
 
-    // Find out what the new VAPID key is.
-
-    let boot = app.boot().snapshot().await.expect("boot always succeeds");
-    let fresh_vapid = boot
-        .events
-        .into_iter()
-        .filter_map(event::vapid)
-        .filter_map(event::vapid::changed)
-        .next_back()
-        .expect("the application will have a vapid key after a refresh");
+    let fresh_vapid = fixtures::vapid::key(&app).await;
 
     // Create a dummy subscription with the original key.
 
@@ -221,7 +150,7 @@ async fn rejects_stale_vapid_key() {
                 auth: String::from("test-auth-value"),
             },
         },
-        vapid: vapid.key,
+        vapid: stale_vapid,
     };
     let response = super::handler(State(app.push()), subscriber, Json(request))
         .await
@@ -231,6 +160,6 @@ async fn rejects_stale_vapid_key() {
 
     assert!(matches!(
         response,
-        super::Error(SubscribeError::StaleVapidKey(key)) if key == fresh_vapid.key
+        super::Error(SubscribeError::StaleVapidKey(key)) if key == fresh_vapid
     ));
 }