From f9cbf95e5b850a7407c34f936c0f858520682a5d Mon Sep 17 00:00:00 2001 From: Owen Jacobson Date: Thu, 24 Oct 2024 19:49:54 -0400 Subject: Tests for retrieving invites --- src/channel/routes/channel/test/delete.rs | 10 ++--- src/clock.rs | 2 +- src/invite/app.rs | 6 +-- src/invite/mod.rs | 2 +- src/invite/routes/invite/mod.rs | 2 + src/invite/routes/invite/test/get.rs | 65 +++++++++++++++++++++++++++++++ src/invite/routes/invite/test/mod.rs | 1 + src/invite/routes/mod.rs | 2 + src/invite/routes/post.rs | 2 +- src/invite/routes/test.rs | 28 +++++++++++++ src/test/fixtures/invite.rs | 17 ++++++++ src/test/fixtures/mod.rs | 1 + 12 files changed, 125 insertions(+), 13 deletions(-) create mode 100644 src/invite/routes/invite/test/get.rs create mode 100644 src/invite/routes/invite/test/mod.rs create mode 100644 src/invite/routes/test.rs create mode 100644 src/test/fixtures/invite.rs (limited to 'src') diff --git a/src/channel/routes/channel/test/delete.rs b/src/channel/routes/channel/test/delete.rs index e9af12f..e1210fd 100644 --- a/src/channel/routes/channel/test/delete.rs +++ b/src/channel/routes/channel/test/delete.rs @@ -9,7 +9,7 @@ use crate::{ }; #[tokio::test] -pub async fn delete_channel() { +pub async fn valid_channel() { // Set up the environment let app = fixtures::scratch_app().await; @@ -38,7 +38,7 @@ pub async fn delete_channel() { } #[tokio::test] -pub async fn delete_invalid_channel_id() { +pub async fn invalid_channel_id() { // Set up the environment let app = fixtures::scratch_app().await; @@ -62,7 +62,7 @@ pub async fn delete_invalid_channel_id() { } #[tokio::test] -pub async fn delete_deleted() { +pub async fn channel_deleted() { // Set up the environment let app = fixtures::scratch_app().await; @@ -91,7 +91,7 @@ pub async fn delete_deleted() { } #[tokio::test] -pub async fn delete_expired() { +pub async fn channel_expired() { // Set up the environment let app = fixtures::scratch_app().await; @@ -120,7 +120,7 @@ pub async fn delete_expired() { } #[tokio::test] -pub async fn delete_purged() { +pub async fn channel_purged() { // Set up the environment let app = fixtures::scratch_app().await; diff --git a/src/clock.rs b/src/clock.rs index 9ffef82..242bcdf 100644 --- a/src/clock.rs +++ b/src/clock.rs @@ -12,7 +12,7 @@ pub type DateTime = chrono::DateTime; // calculated once per request, even if the extractor is used in multiple // places. This requires the [middleware] function to be installed with // [axum::middleware::from_fn] around the current route. -#[derive(Clone)] +#[derive(Debug, Clone)] pub struct RequestedAt(pub DateTime); impl RequestedAt { diff --git a/src/invite/app.rs b/src/invite/app.rs index 64ba753..65e7721 100644 --- a/src/invite/app.rs +++ b/src/invite/app.rs @@ -20,11 +20,7 @@ impl<'a> Invites<'a> { Self { db } } - pub async fn create( - &self, - issuer: &Login, - issued_at: &DateTime, - ) -> Result { + pub async fn issue(&self, issuer: &Login, issued_at: &DateTime) -> Result { let mut tx = self.db.begin().await?; let invite = tx.invites().create(issuer, issued_at).await?; tx.commit().await?; diff --git a/src/invite/mod.rs b/src/invite/mod.rs index d59fb9c..53ca984 100644 --- a/src/invite/mod.rs +++ b/src/invite/mod.rs @@ -14,7 +14,7 @@ pub struct Invite { pub issued_at: DateTime, } -#[derive(serde::Serialize)] +#[derive(Debug, serde::Serialize)] pub struct Summary { pub id: Id, pub issuer: nfc::String, diff --git a/src/invite/routes/invite/mod.rs b/src/invite/routes/invite/mod.rs index 04593fd..c22029a 100644 --- a/src/invite/routes/invite/mod.rs +++ b/src/invite/routes/invite/mod.rs @@ -1,4 +1,6 @@ pub mod get; pub mod post; +#[cfg(test)] +pub mod test; type PathInfo = crate::invite::Id; diff --git a/src/invite/routes/invite/test/get.rs b/src/invite/routes/invite/test/get.rs new file mode 100644 index 0000000..c6780ed --- /dev/null +++ b/src/invite/routes/invite/test/get.rs @@ -0,0 +1,65 @@ +use axum::extract::{Json, Path, State}; + +use crate::{invite::routes::invite::get, test::fixtures}; + +#[tokio::test] +async fn valid_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let issuer = fixtures::login::create(&app, &fixtures::now()).await; + let invite = fixtures::invite::issue(&app, &issuer, &fixtures::now()).await; + + // Call endpoint + + let Json(response) = get::handler(State(app), Path(invite.id)) + .await + .expect("get for an existing invite succeeds"); + + // Verify response + + assert_eq!(issuer.name.display(), &response.issuer); + assert_eq!(invite.issued_at, response.issued_at); +} + +#[tokio::test] +async fn nonexistent_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + + // Call endpoint + + let invite = fixtures::invite::fictitious(); + let error = get::handler(State(app), Path(invite.clone())) + .await + .expect_err("get for a nonexistent invite fails"); + + // Verify response + + assert!(matches!(error, get::Error::NotFound(error_id) if invite == error_id)); +} + +#[tokio::test] +async fn expired_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let issuer = fixtures::login::create(&app, &fixtures::ancient()).await; + let invite = fixtures::invite::issue(&app, &issuer, &fixtures::ancient()).await; + + app.invites() + .expire(&fixtures::now()) + .await + .expect("expiring invites never fails"); + + // Call endpoint + + let error = get::handler(State(app), Path(invite.id.clone())) + .await + .expect_err("get for an expired invite fails"); + + // Verify response + + assert!(matches!(error, get::Error::NotFound(error_id) if invite.id == error_id)); +} diff --git a/src/invite/routes/invite/test/mod.rs b/src/invite/routes/invite/test/mod.rs new file mode 100644 index 0000000..a608fbb --- /dev/null +++ b/src/invite/routes/invite/test/mod.rs @@ -0,0 +1 @@ +mod get; diff --git a/src/invite/routes/mod.rs b/src/invite/routes/mod.rs index dae20ba..2f7375c 100644 --- a/src/invite/routes/mod.rs +++ b/src/invite/routes/mod.rs @@ -7,6 +7,8 @@ use crate::app::App; mod invite; mod post; +#[cfg(test)] +mod test; pub fn router() -> Router { Router::new() diff --git a/src/invite/routes/post.rs b/src/invite/routes/post.rs index eb7d706..898081e 100644 --- a/src/invite/routes/post.rs +++ b/src/invite/routes/post.rs @@ -10,7 +10,7 @@ pub async fn handler( identity: Identity, _: Json, ) -> Result, Internal> { - let invite = app.invites().create(&identity.login, &issued_at).await?; + let invite = app.invites().issue(&identity.login, &issued_at).await?; Ok(Json(invite)) } diff --git a/src/invite/routes/test.rs b/src/invite/routes/test.rs new file mode 100644 index 0000000..4d99660 --- /dev/null +++ b/src/invite/routes/test.rs @@ -0,0 +1,28 @@ +use axum::extract::{Json, State}; + +use super::post; +use crate::test::fixtures; + +#[tokio::test] +async fn create_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let issuer = fixtures::identity::create(&app, &fixtures::now()).await; + let issued_at = fixtures::now(); + + // Call the endpoint + + let Json(invite) = post::handler( + State(app), + issued_at.clone(), + issuer.clone(), + Json(post::Request {}), + ) + .await + .expect("creating an invite always succeeds"); + + // Verify the response + assert_eq!(issuer.login.id, invite.issuer); + assert_eq!(&*issued_at, &invite.issued_at); +} diff --git a/src/test/fixtures/invite.rs b/src/test/fixtures/invite.rs new file mode 100644 index 0000000..654d1b4 --- /dev/null +++ b/src/test/fixtures/invite.rs @@ -0,0 +1,17 @@ +use crate::{ + app::App, + clock::DateTime, + invite::{self, Invite}, + login::Login, +}; + +pub async fn issue(app: &App, issuer: &Login, issued_at: &DateTime) -> Invite { + app.invites() + .issue(issuer, issued_at) + .await + .expect("issuing invites never fails") +} + +pub fn fictitious() -> invite::Id { + invite::Id::generate() +} diff --git a/src/test/fixtures/mod.rs b/src/test/fixtures/mod.rs index 9111811..2b7b6af 100644 --- a/src/test/fixtures/mod.rs +++ b/src/test/fixtures/mod.rs @@ -7,6 +7,7 @@ pub mod cookie; pub mod event; pub mod future; pub mod identity; +pub mod invite; pub mod login; pub mod message; -- cgit v1.2.3 From 6321980ac49525d6d56f1fec67f9008e253a9c39 Mon Sep 17 00:00:00 2001 From: Owen Jacobson Date: Thu, 24 Oct 2024 20:17:55 -0400 Subject: Tests for accepting invites --- src/invite/routes/invite/post.rs | 3 +- src/invite/routes/invite/test/mod.rs | 1 + src/invite/routes/invite/test/post.rs | 208 ++++++++++++++++++++++++++++++++++ src/login/password.rs | 2 +- 4 files changed, 212 insertions(+), 2 deletions(-) create mode 100644 src/invite/routes/invite/test/post.rs (limited to 'src') diff --git a/src/invite/routes/invite/post.rs b/src/invite/routes/invite/post.rs index 3ca4e6b..0dd8dba 100644 --- a/src/invite/routes/invite/post.rs +++ b/src/invite/routes/invite/post.rs @@ -36,7 +36,8 @@ pub struct Request { pub password: Password, } -pub struct Error(app::AcceptError); +#[derive(Debug)] +pub struct Error(pub app::AcceptError); impl IntoResponse for Error { fn into_response(self) -> Response { diff --git a/src/invite/routes/invite/test/mod.rs b/src/invite/routes/invite/test/mod.rs index a608fbb..d6c1f06 100644 --- a/src/invite/routes/invite/test/mod.rs +++ b/src/invite/routes/invite/test/mod.rs @@ -1 +1,2 @@ mod get; +mod post; diff --git a/src/invite/routes/invite/test/post.rs b/src/invite/routes/invite/test/post.rs new file mode 100644 index 0000000..65ab61e --- /dev/null +++ b/src/invite/routes/invite/test/post.rs @@ -0,0 +1,208 @@ +use axum::extract::{Json, Path, State}; + +use crate::{ + invite::{app::AcceptError, routes::invite::post}, + name::Name, + test::fixtures, +}; + +#[tokio::test] +async fn valid_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let issuer = fixtures::login::create(&app, &fixtures::now()).await; + let invite = fixtures::invite::issue(&app, &issuer, &fixtures::now()).await; + + // Call the endpoint + + let (name, password) = fixtures::login::propose(); + let identity = fixtures::cookie::not_logged_in(); + let request = post::Request { + name: name.clone(), + password: password.clone(), + }; + let (identity, Json(response)) = post::handler( + State(app.clone()), + fixtures::now(), + identity, + Path(invite.id), + Json(request), + ) + .await + .expect("accepting a valid invite succeeds"); + + // Verify the response + + assert!(identity.secret().is_some()); + assert_eq!(name, response.name); + + // Verify that the issued token is valid + + let secret = identity + .secret() + .expect("newly-issued identity has a token secret"); + let (_, login) = app + .tokens() + .validate(&secret, &fixtures::now()) + .await + .expect("newly-issued identity cookie is valid"); + assert_eq!(response, login); + + // Verify that the given credentials can log in + + let (login, _) = app + .tokens() + .login(&name, &password, &fixtures::now()) + .await + .expect("credentials given on signup are valid"); + assert_eq!(response, login); +} + +#[tokio::test] +async fn nonexistent_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let invite = fixtures::invite::fictitious(); + + // Call the endpoint + + let (name, password) = fixtures::login::propose(); + let identity = fixtures::cookie::not_logged_in(); + let request = post::Request { + name: name.clone(), + password: password.clone(), + }; + let post::Error(error) = post::handler( + State(app.clone()), + fixtures::now(), + identity, + Path(invite.clone()), + Json(request), + ) + .await + .expect_err("accepting a nonexistent invite fails"); + + // Verify the response + + assert!(matches!(error, AcceptError::NotFound(error_id) if error_id == invite)); +} + +#[tokio::test] +async fn expired_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let issuer = fixtures::login::create(&app, &fixtures::ancient()).await; + let invite = fixtures::invite::issue(&app, &issuer, &fixtures::ancient()).await; + + app.invites() + .expire(&fixtures::now()) + .await + .expect("expiring invites never fails"); + + // Call the endpoint + + let (name, password) = fixtures::login::propose(); + let identity = fixtures::cookie::not_logged_in(); + let request = post::Request { + name: name.clone(), + password: password.clone(), + }; + let post::Error(error) = post::handler( + State(app.clone()), + fixtures::now(), + identity, + Path(invite.id.clone()), + Json(request), + ) + .await + .expect_err("accepting a nonexistent invite fails"); + + // Verify the response + + assert!(matches!(error, AcceptError::NotFound(error_id) if error_id == invite.id)); +} + +#[tokio::test] +async fn accepted_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let issuer = fixtures::login::create(&app, &fixtures::ancient()).await; + let invite = fixtures::invite::issue(&app, &issuer, &fixtures::ancient()).await; + + let (name, password) = fixtures::login::propose(); + app.invites() + .accept(&invite.id, &name, &password, &fixtures::now()) + .await + .expect("accepting a valid invite succeeds"); + + // Call the endpoint + + let (name, password) = fixtures::login::propose(); + let identity = fixtures::cookie::not_logged_in(); + let request = post::Request { + name: name.clone(), + password: password.clone(), + }; + let post::Error(error) = post::handler( + State(app.clone()), + fixtures::now(), + identity, + Path(invite.id.clone()), + Json(request), + ) + .await + .expect_err("accepting a nonexistent invite fails"); + + // Verify the response + + assert!(matches!(error, AcceptError::NotFound(error_id) if error_id == invite.id)); +} + +#[tokio::test] +async fn conflicting_name() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let issuer = fixtures::login::create(&app, &fixtures::ancient()).await; + let invite = fixtures::invite::issue(&app, &issuer, &fixtures::ancient()).await; + + let existing_name = Name::from("rijksmuseum"); + app.logins() + .create( + &existing_name, + &fixtures::login::propose_password(), + &fixtures::now(), + ) + .await + .expect("creating a login in an empty environment succeeds"); + + // Call the endpoint + + let conflicting_name = Name::from("r\u{0133}ksmuseum"); + let password = fixtures::login::propose_password(); + + let identity = fixtures::cookie::not_logged_in(); + let request = post::Request { + name: conflicting_name.clone(), + password: password.clone(), + }; + let post::Error(error) = post::handler( + State(app.clone()), + fixtures::now(), + identity, + Path(invite.id.clone()), + Json(request), + ) + .await + .expect_err("accepting a nonexistent invite fails"); + + // Verify the response + + assert!( + matches!(error, AcceptError::DuplicateLogin(error_name) if error_name == conflicting_name) + ); +} diff --git a/src/login/password.rs b/src/login/password.rs index c27c950..e1d164e 100644 --- a/src/login/password.rs +++ b/src/login/password.rs @@ -31,7 +31,7 @@ impl fmt::Debug for StoredHash { } } -#[derive(serde::Deserialize)] +#[derive(Clone, serde::Deserialize)] #[serde(transparent)] pub struct Password(nfc::String); -- cgit v1.2.3 From d97f8ac1858fbd46b77b541a612cc37b07fc0b5d Mon Sep 17 00:00:00 2001 From: Owen Jacobson Date: Thu, 24 Oct 2024 20:53:40 -0400 Subject: Tests for initial setup --- src/setup/routes/mod.rs | 2 ++ src/setup/routes/test.rs | 69 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+) create mode 100644 src/setup/routes/test.rs (limited to 'src') diff --git a/src/setup/routes/mod.rs b/src/setup/routes/mod.rs index e1e1711..6054983 100644 --- a/src/setup/routes/mod.rs +++ b/src/setup/routes/mod.rs @@ -3,6 +3,8 @@ use axum::{routing::post, Router}; use crate::app::App; mod post; +#[cfg(test)] +mod test; pub fn router() -> Router { Router::new().route("/api/setup", post(post::handler)) diff --git a/src/setup/routes/test.rs b/src/setup/routes/test.rs new file mode 100644 index 0000000..f7562ae --- /dev/null +++ b/src/setup/routes/test.rs @@ -0,0 +1,69 @@ +use axum::extract::{Json, State}; + +use super::post; +use crate::{setup::app, test::fixtures}; + +#[tokio::test] +async fn fresh_instance() { + // Set up the environment + + let app = fixtures::scratch_app().await; + + // Call the endpoint + let identity = fixtures::cookie::not_logged_in(); + let (name, password) = fixtures::login::propose(); + let request = post::Request { + name: name.clone(), + password: password.clone(), + }; + let (identity, Json(response)) = + post::handler(State(app.clone()), fixtures::now(), identity, Json(request)) + .await + .expect("setup in a fresh app succeeds"); + + // Verify the response + + assert_eq!(name, response.name); + + // Verify that the issued token is valid + + let secret = identity + .secret() + .expect("newly-issued identity has a token secret"); + let (_, login) = app + .tokens() + .validate(&secret, &fixtures::now()) + .await + .expect("newly-issued identity cookie is valid"); + assert_eq!(response, login); + + // Verify that the given credentials can log in + + let (login, _) = app + .tokens() + .login(&name, &password, &fixtures::now()) + .await + .expect("credentials given on signup are valid"); + assert_eq!(response, login); +} + +#[tokio::test] +async fn login_exists() { + // Set up the environment + + let app = fixtures::scratch_app().await; + fixtures::login::create(&app, &fixtures::now()).await; + + // Call the endpoint + let identity = fixtures::cookie::not_logged_in(); + let (name, password) = fixtures::login::propose(); + let request = post::Request { name, password }; + let post::Error(error) = + post::handler(State(app.clone()), fixtures::now(), identity, Json(request)) + .await + .expect_err("setup in a populated app fails"); + + // Verify the response + + assert!(matches!(error, app::Error::SetupCompleted)); +} -- cgit v1.2.3 From 0bb17bd01640492db2685e67bacac12dd54a9f59 Mon Sep 17 00:00:00 2001 From: Owen Jacobson 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/app.rs | 2 +- src/event/routes/test.rs | 459 --------------------------------------- src/event/routes/test/channel.rs | 212 ++++++++++++++++++ src/event/routes/test/invite.rs | 82 +++++++ src/event/routes/test/message.rs | 316 +++++++++++++++++++++++++++ src/event/routes/test/mod.rs | 6 + src/event/routes/test/resume.rs | 220 +++++++++++++++++++ src/event/routes/test/setup.rs | 46 ++++ src/event/routes/test/token.rs | 97 +++++++++ src/invite/app.rs | 10 +- src/test/fixtures/channel.rs | 7 + src/test/fixtures/event.rs | 8 - src/test/fixtures/login.rs | 16 ++ src/test/fixtures/message.rs | 16 +- src/test/fixtures/mod.rs | 1 - 15 files changed, 1025 insertions(+), 473 deletions(-) delete mode 100644 src/event/routes/test.rs create mode 100644 src/event/routes/test/channel.rs create mode 100644 src/event/routes/test/invite.rs create mode 100644 src/event/routes/test/message.rs create mode 100644 src/event/routes/test/mod.rs create mode 100644 src/event/routes/test/resume.rs create mode 100644 src/event/routes/test/setup.rs create mode 100644 src/event/routes/test/token.rs delete mode 100644 src/test/fixtures/event.rs (limited to 'src') diff --git a/src/app.rs b/src/app.rs index 6d71259..bc1daa5 100644 --- a/src/app.rs +++ b/src/app.rs @@ -44,7 +44,7 @@ impl App { } pub const fn invites(&self) -> Invites { - Invites::new(&self.db) + Invites::new(&self.db, &self.events) } #[cfg(not(test))] diff --git a/src/event/routes/test.rs b/src/event/routes/test.rs deleted file mode 100644 index 49f8094..0000000 --- a/src/event/routes/test.rs +++ /dev/null @@ -1,459 +0,0 @@ -use axum::extract::State; -use axum_extra::extract::Query; -use futures::{ - future, - stream::{self, StreamExt as _}, -}; - -use super::get; -use crate::{ - event::Sequenced as _, - test::fixtures::{self, future::Immediately as _}, -}; - -#[tokio::test] -async fn includes_historical_message() { - // Set up the environment - - let app = fixtures::scratch_app().await; - let sender = fixtures::login::create(&app, &fixtures::now()).await; - let channel = fixtures::channel::create(&app, &fixtures::now()).await; - let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).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 event = events - .filter_map(fixtures::message::events) - .next() - .immediately() - .await - .expect("delivered stored message"); - - assert!(fixtures::event::message_sent(&event, &message)); -} - -#[tokio::test] -async fn includes_live_message() { - // 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"); - - // Verify the semantics - - let sender = fixtures::login::create(&app, &fixtures::now()).await; - let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await; - - let event = events - .filter_map(fixtures::message::events) - .next() - .immediately() - .await - .expect("delivered live message"); - - assert!(fixtures::event::message_sent(&event, &message)); -} - -#[tokio::test] -async fn includes_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::>() - .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) - .take(messages.len()) - .collect::>() - .immediately() - .await; - - for message in &messages { - assert!(events - .iter() - .any(|event| fixtures::event::message_sent(event, message))); - } -} - -#[tokio::test] -async fn sequential_messages() { - // 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, - ]; - - // 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 mut events = events - .filter_map(fixtures::message::events) - .filter(|event| { - future::ready( - messages - .iter() - .any(|message| fixtures::event::message_sent(event, message)), - ) - }); - - // Verify delivery in order - for message in &messages { - let event = events - .next() - .immediately() - .await - .expect("undelivered messages remaining"); - - assert!(fixtures::event::message_sent(&event, message)); - } -} - -#[tokio::test] -async fn resumes_from() { - // 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 initial_message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await; - - let later_messages = vec![ - fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, - fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, - ]; - - // Call the endpoint - - let subscriber = fixtures::identity::create(&app, &fixtures::now()).await; - - let resume_at = { - // First subscription - let get::Response(events) = get::handler( - State(app.clone()), - subscriber.clone(), - None, - Query::default(), - ) - .await - .expect("subscribe never fails"); - - let event = events - .filter_map(fixtures::message::events) - .next() - .immediately() - .await - .expect("delivered events"); - - assert!(fixtures::event::message_sent(&event, &initial_message)); - - event.sequence() - }; - - // Resume after disconnect - let get::Response(resumed) = get::handler( - State(app), - subscriber, - Some(resume_at.into()), - Query::default(), - ) - .await - .expect("subscribe never fails"); - - // Verify the structure of the response. - - let events = resumed - .filter_map(fixtures::message::events) - .take(later_messages.len()) - .collect::>() - .immediately() - .await; - - for message in &later_messages { - assert!(events - .iter() - .any(|event| fixtures::event::message_sent(event, message))); - } -} - -// This test verifies a real bug I hit developing the vector-of-sequences -// approach to resuming events. A small omission caused the event IDs in a -// resumed stream to _omit_ channels that were in the original stream until -// those channels also appeared in the resumed stream. -// -// Clients would see something like -// * In the original stream, Cfoo=5,Cbar=8 -// * In the resumed stream, Cfoo=6 (no Cbar sequence number) -// -// Disconnecting and reconnecting a second time, using event IDs from that -// initial period of the first resume attempt, would then cause the second -// resume attempt to restart all other channels from the beginning, and not -// from where the first disconnection happened. -// -// This is a real and valid behaviour for clients! -#[tokio::test] -async fn serial_resume() { - // Set up the environment - - let app = fixtures::scratch_app().await; - let sender = fixtures::login::create(&app, &fixtures::now()).await; - let channel_a = fixtures::channel::create(&app, &fixtures::now()).await; - let channel_b = fixtures::channel::create(&app, &fixtures::now()).await; - - // Call the endpoint - - let subscriber = fixtures::identity::create(&app, &fixtures::now()).await; - - let resume_at = { - let initial_messages = [ - fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, - fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await, - ]; - - // First subscription - let get::Response(events) = get::handler( - State(app.clone()), - subscriber.clone(), - None, - Query::default(), - ) - .await - .expect("subscribe never fails"); - - let events = events - .filter_map(fixtures::message::events) - .take(initial_messages.len()) - .collect::>() - .immediately() - .await; - - for message in &initial_messages { - assert!(events - .iter() - .any(|event| fixtures::event::message_sent(event, message))); - } - - let event = events.last().expect("this vec is non-empty"); - - event.sequence() - }; - - // Resume after disconnect - let resume_at = { - let resume_messages = [ - // Note that channel_b does not appear here. The buggy behaviour - // would be masked if channel_b happened to send a new message - // into the resumed event stream. - fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, - fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, - ]; - - // Second subscription - let get::Response(events) = get::handler( - State(app.clone()), - subscriber.clone(), - Some(resume_at.into()), - Query::default(), - ) - .await - .expect("subscribe never fails"); - - let events = events - .filter_map(fixtures::message::events) - .take(resume_messages.len()) - .collect::>() - .immediately() - .await; - - for message in &resume_messages { - assert!(events - .iter() - .any(|event| fixtures::event::message_sent(event, message))); - } - - let event = events.last().expect("this vec is non-empty"); - - event.sequence() - }; - - // Resume after disconnect a second time - { - // At this point, we can send on either channel and demonstrate the - // problem. The resume point should before both of these messages, but - // after _all_ prior messages. - let final_messages = [ - fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, - fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await, - ]; - - // Third subscription - let get::Response(events) = get::handler( - State(app.clone()), - subscriber.clone(), - Some(resume_at.into()), - Query::default(), - ) - .await - .expect("subscribe never fails"); - - let events = events - .filter_map(fixtures::message::events) - .take(final_messages.len()) - .collect::>() - .immediately() - .await; - - // This set of messages, in particular, _should not_ include any prior - // messages from `initial_messages` or `resume_messages`. - for message in &final_messages { - assert!(events - .iter() - .any(|event| fixtures::event::message_sent(event, message))); - } - }; -} - -#[tokio::test] -async fn terminates_on_token_expiry() { - // 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; - - // Subscribe via the endpoint - - let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; - 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"); - - // Verify the resulting stream's behaviour - - app.tokens() - .expire(&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, - ]; - - assert!(events - .filter_map(fixtures::message::events) - .filter(|event| future::ready( - messages - .iter() - .any(|message| fixtures::event::message_sent(event, message)) - )) - .next() - .immediately() - .await - .is_none()); -} - -#[tokio::test] -async fn terminates_on_logout() { - // 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; - - // Subscribe via the endpoint - - let subscriber = fixtures::identity::create(&app, &fixtures::now()).await; - - let get::Response(events) = get::handler( - State(app.clone()), - subscriber.clone(), - None, - Query::default(), - ) - .await - .expect("subscribe never fails"); - - // Verify the resulting stream's behaviour - - app.tokens() - .logout(&subscriber.token) - .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, - ]; - - assert!(events - .filter_map(fixtures::message::events) - .filter(|event| future::ready( - messages - .iter() - .any(|message| fixtures::event::message_sent(event, message)) - )) - .next() - .immediately() - .await - .is_none()); -} diff --git a/src/event/routes/test/channel.rs b/src/event/routes/test/channel.rs new file mode 100644 index 0000000..ac45bfc --- /dev/null +++ b/src/event/routes/test/channel.rs @@ -0,0 +1,212 @@ +use axum::extract::State; +use axum_extra::extract::Query; +use futures::{future, stream::StreamExt as _}; + +use crate::{ + event::routes::get, + test::fixtures::{self, future::Immediately as _}, +}; + +#[tokio::test] +async fn creating() { + // Set up the environment + + let app = fixtures::scratch_app().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"); + + // Create a channel + + let name = fixtures::channel::propose(); + let channel = app + .channels() + .create(&name, &fixtures::now()) + .await + .expect("creating a channel succeeds"); + + // Verify channel created event + + let _ = events + .filter_map(fixtures::channel::events) + .filter_map(fixtures::channel::created) + .filter(|event| future::ready(event.channel == channel)) + .next() + .immediately() + .await + .expect("channel created event is delivered"); +} + +#[tokio::test] +async fn previously_created() { + // Set up the environment + + let app = fixtures::scratch_app().await; + + // Create a channel + + let name = fixtures::channel::propose(); + let channel = app + .channels() + .create(&name, &fixtures::now()) + .await + .expect("creating a channel 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"); + + // Verify channel created event + + let _ = events + .filter_map(fixtures::channel::events) + .filter_map(fixtures::channel::created) + .filter(|event| future::ready(event.channel == channel)) + .next() + .immediately() + .await + .expect("channel created event is delivered"); +} + +#[tokio::test] +async fn expiring() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let channel = fixtures::channel::create(&app, &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 channels + + app.channels() + .expire(&fixtures::now()) + .await + .expect("expiring channels always succeeds"); + + // Check for expiry event + let _ = events + .filter_map(fixtures::channel::events) + .filter_map(fixtures::channel::deleted) + .filter(|event| future::ready(event.id == channel.id)) + .next() + .immediately() + .await + .expect("a deleted channel 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; + + // Expire channels + + app.channels() + .expire(&fixtures::now()) + .await + .expect("expiring 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 + let _ = events + .filter_map(fixtures::channel::events) + .filter_map(fixtures::channel::deleted) + .filter(|event| future::ready(event.id == channel.id)) + .next() + .immediately() + .await + .expect("a deleted channel 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; + + // 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 channel + + app.channels() + .delete(&channel.id, &fixtures::now()) + .await + .expect("deleting a valid channel succeeds"); + + // Check for delete event + let _ = events + .filter_map(fixtures::channel::events) + .filter_map(fixtures::channel::deleted) + .filter(|event| future::ready(event.id == channel.id)) + .next() + .immediately() + .await + .expect("a deleted channel 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 channel + + app.channels() + .delete(&channel.id, &fixtures::now()) + .await + .expect("deleting a valid channel 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 will be delivered"); +} diff --git a/src/event/routes/test/invite.rs b/src/event/routes/test/invite.rs new file mode 100644 index 0000000..10e4521 --- /dev/null +++ b/src/event/routes/test/invite.rs @@ -0,0 +1,82 @@ +use axum::extract::State; +use axum_extra::extract::Query; +use futures::{future, stream::StreamExt as _}; + +use crate::{ + event::routes::get, + test::fixtures::{self, future::Immediately as _}, +}; + +#[tokio::test] +async fn accepting_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let issuer = fixtures::login::create(&app, &fixtures::now()).await; + let invite = fixtures::invite::issue(&app, &issuer, &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"); + + // Accept the invite + + let (name, password) = fixtures::login::propose(); + let (joiner, _) = app + .invites() + .accept(&invite.id, &name, &password, &fixtures::now()) + .await + .expect("accepting an invite succeeds"); + + // Expect a login created event + + let _ = events + .filter_map(fixtures::login::events) + .filter_map(fixtures::login::created) + .filter(|event| future::ready(event.login == joiner)) + .next() + .immediately() + .await + .expect("a login created event is sent"); +} + +#[tokio::test] +async fn previously_accepted_invite() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let issuer = fixtures::login::create(&app, &fixtures::now()).await; + let invite = fixtures::invite::issue(&app, &issuer, &fixtures::now()).await; + + // Accept the invite + + let (name, password) = fixtures::login::propose(); + let (joiner, _) = app + .invites() + .accept(&invite.id, &name, &password, &fixtures::now()) + .await + .expect("accepting an invite 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"); + + // Expect a login created event + + let _ = events + .filter_map(fixtures::login::events) + .filter_map(fixtures::login::created) + .filter(|event| future::ready(event.login == joiner)) + .next() + .immediately() + .await + .expect("a login created event is sent"); +} 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::>() + .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::>() + .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"); +} diff --git a/src/event/routes/test/mod.rs b/src/event/routes/test/mod.rs new file mode 100644 index 0000000..e7e35f1 --- /dev/null +++ b/src/event/routes/test/mod.rs @@ -0,0 +1,6 @@ +mod channel; +mod invite; +mod message; +mod resume; +mod setup; +mod token; diff --git a/src/event/routes/test/resume.rs b/src/event/routes/test/resume.rs new file mode 100644 index 0000000..c393d38 --- /dev/null +++ b/src/event/routes/test/resume.rs @@ -0,0 +1,220 @@ +use std::future; + +use axum::extract::State; +use axum_extra::extract::Query; +use futures::stream::{self, StreamExt as _}; + +use crate::{ + event::{routes::get, Sequenced as _}, + test::fixtures::{self, future::Immediately as _}, +}; + +#[tokio::test] +async fn resume() { + // 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 initial_message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await; + + let later_messages = vec![ + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await, + ]; + + // Call the endpoint + + let subscriber = fixtures::identity::create(&app, &fixtures::now()).await; + + let resume_at = { + // First subscription + let get::Response(events) = get::handler( + State(app.clone()), + subscriber.clone(), + None, + Query::default(), + ) + .await + .expect("subscribe never fails"); + + let event = events + .filter_map(fixtures::message::events) + .filter_map(fixtures::message::sent) + .filter(|event| future::ready(event.message == initial_message)) + .next() + .immediately() + .await + .expect("delivered events"); + + event.sequence() + }; + + // Resume after disconnect + let get::Response(resumed) = get::handler( + State(app), + subscriber, + Some(resume_at.into()), + Query::default(), + ) + .await + .expect("subscribe never fails"); + + // Verify final events + + let mut events = resumed + .filter_map(fixtures::message::events) + .filter_map(fixtures::message::sent) + .zip(stream::iter(later_messages)); + + while let Some((event, message)) = events.next().immediately().await { + assert_eq!(message, event.message); + } +} + +// This test verifies a real bug I hit developing the vector-of-sequences +// approach to resuming events. A small omission caused the event IDs in a +// resumed stream to _omit_ channels that were in the original stream until +// those channels also appeared in the resumed stream. +// +// Clients would see something like +// * In the original stream, Cfoo=5,Cbar=8 +// * In the resumed stream, Cfoo=6 (no Cbar sequence number) +// +// Disconnecting and reconnecting a second time, using event IDs from that +// initial period of the first resume attempt, would then cause the second +// resume attempt to restart all other channels from the beginning, and not +// from where the first disconnection happened. +// +// As we have switched to a single global event sequence number, this scenario +// can no longer arise, but this test is preserved because the actual behaviour +// _is_ a valid way for clients to behave, and should work. We might as well +// keep testing it. +#[tokio::test] +async fn serial_resume() { + // Set up the environment + + let app = fixtures::scratch_app().await; + let sender = fixtures::login::create(&app, &fixtures::now()).await; + let channel_a = fixtures::channel::create(&app, &fixtures::now()).await; + let channel_b = fixtures::channel::create(&app, &fixtures::now()).await; + + // Call the endpoint + + let subscriber = fixtures::identity::create(&app, &fixtures::now()).await; + + let resume_at = { + let initial_messages = [ + fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await, + ]; + + // First subscription + + let get::Response(events) = get::handler( + State(app.clone()), + subscriber.clone(), + None, + Query::default(), + ) + .await + .expect("subscribe never fails"); + + // Check for expected events + + let events = events + .filter_map(fixtures::message::events) + .filter_map(fixtures::message::sent) + .zip(stream::iter(initial_messages)) + .collect::>() + .immediately() + .await; + + assert!(events + .iter() + .all(|(event, message)| message == &event.message)); + + let (event, _) = events.last().expect("this vec is non-empty"); + + // Take the last one's resume point + + event.sequence() + }; + + // Resume after disconnect + let resume_at = { + let resume_messages = [ + // Note that channel_b does not appear here. The buggy behaviour + // would be masked if channel_b happened to send a new message + // into the resumed event stream. + fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, + ]; + + // Second subscription + let get::Response(events) = get::handler( + State(app.clone()), + subscriber.clone(), + Some(resume_at.into()), + Query::default(), + ) + .await + .expect("subscribe never fails"); + + // Check for expected events + + let events = events + .filter_map(fixtures::message::events) + .filter_map(fixtures::message::sent) + .zip(stream::iter(resume_messages)) + .collect::>() + .immediately() + .await; + + assert!(events + .iter() + .all(|(event, message)| message == &event.message)); + + let (event, _) = events.last().expect("this vec is non-empty"); + + // Take the last one's resume point + + event.sequence() + }; + + // Resume after disconnect a second time + { + // At this point, we can send on either channel and demonstrate the + // problem. The resume point should before both of these messages, but + // after _all_ prior messages. + let final_messages = [ + fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await, + fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await, + ]; + + // Third subscription + let get::Response(events) = get::handler( + State(app.clone()), + subscriber.clone(), + Some(resume_at.into()), + Query::default(), + ) + .await + .expect("subscribe never fails"); + + // Check for expected events + + let events = events + .filter_map(fixtures::message::events) + .filter_map(fixtures::message::sent) + .zip(stream::iter(final_messages)) + .collect::>() + .immediately() + .await; + + assert!(events + .iter() + .all(|(event, message)| message == &event.message)); + }; +} diff --git a/src/event/routes/test/setup.rs b/src/event/routes/test/setup.rs new file mode 100644 index 0000000..234c2d9 --- /dev/null +++ b/src/event/routes/test/setup.rs @@ -0,0 +1,46 @@ +use axum::extract::State; +use axum_extra::extract::Query; +use futures::{future, stream::StreamExt as _}; + +use crate::{ + event::routes::get, + test::fixtures::{self, future::Immediately as _}, +}; + +// There's no test for this in subscribe-then-setup order because creating an +// identity to subscribe with also completes initial setup, preventing the +// test from running. That is also a can't-happen scenario in reality. +#[tokio::test] +async fn previously_completed() { + // Set up the environment + + let app = fixtures::scratch_app().await; + + // Complete initial setup + + let (name, password) = fixtures::login::propose(); + let (owner, _) = app + .setup() + .initial(&name, &password, &fixtures::now()) + .await + .expect("initial setup in an empty app succeeds"); + + // Subscribe to events + + 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"); + + // Expect a login created event + + let _ = events + .filter_map(fixtures::login::events) + .filter_map(fixtures::login::created) + .filter(|event| future::ready(event.login == owner)) + .next() + .immediately() + .await + .expect("a login created event is sent"); +} diff --git a/src/event/routes/test/token.rs b/src/event/routes/test/token.rs new file mode 100644 index 0000000..a545988 --- /dev/null +++ b/src/event/routes/test/token.rs @@ -0,0 +1,97 @@ +use axum::extract::State; +use axum_extra::extract::Query; +use futures::{future, stream::StreamExt as _}; + +use crate::{ + event::routes::get, + test::fixtures::{self, future::Immediately as _}, +}; + +#[tokio::test] +async fn terminates_on_token_expiry() { + // 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; + + // Subscribe via the endpoint + + let subscriber_creds = fixtures::login::create_with_password(&app, &fixtures::now()).await; + 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"); + + // Verify the resulting stream's behaviour + + app.tokens() + .expire(&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, + ]; + + assert!(events + .filter_map(fixtures::message::events) + .filter_map(fixtures::message::sent) + .filter(|event| future::ready(messages.iter().any(|message| &event.message == message))) + .next() + .immediately() + .await + .is_none()); +} + +#[tokio::test] +async fn terminates_on_logout() { + // 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; + + // Subscribe via the endpoint + + let subscriber = fixtures::identity::create(&app, &fixtures::now()).await; + + let get::Response(events) = get::handler( + State(app.clone()), + subscriber.clone(), + None, + Query::default(), + ) + .await + .expect("subscribe never fails"); + + // Verify the resulting stream's behaviour + + app.tokens() + .logout(&subscriber.token) + .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, + ]; + + assert!(events + .filter_map(fixtures::message::events) + .filter_map(fixtures::message::sent) + .filter(|event| future::ready(messages.iter().any(|message| &event.message == message))) + .next() + .immediately() + .await + .is_none()); +} diff --git a/src/invite/app.rs b/src/invite/app.rs index 65e7721..176075f 100644 --- a/src/invite/app.rs +++ b/src/invite/app.rs @@ -5,7 +5,7 @@ use super::{repo::Provider as _, Id, Invite, Summary}; use crate::{ clock::DateTime, db::{Duplicate as _, NotFound as _}, - event::repo::Provider as _, + event::{repo::Provider as _, Broadcaster, Event}, login::{repo::Provider as _, Login, Password}, name::Name, token::{repo::Provider as _, Secret}, @@ -13,11 +13,12 @@ use crate::{ pub struct Invites<'a> { db: &'a SqlitePool, + events: &'a Broadcaster, } impl<'a> Invites<'a> { - pub const fn new(db: &'a SqlitePool) -> Self { - Self { db } + pub const fn new(db: &'a SqlitePool, events: &'a Broadcaster) -> Self { + Self { db, events } } pub async fn issue(&self, issuer: &Login, issued_at: &DateTime) -> Result { @@ -69,6 +70,9 @@ impl<'a> Invites<'a> { let secret = tx.tokens().issue(&login, accepted_at).await?; tx.commit().await?; + self.events + .broadcast(login.events().map(Event::from).collect::>()); + Ok((login.as_created(), secret)) } diff --git a/src/test/fixtures/channel.rs b/src/test/fixtures/channel.rs index 8cb38ae..1fd8d23 100644 --- a/src/test/fixtures/channel.rs +++ b/src/test/fixtures/channel.rs @@ -45,6 +45,13 @@ pub fn created(event: channel::Event) -> future::Ready future::Ready> { + 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 deleted file mode 100644 index fa4fbc0..0000000 --- a/src/test/fixtures/event.rs +++ /dev/null @@ -1,8 +0,0 @@ -use crate::message::{Event, Message}; - -pub fn message_sent(event: &Event, message: &Message) -> bool { - matches!( - &event, - Event::Sent(event) if message == &event.into() - ) -} diff --git a/src/test/fixtures/login.rs b/src/test/fixtures/login.rs index e308289..cbcbdd4 100644 --- a/src/test/fixtures/login.rs +++ b/src/test/fixtures/login.rs @@ -1,9 +1,12 @@ +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, }; @@ -45,3 +48,16 @@ fn propose_name() -> Name { pub fn propose_password() -> Password { Uuid::new_v4().to_string().into() } + +pub fn events(event: Event) -> Ready> { + future::ready(match event { + Event::Login(event) => Some(event), + _ => None, + }) +} + +pub fn created(event: login::Event) -> Ready> { + 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 3aebdd9..8cb50c1 100644 --- a/src/test/fixtures/message.rs +++ b/src/test/fixtures/message.rs @@ -8,7 +8,7 @@ use crate::{ clock::RequestedAt, event::Event, login::Login, - message::{self, Body, Message}, + message::{self, event, Body, Message}, }; pub async fn send(app: &App, channel: &Channel, sender: &Login, sent_at: &RequestedAt) -> Message { @@ -31,6 +31,20 @@ pub fn events(event: Event) -> future::Ready> { }) } +pub fn sent(event: message::Event) -> future::Ready> { + future::ready(match event { + message::Event::Sent(event) => Some(event), + message::Event::Deleted(_) => None, + }) +} + +pub fn deleted(event: message::Event) -> future::Ready> { + 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 2b7b6af..cf30e02 100644 --- a/src/test/fixtures/mod.rs +++ b/src/test/fixtures/mod.rs @@ -4,7 +4,6 @@ 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 eae0edb57e9ade7c73affb78baf2ae267b6290b8 Mon Sep 17 00:00:00 2001 From: Owen Jacobson 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') 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::>().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::>() .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::>() .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::>() .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::>() .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> { - future::ready(match event { - Event::Channel(channel) => Some(channel), - _ => None, - }) -} - -pub fn created(event: channel::Event) -> future::Ready> { - future::ready(match event { - channel::Event::Created(event) => Some(event), - channel::Event::Deleted(_) => None, - }) -} - -pub fn deleted(event: channel::Event) -> future::Ready> { - 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> { + future::ready(match event { + Event::Channel(channel) => Some(channel), + _ => None, + }) +} + +pub fn message(event: Event) -> Ready> { + future::ready(match event { + Event::Message(event) => Some(event), + _ => None, + }) +} + +pub fn login(event: Event) -> Ready> { + 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> { + future::ready(match event { + Event::Created(event) => Some(event), + Event::Deleted(_) => None, + }) + } + + pub fn deleted(event: Event) -> Ready> { + 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> { + future::ready(match event { + Event::Sent(event) => Some(event), + Event::Deleted(_) => None, + }) + } + + pub fn deleted(event: Event) -> future::Ready> { + 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> { + 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> { - future::ready(match event { - Event::Login(event) => Some(event), - _ => None, - }) -} - -pub fn created(event: login::Event) -> Ready> { - 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::().to_string().into() } -pub fn events(event: Event) -> future::Ready> { - future::ready(match event { - Event::Message(event) => Some(event), - _ => None, - }) -} - -pub fn sent(event: message::Event) -> future::Ready> { - future::ready(match event { - message::Event::Sent(event) => Some(event), - message::Event::Deleted(_) => None, - }) -} - -pub fn deleted(event: message::Event) -> future::Ready> { - 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 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`). --- 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') 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, Error> { + pub async fn get(&self, channel: &Id) -> Result { + 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::>().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::>() - .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::>() - .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::>() - .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::>() - .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(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 + 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 + 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(self, message: &str) -> Some + where + Self: Future>; + + // 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(self, message: &str) -> None + where + Self: Future>; +} + +impl<'a, St> Expect for stream::Next<'a, St> { + fn expect_ready(self, message: &str) -> Ready { + Ready { + future: self, + message, + } + } + + fn expect_wait(self, message: &str) -> Wait { + Wait { + future: self, + message, + } + } + + fn expect_some(self, message: &str) -> Some + where + Self: Future>, + { + Some { + future: self, + message, + } + } + + fn expect_none(self, message: &str) -> None + where + Self: Future>, + { + None { + future: self, + message, + } + } +} + +impl Expect for stream::Collect { + fn expect_ready(self, message: &str) -> Ready { + Ready { + future: self, + message, + } + } + + fn expect_wait(self, message: &str) -> Wait { + Wait { + future: self, + message, + } + } + + fn expect_some(self, message: &str) -> Some + where + Self: Future>, + { + Some { + future: self, + message, + } + } + + fn expect_none(self, message: &str) -> None + where + Self: Future>, + { + 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 { + 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 { + 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> + std::fmt::Debug, { - type Output = Option<::Item>; + type Output = T; + + fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll { + 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 Immediately for stream::Collect +#[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<::Item>, + F: Future> + std::fmt::Debug, { - type Output = C; + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll { + 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

