Some test functionality organization.

Author: j7nw4r

tests/producer.rs

@@ -0,0 +1,18 @@
+use crate::containers::KafkaContext;
+use crate::utils::logging::init_test_logger;
+
+mod containers;
+mod utils;
+
+#[tokio::test]
+pub async fn test_basic_produce() {
+    init_test_logger();
+
+    let kafka_context_result = KafkaContext::new().await;
+    let Ok(_kafka_context) = kafka_context_result else {
+        panic!(
+            "could not create kafka context: {}",
+            kafka_context_result.unwrap_err()
+        );
+    };
+}

tests/test_admin.rs

@@ -15,6 +15,8 @@ use rdkafka::metadata::Metadata;
 use rdkafka::producer::{FutureProducer, FutureRecord, Producer};
 use rdkafka::{ClientConfig, Offset, TopicPartitionList};
 
+use crate::utils::logging::init_test_logger;
+use crate::utils::rand::{rand_test_group, rand_test_topic};
 use crate::utils::*;
 
 mod utils;
@@ -26,7 +28,7 @@ fn create_config() -> ClientConfig {
 }
 
 fn create_admin_client() -> AdminClient<DefaultClientContext> {
-    configure_logging_for_tests();
+    init_test_logger();
     create_config()
         .create()
         .expect("admin client creation failed")

tests/test_high_consumers.rs

@@ -17,6 +17,7 @@ use rdkafka::util::current_time_millis;
 use rdkafka::{Message, Timestamp};
 use rdkafka_sys::types::RDKafkaConfRes;
 
+use crate::utils::rand::*;
 use crate::utils::*;
 
 mod utils;
@@ -46,7 +47,7 @@ where
 #[tokio::test]
 async fn test_invalid_max_poll_interval() {
     let res: Result<StreamConsumer, _> = consumer_config(
-        &rand_test_group(),
+        &crate::utils::rand::rand_test_group(),
         Some(hashmap! { "max.poll.interval.ms" => "-1" }),
     )
     .create();

tests/test_high_producers.rs

@@ -13,7 +13,7 @@ use rdkafka::producer::{FutureProducer, FutureRecord, Producer};
 use rdkafka::util::Timeout;
 use rdkafka::Timestamp;
 
-use crate::utils::*;
+use crate::utils::rand::*;
 
 mod utils;
 

tests/test_low_consumers.rs

@@ -13,6 +13,7 @@ use rdkafka::topic_partition_list::{Offset, TopicPartitionList};
 use rdkafka::util::{current_time_millis, Timeout};
 use rdkafka::{ClientConfig, Message, Timestamp};
 
+use crate::utils::rand::*;
 use crate::utils::*;
 
 mod utils;

tests/test_low_producers.rs

@@ -21,6 +21,8 @@ use rdkafka::types::RDKafkaRespErr;
 use rdkafka::util::current_time_millis;
 use rdkafka::{ClientContext, Statistics};
 
+use crate::utils::logging::init_test_logger;
+use crate::utils::rand::*;
 use crate::utils::*;
 
 mod utils;
@@ -162,7 +164,7 @@ fn base_producer_with_context<Part: Partitioner, C: ProducerContext<Part>>(
     context: C,
     config_overrides: HashMap<&str, &str>,
 ) -> BaseProducer<C, Part> {
-    configure_logging_for_tests();
+    init_test_logger();
     default_config(config_overrides)
         .create_with_context::<C, BaseProducer<_, Part>>(context)
         .unwrap()
@@ -183,7 +185,7 @@ where
     Part: Partitioner + Send + Sync + 'static,
     C: ProducerContext<Part>,
 {
-    configure_logging_for_tests();
+    init_test_logger();
     default_config(config_overrides)
         .create_with_context::<C, ThreadedProducer<_, _>>(context)
         .unwrap()

tests/test_metadata.rs

@@ -9,6 +9,7 @@ use rdkafka::topic_partition_list::TopicPartitionList;
 
 use rdkafka_sys::types::RDKafkaConfRes;
 
+use crate::utils::rand::*;
 use crate::utils::*;
 
 mod utils;

tests/test_transactions.rs

@@ -15,19 +15,21 @@ use rdkafka::producer::{BaseProducer, BaseRecord, Producer};
 use rdkafka::topic_partition_list::{Offset, TopicPartitionList};
 use rdkafka::util::Timeout;
 
+use crate::utils::logging::init_test_logger;
+use crate::utils::rand::*;
 use utils::*;
 
 mod utils;
 
 fn create_consumer(
     config_overrides: Option<HashMap<&str, &str>>,
 ) -> Result<BaseConsumer, KafkaError> {
-    configure_logging_for_tests();
+    init_test_logger();
     consumer_config(&rand_test_group(), config_overrides).create()
 }
 
 fn create_producer() -> Result<BaseProducer, KafkaError> {
-    configure_logging_for_tests();
+    init_test_logger();
     let mut config = ClientConfig::new();
     config
         .set("bootstrap.servers", get_bootstrap_server())

tests/utils/logging.rs

@@ -0,0 +1,9 @@
+use std::sync::Once;
+
+static INIT: Once = Once::new();
+
+pub fn init_test_logger() {
+    INIT.call_once(|| {
+        env_logger::try_init().expect("Failed to initialize env_logger");
+    });
+}

tests/utils.rs renamed to tests/utils/mod.rs

@@ -1,11 +1,12 @@
 #![allow(dead_code)]
 
+pub mod logging;
+pub mod rand;
+
 use std::collections::HashMap;
 use std::env::{self, VarError};
-use std::sync::Once;
 use std::time::Duration;
 
-use rand::distr::{Alphanumeric, SampleString};
 use regex::Regex;
 
 use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
@@ -18,21 +19,6 @@ use rdkafka::producer::{FutureProducer, FutureRecord};
 use rdkafka::statistics::Statistics;
 use rdkafka::TopicPartitionList;
 
-pub fn rand_test_topic(test_name: &str) -> String {
-    let id = Alphanumeric.sample_string(&mut rand::rng(), 10);
-    format!("__{}_{}", test_name, id)
-}
-
-pub fn rand_test_group() -> String {
-    let id = Alphanumeric.sample_string(&mut rand::rng(), 10);
-    format!("__test_{}", id)
-}
-
-pub fn rand_test_transactional_id() -> String {
-    let id = Alphanumeric.sample_string(&mut rand::rng(), 10);
-    format!("__test_{}", id)
-}
-
 pub fn get_bootstrap_server() -> String {
     env::var("KAFKA_HOST").unwrap_or_else(|_| "localhost:9092".to_owned())
 }
@@ -195,21 +181,13 @@ pub fn consumer_config(
     config
 }
 
-static INIT: Once = Once::new();
-
-pub fn configure_logging_for_tests() {
-    INIT.call_once(|| {
-        env_logger::try_init().expect("Failed to initialize env_logger");
-    });
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
 
     #[tokio::test]
     async fn test_populate_topic() {
-        let topic_name = rand_test_topic("test_populate_topic");
+        let topic_name = rand::rand_test_topic("test_populate_topic");
         let message_map = populate_topic(&topic_name, 100, &value_fn, &key_fn, Some(0), None).await;
 
         let total_messages = message_map