Add testing for notify behaviour

src/ingestion/db.rs

@@ -1,8 +1,7 @@
 use crate::error::AppError;
 use anyhow::anyhow;
 use entity::{expenditure, transaction};
-use migration::PostgresQueryBuilder;
-use sea_orm::sea_query::OnConflict;
+use sea_orm::sea_query::{OnConflict, PostgresQueryBuilder};
 use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, Iterable, TransactionTrait};
 use sea_orm::{
     ConnectionTrait, DatabaseBackend, DatabaseTransaction, DbErr, QueryFilter, QueryTrait,
@@ -108,8 +107,60 @@ async fn notify_new_transactions(
 ) -> Result<(), AppError> {
     let payload = serde_json::to_string(&new_transaction_ids).map_err(|e| anyhow!(e))?;
 
-    db.execute_unprepared(&format!(r#"NOTIFY monzo_new_transactions, '{payload}'"#,))
+    db.execute_unprepared(&format!(r#"NOTIFY monzo_new_transactions, '{payload}'"#))
         .await?;
 
     Ok(())
 }
+
+mod tests {
+    use anyhow::Error;
+    use sea_orm::{DatabaseConnection};
+    use sqlx::{PgPool};
+    use sqlx::postgres::PgListener;
+    use testcontainers::ContainerAsync;
+    use migration::MigratorTrait;
+    use testcontainers::runners::AsyncRunner;
+    use super::notify_new_transactions;
+
+    async fn initialise() -> Result<(ContainerAsync<testcontainers_modules::postgres::Postgres>, DatabaseConnection, PgPool), Error> {
+        let container = testcontainers_modules::postgres::Postgres::default()
+            .start()
+            .await?;
+
+        // prepare connection string
+        let connection_string = &format!(
+            "postgres://postgres:postgres@127.0.0.1:{}/postgres",
+            container.get_host_port_ipv4(5432).await?
+        );
+
+        let db: DatabaseConnection = sea_orm::Database::connect(connection_string).await?;
+        migration::Migrator::up(&db, None).await?;
+
+        let pool = PgPool::connect(connection_string)
+            .await?;
+
+        Ok((container, db, pool))
+    }
+
+    #[tokio::test]
+    async fn test_notify() -> Result<(), Error> {
+        let (_container, db, pool) = initialise().await?;
+        let mut listener = PgListener::connect_with(&pool).await?;
+        listener.listen("monzo_new_transactions").await?;
+
+        let ids = vec!["test1".to_string(), "test2".to_string(), "test3".to_string()];
+
+        notify_new_transactions(
+            &db,
+            &ids,
+        ).await?;
+
+        let notification = listener.recv().await?;
+        let payload = notification.payload();
+        println!("Payload: {}", payload);
+        assert_eq!(serde_json::from_str::<Vec<String>>(&payload)?, ids, "Payloads do not match");
+
+        Ok(())
+    }
+}