fix(cassandra): fix cassandra container

Cargo.toml

@@ -150,6 +150,7 @@ azure_storage_blobs = "0.21.0"
 azure_storage = "0.21.0"
 base64 = "0.22.1"
 fantoccini = "0.21"
+cdrs-tokio = "9.0.2"
 
 [[example]]
 name = "anvil"

src/cassandra/mod.rs

@@ -1,3 +1,5 @@
+use std::borrow::Cow;
+
 use testcontainers::{core::WaitFor, Image};
 
 const NAME: &str = "cassandra";
@@ -11,27 +13,16 @@ const TAG: &str = "5.0.6";
 /// ```
 /// use std::time::Duration;
 ///
-/// use scylla::client::{session::Session, session_builder::SessionBuilder};
 /// use testcontainers::{runners::AsyncRunner, ImageExt};
 ///
 /// #[tokio::test]
 /// async fn default_cassandra() -> Result<(), Box<dyn std::error::Error + 'static>> {
-///     let image = ScyllaDB::default();
+///     let image = Cassandra::default().with_startup_timeout(Duration::from_secs(120));
 ///     let instance = image.start().await?;
 ///     let host = instance.get_host().await?;
 ///     let port = instance.get_host_port_ipv4(9042).await?;
 ///     let hostname = format!("{host}:{port}");
-///     let session: Session = SessionBuilder::new().known_node(hostname).build().await?;
-///
-///     let prepared_statement = session
-///         .prepare("SELECT release_version FROM system.local")
-///         .await?;
-///     let rows = session
-///         .execute_unpaged(&prepared_statement, &[])
-///         .await?
-///         .into_rows_result()?;
-///     let (version,) = rows.single_row::<(String,)>()?;
-///     assert_eq!(version, "5.0.6");
+///     // do something using a driver
 ///     Ok(())
 /// }
 /// ```
@@ -50,6 +41,22 @@ impl Image for Cassandra {
         TAG
     }
 
+    fn env_vars(
+        &self,
+    ) -> impl IntoIterator<Item = (impl Into<Cow<'_, str>>, impl Into<Cow<'_, str>>)> {
+        [
+            (
+                "JVM_EXTRA_OPTS",
+                "-Dcassandra.skip_wait_for_gossip_to_settle=0 -Dcassandra.initial_token=0",
+            ),
+            ("CASSANDRA_DC", "dc1"),
+            ("CASSANDRA_SNITCH", "GossipingPropertyFileSnitch"),
+            ("CASSANDRA_ENDPOINT_SNITCH", "GossipingPropertyFileSnitch"),
+            ("HEAP_NEWSIZE", "128M"),
+            ("MAX_HEAP_SIZE", "1024M"),
+        ]
+    }
+
     fn ready_conditions(&self) -> Vec<WaitFor> {
         vec![WaitFor::message_on_either_std("Startup complete")]
     }
@@ -59,29 +66,50 @@ impl Image for Cassandra {
 mod tests {
     use std::time::Duration;
 
-    use scylla::client::{session::Session, session_builder::SessionBuilder};
+    use cdrs_tokio::{
+        cluster::{
+            session::{SessionBuilder, TcpSessionBuilder},
+            NodeTcpConfigBuilder,
+        },
+        load_balancing::RoundRobinLoadBalancingStrategy,
+        types::ByName,
+    };
     use testcontainers::{runners::AsyncRunner, ImageExt};
 
     use super::*;
 
     #[tokio::test]
     async fn cassandra_select_version() -> Result<(), Box<dyn std::error::Error + 'static>> {
-        let image = Cassandra::default().with_startup_timeout(Duration::from_secs(240));
+        pretty_env_logger::init();
+        let image = Cassandra::default().with_startup_timeout(Duration::from_secs(120));
         let instance = image.start().await?;
         let host = instance.get_host().await?;
         let port = instance.get_host_port_ipv4(9042).await?;
         let hostname = format!("{host}:{port}");
-        let session: Session = SessionBuilder::new().known_node(hostname).build().await?;
 
-        let prepared_statement = session
-            .prepare("SELECT release_version FROM system.local")
+        let cluster_config = NodeTcpConfigBuilder::new()
+            .with_contact_point(hostname.into())
+            .build()
             .await?;
-        let rows = session
-            .execute_unpaged(&prepared_statement, &[])
-            .await?
-            .into_rows_result()?;
-        let (version,) = rows.single_row::<(String,)>()?;
-        assert_eq!(version, "5.0.6");
+
+        let session =
+            TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), cluster_config)
+                .build()
+                .await?;
+
+        let result = session
+            .query("SELECT release_version FROM system.local")
+            .await?;
+
+        let body = result.response_body()?;
+        let rows = body.into_rows().unwrap();
+        let version = rows
+            .first()
+            .unwrap()
+            .by_name::<String>("release_version")?
+            .unwrap();
+
+        assert_eq!(version, TAG);
         Ok(())
     }
 }