test: adding unit test for kill signal on database

integrationos-database/Cargo.toml

@@ -33,4 +33,5 @@ tracing-subscriber.workspace = true
 tracing.workspace = true
 
 [dev-dependencies]
+mockito.workspace = true
 testcontainers-modules = { workspace = true, features = ["postgres"] }

integrationos-database/src/main.rs

@@ -26,23 +26,14 @@ fn main() -> Result<()> {
         .block_on(async move {
             match config.database_connection_type {
                 DatabaseConnectionType::PostgreSql => {
-                    match PostgresDatabaseConnection::init(&config).await {
-                        Ok(server) => {
-                            if let Err(e) = server.run().await {
-                                PostgresDatabaseConnection::kill(&config, e.to_string()).await?;
-                                return Err(anyhow::anyhow!("Could not run server: {e}"));
-                            }
+                    let server = PostgresDatabaseConnection::init(&config).await?;
 
-                            Ok(())
-                        }
-                        Err(e) => {
-                            tracing::error!("Could not initialize storage: {e}");
-
-                            PostgresDatabaseConnection::kill(&config, e.to_string()).await?;
-
-                            Err(anyhow::anyhow!("Could not initialize storage: {e}"))
-                        }
+                    if let Err(e) = server.run().await {
+                        PostgresDatabaseConnection::kill(&config, e.to_string()).await?;
+                        return Err(e);
                     }
+
+                    Ok(())
                 }
             }
         })

integrationos-database/src/service/init.rs

@@ -20,15 +20,24 @@ pub trait Initializer {
 #[async_trait]
 impl Initializer for PostgresDatabaseConnection {
     async fn init(config: &DatabaseConnectionConfig) -> Result<Server, anyhow::Error> {
-        let postgres: PostgresDatabaseConnection = PostgresDatabaseConnection::new(config).await?;
-        let storage: Arc<dyn Storage> = Arc::new(postgres);
-
-        Ok(Server {
-            state: Arc::new(AppState {
-                config: config.clone(),
-                storage,
-            }),
-        })
+        let postgres = PostgresDatabaseConnection::new(config).await;
+
+        match postgres {
+            Ok(postgres) => {
+                let storage: Arc<dyn Storage> = Arc::new(postgres);
+
+                Ok(Server {
+                    state: Arc::new(AppState {
+                        config: config.clone(),
+                        storage,
+                    }),
+                })
+            }
+            Err(e) => {
+                PostgresDatabaseConnection::kill(config, e.to_string()).await?;
+                Err(e)
+            }
+        }
     }
 
     async fn kill(
@@ -50,10 +59,12 @@ impl Initializer for PostgresDatabaseConnection {
         client
             .post(format!("{}/v1/emit", emit_url))
             .header("content-type", "application/json")
-            .body(value.to_string())
+            .json(&value)
             .send()
             .await?;
 
+        tracing::info!("Event for dispose of connection {connection_id} emitted");
+
         Ok(())
     }
 }

integrationos-database/tests/context.rs

@@ -25,6 +25,7 @@ static TRACING: OnceLock<()> = OnceLock::new();
 pub struct TestServer {
     pub port: u16,
     pub client: reqwest::Client,
+    // pub mock_server: ServerGuard,
 }
 
 #[derive(Debug, Clone, Eq, PartialEq)]
@@ -34,7 +35,7 @@ pub struct ApiResponse<T: DeserializeOwned = Value> {
 }
 
 impl TestServer {
-    pub async fn new() -> Result<Self, IntegrationOSError> {
+    pub async fn new(r#override: HashMap<String, String>) -> Result<Self, IntegrationOSError> {
         TRACING.get_or_init(|| {
             let filter = EnvFilter::builder()
                 .with_default_directive(LevelFilter::INFO.into())
@@ -53,7 +54,7 @@ impl TestServer {
             .expect("Failed to get local address")
             .port();
 
-        let config = DatabaseConnectionConfig::init_from_hashmap(&HashMap::from([
+        let config_map: HashMap<String, String> = HashMap::from([
             (
                 "INTERNAL_SERVER_ADDRESS".to_string(),
                 format!("0.0.0.0:{server_port}"),
@@ -71,12 +72,15 @@ impl TestServer {
             ("POSTGRES_HOST".to_string(), "localhost".to_string()),
             ("POSTGRES_PORT".to_string(), port.to_string()),
             ("POSTGRES_NAME".to_string(), "postgres".to_string()),
-        ]))
-        .expect("Failed to initialize storage config");
+        ])
+        .into_iter()
+        .chain(r#override.into_iter())
+        .collect::<HashMap<String, String>>();
 
-        let server = PostgresDatabaseConnection::init(&config)
-            .await
-            .expect("Failed to initialize storage");
+        let config = DatabaseConnectionConfig::init_from_hashmap(&config_map)
+            .expect("Failed to initialize storage config");
+
+        let server = PostgresDatabaseConnection::init(&config).await?;
 
         tokio::task::spawn(async move { server.run().await });
 
@@ -87,6 +91,7 @@ impl TestServer {
         Ok(Self {
             port: server_port,
             client,
+            // mock_server,
         })
     }
 

integrationos-database/tests/http/connection.rs

@@ -1,11 +1,12 @@
 use crate::context::TestServer;
 use http::{Method, StatusCode};
-use integrationos_domain::IntegrationOSError;
+use integrationos_domain::{IntegrationOSError, Unit};
 use serde_json::Value;
+use std::collections::HashMap;
 
 #[tokio::test]
-async fn test_execute_probe() -> Result<(), IntegrationOSError> {
-    let server = TestServer::new().await?;
+async fn test_execute_probe() -> Result<Unit, IntegrationOSError> {
+    let server = TestServer::new(HashMap::new()).await?;
     let result = server
         .send_request::<Value, Value>("database/probe", Method::GET, None)
         .await?;
@@ -18,8 +19,8 @@ async fn test_execute_probe() -> Result<(), IntegrationOSError> {
 }
 
 #[tokio::test]
-async fn test_execute_raw() -> Result<(), IntegrationOSError> {
-    let server = TestServer::new().await?;
+async fn test_execute_raw() -> Result<Unit, IntegrationOSError> {
+    let server = TestServer::new(HashMap::new()).await?;
 
     let create_query =
         "CREATE TABLE IF NOT EXISTS users (id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL);";

integrationos-database/tests/http/mod.rs

@@ -1 +1,2 @@
 pub mod connection;
+pub mod signal;

integrationos-emit/tests/context.rs

@@ -39,7 +39,7 @@ pub struct ApiResponse<T: DeserializeOwned = Value> {
 }
 
 impl TestServer {
-    pub async fn new(stream: bool) -> Result<Self, IntegrationOSError> {
+    pub async fn new() -> Result<Self, IntegrationOSError> {
         TRACING.get_or_init(|| {
             let filter = EnvFilter::builder()
                 .with_default_directive(LevelFilter::INFO.into())
@@ -69,7 +69,9 @@ impl TestServer {
             .expect("Failed to get local address")
             .port();
 
-        let mut config = vec![
+        let mock_server = MockServer::new_async().await;
+        let mock_uri = mock_server.url();
+        let config = vec![
             (
                 "INTERNAL_SERVER_ADDRESS".to_string(),
                 format!("0.0.0.0:{server_port}"),
@@ -90,32 +92,25 @@ impl TestServer {
             ),
             ("PARTITION_COUNT".to_string(), "1".to_string()),
             ("ENVIRONMENT".to_string(), "test".to_string()),
-        ];
-
-        let mock_server = MockServer::new_async().await;
-
-        if stream {
-            let uri = mock_server.url();
-
-            config.push(("EVENT_STREAM_PROVIDER".to_string(), "fluvio".to_string()));
-            config.push(("EVENT_STREAM_PORT".to_string(), "9103".to_string()));
-            config.push((
+            ("EVENT_STREAM_PROVIDER".to_string(), "fluvio".to_string()),
+            ("EVENT_STREAM_PORT".to_string(), "9103".to_string()),
+            (
                 "EVENT_STREAM_PRODUCER_TOPIC".to_string(),
                 "events".to_string(),
-            ));
-            config.push((
+            ),
+            (
                 "EVENT_STREAM_CONSUMER_TOPIC".to_string(),
                 "events".to_string(),
-            ));
-            config.push((
+            ),
+            (
                 "EVENT_STREAM_CONSUMER_GROUP".to_string(),
                 "event-all-partitions-consumer".to_string(),
-            ));
-            config.push((
+            ),
+            (
                 "EVENT_CALLBACK_URL".to_string(),
-                format!("{uri}/v1/event-callbacks"),
-            ));
-        }
+                format!("{mock_uri}/v1/event-callbacks"),
+            ),
+        ];
 
         let config = EmitterConfig::init_from_hashmap(&HashMap::from_iter(config))
             .expect("Failed to initialize storage config");

integrationos-emit/tests/http/emitter.rs

@@ -14,7 +14,7 @@ const PARALLEL_REQUESTS: usize = 10;
 
 #[tokio::test]
 async fn test_concurrent_requests() -> Result<Unit, IntegrationOSError> {
-    let server = TestServer::new(true).await?;
+    let server = TestServer::new().await?;
     let payload = json!({
         "type": "DatabaseConnectionLost",
         "connectionId": "conn::GAL2svWJp9k::MtmXaau5Qf6R5n3Y-L9ejQ"
@@ -76,7 +76,7 @@ async fn test_concurrent_requests() -> Result<Unit, IntegrationOSError> {
 
 #[tokio::test]
 async fn test_event_processed() -> Result<Unit, IntegrationOSError> {
-    let mut server = TestServer::new(true).await?;
+    let mut server = TestServer::new().await?;
 
     let id = Id::now(IdPrefix::Connection).to_string();
     let payload = json!({