(path: P) -> Result where P: AsRef, 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, ) -> Result { 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) -> Self { - Self::Internal(err.into()) + impl From 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 From 3a2f37e41681c2b233728c3cbddaea3f9fc74c08 Mon Sep 17 00:00:00 2001 From: Owen Jacobson Date: Fri, 25 Oct 2024 01:19:49 -0400 Subject: To make it easier to correlate deletes to the event stream, have deletes return the ID of the affected entity. --- ...c2e965e2294c1a6932d72a8b04d258d06d0f8938bd.json | 4 +-- docs/api/channels-messages.md | 32 +++++++++++++++++++--- src/channel/routes/channel/delete.rs | 23 ++++++++++++---- src/channel/routes/channel/test/delete.rs | 7 ++--- src/message/routes/message/mod.rs | 21 ++++++++++---- src/message/routes/message/test.rs | 7 ++--- 6 files changed, 67 insertions(+), 27 deletions(-) (limited to 'src') diff --git a/.sqlx/query-a40496319887752f5845c0c2e965e2294c1a6932d72a8b04d258d06d0f8938bd.json b/.sqlx/query-a40496319887752f5845c0c2e965e2294c1a6932d72a8b04d258d06d0f8938bd.json index 647e5ad..436e1dd 100644 --- a/.sqlx/query-a40496319887752f5845c0c2e965e2294c1a6932d72a8b04d258d06d0f8938bd.json +++ b/.sqlx/query-a40496319887752f5845c0c2e965e2294c1a6932d72a8b04d258d06d0f8938bd.json @@ -11,12 +11,12 @@ { "name": "display_name: String", "ordinal": 1, - "type_info": "Text" + "type_info": "Null" }, { "name": "canonical_name: String", "ordinal": 2, - "type_info": "Text" + "type_info": "Null" }, { "name": "created_at: DateTime", diff --git a/docs/api/channels-messages.md b/docs/api/channels-messages.md index 9ef4e66..9854d22 100644 --- a/docs/api/channels-messages.md +++ b/docs/api/channels-messages.md @@ -162,9 +162,21 @@ This endpoint requires the following path parameter: ### Success -This endpoint will respond with a status of `202 Accepted` when successful. The response will not include a body. +This endpoint will respond with a status of `202 Accepted` when successful. The body of the response will be a JSON object describing the deleted channel: -When completed, the service will emit a [channel deleted](events.md#channel-deleted) event with the channel's ID. In addition, the service will emit a [message deleted](events.md#message-deleted) event for each message deleted. +```json +{ + "id": "Cfqdn1234" +} +``` + +The response will have the following fields: + +| Field | Type | Description | +|:----------|:----------|:--| +| `id` | string | The channel's ID. | + +When completed, the service will emit a [message deleted](events.md#message-deleted) event for each message in the channel, followed by a [channel deleted](events.md#channel-deleted) event with the channel's ID. ### Invalid channel ID @@ -184,9 +196,21 @@ This endpoint requires the following path parameter: ### Success -This endpoint will respond with a status of `202 Accepted` when successful. The response will not include a body. +This endpoint will respond with a status of `202 Accepted` when successful. The body of the response will be a JSON object describing the deleted message: + +```json +{ + "id": "Mgh98yp75" +} +``` + +The response will have the following fields: + +| Field | Type | Description | +|:----------|:----------|:--| +| `id` | string | The message's ID. | -When completed, the service will emit a [message deleted](events.md#message-deleted) event with the channel's ID. +When completed, the service will emit a [message deleted](events.md#message-deleted) event with the message's ID. ### Invalid message ID diff --git a/src/channel/routes/channel/delete.rs b/src/channel/routes/channel/delete.rs index 91eb506..2d2b5f1 100644 --- a/src/channel/routes/channel/delete.rs +++ b/src/channel/routes/channel/delete.rs @@ -1,12 +1,12 @@ use axum::{ - extract::{Path, State}, + extract::{Json, Path, State}, http::StatusCode, - response::{IntoResponse, Response}, + response::{self, IntoResponse}, }; use crate::{ app::App, - channel::app, + channel::{self, app}, clock::RequestedAt, error::{Internal, NotFound}, token::extract::Identity, @@ -17,10 +17,21 @@ pub async fn handler( Path(channel): Path, RequestedAt(deleted_at): RequestedAt, _: Identity, -) -> Result { +) -> Result { app.channels().delete(&channel, &deleted_at).await?; - Ok(StatusCode::ACCEPTED) + Ok(Response { id: channel }) +} + +#[derive(Debug, serde::Serialize)] +pub struct Response { + pub id: channel::Id, +} + +impl IntoResponse for Response { + fn into_response(self) -> response::Response { + (StatusCode::ACCEPTED, Json(self)).into_response() + } } #[derive(Debug, thiserror::Error)] @@ -28,7 +39,7 @@ pub async fn handler( pub struct Error(#[from] pub app::Error); impl IntoResponse for Error { - fn into_response(self) -> Response { + fn into_response(self) -> response::Response { let Self(error) = self; #[allow(clippy::match_wildcard_for_single_variants)] match error { diff --git a/src/channel/routes/channel/test/delete.rs b/src/channel/routes/channel/test/delete.rs index e1210fd..0371b0a 100644 --- a/src/channel/routes/channel/test/delete.rs +++ b/src/channel/routes/channel/test/delete.rs @@ -1,7 +1,4 @@ -use axum::{ - extract::{Path, State}, - http::StatusCode, -}; +use axum::extract::{Path, State}; use crate::{ channel::{app, routes::channel::delete}, @@ -29,7 +26,7 @@ pub async fn valid_channel() { // Verify the response - assert_eq!(response, StatusCode::ACCEPTED); + assert_eq!(channel.id, response.id); // Verify the semantics diff --git a/src/message/routes/message/mod.rs b/src/message/routes/message/mod.rs index 545ad26..45a7e9d 100644 --- a/src/message/routes/message/mod.rs +++ b/src/message/routes/message/mod.rs @@ -3,9 +3,9 @@ mod test; pub mod delete { use axum::{ - extract::{Path, State}, + extract::{Json, Path, State}, http::StatusCode, - response::{IntoResponse, Response}, + response::{self, IntoResponse}, }; use crate::{ @@ -21,10 +21,21 @@ pub mod delete { Path(message): Path, RequestedAt(deleted_at): RequestedAt, _: Identity, - ) -> Result { + ) -> Result { app.messages().delete(&message, &deleted_at).await?; - Ok(StatusCode::ACCEPTED) + Ok(Response { id: message }) + } + + #[derive(Debug, serde::Serialize)] + pub struct Response { + pub id: message::Id, + } + + impl IntoResponse for Response { + fn into_response(self) -> response::Response { + (StatusCode::ACCEPTED, Json(self)).into_response() + } } #[derive(Debug, thiserror::Error)] @@ -32,7 +43,7 @@ pub mod delete { pub struct Error(#[from] pub DeleteError); impl IntoResponse for Error { - fn into_response(self) -> Response { + fn into_response(self) -> response::Response { let Self(error) = self; #[allow(clippy::match_wildcard_for_single_variants)] match error { diff --git a/src/message/routes/message/test.rs b/src/message/routes/message/test.rs index 2016fb8..ae89506 100644 --- a/src/message/routes/message/test.rs +++ b/src/message/routes/message/test.rs @@ -1,7 +1,4 @@ -use axum::{ - extract::{Path, State}, - http::StatusCode, -}; +use axum::extract::{Path, State}; use super::delete; use crate::{message::app, test::fixtures}; @@ -29,7 +26,7 @@ pub async fn delete_message() { // Verify the response - assert_eq!(response, StatusCode::ACCEPTED); + assert_eq!(message.id, response.id); // Verify the semantics -- cgit v1.2.3