fix integration tests

api/tests/context/environment.rs

@@ -9,10 +9,13 @@ pub struct Context {
 impl Context {
 	pub fn new() -> Self {
 		Self {
-			_guards: vec![set_env(
-				OsString::from("HASURA_GRAPHQL_JWT_SECRET"),
-				r#"{"type":"HS256","key":"secret","issuer":"hasura-auth-unit-tests"}"#,
-			)],
+			_guards: vec![
+				set_env(
+					OsString::from("HASURA_GRAPHQL_JWT_SECRET"),
+					r#"{"type":"HS256","key":"secret","issuer":"hasura-auth-unit-tests"}"#,
+				),
+				set_env(OsString::from("QUOTES_SYNCER_SLEEP_DURATION"), "0"),
+			],
 		}
 	}
 }

api/tests/context/mod.rs

@@ -3,7 +3,6 @@ use std::{env, sync::Arc};
 use anyhow::Result;
 use api::{
 	domain::projectors::{self, projections},
-	presentation::bootstrap,
 	Config,
 };
 use domain::{CompositePublisher, Event, EventPublisher, Publisher};
@@ -14,6 +13,7 @@ use rstest::fixture;
 use testcontainers::clients::Cli;
 use testing::context::{amqp, coinmarketcap, database, github};
 use tokio::task::JoinHandle;
+use tokio_cron_scheduler::Job;
 
 pub mod environment;
 pub mod indexer;
@@ -39,16 +39,18 @@ pub struct Context<'a> {
 	pub indexer: indexer::Context<'a>,
 	pub web3: web3::Context<'a>,
 	pub coinmarketcap: coinmarketcap::Context<'a>,
-	pub quotes_syncer: api::presentation::cron::quotes_syncer::Cron,
+	pub quotes_syncer: Job,
 	pub event_publisher: Arc<dyn Publisher<Event>>,
-	_event_listeners: Vec<JoinHandle<()>>,
+	_event_listeners: JoinHandle<Result<()>>,
 	_environment: environment::Context,
 }
 
 impl<'a> Context<'a> {
 	pub async fn new(docker: &'a Cli) -> Result<Context<'a>> {
 		tracing_subscriber::fmt::init();
 
+		let environment = environment::Context::new();
+
 		let database = database::Context::new(docker)?;
 		let amqp = amqp::Context::new(docker, vec![], vec![EXCHANGE_NAME]).await?;
 		let simple_storage = simple_storage::Context::new(docker)?;
@@ -111,6 +113,10 @@ impl<'a> Context<'a> {
 			coinmarketcap: coinmarketcap.config.clone(),
 		};
 
+		let event_listeners = tokio::spawn(api::presentation::event_listeners::_bootstrap(
+			config.clone(),
+		));
+
 		let event_publisher = CompositePublisher::new(vec![
 			Arc::new(EventPublisher::new(
 				projectors::event_store::Projector::new(database.client.clone()),
@@ -133,10 +139,9 @@ impl<'a> Context<'a> {
 			))),
 		]);
 
-		let (http_server, event_listeners, cron) = bootstrap(config.clone()).await?;
-
 		Ok(Self {
-			http_client: Client::tracked(http_server).await?,
+			http_client: Client::tracked(api::presentation::http::bootstrap(config.clone()).await?)
+				.await?,
 			database,
 			amqp,
 			simple_storage,
@@ -145,16 +150,17 @@ impl<'a> Context<'a> {
 			indexer,
 			web3,
 			coinmarketcap,
-			quotes_syncer: cron,
+			quotes_syncer: api::presentation::cron::quotes_syncer::bootstrap(config.clone())
+				.await?,
 			event_publisher: Arc::new(event_publisher),
 			_event_listeners: event_listeners,
-			_environment: environment::Context::new(),
+			_environment: environment,
 		})
 	}
 }
 
 impl<'a> Drop for Context<'a> {
 	fn drop(&mut self) {
-		self._event_listeners.iter().for_each(JoinHandle::abort);
+		self._event_listeners.abort();
 	}
 }

api/tests/crypto_quotes_sync_it.rs

@@ -1,4 +1,5 @@
 mod context;
+mod fixtures;
 mod models;
 
 use anyhow::Result;
@@ -12,7 +13,10 @@ use rust_decimal::Decimal;
 use rust_decimal_macros::dec;
 use testcontainers::clients::Cli;
 
-use crate::context::{docker, Context};
+use crate::{
+	context::{docker, Context},
+	fixtures::job::Runnable,
+};
 
 #[macro_use]
 extern crate diesel;
@@ -58,7 +62,7 @@ impl<'a> Test<'a> {
 
 		// When
 		let before = Utc::now().naive_utc();
-		self.context.quotes_syncer.run_once().await;
+		self.context.quotes_syncer.run().await?;
 		let after = Utc::now().naive_utc();
 
 		// Then

api/tests/fixtures/job.rs

@@ -0,0 +1,35 @@
+use anyhow::Result;
+use async_trait::async_trait;
+use rocket::http::hyper::body::HttpBody;
+use tokio::sync::mpsc;
+use tokio_cron_scheduler::{Job, JobScheduler, OnJobNotification};
+
+#[async_trait::async_trait]
+pub trait Runnable {
+	async fn run(&self) -> Result<()>;
+}
+
+#[async_trait]
+impl Runnable for Job {
+	async fn run(&self) -> Result<()> {
+		let mut job = self.clone();
+		let mut scheduler = JobScheduler::new().await?;
+		let (tx, mut rx) = mpsc::unbounded_channel();
+
+		let on_job_done: Box<OnJobNotification> = Box::new(move |_, _, _| {
+			let cloned_tx = tx.clone();
+			Box::pin(async move {
+				cloned_tx.clone().send(()).unwrap();
+			})
+		});
+
+		job.on_done_notification_add(&scheduler, on_job_done).await?;
+
+		scheduler.add(job).await?;
+		scheduler.start().await?;
+		rx.recv().await; // Wait for job completion
+		scheduler.shutdown().await?;
+
+		Ok(())
+	}
+}

api/tests/fixtures/mod.rs

@@ -1,5 +1,7 @@
 #![allow(unused)]
 
+pub mod job;
+
 use std::time::Duration;
 
 pub async fn retry<R, E, F: FnMut() -> std::result::Result<Vec<R>, E>, C: FnMut(&[R]) -> bool>(