GKaszewski/movies-diary
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: move inline tests to separate files via #[path]

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-05-12 16:39:58 +02:00
parent 00218366da
commit 763d622601
58 changed files with 3267 additions and 3267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

347
crates/application/src/movie_resolver.rs
View File

@@ -165,348 +165,5 @@ impl ResolutionStrategy for ManualMovieStrategy {
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::NaiveDate;
use domain::{
errors::DomainError,
models::Movie,
ports::{MetadataSearchCriteria, MovieRepository},
value_objects::{ExternalMetadataId, MovieId, MovieTitle, PosterUrl, ReleaseYear},
};
fn make_cmd(ext_id: Option<&str>, title: Option<&str>, year: Option<u16>) -> LogReviewCommand {
LogReviewCommand {
external_metadata_id: ext_id.map(String::from),
manual_title: title.map(String::from),
manual_release_year: year,
manual_director: None,
user_id: uuid::Uuid::new_v4(),
rating: 4,
comment: None,
watched_at: NaiveDate::from_ymd_opt(2024, 1, 1)
.unwrap()
.and_hms_opt(0, 0, 0)
.unwrap(),
}
}
fn make_movie() -> Movie {
Movie::new(
None,
MovieTitle::new("Inception".to_string()).unwrap(),
ReleaseYear::new(2010).unwrap(),
None,
None,
)
}
struct RepoWithExternalMovie(Movie);
struct RepoEmpty;
struct RepoWithTitleMatch(Movie);
#[async_trait]
impl MovieRepository for RepoWithExternalMovie {
async fn get_movie_by_external_id(
&self,
_: &ExternalMetadataId,
) -> Result<Option<Movie>, DomainError> {
Ok(Some(self.0.clone()))
}
async fn get_movie_by_id(&self, _: &MovieId) -> Result<Option<Movie>, DomainError> {
panic!("unexpected")
}
async fn get_movies_by_title_and_year(
&self,
_: &MovieTitle,
_: &ReleaseYear,
) -> Result<Vec<Movie>, DomainError> {
panic!("unexpected")
}
async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> {
panic!("unexpected")
}
async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> { panic!("unexpected") }
async fn list_movies(&self, _: &domain::models::collections::PageParams, _: Option<&str>) -> Result<domain::models::collections::Paginated<Movie>, DomainError> { panic!("unexpected") }
}
#[async_trait]
impl MovieRepository for RepoEmpty {
async fn get_movie_by_external_id(
&self,
_: &ExternalMetadataId,
) -> Result<Option<Movie>, DomainError> {
Ok(None)
}
async fn get_movie_by_id(&self, _: &MovieId) -> Result<Option<Movie>, DomainError> {
panic!("unexpected")
}
async fn get_movies_by_title_and_year(
&self,
_: &MovieTitle,
_: &ReleaseYear,
) -> Result<Vec<Movie>, DomainError> {
Ok(vec![])
}
async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> { panic!("unexpected") }
async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> { panic!("unexpected") }
async fn list_movies(&self, _: &domain::models::collections::PageParams, _: Option<&str>) -> Result<domain::models::collections::Paginated<Movie>, DomainError> { panic!("unexpected") }
}
#[async_trait]
impl MovieRepository for RepoWithTitleMatch {
async fn get_movie_by_external_id(
&self,
_: &ExternalMetadataId,
) -> Result<Option<Movie>, DomainError> {
panic!("unexpected")
}
async fn get_movie_by_id(&self, _: &MovieId) -> Result<Option<Movie>, DomainError> {
panic!("unexpected")
}
async fn get_movies_by_title_and_year(
&self,
_: &MovieTitle,
_: &ReleaseYear,
) -> Result<Vec<Movie>, DomainError> {
Ok(vec![self.0.clone()])
}
async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> { panic!("unexpected") }
async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> { panic!("unexpected") }
async fn list_movies(&self, _: &domain::models::collections::PageParams, _: Option<&str>) -> Result<domain::models::collections::Paginated<Movie>, DomainError> { panic!("unexpected") }
}
struct MetaReturnsMovie(Movie);
struct MetaErrors;
#[async_trait]
impl MetadataClient for MetaReturnsMovie {
async fn fetch_movie_metadata(
&self,
_: &MetadataSearchCriteria,
) -> Result<Movie, DomainError> {
Ok(self.0.clone())
}
async fn get_poster_url(
&self,
_: &ExternalMetadataId,
) -> Result<Option<PosterUrl>, DomainError> {
panic!("unexpected")
}
}
#[async_trait]
impl MetadataClient for MetaErrors {
async fn fetch_movie_metadata(
&self,
_: &MetadataSearchCriteria,
) -> Result<Movie, DomainError> {
Err(DomainError::InfrastructureError(
"metadata unavailable".into(),
))
}
async fn get_poster_url(
&self,
_: &ExternalMetadataId,
) -> Result<Option<PosterUrl>, DomainError> {
panic!("unexpected")
}
}
// --- ExternalIdStrategy ---
#[test]
fn external_id_strategy_can_handle_cmd_with_id() {
let cmd = make_cmd(Some("tt123"), None, None);
assert!(ExternalIdStrategy.can_handle(&cmd));
}
#[test]
fn external_id_strategy_cannot_handle_cmd_without_id() {
let cmd = make_cmd(None, Some("Inception"), Some(2010));
assert!(!ExternalIdStrategy.can_handle(&cmd));
}
#[tokio::test]
async fn external_id_strategy_returns_cached_movie() {
let movie = make_movie();
let repo = RepoWithExternalMovie(movie.clone());
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), None, None);
let result = ExternalIdStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, false))));
}
#[tokio::test]
async fn external_id_strategy_fetches_from_metadata_when_not_cached() {
let movie = make_movie();
let repo = RepoEmpty;
let meta = MetaReturnsMovie(movie);
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), None, None);
let result = ExternalIdStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, true))));
}
#[tokio::test]
async fn external_id_strategy_falls_through_on_metadata_error() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), None, None);
let result = ExternalIdStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(result.is_none());
}
// --- TitleSearchStrategy ---
#[test]
fn title_strategy_can_handle_cmd_with_title() {
let cmd = make_cmd(None, Some("Inception"), Some(2010));
assert!(TitleSearchStrategy.can_handle(&cmd));
}
#[test]
fn title_strategy_cannot_handle_cmd_without_title() {
let cmd = make_cmd(Some("tt123"), None, None);
assert!(!TitleSearchStrategy.can_handle(&cmd));
}
#[tokio::test]
async fn title_strategy_fetches_from_metadata() {
let movie = make_movie();
let repo = RepoEmpty;
let meta = MetaReturnsMovie(movie);
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), Some(2010));
let result = TitleSearchStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, true))));
}
#[tokio::test]
async fn title_strategy_falls_through_on_metadata_error() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), Some(2010));
let result = TitleSearchStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(result.is_none());
}
// --- ManualMovieStrategy ---
#[test]
fn manual_strategy_can_handle_cmd_with_title() {
let cmd = make_cmd(None, Some("Inception"), Some(2010));
assert!(ManualMovieStrategy.can_handle(&cmd));
}
#[test]
fn manual_strategy_cannot_handle_cmd_without_title() {
let cmd = make_cmd(Some("tt123"), None, None);
assert!(!ManualMovieStrategy.can_handle(&cmd));
}
#[tokio::test]
async fn manual_strategy_returns_existing_movie() {
let movie = make_movie();
let repo = RepoWithTitleMatch(movie.clone());
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), Some(2010));
let result = ManualMovieStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, false))));
}
#[tokio::test]
async fn manual_strategy_creates_new_movie_when_no_match() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), Some(2010));
let result = ManualMovieStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, true))));
}
#[tokio::test]
async fn manual_strategy_errors_without_year() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), None);
assert!(ManualMovieStrategy.resolve(&cmd, &deps).await.is_err());
}
// --- MovieResolver pipeline ---
#[tokio::test]
async fn resolver_returns_error_when_no_strategy_matches() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, None, None);
let result = MovieResolver::default_pipeline().resolve(&cmd, &deps).await;
assert!(result.is_err());
}
#[tokio::test]
async fn resolver_uses_cached_movie_when_external_id_matches() {
let movie = make_movie();
let repo = RepoWithExternalMovie(movie.clone());
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), None, None);
let (_, is_new) = MovieResolver::default_pipeline()
.resolve(&cmd, &deps)
.await
.unwrap();
assert!(!is_new);
}
#[tokio::test]
async fn resolver_falls_through_to_manual_when_external_and_title_both_fail() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), Some("Inception"), Some(2010));
let (_, is_new) = MovieResolver::default_pipeline()
.resolve(&cmd, &deps)
.await
.unwrap();
assert!(is_new);
}
}
#[path = "tests/movie_resolver.rs"]
mod tests;

