test: move small adapter tests to tests/
event-publisher, export, rss, media-server, object-storage, nats(config).
Kept inline (private fn, no public path): nats src/subject.rs
tests::{review_logged_subject,review_updated_subject,movie_discovered_subject,
consumer_subject_filter_appends_wildcard} - event_to_subject/consumer_subject_filter private.
object-storage build_local_store tests rewritten via StorageConfig::from_config.
This commit is contained in:
58
crates/adapters/event-publisher/tests/event_channel_test.rs
Normal file
58
crates/adapters/event-publisher/tests/event_channel_test.rs
Normal file
@@ -0,0 +1,58 @@
|
||||
use domain::ports::{EventConsumer, EventPublisher};
|
||||
use domain::{
|
||||
events::DomainEvent,
|
||||
value_objects::{ExternalMetadataId, MovieId},
|
||||
};
|
||||
use event_publisher::{EventPublisherConfig, create_event_channel};
|
||||
use futures::StreamExt;
|
||||
|
||||
fn movie_discovered() -> DomainEvent {
|
||||
DomainEvent::MovieDiscovered {
|
||||
movie_id: MovieId::generate(),
|
||||
external_metadata_id: ExternalMetadataId::new("tt1234567".into()).unwrap(),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn consumer_yields_published_events() {
|
||||
let config = EventPublisherConfig { channel_buffer: 8 };
|
||||
let (publisher, consumer) = create_event_channel(config);
|
||||
|
||||
publisher.publish(&movie_discovered()).await.unwrap();
|
||||
drop(publisher);
|
||||
|
||||
let mut stream = consumer.consume();
|
||||
let envelope = stream.next().await.unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
envelope.event,
|
||||
DomainEvent::MovieDiscovered { .. }
|
||||
));
|
||||
assert!(stream.next().await.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn consumer_yields_multiple_events_in_order() {
|
||||
let config = EventPublisherConfig { channel_buffer: 8 };
|
||||
let (publisher, consumer) = create_event_channel(config);
|
||||
|
||||
publisher.publish(&movie_discovered()).await.unwrap();
|
||||
publisher.publish(&movie_discovered()).await.unwrap();
|
||||
drop(publisher);
|
||||
|
||||
let mut stream = consumer.consume();
|
||||
let first = stream.next().await.unwrap().unwrap();
|
||||
let second = stream.next().await.unwrap().unwrap();
|
||||
assert!(matches!(first.event, DomainEvent::MovieDiscovered { .. }));
|
||||
assert!(matches!(second.event, DomainEvent::MovieDiscovered { .. }));
|
||||
assert!(stream.next().await.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stream_ends_when_publisher_dropped() {
|
||||
let config = EventPublisherConfig { channel_buffer: 8 };
|
||||
let (publisher, consumer) = create_event_channel(config);
|
||||
drop(publisher);
|
||||
|
||||
let mut stream = consumer.consume();
|
||||
assert!(stream.next().await.is_none());
|
||||
}
|
||||
Reference in New Issue
Block a user