cassandra_int_tests: Refactor CassandraConnection execute methods

shotover-proxy/tests/cassandra_int_tests/mod.rs

@@ -6,8 +6,6 @@ use crate::helpers::cassandra::{
 };
 use crate::helpers::ShotoverManager;
 #[cfg(feature = "cassandra-cpp-driver-tests")]
-use cassandra_cpp::{Error, ErrorKind};
-#[cfg(feature = "cassandra-cpp-driver-tests")]
 use cassandra_protocol::frame::message_error::{ErrorBody, ErrorType};
 use cdrs_tokio::frame::events::{
  SchemaChange, SchemaChangeOptions, SchemaChangeTarget, SchemaChangeType, ServerEvent,
@@ -472,7 +470,8 @@ async fn peers_rewrite_v4(#[case] driver: CassandraDriver) {
 
 #[cfg(feature = "cassandra-cpp-driver-tests")]
 #[rstest]
-//#[case::cdrs(CdrsTokio)] // TODO
+#[case::cdrs(CdrsTokio)]
+#[case::scylla(Scylla)]
 #[cfg_attr(feature = "cassandra-cpp-driver-tests", case::datastax(Datastax))]
 #[tokio::test(flavor = "multi_thread")]
 #[serial]
@@ -491,20 +490,21 @@ async fn peers_rewrite_v3(#[case] driver: CassandraDriver) {
 
  // Assert that the error cassandra gives because system.peers_v2 does not exist on cassandra v3
  // is passed through shotover unchanged.
- let statement = "SELECT data_center, native_port, rack FROM system.peers_v2;";
- let result = connection.execute_expect_err(statement).await;
  assert_eq!(
- result,
- ErrorBody {
+ connection
+ .execute_fallible("SELECT data_center, native_port, rack FROM system.peers_v2;")
+ .await,
+ Err(ErrorBody {
  ty: ErrorType::Invalid,
  message: "unconfigured table peers_v2".into()
- }
+ })
  );
 }
 
 #[cfg(feature = "cassandra-cpp-driver-tests")]
 #[rstest]
-//#[case::cdrs(CdrsTokio)] // TODO
+#[case::cdrs(CdrsTokio)]
+#[case::scylla(Scylla)]
 #[cfg_attr(feature = "cassandra-cpp-driver-tests", case::datastax(Datastax))]
 #[tokio::test(flavor = "multi_thread")]
 #[serial]
@@ -545,10 +545,10 @@ async fn request_throttling(#[case] driver: CassandraDriver) {
  let mut results = join_all(future_list).await;
  results.retain(|result| match result {
  Ok(_) => true,
- Err(Error(
- ErrorKind::CassErrorResult(cassandra_cpp::CassErrorCode::SERVER_OVERLOADED, ..),
- _,
- )) => false,
+ Err(ErrorBody {
+ ty: ErrorType::Overloaded,
+ ..
+ }) => false,
  Err(e) => panic!(
  "wrong error returned, got {:?}, expected SERVER_OVERLOADED",
  e
@@ -584,7 +584,10 @@ async fn request_throttling(#[case] driver: CassandraDriver) {
  for i in 0..60 {
  queries.push(format!("INSERT INTO test_keyspace.my_table (id, lastname, firstname) VALUES ({}, 'text', 'text')", i));
  }
- let result = connection.execute_batch_expect_err(queries).await;
+ let result = connection
+ .execute_batch_fallible(queries)
+ .await
+ .unwrap_err();
  assert_eq!(
  result,
  ErrorBody {

shotover-proxy/tests/cassandra_int_tests/prepared_statements_all.rs

@@ -55,15 +55,15 @@ async fn insert(connection: &CassandraConnection) {
  ]
  )
  .await,
- Vec::<Vec<_>>::new()
+ Ok(Vec::<Vec<_>>::new())
  );
  } else {
  let prepared = connection
  .prepare("INSERT INTO test_prepare_statements_all.test (id, v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);")
  .await;
  assert_eq!(
  connection.execute_prepared(&prepared, &values()).await,
- Vec::<Vec<_>>::new()
+ Ok(Vec::<Vec<_>>::new())
  );
  }
 }

shotover-proxy/tests/cassandra_int_tests/prepared_statements_simple.rs

@@ -13,7 +13,7 @@ async fn delete(session: &CassandraConnection) {
  session
  .execute_prepared(&prepared, &[ResultValue::Int(1)])
  .await,
- Vec::<Vec<ResultValue>>::new()
+ Ok(Vec::<Vec<ResultValue>>::new())
  );
 
  assert_query_result(
@@ -33,21 +33,21 @@ async fn insert(session: &CassandraConnection) {
  session
  .execute_prepared(&prepared, &[ResultValue::Int(1)])
  .await,
- Vec::<Vec<ResultValue>>::new()
+ Ok(Vec::<Vec<ResultValue>>::new())
  );
 
  assert_eq!(
  session
  .execute_prepared(&prepared, &[ResultValue::Int(2)])
  .await,
- Vec::<Vec<ResultValue>>::new()
+ Ok(Vec::<Vec<ResultValue>>::new())
  );
 
  assert_eq!(
  session
  .execute_prepared(&prepared, &[ResultValue::Int(3)])
  .await,
- Vec::<Vec<ResultValue>>::new()
+ Ok(Vec::<Vec<ResultValue>>::new())
  );
 }
 
@@ -58,7 +58,8 @@ async fn select(session: &CassandraConnection) {
 
  let result_rows = session
  .execute_prepared(&prepared, &[ResultValue::Int(1)])
- .await;
+ .await
+ .unwrap();
 
  assert_rows(result_rows, &[&[ResultValue::Int(1)]]);
 }
@@ -80,13 +81,15 @@ async fn select_cross_connection<Fut>(
  assert_rows(
  connection_before
  .execute_prepared(&prepared, &[ResultValue::Int(1)])
- .await,
+ .await
+ .unwrap(),
  &[&[ResultValue::Int(1), ResultValue::Int(1)]],
  );
  assert_rows(
  connection_after
  .execute_prepared(&prepared, &[ResultValue::Int(1)])
- .await,
+ .await
+ .unwrap(),
  &[&[ResultValue::Int(1), ResultValue::Int(1)]],
  );
 }
@@ -106,7 +109,7 @@ async fn use_statement(session: &CassandraConnection) {
  session
  .execute_prepared(&prepared, &[ResultValue::Int(358)])
  .await,
- Vec::<Vec<ResultValue>>::new()
+ Ok(Vec::<Vec<ResultValue>>::new())
  );
 
  // observe that the query succeeded despite the keyspace being incorrect at the time.
@@ -139,6 +142,6 @@ where
  if session.is(&[CassandraDriver::Scylla, CassandraDriver::CdrsTokio]) {
  let cql = "SELECT * FROM system.local WHERE key = 'local'";
  let prepared = session.prepare(cql).await;
- session.execute_prepared(&prepared, &[]).await;
+ session.execute_prepared(&prepared, &[]).await.unwrap();
  }
 }