tests: init utils

Author: alissa-tung

src/hstreamdb/src/producer.rs

@@ -71,6 +71,14 @@ impl FlushSettings {
     pub fn builder() -> FlushSettingsBuilder {
         default()
     }
+
+    pub fn sync() -> Self {
+        Self {
+            len: 0,
+            size: 0,
+            deadline: None,
+        }
+    }
 }
 
 #[derive(Default)]

src/hstreamdb/tests/common.rs

@@ -0,0 +1,108 @@
+use std::env;
+
+use hstreamdb::appender::Appender;
+use hstreamdb::common::{CompressionType, SpecialOffset, Stream};
+use hstreamdb::producer::{FlushSettings, Producer};
+use hstreamdb::{ChannelProviderSettings, Record, Subscription};
+use hstreamdb_test_utils::rand_alphanumeric;
+
+pub async fn init_client() -> anyhow::Result<Client> {
+    let server_url = env::var("TEST_SERVER_ADDR").unwrap();
+    let channel_provider_settings = ChannelProviderSettings::default();
+    let client = hstreamdb::Client::new(server_url, channel_provider_settings).await?;
+    Ok(Client(client))
+}
+
+pub struct Client(hstreamdb::Client);
+
+impl Client {
+    pub async fn new_stream(&self) -> anyhow::Result<Stream> {
+        let stream_name = rand_alphanumeric(20);
+        let stream = self
+            .0
+            .create_stream(Stream {
+                stream_name,
+                replication_factor: 3,
+                backlog_duration: 60 * 60 * 24,
+                shard_count: 20,
+                creation_time: None,
+            })
+            .await?;
+        Ok(stream)
+    }
+
+    pub async fn new_subscription<T: Into<String>>(
+        &self,
+        stream_name: T,
+    ) -> anyhow::Result<Subscription> {
+        let subscription = self
+            .0
+            .create_subscription(Subscription {
+                subscription_id: rand_alphanumeric(20),
+                stream_name: stream_name.into(),
+                ack_timeout_seconds: 60 * 15,
+                max_unacked_records: 1000,
+                offset: SpecialOffset::Earliest,
+                creation_time: None,
+            })
+            .await?;
+        Ok(subscription)
+    }
+
+    pub async fn new_sync_producer<T: Into<String>>(
+        &self,
+        stream_name: T,
+    ) -> anyhow::Result<(Appender, Producer)> {
+        let producer = self
+            .0
+            .new_producer(
+                stream_name.into(),
+                CompressionType::None,
+                None,
+                FlushSettings::sync(),
+                ChannelProviderSettings::default(),
+                None,
+            )
+            .await?;
+        Ok(producer)
+    }
+
+    pub async fn write_rand<T: Into<String>>(
+        &self,
+        stream_name: T,
+        appender_num: usize,
+        record_num: usize,
+        payload_size: usize,
+    ) -> anyhow::Result<()> {
+        let (appender, producer) = self.new_sync_producer(stream_name).await?;
+
+        for _ in 0..appender_num {
+            let appender = appender.clone();
+            tokio::spawn(async move {
+                for _ in 0..record_num {
+                    let payload = rand_alphanumeric(payload_size);
+                    match appender
+                        .append(Record {
+                            partition_key: "".to_string(),
+                            payload: hstreamdb::Payload::RawRecord(payload.into_bytes()),
+                        })
+                        .await
+                    {
+                        Ok(_) => (),
+                        Err(err) => log::error!("{}", err),
+                    };
+                }
+            });
+        }
+        drop(appender);
+        producer.start().await;
+        Ok(())
+    }
+
+    pub async fn new_stream_subscription(&self) -> anyhow::Result<(Stream, Subscription)> {
+        let stream = self.new_stream().await?;
+        let stream_name = stream.stream_name.clone();
+        let subscription = self.new_subscription(stream_name).await?;
+        Ok((stream, subscription))
+    }
+}

src/hstreamdb/tests/integration_test.rs

@@ -0,0 +1,23 @@
+mod common;
+
+use common::init_client;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn utils_base_test() {
+    let client = init_client().await.unwrap();
+
+    let stream = client.new_stream().await.unwrap();
+    let stream_name = &stream.stream_name;
+    let _subscription = client.new_subscription(stream_name).await.unwrap();
+    let _producer = client.new_sync_producer(stream_name).await.unwrap();
+
+    let appender_num = 5;
+    let record_num = 50;
+    let payload_size = 2000;
+    client
+        .write_rand(stream_name, appender_num, record_num, payload_size)
+        .await
+        .unwrap();
+
+    client.new_stream_subscription().await.unwrap();
+}

src/hstreamdb/tests/tls_test.rs

@@ -1,32 +1,39 @@
-// use std::{env, fs};
+use std::env;
 
-// use hstreamdb::tls::{Certificate, ClientTlsConfig, Identity};
-// use hstreamdb::{ChannelProviderSettings, Client};
+use hstreamdb::tls::{Certificate, ClientTlsConfig, Identity};
+use hstreamdb::{ChannelProviderSettings, Client};
 
-// async fn _test_tls_impl() {
-//     env::set_var("RUST_LOG", "DEBUG");
-//     env_logger::init();
+#[tokio::test(flavor = "multi_thread")]
+async fn test_tls() {
+    if let Ok(_) = env::var("ENDPOINT") {
+        test_tls_impl().await
+    } else {
+        log::warn!("cloud endpoint is not presented");
+        log::warn!("ignore tls tests");
+    }
+}
 
-//     let server_url: &str = todo!();
-//     let tls_dir: &str = todo!();
+async fn test_tls_impl() {
+    env::set_var("RUST_LOG", "DEBUG");
+    env_logger::init();
 
-//     let ca_certificate =
-// Certificate::from_pem(fs::read(format!("{tls_dir}/root_ca.crt")).unwrap());
-//     let cert = fs::read(format!("{tls_dir}/client.crt")).unwrap();
-//     let key = fs::read(format!("{tls_dir}/client.key")).unwrap();
+    let server_url: String = env::var("ENDPOINT").unwrap();
+    let ca_certificate: String = env::var("ROOT_CA").unwrap();
+    let cert = env::var("CLIENT_CRT").unwrap();
+    let key = env::var("CLIENT_KEY").unwrap();
 
-//     let client = Client::new(
-//         server_url,
-//         ChannelProviderSettings::builder()
-//             .set_tls_config(
-//                 ClientTlsConfig::new()
-//                     .ca_certificate(ca_certificate)
-//                     .identity(Identity::from_pem(cert, key)),
-//             )
-//             .build(),
-//     )
-//     .await
-//     .unwrap();
+    let client = Client::new(
+        server_url,
+        ChannelProviderSettings::builder()
+            .set_tls_config(
+                ClientTlsConfig::new()
+                    .ca_certificate(Certificate::from_pem(ca_certificate))
+                    .identity(Identity::from_pem(cert, key)),
+            )
+            .build(),
+    )
+    .await
+    .unwrap();
 
-//     log::info!("{:?}", client.list_streams().await.unwrap());
-// }
+    log::info!("{:?}", client.list_streams().await.unwrap());
+}