tests: Add sanity integration test

The test creates, indexes, searches and then drops a small hdfs
logs file.

Using `testcontainers-rs`, the test runs a postgres instance inside
a container.

Author: tontinton

Cargo.lock

@@ -34,3 +34,6 @@ uuid = { version = "1.8.0", features = ["v7"] }
 
 [dev-dependencies]
 async-tempfile = "0.5.0"
+ctor = "0.2.8"
+testcontainers = "0.17.0"
+testcontainers-modules = { version = "0.5.0", features = ["postgres"] }

Cargo.toml

@@ -6,7 +6,7 @@ pub mod number;
 pub mod static_object;
 pub mod text;
 
-use std::{ops::Deref, path::Path, vec::IntoIter};
+use std::{ops::Deref, path::Path, str::FromStr, vec::IntoIter};
 
 use color_eyre::eyre::Result;
 use serde::{Deserialize, Serialize};
@@ -232,7 +232,7 @@ pub struct IndexSchema {
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
-pub(crate) struct IndexConfig {
+pub struct IndexConfig {
     pub name: String,
     pub path: String,
 
@@ -246,6 +246,14 @@ pub(crate) struct IndexConfig {
 impl IndexConfig {
     pub async fn from_path<P: AsRef<Path>>(path: P) -> Result<Self> {
         let config_str = read_to_string(path).await?;
-        Ok(serde_yaml::from_str(&config_str)?)
+        Self::from_str(&config_str)
+    }
+}
+
+impl FromStr for IndexConfig {
+    type Err = color_eyre::Report;
+
+    fn from_str(s: &str) -> Result<Self> {
+        Ok(serde_yaml::from_str(s)?)
     }
 }

src/config/mod.rs

@@ -0,0 +1,55 @@
+use std::{
+    fs::canonicalize,
+    path::{Path, PathBuf},
+};
+
+use color_eyre::Result;
+use sqlx::{migrate::Migrator, postgres::PgPoolOptions, PgPool};
+use testcontainers::{runners::AsyncRunner, ContainerAsync};
+use testcontainers_modules::postgres::Postgres as PostgresContainer;
+
+static MIGRATOR: Migrator = sqlx::migrate!();
+
+const MAX_DB_CONNECTIONS: u32 = 100;
+
+pub struct Postgres {
+    /// Keep container alive (container is deleted on drop).
+    _container: ContainerAsync<PostgresContainer>,
+
+    /// The underlying sqlx connection to the postgres inside the container.
+    pub pool: PgPool,
+}
+
+async fn open_db_pool(url: &str) -> Result<PgPool> {
+    Ok(PgPoolOptions::new()
+        .max_connections(MAX_DB_CONNECTIONS)
+        .connect(url)
+        .await?)
+}
+
+pub async fn run_postgres() -> Result<Postgres> {
+    let container = PostgresContainer::default().start().await?;
+    let pool = open_db_pool(&format!(
+        "postgres://postgres:postgres@127.0.0.1:{}/postgres",
+        container.get_host_port_ipv4(5432).await?
+    ))
+    .await?;
+
+    MIGRATOR.run(&pool).await?;
+
+    Ok(Postgres {
+        _container: container,
+        pool,
+    })
+}
+
+pub fn get_test_file_path(test_file: &str) -> PathBuf {
+    canonicalize(&Path::new(file!()))
+        .unwrap()
+        .parent()
+        .unwrap()
+        .parent()
+        .unwrap()
+        .join("test_files")
+        .join(test_file)
+}

tests/common/mod.rs

@@ -0,0 +1,71 @@
+mod common;
+
+use std::str::FromStr;
+
+use clap::Parser;
+use color_eyre::Result;
+use ctor::ctor;
+use pretty_env_logger::formatted_timed_builder;
+use tokio::sync::mpsc;
+use toshokan::{
+    args::{DropArgs, IndexArgs, SearchArgs},
+    commands::{
+        create::run_create_from_config, drop::run_drop, index::run_index,
+        search::run_search_with_callback,
+    },
+    config::IndexConfig,
+};
+
+use crate::common::{get_test_file_path, run_postgres};
+
+#[ctor]
+fn init() {
+    color_eyre::install().unwrap();
+
+    let mut log_builder = formatted_timed_builder();
+    log_builder.parse_filters("toshokan=trace,opendal::services=info");
+    log_builder.try_init().unwrap();
+}
+
+#[tokio::test]
+async fn test_example_config() -> Result<()> {
+    let postgres = run_postgres().await?;
+    let config = IndexConfig::from_str(include_str!("../example_config.yaml"))?;
+
+    run_create_from_config(&config, &postgres.pool).await?;
+
+    run_index(
+        IndexArgs::parse_from([
+            "",
+            &config.name,
+            &get_test_file_path("hdfs-logs-multitenants-2.json").to_string_lossy(),
+        ]),
+        &postgres.pool,
+    )
+    .await?;
+
+    let (tx, mut rx) = mpsc::channel(1);
+    run_search_with_callback(
+        SearchArgs::parse_from([
+            "",
+            &config.name,
+            "tenant_id:>50 AND severity_text:INFO",
+            "--limit",
+            "1",
+        ]),
+        &postgres.pool,
+        Box::new(move |doc| {
+            tx.try_send(doc).unwrap();
+        }),
+    )
+    .await?;
+
+    assert_eq!(
+        rx.recv().await.unwrap(),
+        r#"{"attributes":{"class":"org.apache.hadoop.hdfs.server.datanode.DataNode"},"body":"PacketResponder: BP-108841162-10.10.34.11-1440074360971:blk_1074072698_331874, type=HAS_DOWNSTREAM_IN_PIPELINE terminating","resource":{"service":"datanode/01"},"severity_text":"INFO","tenant_id":58,"timestamp":"2016-04-13T06:46:53Z"}"#
+    );
+
+    run_drop(DropArgs::parse_from(["", &config.name]), &postgres.pool).await?;
+
+    Ok(())
+}

tests/config.rs

@@ -0,0 +1,2 @@
+{"timestamp":1460530013,"severity_text":"INFO","body":"PacketResponder: BP-108841162-10.10.34.11-1440074360971:blk_1074072698_331874, type=HAS_DOWNSTREAM_IN_PIPELINE terminating","resource":{"service":"datanode/01"},"attributes":{"class":"org.apache.hadoop.hdfs.server.datanode.DataNode"},"tenant_id":58}
+{"timestamp":1460530014,"severity_text":"INFO","body":"Receiving BP-108841162-10.10.34.11-1440074360971:blk_1074072706_331882 src: /10.10.34.33:42666 dest: /10.10.34.11:50010","resource":{"service":"datanode/01"},"attributes":{"class":"org.apache.hadoop.hdfs.server.datanode.DataNode"},"tenant_id":46}