343
crates/application/src/tests/movie_resolver.rs Normal file
View File

@@ -0,0 +1,343 @@
use super::*;
use chrono::NaiveDate;
use domain::{
errors::DomainError,
models::Movie,
ports::{MetadataSearchCriteria, MovieRepository},
value_objects::{ExternalMetadataId, MovieId, MovieTitle, PosterUrl, ReleaseYear},
};
fn make_cmd(ext_id: Option<&str>, title: Option<&str>, year: Option<u16>) -> LogReviewCommand {
LogReviewCommand {
external_metadata_id: ext_id.map(String::from),
manual_title: title.map(String::from),
manual_release_year: year,
manual_director: None,
user_id: uuid::Uuid::new_v4(),
rating: 4,
comment: None,
watched_at: NaiveDate::from_ymd_opt(2024, 1, 1)
.unwrap()
.and_hms_opt(0, 0, 0)
.unwrap(),
}
}
fn make_movie() -> Movie {
Movie::new(
None,
MovieTitle::new("Inception".to_string()).unwrap(),
ReleaseYear::new(2010).unwrap(),
None,
None,
)
}
struct RepoWithExternalMovie(Movie);
struct RepoEmpty;
struct RepoWithTitleMatch(Movie);
#[async_trait::async_trait]
impl MovieRepository for RepoWithExternalMovie {
async fn get_movie_by_external_id(
&self,
_: &ExternalMetadataId,
) -> Result<Option<Movie>, DomainError> {
Ok(Some(self.0.clone()))
}
async fn get_movie_by_id(&self, _: &MovieId) -> Result<Option<Movie>, DomainError> {
panic!("unexpected")
}
async fn get_movies_by_title_and_year(
&self,
_: &MovieTitle,
_: &ReleaseYear,
) -> Result<Vec<Movie>, DomainError> {
panic!("unexpected")
}
async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> {
panic!("unexpected")
}
async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> { panic!("unexpected") }
async fn list_movies(&self, _: &domain::models::collections::PageParams, _: Option<&str>) -> Result<domain::models::collections::Paginated<Movie>, DomainError> { panic!("unexpected") }
}
#[async_trait::async_trait]
impl MovieRepository for RepoEmpty {
async fn get_movie_by_external_id(
&self,
_: &ExternalMetadataId,
) -> Result<Option<Movie>, DomainError> {
Ok(None)
}
async fn get_movie_by_id(&self, _: &MovieId) -> Result<Option<Movie>, DomainError> {
panic!("unexpected")
}
async fn get_movies_by_title_and_year(
&self,
_: &MovieTitle,
_: &ReleaseYear,
) -> Result<Vec<Movie>, DomainError> {
Ok(vec![])
}
async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> { panic!("unexpected") }
async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> { panic!("unexpected") }
async fn list_movies(&self, _: &domain::models::collections::PageParams, _: Option<&str>) -> Result<domain::models::collections::Paginated<Movie>, DomainError> { panic!("unexpected") }
}
#[async_trait::async_trait]
impl MovieRepository for RepoWithTitleMatch {
async fn get_movie_by_external_id(
&self,
_: &ExternalMetadataId,
) -> Result<Option<Movie>, DomainError> {
panic!("unexpected")
}
async fn get_movie_by_id(&self, _: &MovieId) -> Result<Option<Movie>, DomainError> {
panic!("unexpected")
}
async fn get_movies_by_title_and_year(
&self,
_: &MovieTitle,
_: &ReleaseYear,
) -> Result<Vec<Movie>, DomainError> {
Ok(vec![self.0.clone()])
}
async fn upsert_movie(&self, _: &Movie) -> Result<(), DomainError> { panic!("unexpected") }
async fn delete_movie(&self, _: &MovieId) -> Result<(), DomainError> { panic!("unexpected") }
async fn list_movies(&self, _: &domain::models::collections::PageParams, _: Option<&str>) -> Result<domain::models::collections::Paginated<Movie>, DomainError> { panic!("unexpected") }
}
struct MetaReturnsMovie(Movie);
struct MetaErrors;
#[async_trait::async_trait]
impl MetadataClient for MetaReturnsMovie {
async fn fetch_movie_metadata(
&self,
_: &MetadataSearchCriteria,
) -> Result<Movie, DomainError> {
Ok(self.0.clone())
}
async fn get_poster_url(
&self,
_: &ExternalMetadataId,
) -> Result<Option<PosterUrl>, DomainError> {
panic!("unexpected")
}
}
#[async_trait::async_trait]
impl MetadataClient for MetaErrors {
async fn fetch_movie_metadata(
&self,
_: &MetadataSearchCriteria,
) -> Result<Movie, DomainError> {
Err(DomainError::InfrastructureError(
"metadata unavailable".into(),
))
}
async fn get_poster_url(
&self,
_: &ExternalMetadataId,
) -> Result<Option<PosterUrl>, DomainError> {
panic!("unexpected")
}
}
// --- ExternalIdStrategy ---
#[test]
fn external_id_strategy_can_handle_cmd_with_id() {
let cmd = make_cmd(Some("tt123"), None, None);
assert!(ExternalIdStrategy.can_handle(&cmd));
}
#[test]
fn external_id_strategy_cannot_handle_cmd_without_id() {
let cmd = make_cmd(None, Some("Inception"), Some(2010));
assert!(!ExternalIdStrategy.can_handle(&cmd));
}
#[tokio::test]
async fn external_id_strategy_returns_cached_movie() {
let movie = make_movie();
let repo = RepoWithExternalMovie(movie.clone());
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), None, None);
let result = ExternalIdStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, false))));
}
#[tokio::test]
async fn external_id_strategy_fetches_from_metadata_when_not_cached() {
let movie = make_movie();
let repo = RepoEmpty;
let meta = MetaReturnsMovie(movie);
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), None, None);
let result = ExternalIdStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, true))));
}
#[tokio::test]
async fn external_id_strategy_falls_through_on_metadata_error() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), None, None);
let result = ExternalIdStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(result.is_none());
}
// --- TitleSearchStrategy ---
#[test]
fn title_strategy_can_handle_cmd_with_title() {
let cmd = make_cmd(None, Some("Inception"), Some(2010));
assert!(TitleSearchStrategy.can_handle(&cmd));
}
#[test]
fn title_strategy_cannot_handle_cmd_without_title() {
let cmd = make_cmd(Some("tt123"), None, None);
assert!(!TitleSearchStrategy.can_handle(&cmd));
}
#[tokio::test]
async fn title_strategy_fetches_from_metadata() {
let movie = make_movie();
let repo = RepoEmpty;
let meta = MetaReturnsMovie(movie);
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), Some(2010));
let result = TitleSearchStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, true))));
}
#[tokio::test]
async fn title_strategy_falls_through_on_metadata_error() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), Some(2010));
let result = TitleSearchStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(result.is_none());
}
// --- ManualMovieStrategy ---
#[test]
fn manual_strategy_can_handle_cmd_with_title() {
let cmd = make_cmd(None, Some("Inception"), Some(2010));
assert!(ManualMovieStrategy.can_handle(&cmd));
}
#[test]
fn manual_strategy_cannot_handle_cmd_without_title() {
let cmd = make_cmd(Some("tt123"), None, None);
assert!(!ManualMovieStrategy.can_handle(&cmd));
}
#[tokio::test]
async fn manual_strategy_returns_existing_movie() {
let movie = make_movie();
let repo = RepoWithTitleMatch(movie.clone());
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), Some(2010));
let result = ManualMovieStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, false))));
}
#[tokio::test]
async fn manual_strategy_creates_new_movie_when_no_match() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), Some(2010));
let result = ManualMovieStrategy.resolve(&cmd, &deps).await.unwrap();
assert!(matches!(result, Some((_, true))));
}
#[tokio::test]
async fn manual_strategy_errors_without_year() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, Some("Inception"), None);
assert!(ManualMovieStrategy.resolve(&cmd, &deps).await.is_err());
}
// --- MovieResolver pipeline ---
#[tokio::test]
async fn resolver_returns_error_when_no_strategy_matches() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(None, None, None);
let result = MovieResolver::default_pipeline().resolve(&cmd, &deps).await;
assert!(result.is_err());
}
#[tokio::test]
async fn resolver_uses_cached_movie_when_external_id_matches() {
let movie = make_movie();
let repo = RepoWithExternalMovie(movie.clone());
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), None, None);
let (_, is_new) = MovieResolver::default_pipeline()
.resolve(&cmd, &deps)
.await
.unwrap();
assert!(!is_new);
}
#[tokio::test]
async fn resolver_falls_through_to_manual_when_external_and_title_both_fail() {
let repo = RepoEmpty;
let meta = MetaErrors;
let deps = MovieResolverDeps {
repository: &repo,
metadata_client: &meta,
};
let cmd = make_cmd(Some("tt123"), Some("Inception"), Some(2010));
let (_, is_new) = MovieResolver::default_pipeline()
.resolve(&cmd, &deps)
.await
.unwrap();
assert!(is_new);
}

