fix(tests): restore Rust integration coverage

docker-compose.yml

@@ -29,10 +29,10 @@ services:
       - vars.env
     environment:
       - EVENTSTORE_GOSSIP_SEED=172.30.240.12:2113,172.30.240.13:2113
-      - EVENTSTORE_INT_IP=172.30.240.11
+      - EVENTSTORE_REPLICATION_IP=172.30.240.11
       - EVENTSTORE_CERTIFICATE_FILE=/etc/eventstore/certs/node1/node.crt
       - EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/eventstore/certs/node1/node.key
-      - EVENTSTORE_ADVERTISE_HTTP_PORT_TO_CLIENT_AS=2111
+      - EVENTSTORE_ADVERTISE_NODE_PORT_TO_CLIENT_AS=2111
     ports:
       - 2111:2113
     networks:
@@ -48,10 +48,10 @@ services:
     <<: *template
     environment:
       - EVENTSTORE_GOSSIP_SEED=172.30.240.11:2113,172.30.240.13:2113
-      - EVENTSTORE_INT_IP=172.30.240.12
+      - EVENTSTORE_REPLICATION_IP=172.30.240.12
       - EVENTSTORE_CERTIFICATE_FILE=/etc/eventstore/certs/node2/node.crt
       - EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/eventstore/certs/node2/node.key
-      - EVENTSTORE_ADVERTISE_HTTP_PORT_TO_CLIENT_AS=2112
+      - EVENTSTORE_ADVERTISE_NODE_PORT_TO_CLIENT_AS=2112
     ports:
       - 2112:2113
     networks:
@@ -62,10 +62,10 @@ services:
     <<: *template
     environment:
       - EVENTSTORE_GOSSIP_SEED=172.30.240.11:2113,172.30.240.12:2113
-      - EVENTSTORE_INT_IP=172.30.240.13
+      - EVENTSTORE_REPLICATION_IP=172.30.240.13
       - EVENTSTORE_CERTIFICATE_FILE=/etc/eventstore/certs/node3/node.crt
       - EVENTSTORE_CERTIFICATE_PRIVATE_KEY_FILE=/etc/eventstore/certs/node3/node.key
-      - EVENTSTORE_ADVERTISE_HTTP_PORT_TO_CLIENT_AS=2113
+      - EVENTSTORE_ADVERTISE_NODE_PORT_TO_CLIENT_AS=2113
     ports:
       - 2113:2113
     networks:

trogon-eventstore/tests/fixtures/mod.rs

@@ -0,0 +1,61 @@
+use std::{fs, path::Path};
+
+use testcontainers::{Image, core::MountType};
+
+use crate::images::EventStoreDB;
+
+fn fixture_root() -> &'static Path {
+    Path::new(env!("CARGO_MANIFEST_DIR"))
+        .parent()
+        .expect("workspace root")
+}
+
+#[test]
+fn cluster_uses_supported_server_options() {
+    let compose = fs::read_to_string(fixture_root().join("docker-compose.yml")).unwrap();
+    let shared = fs::read_to_string(fixture_root().join("vars.env")).unwrap();
+
+    for option in [
+        "EVENTSTORE_REPLICATION_IP=",
+        "EVENTSTORE_ADVERTISE_NODE_PORT_TO_CLIENT_AS=",
+    ] {
+        assert!(compose.contains(option), "missing {option}");
+    }
+
+    for option in ["EVENTSTORE_REPLICATION_PORT=", "EVENTSTORE_NODE_PORT="] {
+        assert!(shared.contains(option), "missing {option}");
+    }
+
+    for option in [
+        "EVENTSTORE_INT_IP=",
+        "EVENTSTORE_INT_TCP_PORT=",
+        "EVENTSTORE_HTTP_PORT=",
+        "EVENTSTORE_ADVERTISE_HTTP_PORT_TO_CLIENT_AS=",
+        "EVENTSTORE_ENABLE_ATOM_PUB_OVER_HTTP=",
+    ] {
+        assert!(!compose.contains(option), "obsolete {option}");
+        assert!(!shared.contains(option), "obsolete {option}");
+    }
+}
+
+#[test]
+fn invalid_root_certificate_uses_generated_untrusted_ca() {
+    let generator = fs::read_to_string(fixture_root().join("configure-tls-for-tests.yml")).unwrap();
+    let test = fs::read_to_string(
+        fixture_root().join("trogon-eventstore/tests/misc/root_certificates.rs"),
+    )
+    .unwrap();
+
+    assert!(generator.contains("create-ca -out ./untrusted-ca"));
+    assert!(test.contains("certs/untrusted-ca/ca.crt"));
+}
+
+#[test]
+fn database_directory_uses_named_volume_mount() {
+    let image = EventStoreDB::default().attach_volume_to_db_directory("fixture-volume".into());
+    let mount = image.mounts().into_iter().next().expect("database mount");
+
+    assert!(matches!(mount.mount_type(), MountType::Volume));
+    assert_eq!(mount.source(), Some("fixture-volume"));
+    assert_eq!(mount.target(), Some("/var/lib/eventstore"));
+}

