use axum::extract::State; use axum_extra::extract::Query; use futures::{future, stream::StreamExt as _}; use crate::{ event::routes::get, test::fixtures::{self, future::Expect as _}, }; #[tokio::test] async fn creating() { // Set up the environment let app = fixtures::scratch_app().await; let resume_point = fixtures::boot::resume_point(&app).await; // Subscribe let subscriber = fixtures::identity::create(&app, &fixtures::now()).await; let get::Response(events) = get::handler( State(app.clone()), subscriber, None, Query(get::QueryParams { resume_point }), ) .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 events .filter_map(fixtures::event::channel) .filter_map(fixtures::event::channel::created) .filter(|event| future::ready(event.channel == channel)) .next() .expect_some("channel created event is delivered") .await; } #[tokio::test] async fn previously_created() { // Set up the environment let app = fixtures::scratch_app().await; let resume_point = fixtures::boot::resume_point(&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(get::QueryParams { resume_point }), ) .await .expect("subscribe never fails"); // Verify channel created event let _ = events .filter_map(fixtures::event::channel) .filter_map(fixtures::event::channel::created) .filter(|event| future::ready(event.channel == channel)) .next() .expect_some("channel created event is delivered") .await; } #[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 resume_point = fixtures::boot::resume_point(&app).await; // Subscribe let subscriber = fixtures::identity::create(&app, &fixtures::now()).await; let get::Response(events) = get::handler( State(app.clone()), subscriber, None, Query(get::QueryParams { resume_point }), ) .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::event::channel) .filter_map(fixtures::event::channel::deleted) .filter(|event| future::ready(event.id == channel.id)) .next() .expect_some("a deleted channel event will be delivered") .await; } #[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 resume_point = fixtures::boot::resume_point(&app).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(get::QueryParams { resume_point }), ) .await .expect("subscribe never fails"); // Check for expiry event let _ = events .filter_map(fixtures::event::channel) .filter_map(fixtures::event::channel::deleted) .filter(|event| future::ready(event.id == channel.id)) .next() .expect_some("a deleted channel event will be delivered") .await; } #[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 resume_point = fixtures::boot::resume_point(&app).await; // Subscribe let subscriber = fixtures::identity::create(&app, &fixtures::now()).await; let get::Response(events) = get::handler( State(app.clone()), subscriber, None, Query(get::QueryParams { resume_point }), ) .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::event::channel) .filter_map(fixtures::event::channel::deleted) .filter(|event| future::ready(event.id == channel.id)) .next() .expect_some("a deleted channel event will be delivered") .await; } #[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 resume_point = fixtures::boot::resume_point(&app).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(get::QueryParams { resume_point }), ) .await .expect("subscribe never fails"); // Check for expiry event let _ = events .filter_map(fixtures::event::channel) .filter_map(fixtures::event::channel::deleted) .filter(|event| future::ready(event.id == channel.id)) .next() .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; let resume_point = fixtures::boot::resume_point(&app).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("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(get::QueryParams { resume_point }), ) .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; }