169
crates/application/src/tests/worker.rs Normal file
View File

@@ -0,0 +1,169 @@
use super::*;
use async_trait::async_trait;
use domain::{errors::DomainError, events::{AckHandle, DomainEvent}};
use domain::value_objects::{ExternalMetadataId, MovieId};
use futures::{stream, stream::BoxStream};
use std::sync::{Arc, Mutex};
struct NoopAck;
#[async_trait]
impl AckHandle for NoopAck {
async fn ack(&self) -> Result<(), DomainError> { Ok(()) }
async fn nack(&self) -> Result<(), DomainError> { Ok(()) }
}
struct VecConsumer {
events: Vec<DomainEvent>,
}
impl EventConsumer for VecConsumer {
fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
let envelopes: Vec<Result<EventEnvelope, DomainError>> = self
.events
.iter()
.cloned()
.map(|e| Ok(EventEnvelope::new(e, Box::new(NoopAck))))
.collect();
Box::pin(stream::iter(envelopes))
}
}
struct RecordingHandler {
calls: Arc<Mutex<Vec<&'static str>>>,
}
#[async_trait]
impl EventHandler for RecordingHandler {
async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
let label = match event {
DomainEvent::MovieDiscovered { .. } => "movie_discovered",
DomainEvent::ReviewLogged { .. } => "review_logged",
DomainEvent::ReviewUpdated { .. } => "review_updated",
DomainEvent::ReviewDeleted { .. } => "review_deleted",
DomainEvent::MovieDeleted { .. } => "movie_deleted",
DomainEvent::UserUpdated { .. } => "user_updated",
DomainEvent::MovieEnrichmentRequested { .. } => "movie_enrichment_requested",
DomainEvent::ImageStored { .. } => "image_stored",
};
self.calls.lock().unwrap().push(label);
Ok(())
}
}
fn movie_discovered() -> DomainEvent {
DomainEvent::MovieDiscovered {
movie_id: MovieId::generate(),
external_metadata_id: ExternalMetadataId::new("tt1234567".into()).unwrap(),
}
}
#[tokio::test]
async fn dispatches_to_all_handlers() {
let calls = Arc::new(Mutex::new(vec![]));
let consumer = VecConsumer { events: vec![movie_discovered()] };
let handler = RecordingHandler { calls: Arc::clone(&calls) };
WorkerService::new(Arc::new(consumer), vec![Arc::new(handler)])
.run()
.await;
assert_eq!(*calls.lock().unwrap(), vec!["movie_discovered"]);
}
#[tokio::test]
async fn nacks_when_handler_fails() {
let nack_called = Arc::new(Mutex::new(false));
struct TrackingAck {
nack_called: Arc<Mutex<bool>>,
}
#[async_trait]
impl AckHandle for TrackingAck {
async fn ack(&self) -> Result<(), DomainError> { Ok(()) }
async fn nack(&self) -> Result<(), DomainError> {
*self.nack_called.lock().unwrap() = true;
Ok(())
}
}
struct TrackingConsumer {
event: DomainEvent,
nack_called: Arc<Mutex<bool>>,
}
impl EventConsumer for TrackingConsumer {
fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
let envelope = EventEnvelope::new(
self.event.clone(),
Box::new(TrackingAck { nack_called: Arc::clone(&self.nack_called) }),
);
Box::pin(stream::iter(vec![Ok(envelope)]))
}
}
struct FailingHandler;
#[async_trait]
impl EventHandler for FailingHandler {
async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> {
Err(DomainError::InfrastructureError("boom".into()))
}
}
let consumer = TrackingConsumer {
event: movie_discovered(),
nack_called: Arc::clone(&nack_called),
};
WorkerService::new(Arc::new(consumer), vec![Arc::new(FailingHandler)])
.run()
.await;
assert!(*nack_called.lock().unwrap());
}
#[tokio::test]
async fn acks_when_all_handlers_succeed() {
let ack_called = Arc::new(Mutex::new(false));
struct TrackingAck {
ack_called: Arc<Mutex<bool>>,
}
#[async_trait]
impl AckHandle for TrackingAck {
async fn ack(&self) -> Result<(), DomainError> {
*self.ack_called.lock().unwrap() = true;
Ok(())
}
async fn nack(&self) -> Result<(), DomainError> { Ok(()) }
}
struct TrackingConsumer {
event: DomainEvent,
ack_called: Arc<Mutex<bool>>,
}
impl EventConsumer for TrackingConsumer {
fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
let envelope = EventEnvelope::new(
self.event.clone(),
Box::new(TrackingAck { ack_called: Arc::clone(&self.ack_called) }),
);
Box::pin(stream::iter(vec![Ok(envelope)]))
}
}
let consumer = TrackingConsumer {
event: movie_discovered(),
ack_called: Arc::clone(&ack_called),
};
WorkerService::new(Arc::new(consumer), vec![])
.run()
.await;
assert!(*ack_called.lock().unwrap());
}