trogon-eventstore/tests/images.rs

@@ -72,8 +72,10 @@ impl EventStoreDB {
     }
 
     pub fn attach_volume_to_db_directory(mut self, volume: String) -> Self {
-        self.mounts
-            .push(Mount::bind_mount(volume, "/var/lib/eventstore".to_string()));
+        self.mounts.push(Mount::volume_mount(
+            volume,
+            "/var/lib/eventstore".to_string(),
+        ));
 
         self
     }

trogon-eventstore/tests/integration.rs

@@ -1,16 +1,16 @@
 mod api;
 mod common;
+mod fixtures;
 mod images;
 mod misc;
 mod plugins;
 
 use crate::common::{fresh_stream_id, generate_events};
-use futures::channel::oneshot;
 use std::time::Duration;
 use testcontainers::{ImageExt, core::ContainerPort, runners::AsyncRunner};
 use tracing::{debug, error};
 use tracing_subscriber::EnvFilter;
-use trogon_eventstore::{Client, ClientSettings};
+use trogon_eventstore::{Client, ClientSettings, Subscription, SubscriptionEvent};
 
 fn configure_logging() {
     tracing_subscriber::fmt::fmt()
@@ -23,19 +23,78 @@ fn configure_logging() {
         .init();
 }
 
-type VolumeName = String;
+struct TestVolume<C: FnMut(&str) = fn(&str)> {
+    name: String,
+    cleanup: C,
+}
+
+impl TestVolume {
+    fn create() -> eyre::Result<Self> {
+        let name = format!("dir-{}", uuid::Uuid::new_v4());
+
+        let status = std::process::Command::new("docker")
+            .arg("volume")
+            .arg("create")
+            .arg(&name)
+            .status()?;
 
-fn create_unique_volume() -> eyre::Result<VolumeName> {
-    let dir_name = uuid::Uuid::new_v4();
-    let dir_name = format!("dir-{}", dir_name);
+        if !status.success() {
+            eyre::bail!("failed to create Docker volume {name}");
+        }
+
+        Ok(Self {
+            name,
+            cleanup: remove_test_volume,
+        })
+    }
+}
+
+impl<C: FnMut(&str)> TestVolume<C> {
+    fn name(&self) -> &str {
+        &self.name
+    }
+}
 
-    std::process::Command::new("docker")
+impl<C: FnMut(&str)> Drop for TestVolume<C> {
+    fn drop(&mut self) {
+        (self.cleanup)(&self.name);
+    }
+}
+
+fn remove_test_volume(name: &str) {
+    match std::process::Command::new("docker")
         .arg("volume")
-        .arg("create")
-        .arg(format!("--name {}", dir_name))
-        .output()?;
+        .arg("rm")
+        .arg(name)
+        .status()
+    {
+        Ok(status) if status.success() => {}
+        Ok(status) => error!("Failed to remove Docker volume {name}: {status}"),
+        Err(err) => error!("Failed to remove Docker volume {name}: {err}"),
+    }
+}
+
+#[cfg(test)]
+mod test_volume_tests {
+    use super::TestVolume;
+    use std::sync::{Arc, Mutex};
+
+    #[test]
+    fn cleanup_runs_when_test_volume_is_dropped() {
+        let cleaned = Arc::new(Mutex::new(Vec::new()));
+
+        {
+            let cleaned = Arc::clone(&cleaned);
+            let volume = TestVolume {
+                name: "test-volume".to_string(),
+                cleanup: move |name: &str| cleaned.lock().unwrap().push(name.to_string()),
+            };
+
+            assert_eq!(volume.name(), "test-volume");
+        }
 
-    Ok(dir_name)
+        assert_eq!(*cleaned.lock().unwrap(), ["test-volume"]);
+    }
 }
 
 async fn wait_node_is_alive(
@@ -101,6 +160,32 @@ async fn wait_node_is_alive(
     }
 }
 
+async fn wait_for_subscription_confirmation(
+    subscription: &mut Subscription,
+) -> trogon_eventstore::Result<()> {
+    loop {
+        if let SubscriptionEvent::Confirmed(_) = subscription.next_subscription_event().await? {
+            return Ok(());
+        }
+    }
+}
+
+async fn assert_subscription_batch(
+    subscription: &mut Subscription,
+    event_type: &str,
+    revisions: std::ops::Range<u64>,
+) -> trogon_eventstore::Result<()> {
+    for revision in revisions {
+        let event = subscription.next().await?;
+        let event = event.get_original_event();
+
+        assert_eq!(event.event_type, event_type);
+        assert_eq!(event.revision, revision);
+    }
+
+    Ok(())
+}
+
 // This function assumes that we are using the admin credentials. It's possible during CI that
 // the cluster hasn't created the admin user yet, leading to failing the tests.
 async fn wait_for_admin_to_be_available(client: &Client) -> trogon_eventstore::Result<()> {
@@ -361,10 +446,10 @@ async fn single_node_discover_error() -> eyre::Result<()> {
 
 #[tokio::test(flavor = "multi_thread")]
 async fn single_node_auto_resub_on_connection_drop() -> eyre::Result<()> {
-    let volume = create_unique_volume()?;
+    let volume = TestVolume::create()?;
     let image = images::EventStoreDB::default()
         .insecure_mode()
-        .attach_volume_to_db_directory(volume);
+        .attach_volume_to_db_directory(volume.name().to_owned());
 
     let container = image
         .clone()
@@ -385,34 +470,25 @@ async fn single_node_auto_resub_on_connection_drop() -> eyre::Result<()> {
     let mut stream = client
         .subscribe_to_stream(stream_name.as_str(), &options)
         .await;
-    let max = 6usize;
-    let (tx, recv) = oneshot::channel();
 
-    tokio::spawn(async move {
-        let mut count = 0usize;
+    tokio::time::timeout(
+        Duration::from_secs(60),
+        wait_for_subscription_confirmation(&mut stream),
+    )
+    .await??;
 
-        loop {
-            if let Err(e) = stream.next().await {
-                error!("Subscription exited with: {}", e);
-                break;
-            }
-
-            count += 1;
-
-            if count == max {
-                break;
-            }
-        }
-
-        tx.send(count).unwrap();
-    });
-
-    let events = generate_events("reconnect", 3);
+    let events = generate_events("reconnect-before", 3);
 
     let _ = client
         .append_to_stream(stream_name.as_str(), &Default::default(), events)
         .await?;
 
+    tokio::time::timeout(
+        Duration::from_secs(60),
+        assert_subscription_batch(&mut stream, "reconnect-before", 0..3),
+    )
+    .await??;
+
     container.stop().await?;
     debug!("Server is stopped, restarting...");
     let _container = image
@@ -423,19 +499,17 @@ async fn single_node_auto_resub_on_connection_drop() -> eyre::Result<()> {
     wait_node_is_alive(&cloned_setts, 3_113).await?;
     debug!("Server is up again");
 
-    let events = generate_events("reconnect", 3);
+    let events = generate_events("reconnect-after", 3);
 
     let _ = client
         .append_to_stream(stream_name.as_str(), &Default::default(), events)
         .await?;
 
-    let test_count = tokio::time::timeout(std::time::Duration::from_secs(60), recv).await??;
-
-    assert_eq!(
-        test_count, 6,
-        "We are testing proper state after subscription upon reconnection: got {} expected {}.",
-        test_count, 6
-    );
+    tokio::time::timeout(
+        Duration::from_secs(60),
+        assert_subscription_batch(&mut stream, "reconnect-after", 3..6),
+    )
+    .await??;
 
     Ok(())
 }

trogon-eventstore/tests/misc/root_certificates.rs

@@ -19,7 +19,7 @@ async fn test_with_valid_root_certificate(port: u16) -> eyre::Result<()> {
 
 async fn test_with_invalid_certificate(port: u16) -> eyre::Result<()> {
     // invalid root certificate
-    let root_cert = "certs/node1/node.crt";
+    let root_cert = "certs/untrusted-ca/ca.crt";
 
     let setts = format!(
         "esdb://admin:changeit@localhost:{}?tlsVerifyCert=true&tls=true&tlsCaFile={}",

vars.env

@@ -1,8 +1,7 @@
 EVENTSTORE_CLUSTER_SIZE=3
 EVENTSTORE_RUN_PROJECTIONS=All
-EVENTSTORE_INT_TCP_PORT=1112
-EVENTSTORE_HTTP_PORT=2113
+EVENTSTORE_REPLICATION_PORT=1112
+EVENTSTORE_NODE_PORT=2113
 EVENTSTORE_TRUSTED_ROOT_CERTIFICATES_PATH=/etc/eventstore/certs/ca
 EVENTSTORE_DISCOVER_VIA_DNS=false
-EVENTSTORE_ENABLE_ATOM_PUB_OVER_HTTP=true
-EVENTSTORE_ADVERTISE_HOST_TO_CLIENT_AS=localhost
+EVENTSTORE_ADVERTISE_HOST_TO_CLIENT_AS=localhost