cassandra_int_tests: add node lost test case

shotover-proxy/tests/cassandra_int_tests/cluster/single_rack_v4.rs

@@ -254,6 +254,7 @@ pub async fn test_node_going_down(
  compose: DockerCompose,
  shotover_manager: ShotoverManager,
  driver: CassandraDriver,
+ kill: bool,
 ) {
  let mut connection_shotover = CassandraConnection::new("127.0.0.1", 9042, driver).await;
  connection_shotover
@@ -277,7 +278,11 @@ pub async fn test_node_going_down(
 
  // stop one of the containers to trigger a status change event.
  // event_connection_direct is connecting to cassandra-one, so make sure to instead kill caassandra-two.
- compose.stop_service("cassandra-two").await;
+ if kill {
+ compose.kill_service("cassandra-two").await;
+ } else {
+ compose.stop_service("cassandra-two").await;
+ }
 
  loop {
  // The direct connection should allow all events to pass through

shotover-proxy/tests/cassandra_int_tests/mod.rs

@@ -199,7 +199,22 @@ async fn cluster_single_rack_v4(#[case] driver: CassandraDriver) {
 
  let shotover_manager =
  ShotoverManager::from_topology_file("example-configs/cassandra-cluster/topology-v4.yaml");
- cluster::single_rack_v4::test_node_going_down(compose, shotover_manager, driver).await;
+ cluster::single_rack_v4::test_node_going_down(compose, shotover_manager, driver, false).await;
+}
+
+#[cfg(feature = "cassandra-cpp-driver-tests")]
+#[rstest]
+//#[case::cdrs(CdrsTokio)]
+#[cfg_attr(feature = "cassandra-cpp-driver-tests", case(Datastax))]
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn cluster_single_rack_node_lost(#[case] driver: CassandraDriver) {
+ let compose =
+ DockerCompose::new("example-configs/cassandra-cluster/docker-compose-cassandra-v4.yaml");
+
+ let shotover_manager =
+ ShotoverManager::from_topology_file("example-configs/cassandra-cluster/topology-v4.yaml");
+ cluster::single_rack_v4::test_node_going_down(compose, shotover_manager, driver, true).await;
 }
 
 #[cfg(feature = "cassandra-cpp-driver-tests")]

test-helpers/src/docker_compose.rs

@@ -1,5 +1,5 @@
 use anyhow::{anyhow, Result};
-use docker_api::Docker;
+use docker_api::{Container, Docker};
 use std::io::ErrorKind;
 use std::process::Command;
 use std::time::{self, Duration};
@@ -91,10 +91,24 @@ impl DockerCompose {
 
  /// Stops the container with the provided service name
  pub async fn stop_service(&self, service_name: &str) {
+ for container in self.get_containers_with_service_name(service_name).await {
+ container.stop(None).await.unwrap();
+ }
+ }
+
+ /// Kills the container with the provided service name
+ pub async fn kill_service(&self, service_name: &str) {
+ for container in self.get_containers_with_service_name(service_name).await {
+ container.kill(None).await.unwrap();
+ }
+ }
+
+ async fn get_containers_with_service_name(&self, service_name: &str) -> Vec<Container> {
  let docker = Docker::new("unix:///var/run/docker.sock").unwrap();
  let containers = docker.containers();
  let mut found = false;
  let mut all_names: Vec<String> = vec![];
+ let mut result = vec![];
  for container in containers.list(&Default::default()).await.unwrap() {
  let compose_service = container
  .labels
@@ -104,8 +118,7 @@ impl DockerCompose {
  .to_string();
  if compose_service == service_name {
  found = true;
- let container = containers.get(container.id.unwrap());
- container.stop(None).await.unwrap();
+ result.push(containers.get(container.id.unwrap()));
  }
  all_names.push(compose_service);
  }
@@ -114,6 +127,7 @@ impl DockerCompose {
  "container was not found with expected docker compose service name, actual names were {:?}",
  all_names
  );
+ result
  }
 
  fn wait_for_containers_to_startup(&self) {