173
crates/application/src/worker.rs
View File

@@ -50,174 +50,5 @@ impl WorkerService {
}
#[cfg(test)]
mod tests {
use super::*;
use async_trait::async_trait;
use domain::{errors::DomainError, events::{AckHandle, DomainEvent}};
use domain::value_objects::{ExternalMetadataId, MovieId};
use futures::{stream, stream::BoxStream};
use std::sync::{Arc, Mutex};
struct NoopAck;
#[async_trait]
impl AckHandle for NoopAck {
async fn ack(&self) -> Result<(), DomainError> { Ok(()) }
async fn nack(&self) -> Result<(), DomainError> { Ok(()) }
}
struct VecConsumer {
events: Vec<DomainEvent>,
}
impl EventConsumer for VecConsumer {
fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
let envelopes: Vec<Result<EventEnvelope, DomainError>> = self
.events
.iter()
.cloned()
.map(|e| Ok(EventEnvelope::new(e, Box::new(NoopAck))))
.collect();
Box::pin(stream::iter(envelopes))
}
}
struct RecordingHandler {
calls: Arc<Mutex<Vec<&'static str>>>,
}
#[async_trait]
impl EventHandler for RecordingHandler {
async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
let label = match event {
DomainEvent::MovieDiscovered { .. } => "movie_discovered",
DomainEvent::ReviewLogged { .. } => "review_logged",
DomainEvent::ReviewUpdated { .. } => "review_updated",
DomainEvent::ReviewDeleted { .. } => "review_deleted",
DomainEvent::MovieDeleted { .. } => "movie_deleted",
DomainEvent::UserUpdated { .. } => "user_updated",
DomainEvent::MovieEnrichmentRequested { .. } => "movie_enrichment_requested",
DomainEvent::ImageStored { .. } => "image_stored",
};
self.calls.lock().unwrap().push(label);
Ok(())
}
}
fn movie_discovered() -> DomainEvent {
DomainEvent::MovieDiscovered {
movie_id: MovieId::generate(),
external_metadata_id: ExternalMetadataId::new("tt1234567".into()).unwrap(),
}
}
#[tokio::test]
async fn dispatches_to_all_handlers() {
let calls = Arc::new(Mutex::new(vec![]));
let consumer = VecConsumer { events: vec![movie_discovered()] };
let handler = RecordingHandler { calls: Arc::clone(&calls) };
WorkerService::new(Arc::new(consumer), vec![Arc::new(handler)])
.run()
.await;
assert_eq!(*calls.lock().unwrap(), vec!["movie_discovered"]);
}
#[tokio::test]
async fn nacks_when_handler_fails() {
let nack_called = Arc::new(Mutex::new(false));
struct TrackingAck {
nack_called: Arc<Mutex<bool>>,
}
#[async_trait]
impl AckHandle for TrackingAck {
async fn ack(&self) -> Result<(), DomainError> { Ok(()) }
async fn nack(&self) -> Result<(), DomainError> {
*self.nack_called.lock().unwrap() = true;
Ok(())
}
}
struct TrackingConsumer {
event: DomainEvent,
nack_called: Arc<Mutex<bool>>,
}
impl EventConsumer for TrackingConsumer {
fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
let envelope = EventEnvelope::new(
self.event.clone(),
Box::new(TrackingAck { nack_called: Arc::clone(&self.nack_called) }),
);
Box::pin(stream::iter(vec![Ok(envelope)]))
}
}
struct FailingHandler;
#[async_trait]
impl EventHandler for FailingHandler {
async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> {
Err(DomainError::InfrastructureError("boom".into()))
}
}
let consumer = TrackingConsumer {
event: movie_discovered(),
nack_called: Arc::clone(&nack_called),
};
WorkerService::new(Arc::new(consumer), vec![Arc::new(FailingHandler)])
.run()
.await;
assert!(*nack_called.lock().unwrap());
}
#[tokio::test]
async fn acks_when_all_handlers_succeed() {
let ack_called = Arc::new(Mutex::new(false));
struct TrackingAck {
ack_called: Arc<Mutex<bool>>,
}
#[async_trait]
impl AckHandle for TrackingAck {
async fn ack(&self) -> Result<(), DomainError> {
*self.ack_called.lock().unwrap() = true;
Ok(())
}
async fn nack(&self) -> Result<(), DomainError> { Ok(()) }
}
struct TrackingConsumer {
event: DomainEvent,
ack_called: Arc<Mutex<bool>>,
}
impl EventConsumer for TrackingConsumer {
fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
let envelope = EventEnvelope::new(
self.event.clone(),
Box::new(TrackingAck { ack_called: Arc::clone(&self.ack_called) }),
);
Box::pin(stream::iter(vec![Ok(envelope)]))
}
}
let consumer = TrackingConsumer {
event: movie_discovered(),
ack_called: Arc::clone(&ack_called),
};
WorkerService::new(Arc::new(consumer), vec![])
.run()
.await;
assert!(*ack_called.lock().unwrap());
}
}
#[path = "tests/worker.rs"]
mod tests;