You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/02/03 09:23:52 UTC
(iggy) 01/01: refactor(integration): migrate connectors tests to iggy_harness proc macro
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch connect-harness
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 148f7dc977294d5a97636bc1c60cd7e00e4f47c3
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Jan 29 16:10:26 2026 +0100
refactor(integration): migrate connectors tests to iggy_harness proc macro
Eliminates boilerplate in connector tests by leveraging the same
proc macro infrastructure used for MCP tests. Connector-specific
fixtures (postgres, wiremock, random) moved to tests/ since
they're not part of core harness.
MCP and connectors runtime are now owned by ServerHandle rather
than TestHarness, enabling per-node dependent binaries in cluster
setups.
Also switches test output capture from env_logger to
tracing-subscriber for consistency with server logging.
---
Cargo.lock | 34 +-
DEPENDENCIES.md | 2 -
core/harness_derive/src/codegen.rs | 14 +-
core/integration/Cargo.toml | 5 +-
core/integration/src/harness/context.rs | 30 +-
.../src/harness/{fixtures/mod.rs => fixture.rs} | 13 -
core/integration/src/harness/fixtures/random.rs | 36 --
.../src/harness/handle/connectors_runtime.rs | 28 +-
core/integration/src/harness/handle/mcp.rs | 20 +-
core/integration/src/harness/handle/mod.rs | 2 +-
core/integration/src/harness/handle/server.rs | 101 ++++-
core/integration/src/harness/mod.rs | 4 +-
.../src/harness/orchestrator/builder.rs | 24 +-
.../src/harness/orchestrator/harness.rs | 79 ++--
core/integration/tests/connectors/api/endpoints.rs | 107 +++--
core/integration/tests/connectors/api/mod.rs | 9 -
.../postgres => tests/connectors/fixtures}/mod.rs | 17 +-
.../connectors}/fixtures/postgres/container.rs | 25 +-
.../connectors}/fixtures/postgres/mod.rs | 6 +-
.../connectors}/fixtures/postgres/sink.rs | 3 +-
.../connectors}/fixtures/postgres/source.rs | 110 +----
.../connectors/fixtures/quickwit/container.rs | 359 +++++++++++++++
.../connectors/fixtures/quickwit}/mod.rs | 13 +-
.../connectors}/fixtures/wiremock.rs | 15 +-
.../http_config_provider/direct_responses.rs | 151 ++++---
.../tests/connectors/http_config_provider/mod.rs | 82 ----
.../http_config_provider/wrapped_responses.rs | 153 ++++---
core/integration/tests/connectors/mod.rs | 190 +-------
core/integration/tests/connectors/postgres/mod.rs | 502 +--------------------
.../tests/connectors/postgres/postgres_sink.rs | 136 ++++--
.../tests/connectors/postgres/postgres_source.rs | 314 ++++++++++---
core/integration/tests/connectors/quickwit/mod.rs | 265 -----------
.../tests/connectors/quickwit/quickwit_sink.rs | 283 ++++++++----
core/integration/tests/connectors/random/mod.rs | 9 -
.../tests/connectors/random/random_source.rs | 32 +-
core/integration/tests/mod.rs | 73 ++-
36 files changed, 1468 insertions(+), 1778 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 42b1ace77..fab20e3e8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2945,29 +2945,6 @@ dependencies = [
"syn 2.0.114",
]
-[[package]]
-name = "env_filter"
-version = "0.1.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1bf3c259d255ca70051b30e2e95b5446cdb8949ac4cd22c0d7fd634d89f568e2"
-dependencies = [
- "log",
- "regex",
-]
-
-[[package]]
-name = "env_logger"
-version = "0.11.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "13c863f0904021b108aa8b2f55046443e6b1ebde8fd4a15c399893aae4fa069f"
-dependencies = [
- "anstream",
- "anstyle",
- "env_filter",
- "jiff",
- "log",
-]
-
[[package]]
name = "equivalent"
version = "1.0.2"
@@ -4210,7 +4187,7 @@ dependencies = [
"libc",
"percent-encoding",
"pin-project-lite",
- "socket2 0.5.10",
+ "socket2 0.6.2",
"tokio",
"tower-service",
"tracing",
@@ -5005,13 +4982,11 @@ dependencies = [
"async-trait",
"bon",
"bytes",
- "chrono",
"compio",
"configs",
"configs_derive",
"ctor",
"derive_more",
- "env_logger",
"figment",
"futures",
"harness_derive",
@@ -5022,7 +4997,6 @@ dependencies = [
"keyring",
"lazy_static",
"libc",
- "log",
"once_cell",
"predicates",
"rand 0.9.2",
@@ -5038,12 +5012,12 @@ dependencies = [
"socket2 0.6.2",
"sqlx",
"strum",
- "strum_macros",
"tempfile",
"test-case",
"testcontainers-modules",
"tokio",
"tracing",
+ "tracing-subscriber",
"twox-hash",
"uuid",
"zip",
@@ -7128,7 +7102,7 @@ dependencies = [
"quinn-udp",
"rustc-hash",
"rustls",
- "socket2 0.5.10",
+ "socket2 0.6.2",
"thiserror 2.0.18",
"tokio",
"tracing",
@@ -7167,7 +7141,7 @@ dependencies = [
"cfg_aliases",
"libc",
"once_cell",
- "socket2 0.5.10",
+ "socket2 0.6.2",
"tracing",
"windows-sys 0.60.2",
]
diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md
index 63162f765..3a52030ec 100644
--- a/DEPENDENCIES.md
+++ b/DEPENDENCIES.md
@@ -255,8 +255,6 @@ embedded-io: 0.6.1, "Apache-2.0 OR MIT",
encode_unicode: 1.0.0, "Apache-2.0 OR MIT",
encoding_rs: 0.8.35, "(Apache-2.0 OR MIT) AND BSD-3-Clause",
enum_dispatch: 0.3.13, "Apache-2.0 OR MIT",
-env_filter: 0.1.4, "Apache-2.0 OR MIT",
-env_logger: 0.11.8, "Apache-2.0 OR MIT",
equivalent: 1.0.2, "Apache-2.0 OR MIT",
err_trail: 0.11.0, "Apache-2.0",
errno: 0.3.14, "Apache-2.0 OR MIT",
diff --git a/core/harness_derive/src/codegen.rs b/core/harness_derive/src/codegen.rs
index 2d445c942..eb06e655b 100644
--- a/core/harness_derive/src/codegen.rs
+++ b/core/harness_derive/src/codegen.rs
@@ -476,7 +476,7 @@ fn generate_harness_setup(
#config_resolution
let mut __harness = ::integration::harness::TestHarness::builder()
.server(#server_config)
- .primary_client(#client_config)
+ .client(#client_config)
#mcp_builder_call
#connectors_runtime_builder_call
#cluster_builder_call
@@ -526,6 +526,12 @@ fn generate_start_and_seed(attrs: &IggyTestAttrs, fixture_seed: TokenStream) ->
}
}
+fn fixture_var_ident(name: &syn::Ident) -> syn::Ident {
+ let name_str = name.to_string();
+ let clean_name = name_str.trim_start_matches('_');
+ format_ident!("__fixture_{}", clean_name)
+}
+
/// Generate fixture setup calls (before harness setup).
fn generate_fixture_setup(params: &[DetectedParam]) -> TokenStream {
let fixtures = fixture_params(params);
@@ -537,7 +543,7 @@ fn generate_fixture_setup(params: &[DetectedParam]) -> TokenStream {
.iter()
.filter_map(|p| {
if let DetectedParam::Fixture { name, ty } = p {
- let var_name = format_ident!("__fixture_{}", name);
+ let var_name = fixture_var_ident(name);
Some(quote! {
let #var_name = <#ty as ::integration::harness::TestFixture>::setup()
.await
@@ -563,7 +569,7 @@ fn generate_fixture_envs_collection(params: &[DetectedParam]) -> TokenStream {
.iter()
.filter_map(|p| {
if let DetectedParam::Fixture { name, .. } = p {
- let var_name = format_ident!("__fixture_{}", name);
+ let var_name = fixture_var_ident(name);
Some(quote! {
__fixture_envs.extend(
::integration::harness::TestFixture::connectors_runtime_envs(&#var_name)
@@ -622,7 +628,7 @@ fn generate_param_bindings(params: &[DetectedParam]) -> TokenStream {
});
}
DetectedParam::Fixture { name, .. } => {
- let fixture_var = format_ident!("__fixture_{}", name);
+ let fixture_var = fixture_var_ident(name);
bindings.push(quote! {
let #name = #fixture_var;
});
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index 4805192ce..42142e151 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -33,13 +33,11 @@ assert_cmd = { workspace = true }
async-trait = { workspace = true }
bon = { workspace = true }
bytes = { workspace = true }
-chrono = { workspace = true }
compio = { workspace = true }
configs = { workspace = true }
configs_derive = { workspace = true }
ctor = { workspace = true }
derive_more = { workspace = true }
-env_logger = { workspace = true }
figment = { workspace = true }
futures = { workspace = true }
harness_derive = { workspace = true }
@@ -50,7 +48,6 @@ iggy_common = { workspace = true }
keyring = { workspace = true }
lazy_static = { workspace = true }
libc = { workspace = true }
-log = { workspace = true }
once_cell = { workspace = true }
predicates = { workspace = true }
rand = { workspace = true }
@@ -71,12 +68,12 @@ server = { workspace = true }
socket2 = { workspace = true }
sqlx = { workspace = true }
strum = { workspace = true }
-strum_macros = { workspace = true }
tempfile = { workspace = true }
test-case = { workspace = true }
testcontainers-modules = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
twox-hash = { workspace = true }
uuid = { workspace = true }
zip = { workspace = true }
diff --git a/core/integration/src/harness/context.rs b/core/integration/src/harness/context.rs
index 480cfa92e..76fc77eca 100644
--- a/core/integration/src/harness/context.rs
+++ b/core/integration/src/harness/context.rs
@@ -115,24 +115,27 @@ impl TestContext {
&self.base_dir
}
- pub fn mcp_stdout_path(&self) -> PathBuf {
- self.base_dir.join("mcp_stdout.log")
+ pub fn mcp_stdout_path(&self, server_id: u32) -> PathBuf {
+ self.base_dir.join(format!("mcp_{server_id}_stdout.log"))
}
- pub fn mcp_stderr_path(&self) -> PathBuf {
- self.base_dir.join("mcp_stderr.log")
+ pub fn mcp_stderr_path(&self, server_id: u32) -> PathBuf {
+ self.base_dir.join(format!("mcp_{server_id}_stderr.log"))
}
- pub fn connectors_runtime_state_path(&self) -> PathBuf {
- self.base_dir.join("connectors_state")
+ pub fn connectors_runtime_state_path(&self, server_id: u32) -> PathBuf {
+ self.base_dir
+ .join(format!("connectors_runtime_{server_id}_state"))
}
- pub fn connectors_runtime_stdout_path(&self) -> PathBuf {
- self.base_dir.join("connectors_stdout.log")
+ pub fn connectors_runtime_stdout_path(&self, server_id: u32) -> PathBuf {
+ self.base_dir
+ .join(format!("connectors_runtime_{server_id}_stdout.log"))
}
- pub fn connectors_runtime_stderr_path(&self) -> PathBuf {
- self.base_dir.join("connectors_stderr.log")
+ pub fn connectors_runtime_stderr_path(&self, server_id: u32) -> PathBuf {
+ self.base_dir
+ .join(format!("connectors_runtime_{server_id}_stderr.log"))
}
pub fn test_stdout_path(&self) -> PathBuf {
@@ -185,10 +188,11 @@ mod tests {
#[test]
fn test_context_paths() {
let ctx = TestContext::new(Some("test_context_paths".to_string()), true).unwrap();
- assert!(ctx.mcp_stdout_path().ends_with("mcp_stdout.log"));
+ assert!(ctx.mcp_stdout_path(0).ends_with("mcp_0_stdout.log"));
assert!(
- ctx.connectors_runtime_state_path()
- .ends_with("connectors_state")
+ ctx.connectors_runtime_state_path(0)
+ .ends_with("connectors_runtime_0_state")
);
+ assert!(ctx.mcp_stdout_path(1).ends_with("mcp_1_stdout.log"));
}
}
diff --git a/core/integration/src/harness/fixtures/mod.rs b/core/integration/src/harness/fixture.rs
similarity index 79%
rename from core/integration/src/harness/fixtures/mod.rs
rename to core/integration/src/harness/fixture.rs
index 5f46025c1..2e09f843d 100644
--- a/core/integration/src/harness/fixtures/mod.rs
+++ b/core/integration/src/harness/fixture.rs
@@ -17,19 +17,6 @@
* under the License.
*/
-mod postgres;
-mod random;
-mod wiremock;
-
-pub use postgres::{
- PostgresContainer, PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture,
- PostgresSourceByteaFixture, PostgresSourceDeleteFixture, PostgresSourceFixture,
- PostgresSourceJsonFixture, PostgresSourceJsonbFixture, PostgresSourceMarkFixture,
- SinkPayloadFormat, SinkSchema,
-};
-pub use random::RandomSourceFixture;
-pub use wiremock::{WireMockDirectFixture, WireMockWrappedFixture};
-
use crate::harness::error::TestBinaryError;
use crate::harness::seeds::SeedError;
use async_trait::async_trait;
diff --git a/core/integration/src/harness/fixtures/random.rs b/core/integration/src/harness/fixtures/random.rs
deleted file mode 100644
index 3ecc73909..000000000
--- a/core/integration/src/harness/fixtures/random.rs
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use super::TestFixture;
-use crate::harness::error::TestBinaryError;
-use async_trait::async_trait;
-use std::collections::HashMap;
-
-pub struct RandomSourceFixture;
-
-#[async_trait]
-impl TestFixture for RandomSourceFixture {
- async fn setup() -> Result<Self, TestBinaryError> {
- Ok(Self)
- }
-
- fn connectors_runtime_envs(&self) -> HashMap<String, String> {
- HashMap::new()
- }
-}
diff --git a/core/integration/src/harness/handle/connectors_runtime.rs b/core/integration/src/harness/handle/connectors_runtime.rs
index 941353886..34190bb15 100644
--- a/core/integration/src/harness/handle/connectors_runtime.rs
+++ b/core/integration/src/harness/handle/connectors_runtime.rs
@@ -37,6 +37,7 @@ use tokio::time::sleep;
const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
pub struct ConnectorsRuntimeHandle {
+ server_id: u32,
config: ConnectorsRuntimeConfig,
context: Arc<TestContext>,
envs: HashMap<String, String>,
@@ -72,7 +73,7 @@ impl ConnectorsRuntimeHandle {
}
pub fn state_path(&self) -> PathBuf {
- self.context.connectors_runtime_state_path()
+ self.context.connectors_runtime_state_path(self.server_id)
}
pub fn collect_logs(&self) -> (String, String) {
@@ -80,7 +81,7 @@ impl ConnectorsRuntimeHandle {
}
fn build_envs(&mut self) {
- let state_path = self.context.connectors_runtime_state_path();
+ let state_path = self.context.connectors_runtime_state_path(self.server_id);
self.envs.insert(
"IGGY_CONNECTORS_STATE_PATH".to_string(),
state_path.display().to_string(),
@@ -108,15 +109,18 @@ impl ConnectorsRuntimeHandle {
}
}
-impl TestBinary for ConnectorsRuntimeHandle {
- type Config = ConnectorsRuntimeConfig;
-
- fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+impl ConnectorsRuntimeHandle {
+ pub fn with_server_id(
+ config: ConnectorsRuntimeConfig,
+ context: Arc<TestContext>,
+ server_id: u32,
+ ) -> Self {
let reserver =
SinglePortReserver::new().expect("Failed to reserve port for connectors runtime");
let server_address = reserver.address();
Self {
+ server_id,
config,
context,
envs: HashMap::new(),
@@ -128,6 +132,14 @@ impl TestBinary for ConnectorsRuntimeHandle {
port_reserver: Some(reserver),
}
}
+}
+
+impl TestBinary for ConnectorsRuntimeHandle {
+ type Config = ConnectorsRuntimeConfig;
+
+ fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+ Self::with_server_id(config, context, 0)
+ }
fn start(&mut self) -> Result<(), TestBinaryError> {
self.build_envs();
@@ -151,8 +163,8 @@ impl TestBinary for ConnectorsRuntimeHandle {
command.stdout(Stdio::inherit());
command.stderr(Stdio::inherit());
} else {
- let stdout_path = self.context.connectors_runtime_stdout_path();
- let stderr_path = self.context.connectors_runtime_stderr_path();
+ let stdout_path = self.context.connectors_runtime_stdout_path(self.server_id);
+ let stderr_path = self.context.connectors_runtime_stderr_path(self.server_id);
let stdout_file =
File::create(&stdout_path).map_err(|e| TestBinaryError::FileSystemError {
diff --git a/core/integration/src/harness/handle/mcp.rs b/core/integration/src/harness/handle/mcp.rs
index 145d39ecd..2ed3c724a 100644
--- a/core/integration/src/harness/handle/mcp.rs
+++ b/core/integration/src/harness/handle/mcp.rs
@@ -45,6 +45,7 @@ const MCP_HEALTH_CHECK_RETRIES: u32 = common::DEFAULT_HEALTH_CHECK_RETRIES * 3;
pub type McpClient = RunningService<RoleClient, InitializeRequestParams>;
pub struct McpHandle {
+ server_id: u32,
config: McpConfig,
context: Arc<TestContext>,
envs: HashMap<String, String>,
@@ -135,14 +136,13 @@ impl McpHandle {
}
}
-impl TestBinary for McpHandle {
- type Config = McpConfig;
-
- fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+impl McpHandle {
+ pub fn with_server_id(config: McpConfig, context: Arc<TestContext>, server_id: u32) -> Self {
let reserver = SinglePortReserver::new().expect("Failed to reserve port for MCP server");
let server_address = reserver.address();
Self {
+ server_id,
config,
context,
envs: HashMap::new(),
@@ -154,6 +154,14 @@ impl TestBinary for McpHandle {
port_reserver: Some(reserver),
}
}
+}
+
+impl TestBinary for McpHandle {
+ type Config = McpConfig;
+
+ fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+ Self::with_server_id(config, context, 0)
+ }
#[allow(deprecated)]
fn start(&mut self) -> Result<(), TestBinaryError> {
@@ -177,8 +185,8 @@ impl TestBinary for McpHandle {
command.stdout(Stdio::inherit());
command.stderr(Stdio::inherit());
} else {
- let stdout_path = self.context.mcp_stdout_path();
- let stderr_path = self.context.mcp_stderr_path();
+ let stdout_path = self.context.mcp_stdout_path(self.server_id);
+ let stderr_path = self.context.mcp_stderr_path(self.server_id);
let stdout_file =
File::create(&stdout_path).map_err(|e| TestBinaryError::FileSystemError {
diff --git a/core/integration/src/harness/handle/mod.rs b/core/integration/src/harness/handle/mod.rs
index d98a4dfc6..da746d175 100644
--- a/core/integration/src/harness/handle/mod.rs
+++ b/core/integration/src/harness/handle/mod.rs
@@ -28,4 +28,4 @@ pub use client::ClientHandle;
pub use client_builder::ClientBuilder;
pub use connectors_runtime::ConnectorsRuntimeHandle;
pub use mcp::{McpClient, McpHandle};
-pub use server::ServerHandle;
+pub use server::{ServerHandle, ServerLogs};
diff --git a/core/integration/src/harness/handle/server.rs b/core/integration/src/harness/handle/server.rs
index 10a651496..cc0b33e6d 100644
--- a/core/integration/src/harness/handle/server.rs
+++ b/core/integration/src/harness/handle/server.rs
@@ -18,11 +18,13 @@
*/
use super::client_builder::{ClientBuilder, ServerConnection};
-use crate::harness::config::{IpAddrKind, TestServerConfig};
+use super::connectors_runtime::ConnectorsRuntimeHandle;
+use super::mcp::McpHandle;
+use crate::harness::config::{ConnectorsRuntimeConfig, IpAddrKind, McpConfig, TestServerConfig};
use crate::harness::context::TestContext;
use crate::harness::error::TestBinaryError;
use crate::harness::port_reserver::PortReserver;
-use crate::harness::traits::{Restartable, TestBinary};
+use crate::harness::traits::{IggyServerDependent, Restartable, TestBinary};
use assert_cmd::prelude::CommandCargoExt;
use iggy::prelude::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
use iggy_common::TransportProtocol;
@@ -75,6 +77,8 @@ pub struct ServerHandle {
generated_cert_dir: Option<PathBuf>,
port_reserver: Option<PortReserver>,
test_transport: Option<iggy_common::TransportProtocol>,
+ mcp: Option<McpHandle>,
+ connectors_runtime: Option<ConnectorsRuntimeHandle>,
}
impl std::fmt::Debug for ServerHandle {
@@ -500,6 +504,8 @@ impl ServerHandle {
generated_cert_dir: None,
port_reserver: None,
test_transport: None,
+ mcp: None,
+ connectors_runtime: None,
}
}
@@ -517,6 +523,95 @@ impl ServerHandle {
pub fn set_test_transport(&mut self, transport: iggy_common::TransportProtocol) {
self.test_transport = Some(transport);
}
+
+ /// Configure MCP server for this iggy server.
+ pub fn set_mcp_config(&mut self, config: McpConfig) {
+ self.mcp = Some(McpHandle::with_server_id(
+ config,
+ self.context.clone(),
+ self.server_id,
+ ));
+ }
+
+ /// Configure connectors runtime for this iggy server.
+ pub fn set_connectors_runtime_config(&mut self, config: ConnectorsRuntimeConfig) {
+ self.connectors_runtime = Some(ConnectorsRuntimeHandle::with_server_id(
+ config,
+ self.context.clone(),
+ self.server_id,
+ ));
+ }
+
+ /// Get reference to MCP handle if configured.
+ pub fn mcp(&self) -> Option<&McpHandle> {
+ self.mcp.as_ref()
+ }
+
+ /// Get mutable reference to MCP handle if configured.
+ pub fn mcp_mut(&mut self) -> Option<&mut McpHandle> {
+ self.mcp.as_mut()
+ }
+
+ /// Get reference to connectors runtime handle if configured.
+ pub fn connectors_runtime(&self) -> Option<&ConnectorsRuntimeHandle> {
+ self.connectors_runtime.as_ref()
+ }
+
+ /// Get mutable reference to connectors runtime handle if configured.
+ pub fn connectors_runtime_mut(&mut self) -> Option<&mut ConnectorsRuntimeHandle> {
+ self.connectors_runtime.as_mut()
+ }
+
+ /// Start dependent binaries (MCP, connectors runtime) after server is ready.
+ pub async fn start_dependents(&mut self) -> Result<(), TestBinaryError> {
+ let tcp_addr = self.addrs.tcp;
+
+ if let Some(ref mut mcp) = self.mcp {
+ if let Some(addr) = tcp_addr {
+ mcp.set_iggy_address(addr);
+ }
+ mcp.start()?;
+ mcp.wait_ready().await?;
+ }
+
+ if let Some(ref mut connectors_runtime) = self.connectors_runtime {
+ if let Some(addr) = tcp_addr {
+ connectors_runtime.set_iggy_address(addr);
+ }
+ connectors_runtime.start()?;
+ connectors_runtime.wait_ready().await?;
+ }
+
+ Ok(())
+ }
+
+ /// Stop dependent binaries.
+ pub fn stop_dependents(&mut self) -> Result<(), TestBinaryError> {
+ if let Some(ref mut connectors_runtime) = self.connectors_runtime {
+ connectors_runtime.stop()?;
+ }
+ if let Some(ref mut mcp) = self.mcp {
+ mcp.stop()?;
+ }
+ Ok(())
+ }
+
+ /// Collect logs from server and its dependent binaries.
+ pub fn collect_all_logs(&self) -> ServerLogs {
+ ServerLogs {
+ server: self.collect_logs(),
+ mcp: self.mcp.as_ref().map(|m| m.collect_logs()),
+ connectors_runtime: self.connectors_runtime.as_ref().map(|c| c.collect_logs()),
+ }
+ }
+}
+
+/// Collected logs from a server and its dependent binaries.
+#[derive(Debug)]
+pub struct ServerLogs {
+ pub server: (String, String),
+ pub mcp: Option<(String, String)>,
+ pub connectors_runtime: Option<(String, String)>,
}
impl TestBinary for ServerHandle {
@@ -537,6 +632,8 @@ impl TestBinary for ServerHandle {
generated_cert_dir: None,
port_reserver: None,
test_transport: None,
+ mcp: None,
+ connectors_runtime: None,
}
}
diff --git a/core/integration/src/harness/mod.rs b/core/integration/src/harness/mod.rs
index 2d5f4e5d1..13793ab05 100644
--- a/core/integration/src/harness/mod.rs
+++ b/core/integration/src/harness/mod.rs
@@ -45,7 +45,7 @@
pub mod config;
mod context;
mod error;
-pub mod fixtures;
+mod fixture;
pub mod handle;
mod helpers;
mod orchestrator;
@@ -70,4 +70,4 @@ pub use helpers::{
USER_PASSWORD, assert_clean_system, create_user, delete_user, login_root, login_user,
};
-pub use fixtures::TestFixture;
+pub use fixture::TestFixture;
diff --git a/core/integration/src/harness/orchestrator/builder.rs b/core/integration/src/harness/orchestrator/builder.rs
index ab9dfe156..3b503f920 100644
--- a/core/integration/src/harness/orchestrator/builder.rs
+++ b/core/integration/src/harness/orchestrator/builder.rs
@@ -23,7 +23,7 @@ use crate::harness::config::{
};
use crate::harness::context::TestContext;
use crate::harness::error::TestBinaryError;
-use crate::harness::handle::{ConnectorsRuntimeHandle, McpHandle, ServerHandle};
+use crate::harness::handle::ServerHandle;
use crate::harness::port_reserver::ClusterPortReserver;
use crate::harness::traits::TestBinary;
use std::collections::HashMap;
@@ -198,21 +198,21 @@ impl TestHarnessBuilder {
context.ensure_created()?;
let context = Arc::new(context);
- let servers = build_servers(self.server_config, self.cluster_node_count, &context)?;
+ let mut servers = build_servers(self.server_config, self.cluster_node_count, &context)?;
- let mcp = self
- .mcp_config
- .map(|config| McpHandle::with_config(config, context.clone()));
-
- let connectors_runtime = self
- .connectors_runtime_config
- .map(|config| ConnectorsRuntimeHandle::with_config(config, context.clone()));
+ // Attach MCP and connectors runtime to the first server (primary)
+ if let Some(server) = servers.first_mut() {
+ if let Some(config) = self.mcp_config {
+ server.set_mcp_config(config);
+ }
+ if let Some(config) = self.connectors_runtime_config {
+ server.set_connectors_runtime_config(config);
+ }
+ }
Ok(TestHarness {
context,
servers,
- mcp,
- connectors_runtime,
clients: Vec::new(),
client_configs: self.clients,
primary_transport: self.primary_transport,
@@ -383,7 +383,7 @@ mod tests {
.unwrap();
assert_eq!(harness.servers.len(), 1);
- assert!(harness.mcp.is_some());
+ assert!(harness.mcp().is_some());
}
#[test]
diff --git a/core/integration/src/harness/orchestrator/harness.rs b/core/integration/src/harness/orchestrator/harness.rs
index 2c32ef0f2..b2e4630dd 100644
--- a/core/integration/src/harness/orchestrator/harness.rs
+++ b/core/integration/src/harness/orchestrator/harness.rs
@@ -22,9 +22,9 @@ use crate::harness::config::ClientConfig;
use crate::harness::context::TestContext;
use crate::harness::error::TestBinaryError;
use crate::harness::handle::{
- ClientHandle, ConnectorsRuntimeHandle, McpClient, McpHandle, ServerHandle,
+ ClientHandle, ConnectorsRuntimeHandle, McpClient, McpHandle, ServerHandle, ServerLogs,
};
-use crate::harness::traits::{IggyServerDependent, Restartable, TestBinary};
+use crate::harness::traits::{Restartable, TestBinary};
use crate::http_client::HttpClientFactory;
use crate::quic_client::QuicClientFactory;
use crate::tcp_client::TcpClientFactory;
@@ -41,9 +41,7 @@ use std::sync::Arc;
/// Collected logs from all binaries in the harness.
#[derive(Debug)]
pub struct TestLogs {
- pub server: Option<(String, String)>,
- pub mcp: Option<(String, String)>,
- pub connectors_runtime: Option<(String, String)>,
+ pub servers: Vec<ServerLogs>,
}
#[derive(Default)]
@@ -58,8 +56,6 @@ pub(super) struct TlsSettings {
pub struct TestHarness {
pub(super) context: Arc<TestContext>,
pub(super) servers: Vec<ServerHandle>,
- pub(super) mcp: Option<McpHandle>,
- pub(super) connectors_runtime: Option<ConnectorsRuntimeHandle>,
pub(super) clients: Vec<ClientHandle>,
pub(super) client_configs: Vec<ClientConfig>,
pub(super) primary_transport: Option<TransportProtocol>,
@@ -69,12 +65,17 @@ pub struct TestHarness {
impl std::fmt::Debug for TestHarness {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ let has_mcp = self.servers.iter().any(|s| s.mcp().is_some());
+ let has_connectors = self
+ .servers
+ .iter()
+ .any(|s| s.connectors_runtime().is_some());
f.debug_struct("TestHarness")
.field("test_name", &self.context.test_name())
.field("started", &self.started)
.field("server_count", &self.servers.len())
- .field("has_mcp", &self.mcp.is_some())
- .field("has_connectors_runtime", &self.connectors_runtime.is_some())
+ .field("has_mcp", &has_mcp)
+ .field("has_connectors_runtime", &has_connectors)
.field("client_count", &self.clients.len())
.finish()
}
@@ -138,24 +139,9 @@ impl TestHarness {
}
async fn start_dependents(&mut self) -> Result<(), TestBinaryError> {
- let tcp_addr = self.servers.first().and_then(|s| s.tcp_addr());
-
- if let Some(ref mut mcp) = self.mcp {
- if let Some(addr) = tcp_addr {
- mcp.set_iggy_address(addr);
- }
- mcp.start()?;
- mcp.wait_ready().await?;
- }
-
- if let Some(ref mut connectors_runtime) = self.connectors_runtime {
- if let Some(addr) = tcp_addr {
- connectors_runtime.set_iggy_address(addr);
- }
- connectors_runtime.start()?;
- connectors_runtime.wait_ready().await?;
+ for server in &mut self.servers {
+ server.start_dependents().await?;
}
-
Ok(())
}
@@ -166,15 +152,8 @@ impl TestHarness {
}
self.clients.clear();
- if let Some(ref mut connectors_runtime) = self.connectors_runtime {
- connectors_runtime.stop()?;
- }
-
- if let Some(ref mut mcp) = self.mcp {
- mcp.stop()?;
- }
-
for server in self.servers.iter_mut().rev() {
+ server.stop_dependents()?;
server.stop()?;
}
@@ -260,23 +239,39 @@ impl TestHarness {
&mut self.clients
}
- /// Get the MCP handle if configured.
+ /// Get the MCP handle from the primary server if configured.
+ ///
+ /// # Panics
+ /// Panics if called on a cluster (multiple servers). Use `node(i).mcp()` instead.
pub fn mcp(&self) -> Option<&McpHandle> {
- self.mcp.as_ref()
+ assert!(
+ self.servers.len() <= 1,
+ "mcp() is only available for single-server setups. Use node(i).mcp() for clusters."
+ );
+ self.servers.first().and_then(|s| s.mcp())
}
/// Create an MCP client (convenience method).
+ ///
+ /// # Panics
+ /// Panics if called on a cluster (multiple servers). Use `node(i).mcp()` instead.
pub async fn mcp_client(&self) -> Result<McpClient, TestBinaryError> {
- self.mcp
- .as_ref()
+ self.mcp()
.ok_or(TestBinaryError::MissingMcp)?
.create_client()
.await
}
- /// Get the connectors runtime handle if configured.
+ /// Get the connectors runtime handle from the primary server if configured.
+ ///
+ /// # Panics
+ /// Panics if called on a cluster (multiple servers). Use `node(i).connectors_runtime()` instead.
pub fn connectors_runtime(&self) -> Option<&ConnectorsRuntimeHandle> {
- self.connectors_runtime.as_ref()
+ assert!(
+ self.servers.len() <= 1,
+ "connectors_runtime() is only available for single-server setups. Use node(i).connectors_runtime() for clusters."
+ );
+ self.servers.first().and_then(|s| s.connectors_runtime())
}
/// Get the test directory path.
@@ -287,9 +282,7 @@ impl TestHarness {
/// Collect logs from all binaries.
pub fn collect_logs(&self) -> TestLogs {
TestLogs {
- server: self.servers.first().map(|s| s.collect_logs()),
- mcp: self.mcp.as_ref().map(|m| m.collect_logs()),
- connectors_runtime: self.connectors_runtime.as_ref().map(|c| c.collect_logs()),
+ servers: self.servers.iter().map(|s| s.collect_all_logs()).collect(),
}
}
diff --git a/core/integration/tests/connectors/api/endpoints.rs b/core/integration/tests/connectors/api/endpoints.rs
index 7bc9b7588..fe97c2165 100644
--- a/core/integration/tests/connectors/api/endpoints.rs
+++ b/core/integration/tests/connectors/api/endpoints.rs
@@ -17,15 +17,21 @@
* under the License.
*/
-use crate::connectors::api::setup;
+use integration::harness::seeds;
+use integration::iggy_harness;
use reqwest::Client;
const API_KEY: &str = "test-api-key";
-#[tokio::test]
-async fn test_root_endpoint_returns_welcome_message() {
- let runtime = setup().await;
- let api_address = runtime.connectors_api_address().unwrap();
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/api/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn root_endpoint_returns_welcome_message(harness: &TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = Client::new();
let response = client
@@ -39,10 +45,15 @@ async fn test_root_endpoint_returns_welcome_message() {
assert_eq!(body, "Connector Runtime API");
}
-#[tokio::test]
-async fn test_health_endpoint_returns_healthy() {
- let runtime = setup().await;
- let api_address = runtime.connectors_api_address().unwrap();
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/api/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn health_endpoint_returns_healthy(harness: &TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = Client::new();
let response = client
@@ -56,10 +67,15 @@ async fn test_health_endpoint_returns_healthy() {
assert_eq!(body["status"], "healthy");
}
-#[tokio::test]
-async fn test_stats_endpoint_returns_runtime_stats() {
- let runtime = setup().await;
- let api_address = runtime.connectors_api_address().unwrap();
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/api/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn stats_endpoint_returns_runtime_stats(harness: &TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = Client::new();
let response = client
@@ -90,10 +106,15 @@ async fn test_stats_endpoint_returns_runtime_stats() {
assert_eq!(stats["sinks_running"], 0);
}
-#[tokio::test]
-async fn test_metrics_endpoint_returns_prometheus_format() {
- let runtime = setup().await;
- let api_address = runtime.connectors_api_address().unwrap();
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/api/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn metrics_endpoint_returns_prometheus_format(harness: &TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = Client::new();
let response = client
@@ -112,10 +133,15 @@ async fn test_metrics_endpoint_returns_prometheus_format() {
assert!(body.contains("iggy_connectors_sinks_running"));
}
-#[tokio::test]
-async fn test_sources_endpoint_returns_list() {
- let runtime = setup().await;
- let api_address = runtime.connectors_api_address().unwrap();
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/api/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sources_endpoint_returns_list(harness: &TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = Client::new();
let response = client
@@ -132,10 +158,15 @@ async fn test_sources_endpoint_returns_list() {
assert_eq!(sources.as_array().unwrap().len(), 0);
}
-#[tokio::test]
-async fn test_sinks_endpoint_returns_list() {
- let runtime = setup().await;
- let api_address = runtime.connectors_api_address().unwrap();
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/api/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sinks_endpoint_returns_list(harness: &TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = Client::new();
let response = client
@@ -152,10 +183,15 @@ async fn test_sinks_endpoint_returns_list() {
assert_eq!(sinks.as_array().unwrap().len(), 0);
}
-#[tokio::test]
-async fn test_api_key_authentication_required() {
- let runtime = setup().await;
- let api_address = runtime.connectors_api_address().unwrap();
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/api/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn api_key_authentication_required(harness: &TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = Client::new();
let response = client
@@ -191,10 +227,15 @@ async fn test_api_key_authentication_required() {
assert_eq!(response.status(), 401);
}
-#[tokio::test]
-async fn test_api_key_authentication_rejected_with_invalid_key() {
- let runtime = setup().await;
- let api_address = runtime.connectors_api_address().unwrap();
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/api/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn api_key_authentication_rejected_with_invalid_key(harness: &TestHarness) {
+ let api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = Client::new();
let response = client
diff --git a/core/integration/tests/connectors/api/mod.rs b/core/integration/tests/connectors/api/mod.rs
index 7e6f5fdcd..042431056 100644
--- a/core/integration/tests/connectors/api/mod.rs
+++ b/core/integration/tests/connectors/api/mod.rs
@@ -17,13 +17,4 @@
* under the License.
*/
-use crate::connectors::{ConnectorsRuntime, IggySetup, setup_runtime};
-
mod endpoints;
-
-pub(crate) async fn setup() -> ConnectorsRuntime {
- let iggy_setup = IggySetup::default();
- let mut runtime = setup_runtime();
- runtime.init("api/config.toml", None, iggy_setup).await;
- runtime
-}
diff --git a/core/integration/src/harness/fixtures/postgres/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs
similarity index 68%
copy from core/integration/src/harness/fixtures/postgres/mod.rs
copy to core/integration/tests/connectors/fixtures/mod.rs
index dfc195633..2cf2f8a9f 100644
--- a/core/integration/src/harness/fixtures/postgres/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -17,13 +17,14 @@
* under the License.
*/
-mod container;
-mod sink;
-mod source;
+mod postgres;
+mod quickwit;
+mod wiremock;
-pub use container::{PostgresContainer, SinkPayloadFormat, SinkSchema};
-pub use sink::{PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture};
-pub use source::{
- PostgresSourceByteaFixture, PostgresSourceDeleteFixture, PostgresSourceFixture,
- PostgresSourceJsonFixture, PostgresSourceJsonbFixture, PostgresSourceMarkFixture,
+pub use postgres::{
+ PostgresOps, PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture,
+ PostgresSourceByteaFixture, PostgresSourceDeleteFixture, PostgresSourceJsonFixture,
+ PostgresSourceJsonbFixture, PostgresSourceMarkFixture, PostgresSourceOps,
};
+pub use quickwit::{QuickwitFixture, QuickwitOps, QuickwitPreCreatedFixture};
+pub use wiremock::{WireMockDirectFixture, WireMockWrappedFixture};
diff --git a/core/integration/src/harness/fixtures/postgres/container.rs b/core/integration/tests/connectors/fixtures/postgres/container.rs
similarity index 91%
rename from core/integration/src/harness/fixtures/postgres/container.rs
rename to core/integration/tests/connectors/fixtures/postgres/container.rs
index ca016582c..b76a86667 100644
--- a/core/integration/src/harness/fixtures/postgres/container.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/container.rs
@@ -17,7 +17,7 @@
* under the License.
*/
-use crate::harness::error::TestBinaryError;
+use integration::harness::TestBinaryError;
use sqlx::postgres::PgPoolOptions;
use sqlx::{Pool, Postgres};
use testcontainers_modules::{
@@ -90,14 +90,9 @@ pub enum SinkSchema {
}
/// Trait for PostgreSQL fixtures with common container operations.
-#[allow(dead_code)]
pub trait PostgresOps: Sync {
fn container(&self) -> &PostgresContainer;
- fn connection_string(&self) -> &str {
- self.container().connection_string()
- }
-
fn create_pool(
&self,
) -> impl std::future::Future<Output = Result<Pool<Postgres>, TestBinaryError>> + Send {
@@ -106,23 +101,9 @@ pub trait PostgresOps: Sync {
}
/// Extension of `PostgresOps` for source fixtures that operate on a specific table.
-#[allow(dead_code)]
pub trait PostgresSourceOps: PostgresOps {
fn table_name(&self) -> &str;
- fn execute<'a>(
- &'a self,
- pool: &'a Pool<Postgres>,
- query: &'a str,
- ) -> impl std::future::Future<Output = ()> + Send + 'a {
- async move {
- sqlx::query(query)
- .execute(pool)
- .await
- .unwrap_or_else(|e| panic!("Failed to execute query: {e}"));
- }
- }
-
fn count_rows<'a>(
&'a self,
pool: &'a Pool<Postgres>,
@@ -170,10 +151,6 @@ impl PostgresContainer {
})
}
- pub fn connection_string(&self) -> &str {
- &self.connection_string
- }
-
pub async fn create_pool(&self) -> Result<Pool<Postgres>, TestBinaryError> {
PgPoolOptions::new()
.max_connections(1)
diff --git a/core/integration/src/harness/fixtures/postgres/mod.rs b/core/integration/tests/connectors/fixtures/postgres/mod.rs
similarity index 85%
rename from core/integration/src/harness/fixtures/postgres/mod.rs
rename to core/integration/tests/connectors/fixtures/postgres/mod.rs
index dfc195633..7f8b67870 100644
--- a/core/integration/src/harness/fixtures/postgres/mod.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/mod.rs
@@ -21,9 +21,9 @@ mod container;
mod sink;
mod source;
-pub use container::{PostgresContainer, SinkPayloadFormat, SinkSchema};
+pub use container::{PostgresOps, PostgresSourceOps};
pub use sink::{PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture};
pub use source::{
- PostgresSourceByteaFixture, PostgresSourceDeleteFixture, PostgresSourceFixture,
- PostgresSourceJsonFixture, PostgresSourceJsonbFixture, PostgresSourceMarkFixture,
+ PostgresSourceByteaFixture, PostgresSourceDeleteFixture, PostgresSourceJsonFixture,
+ PostgresSourceJsonbFixture, PostgresSourceMarkFixture,
};
diff --git a/core/integration/src/harness/fixtures/postgres/sink.rs b/core/integration/tests/connectors/fixtures/postgres/sink.rs
similarity index 98%
rename from core/integration/src/harness/fixtures/postgres/sink.rs
rename to core/integration/tests/connectors/fixtures/postgres/sink.rs
index 61ffea9bc..559013d63 100644
--- a/core/integration/src/harness/fixtures/postgres/sink.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/sink.rs
@@ -24,9 +24,8 @@ use super::container::{
ENV_SINK_STREAMS_0_TOPICS, ENV_SINK_TARGET_TABLE, PostgresContainer, PostgresOps,
SinkPayloadFormat, SinkSchema,
};
-use crate::harness::error::TestBinaryError;
-use crate::harness::fixtures::TestFixture;
use async_trait::async_trait;
+use integration::harness::{TestBinaryError, TestFixture};
use sqlx::{Pool, Postgres};
use std::collections::HashMap;
use std::time::Duration;
diff --git a/core/integration/src/harness/fixtures/postgres/source.rs b/core/integration/tests/connectors/fixtures/postgres/source.rs
similarity index 83%
rename from core/integration/src/harness/fixtures/postgres/source.rs
rename to core/integration/tests/connectors/fixtures/postgres/source.rs
index 7991c88bf..f5739d88d 100644
--- a/core/integration/src/harness/fixtures/postgres/source.rs
+++ b/core/integration/tests/connectors/fixtures/postgres/source.rs
@@ -23,119 +23,13 @@ use super::container::{
ENV_SOURCE_PAYLOAD_COLUMN, ENV_SOURCE_PAYLOAD_FORMAT, ENV_SOURCE_POLL_INTERVAL,
ENV_SOURCE_PRIMARY_KEY_COLUMN, ENV_SOURCE_PROCESSED_COLUMN, ENV_SOURCE_STREAMS_0_SCHEMA,
ENV_SOURCE_STREAMS_0_STREAM, ENV_SOURCE_STREAMS_0_TOPIC, ENV_SOURCE_TABLES,
- ENV_SOURCE_TRACKING_COLUMN, PostgresContainer, PostgresOps, PostgresSourceOps, SinkSchema,
+ ENV_SOURCE_TRACKING_COLUMN, PostgresContainer, PostgresOps, PostgresSourceOps,
};
-use crate::harness::error::TestBinaryError;
-use crate::harness::fixtures::TestFixture;
use async_trait::async_trait;
+use integration::harness::{TestBinaryError, TestFixture};
use sqlx::{Pool, Postgres};
use std::collections::HashMap;
-/// PostgreSQL source connector fixture.
-///
-/// Starts a PostgreSQL container and provides environment variables
-/// for the source connector to connect to it.
-pub struct PostgresSourceFixture {
- container: PostgresContainer,
- table_name: String,
- schema: SinkSchema,
-}
-
-impl PostgresSourceFixture {
- pub fn connection_string(&self) -> &str {
- self.container.connection_string()
- }
-
- pub async fn create_pool(&self) -> Result<Pool<Postgres>, TestBinaryError> {
- self.container.create_pool().await
- }
-
- pub fn table_name(&self) -> &str {
- &self.table_name
- }
-
- pub async fn execute(&self, pool: &Pool<Postgres>, query: &str) {
- sqlx::query(query)
- .execute(pool)
- .await
- .unwrap_or_else(|e| panic!("Failed to execute query: {e}"));
- }
-
- pub async fn count_rows(&self, pool: &Pool<Postgres>, table: &str) -> i64 {
- let query = format!("SELECT COUNT(*) as count FROM {table}");
- let row: (i64,) = sqlx::query_as(&query)
- .fetch_one(pool)
- .await
- .unwrap_or_else(|e| panic!("Failed to count rows: {e}"));
- row.0
- }
-
- pub async fn count_rows_where(
- &self,
- pool: &Pool<Postgres>,
- table: &str,
- condition: &str,
- ) -> i64 {
- let query = format!("SELECT COUNT(*) as count FROM {table} WHERE {condition}");
- let row: (i64,) = sqlx::query_as(&query)
- .fetch_one(pool)
- .await
- .unwrap_or_else(|e| panic!("Failed to count rows: {e}"));
- row.0
- }
-}
-
-#[async_trait]
-impl TestFixture for PostgresSourceFixture {
- async fn setup() -> Result<Self, TestBinaryError> {
- let container = PostgresContainer::start().await?;
- let table_name = "test_messages".to_string();
- Ok(Self {
- container,
- table_name,
- schema: SinkSchema::Json,
- })
- }
-
- fn connectors_runtime_envs(&self) -> HashMap<String, String> {
- let mut envs = HashMap::new();
-
- envs.insert(
- ENV_SOURCE_CONNECTION_STRING.to_string(),
- self.container.connection_string.clone(),
- );
- envs.insert(
- ENV_SOURCE_TABLES.to_string(),
- format!("[{}]", self.table_name),
- );
- envs.insert(ENV_SOURCE_TRACKING_COLUMN.to_string(), "id".to_string());
- envs.insert(
- ENV_SOURCE_STREAMS_0_STREAM.to_string(),
- DEFAULT_TEST_STREAM.to_string(),
- );
- envs.insert(
- ENV_SOURCE_STREAMS_0_TOPIC.to_string(),
- DEFAULT_TEST_TOPIC.to_string(),
- );
- envs.insert(ENV_SOURCE_POLL_INTERVAL.to_string(), "10ms".to_string());
- envs.insert(
- ENV_SOURCE_PATH.to_string(),
- "../../target/debug/libiggy_connector_postgres_source".to_string(),
- );
-
- let schema_str = match self.schema {
- SinkSchema::Json => "json",
- SinkSchema::Raw => "raw",
- };
- envs.insert(
- ENV_SOURCE_STREAMS_0_SCHEMA.to_string(),
- schema_str.to_string(),
- );
-
- envs
- }
-}
-
/// PostgreSQL source fixture for JSON rows with metadata.
///
/// Creates a table with typed columns that get serialized as JSON with metadata.
diff --git a/core/integration/tests/connectors/fixtures/quickwit/container.rs b/core/integration/tests/connectors/fixtures/quickwit/container.rs
new file mode 100644
index 000000000..8cf33b53a
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/quickwit/container.rs
@@ -0,0 +1,359 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use async_trait::async_trait;
+use integration::harness::{TestBinaryError, TestFixture, seeds};
+use reqwest_middleware::ClientWithMiddleware as HttpClient;
+use reqwest_retry::RetryTransientMiddleware;
+use reqwest_retry::policies::ExponentialBackoff;
+use serde::Deserialize;
+use std::collections::HashMap;
+use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor};
+use testcontainers_modules::testcontainers::runners::AsyncRunner;
+use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt};
+use tracing::info;
+use uuid::Uuid;
+
+const QUICKWIT_IMAGE: &str = "quickwit/quickwit";
+const QUICKWIT_TAG: &str = "0.8.1";
+const QUICKWIT_PORT: u16 = 7280;
+const QUICKWIT_READY_MSG: &str = "REST server is ready";
+const QUICKWIT_LISTEN_ADDRESS: &str = "0.0.0.0";
+
+const ENV_PLUGIN_URL: &str = "IGGY_CONNECTORS_SINK_QUICKWIT_PLUGIN_CONFIG_URL";
+const ENV_PLUGIN_INDEX: &str = "IGGY_CONNECTORS_SINK_QUICKWIT_PLUGIN_CONFIG_INDEX";
+const ENV_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SINK_QUICKWIT_STREAMS_0_STREAM";
+const ENV_STREAMS_0_TOPICS: &str = "IGGY_CONNECTORS_SINK_QUICKWIT_STREAMS_0_TOPICS";
+const ENV_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SINK_QUICKWIT_STREAMS_0_SCHEMA";
+const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_QUICKWIT_PATH";
+
+const INDEXES_API_PATH: &str = "api/v1/indexes";
+const INDEX_TIMESTAMP_FIELD: &str = "timestamp";
+
+#[derive(Deserialize)]
+pub struct QuickwitSearchResponse {
+ pub hits: Vec<serde_json::Value>,
+ pub num_hits: usize,
+}
+
+pub struct QuickwitContainer {
+ #[allow(dead_code)]
+ container: ContainerAsync<GenericImage>,
+ mapped_port: u16,
+}
+
+impl QuickwitContainer {
+ async fn start() -> Result<Self, TestBinaryError> {
+ let unique_network = format!("iggy-quickwit-sink-{}", Uuid::new_v4());
+
+ let container = GenericImage::new(QUICKWIT_IMAGE, QUICKWIT_TAG)
+ .with_exposed_port(0.tcp())
+ .with_wait_for(WaitFor::message_on_stdout(QUICKWIT_READY_MSG))
+ .with_network(unique_network)
+ .with_cmd(["run"])
+ .with_env_var("QW_LISTEN_ADDRESS", QUICKWIT_LISTEN_ADDRESS)
+ .with_mapped_port(0, QUICKWIT_PORT.tcp())
+ .start()
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "QuickwitContainer".to_string(),
+ message: format!("Failed to start container: {e}"),
+ })?;
+
+ info!("Started quickwit container");
+
+ let mapped_port = container
+ .ports()
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "QuickwitContainer".to_string(),
+ message: format!("Failed to get ports: {e}"),
+ })?
+ .map_to_host_port_ipv4(QUICKWIT_PORT)
+ .ok_or_else(|| TestBinaryError::FixtureSetup {
+ fixture_type: "QuickwitContainer".to_string(),
+ message: "No mapping for Quickwit port".to_string(),
+ })?;
+
+ info!("Quickwit container mapped to port {mapped_port}");
+
+ Ok(Self {
+ container,
+ mapped_port,
+ })
+ }
+
+ fn base_url(&self) -> String {
+ format!("http://localhost:{}", self.mapped_port)
+ }
+}
+
+fn create_http_client() -> HttpClient {
+ let retry_policy = ExponentialBackoff::builder().build_with_max_retries(3);
+ reqwest_middleware::ClientBuilder::new(reqwest::Client::new())
+ .with(RetryTransientMiddleware::new_with_policy(retry_policy))
+ .build()
+}
+
+/// Trait for Quickwit fixtures with common HTTP operations.
+pub trait QuickwitOps: Sync {
+ fn container(&self) -> &QuickwitContainer;
+ fn http_client(&self) -> &HttpClient;
+
+ fn create_index(
+ &self,
+ index_config: &str,
+ ) -> impl std::future::Future<Output = Result<(), TestBinaryError>> + Send {
+ async move {
+ let response = self
+ .http_client()
+ .post(format!(
+ "{}/{}",
+ self.container().base_url(),
+ INDEXES_API_PATH
+ ))
+ .header("Content-Type", "application/yaml")
+ .body(index_config.to_string())
+ .send()
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "QuickwitOps".to_string(),
+ message: format!("Failed to create index: {e}"),
+ })?;
+
+ info!("Received create index response");
+
+ if !response.status().is_success() {
+ let status = response.status();
+ let body = response.text().await.unwrap_or_default();
+ return Err(TestBinaryError::FixtureSetup {
+ fixture_type: "QuickwitOps".to_string(),
+ message: format!("Failed to create index: status={status}, body={body}"),
+ });
+ }
+
+ Ok(())
+ }
+ }
+
+ fn flush_index(
+ &self,
+ index_id: &str,
+ ) -> impl std::future::Future<Output = Result<(), TestBinaryError>> + Send {
+ async move {
+ let ingest_url = format!("{}/api/v1/{}/ingest", self.container().base_url(), index_id);
+
+ let response = self
+ .http_client()
+ .post(&ingest_url)
+ .query(&[("commit", "force")])
+ .json("{}")
+ .send()
+ .await
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to flush index: {e}"),
+ })?;
+
+ info!("Received index ingest/flush response");
+
+ if !response.status().is_success() {
+ let status = response.status();
+ let body = response.text().await.unwrap_or_default();
+ return Err(TestBinaryError::InvalidState {
+ message: format!("Failed to flush index: status={status}, body={body}"),
+ });
+ }
+
+ Ok(())
+ }
+ }
+
+ fn search_all(
+ &self,
+ index_id: &str,
+ ) -> impl std::future::Future<Output = Result<QuickwitSearchResponse, TestBinaryError>> + Send
+ {
+ async move {
+ let search_url = format!("{}/api/v1/{}/search", self.container().base_url(), index_id);
+ let descending = "-";
+
+ let response = self
+ .http_client()
+ .get(&search_url)
+ .query(&[("query", "")])
+ .query(&[("sort_by", format!("{descending}{INDEX_TIMESTAMP_FIELD}"))])
+ .send()
+ .await
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to search index: {e}"),
+ })?;
+
+ info!("Received search index response");
+
+ if !response.status().is_success() {
+ let status = response.status();
+ let body = response.text().await.unwrap_or_default();
+ return Err(TestBinaryError::InvalidState {
+ message: format!("Failed to search index: status={status}, body={body}"),
+ });
+ }
+
+ response
+ .json::<QuickwitSearchResponse>()
+ .await
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to parse search response: {e}"),
+ })
+ }
+ }
+}
+
+fn get_index_config(index_id: &str) -> String {
+ format!(
+ r#"
+version: 0.8
+index_id: {index_id}
+doc_mapping:
+ mode: strict
+ field_mappings:
+ - name: id
+ type: u64
+ - name: name
+ type: text
+ tokenizer: raw
+ - name: count
+ type: u64
+ - name: amount
+ type: f64
+ - name: active
+ type: bool
+ - name: {INDEX_TIMESTAMP_FIELD}
+ type: datetime
+ input_formats:
+ - unix_timestamp
+ output_format: unix_timestamp_micros
+ fast_precision: microseconds
+ fast: true
+ timestamp_field: timestamp
+retention:
+ period: 7 days
+ schedule: daily
+"#
+ )
+}
+
+fn build_connector_envs(base_url: &str) -> HashMap<String, String> {
+ HashMap::from([
+ (ENV_PLUGIN_URL.to_string(), base_url.to_string()),
+ (
+ ENV_PLUGIN_INDEX.to_string(),
+ get_index_config(seeds::names::TOPIC),
+ ),
+ (
+ ENV_STREAMS_0_STREAM.to_string(),
+ seeds::names::STREAM.to_string(),
+ ),
+ (
+ ENV_STREAMS_0_TOPICS.to_string(),
+ format!("[{}]", seeds::names::TOPIC),
+ ),
+ (ENV_STREAMS_0_SCHEMA.to_string(), "json".to_string()),
+ (
+ ENV_SINK_PATH.to_string(),
+ "../../target/debug/libiggy_connector_quickwit_sink".to_string(),
+ ),
+ ])
+}
+
+/// Quickwit fixture where the connector auto-creates the index.
+pub struct QuickwitFixture {
+ container: QuickwitContainer,
+ http_client: HttpClient,
+}
+
+impl QuickwitOps for QuickwitFixture {
+ fn container(&self) -> &QuickwitContainer {
+ &self.container
+ }
+
+ fn http_client(&self) -> &HttpClient {
+ &self.http_client
+ }
+}
+
+#[async_trait]
+impl TestFixture for QuickwitFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = QuickwitContainer::start().await?;
+ let http_client = create_http_client();
+
+ Ok(Self {
+ container,
+ http_client,
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ build_connector_envs(&self.container.base_url())
+ }
+}
+
+/// Quickwit fixture where the index is pre-created before the test.
+pub struct QuickwitPreCreatedFixture {
+ inner: QuickwitFixture,
+}
+
+impl std::ops::Deref for QuickwitPreCreatedFixture {
+ type Target = QuickwitFixture;
+ fn deref(&self) -> &Self::Target {
+ &self.inner
+ }
+}
+
+impl QuickwitOps for QuickwitPreCreatedFixture {
+ fn container(&self) -> &QuickwitContainer {
+ &self.inner.container
+ }
+
+ fn http_client(&self) -> &HttpClient {
+ &self.inner.http_client
+ }
+}
+
+#[async_trait]
+impl TestFixture for QuickwitPreCreatedFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = QuickwitContainer::start().await?;
+ let http_client = create_http_client();
+
+ let inner = QuickwitFixture {
+ container,
+ http_client,
+ };
+
+ let index_config = get_index_config(seeds::names::TOPIC);
+ inner.create_index(&index_config).await?;
+
+ Ok(Self { inner })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ self.inner.connectors_runtime_envs()
+ }
+}
diff --git a/core/integration/src/harness/handle/mod.rs b/core/integration/tests/connectors/fixtures/quickwit/mod.rs
similarity index 74%
copy from core/integration/src/harness/handle/mod.rs
copy to core/integration/tests/connectors/fixtures/quickwit/mod.rs
index d98a4dfc6..9b2c59a22 100644
--- a/core/integration/src/harness/handle/mod.rs
+++ b/core/integration/tests/connectors/fixtures/quickwit/mod.rs
@@ -17,15 +17,6 @@
* under the License.
*/
-mod client;
-mod client_builder;
-pub mod common;
-mod connectors_runtime;
-mod mcp;
-mod server;
+mod container;
-pub use client::ClientHandle;
-pub use client_builder::ClientBuilder;
-pub use connectors_runtime::ConnectorsRuntimeHandle;
-pub use mcp::{McpClient, McpHandle};
-pub use server::ServerHandle;
+pub use container::{QuickwitFixture, QuickwitOps, QuickwitPreCreatedFixture};
diff --git a/core/integration/src/harness/fixtures/wiremock.rs b/core/integration/tests/connectors/fixtures/wiremock.rs
similarity index 94%
rename from core/integration/src/harness/fixtures/wiremock.rs
rename to core/integration/tests/connectors/fixtures/wiremock.rs
index 2c9f6adb3..497243ac9 100644
--- a/core/integration/src/harness/fixtures/wiremock.rs
+++ b/core/integration/tests/connectors/fixtures/wiremock.rs
@@ -17,9 +17,8 @@
* under the License.
*/
-use super::TestFixture;
-use crate::harness::error::TestBinaryError;
use async_trait::async_trait;
+use integration::harness::{TestBinaryError, TestFixture};
use std::collections::HashMap;
use testcontainers_modules::testcontainers::core::WaitFor::Healthcheck;
use testcontainers_modules::testcontainers::core::wait::HealthWaitStrategy;
@@ -98,12 +97,6 @@ pub struct WireMockDirectFixture {
container: WireMockContainer,
}
-impl WireMockDirectFixture {
- pub fn base_url(&self) -> &str {
- &self.container.base_url
- }
-}
-
#[async_trait]
impl TestFixture for WireMockDirectFixture {
async fn setup() -> Result<Self, TestBinaryError> {
@@ -126,12 +119,6 @@ pub struct WireMockWrappedFixture {
container: WireMockContainer,
}
-impl WireMockWrappedFixture {
- pub fn base_url(&self) -> &str {
- &self.container.base_url
- }
-}
-
#[async_trait]
impl TestFixture for WireMockWrappedFixture {
async fn setup() -> Result<Self, TestBinaryError> {
diff --git a/core/integration/tests/connectors/http_config_provider/direct_responses.rs b/core/integration/tests/connectors/http_config_provider/direct_responses.rs
index b77c67659..eeebddf4b 100644
--- a/core/integration/tests/connectors/http_config_provider/direct_responses.rs
+++ b/core/integration/tests/connectors/http_config_provider/direct_responses.rs
@@ -16,17 +16,24 @@
* specific language governing permissions and limitations
* under the License.
*/
-use crate::connectors::http_config_provider::WireMockMode;
+
+use crate::connectors::fixtures::WireMockDirectFixture;
+use integration::harness::seeds;
+use integration::iggy_harness;
use reqwest::StatusCode;
-#[tokio::test]
-async fn test_source_configs_list_returns_all_versions() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Direct).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_direct.toml")),
+ seed = seeds::connector_stream
+)]
+async fn source_configs_list_returns_all_versions(
+ harness: &TestHarness,
+ _wiremock: WireMockDirectFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -42,7 +49,6 @@ async fn test_source_configs_list_returns_all_versions() {
let configs = body.as_array().unwrap();
assert_eq!(configs.len(), 2, "Should have 2 config versions");
- // Validate version 0 schema
let v0 = &configs[0];
assert_eq!(v0["key"].as_str().unwrap(), "random");
assert!(v0["enabled"].as_bool().unwrap());
@@ -55,35 +61,36 @@ async fn test_source_configs_list_returns_all_versions() {
.contains("libiggy_connector_random_source")
);
- // Validate streams array (sources have StreamProducerConfig)
let streams = v0["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1, "Should have 1 stream config");
let stream = &streams[0];
assert_eq!(stream["stream"].as_str().unwrap(), "test_stream");
- assert_eq!(stream["topic"].as_str().unwrap(), "test_topic"); // String, not array!
+ assert_eq!(stream["topic"].as_str().unwrap(), "test_topic");
assert_eq!(stream["schema"].as_str().unwrap(), "json");
assert!(stream["batch_length"].as_u64().is_some());
- assert!(stream["linger_time"].as_str().is_some()); // Source-specific field
+ assert!(stream["linger_time"].as_str().is_some());
- // Validate plugin configuration
assert_eq!(v0["plugin_config_format"].as_str().unwrap(), "json");
assert!(v0["plugin_config"].is_object());
- // Validate version 1 exists and is different
let v1 = &configs[1];
assert_eq!(v1["version"].as_u64().unwrap(), 1);
assert_eq!(v1["key"].as_str().unwrap(), "random");
}
-#[tokio::test]
-async fn test_source_config_by_version_returns_specific_version() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Direct).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_direct.toml")),
+ seed = seeds::connector_stream
+)]
+async fn source_config_by_version_returns_specific_version(
+ harness: &TestHarness,
+ _wiremock: WireMockDirectFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -97,26 +104,28 @@ async fn test_source_config_by_version_returns_specific_version() {
let body: serde_json::Value = response.json().await.unwrap();
assert!(body.is_object(), "Should return JSON object");
- // Validate it's version 1
assert_eq!(body["version"].as_u64().unwrap(), 1);
assert_eq!(body["key"].as_str().unwrap(), "random");
assert!(body["enabled"].as_bool().unwrap());
assert_eq!(body["name"].as_str().unwrap(), "Random source");
- // Validate streams structure
let streams = body["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1);
- assert_eq!(streams[0]["topic"].as_str().unwrap(), "test_topic"); // String for sources
+ assert_eq!(streams[0]["topic"].as_str().unwrap(), "test_topic");
}
-#[tokio::test]
-async fn test_source_active_config_returns_current_version() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Direct).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_direct.toml")),
+ seed = seeds::connector_stream
+)]
+async fn source_active_config_returns_current_version(
+ harness: &TestHarness,
+ _wiremock: WireMockDirectFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -130,27 +139,29 @@ async fn test_source_active_config_returns_current_version() {
let body: serde_json::Value = response.json().await.unwrap();
assert!(body.is_object(), "Should return JSON object");
- // Validate it's the active version (version 1)
assert_eq!(body["version"].as_u64().unwrap(), 1);
assert_eq!(body["key"].as_str().unwrap(), "random");
assert!(body["enabled"].as_bool().unwrap());
assert_eq!(body["name"].as_str().unwrap(), "Random source");
- // Validate complete structure
let streams = body["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1);
assert_eq!(body["plugin_config_format"].as_str().unwrap(), "json");
assert!(body["plugin_config"].is_object());
}
-#[tokio::test]
-async fn test_sink_configs_list_returns_all_versions() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Direct).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_direct.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sink_configs_list_returns_all_versions(
+ harness: &TestHarness,
+ _wiremock: WireMockDirectFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -166,7 +177,6 @@ async fn test_sink_configs_list_returns_all_versions() {
let configs = body.as_array().unwrap();
assert_eq!(configs.len(), 1, "Should have 1 config version");
- // Validate version 0 schema
let v0 = &configs[0];
assert_eq!(v0["key"].as_str().unwrap(), "stdout");
assert!(v0["enabled"].as_bool().unwrap());
@@ -179,34 +189,35 @@ async fn test_sink_configs_list_returns_all_versions() {
.contains("libiggy_connector_stdout_sink")
);
- // Validate streams array (sinks have StreamConsumerConfig)
let streams = v0["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1, "Should have 1 stream config");
let stream = &streams[0];
assert_eq!(stream["stream"].as_str().unwrap(), "test_stream");
- // CRITICAL: Sinks have topics as ARRAY, not string!
let topics = stream["topics"].as_array().unwrap();
assert_eq!(topics.len(), 1);
assert_eq!(topics[0].as_str().unwrap(), "test_topic");
assert_eq!(stream["schema"].as_str().unwrap(), "json");
assert!(stream["batch_length"].as_u64().is_some());
- assert!(stream["poll_interval"].as_str().is_some()); // Sink-specific
- assert!(stream["consumer_group"].as_str().is_some()); // Sink-specific
+ assert!(stream["poll_interval"].as_str().is_some());
+ assert!(stream["consumer_group"].as_str().is_some());
- // Validate plugin configuration (can be null)
assert!(v0["plugin_config_format"].is_null());
assert!(v0["plugin_config"].is_object());
}
-#[tokio::test]
-async fn test_sink_config_by_version_returns_specific_version() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Direct).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_direct.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sink_config_by_version_returns_specific_version(
+ harness: &TestHarness,
+ _wiremock: WireMockDirectFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -220,27 +231,29 @@ async fn test_sink_config_by_version_returns_specific_version() {
let body: serde_json::Value = response.json().await.unwrap();
assert!(body.is_object(), "Should return JSON object");
- // Validate it's version 0
assert_eq!(body["version"].as_u64().unwrap(), 0);
assert_eq!(body["key"].as_str().unwrap(), "stdout");
assert!(body["enabled"].as_bool().unwrap());
assert_eq!(body["name"].as_str().unwrap(), "Stdout sink");
- // Validate streams structure (topics as array for sinks)
let streams = body["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1);
let topics = streams[0]["topics"].as_array().unwrap();
assert_eq!(topics[0].as_str().unwrap(), "test_topic");
}
-#[tokio::test]
-async fn test_sink_active_config_returns_current_version() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Direct).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_direct.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sink_active_config_returns_current_version(
+ harness: &TestHarness,
+ _wiremock: WireMockDirectFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -254,13 +267,11 @@ async fn test_sink_active_config_returns_current_version() {
let body: serde_json::Value = response.json().await.unwrap();
assert!(body.is_object(), "Should return JSON object");
- // Validate it's the active version (version 0, only one available)
assert_eq!(body["version"].as_u64().unwrap(), 0);
assert_eq!(body["key"].as_str().unwrap(), "stdout");
assert!(body["enabled"].as_bool().unwrap());
assert_eq!(body["name"].as_str().unwrap(), "Stdout sink");
- // Validate complete structure
let streams = body["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1);
assert!(body["plugin_config"].is_object());
diff --git a/core/integration/tests/connectors/http_config_provider/mod.rs b/core/integration/tests/connectors/http_config_provider/mod.rs
index 60d04e351..31e5235c2 100644
--- a/core/integration/tests/connectors/http_config_provider/mod.rs
+++ b/core/integration/tests/connectors/http_config_provider/mod.rs
@@ -19,85 +19,3 @@
mod direct_responses;
mod wrapped_responses;
-
-use crate::connectors::{ConnectorsRuntime, IggySetup, setup_runtime};
-use std::collections::HashMap;
-use strum_macros::Display;
-use testcontainers_modules::testcontainers::core::WaitFor::Healthcheck;
-use testcontainers_modules::testcontainers::core::wait::HealthWaitStrategy;
-use testcontainers_modules::testcontainers::core::{IntoContainerPort, Mount};
-use testcontainers_modules::testcontainers::runners::AsyncRunner;
-use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt};
-
-async fn setup(mode: WireMockMode) -> WiremockConnectorsRuntime {
- let iggy_setup = IggySetup::default();
- let mut runtime = setup_runtime();
-
- let container = create_wiremock_container(mode).await;
- let base_url = get_base_url(&container).await;
-
- let mut envs = HashMap::new();
- envs.insert("IGGY_CONNECTORS_CONNECTORS_BASE_URL".to_owned(), base_url);
-
- let config_path = match mode {
- WireMockMode::Direct => "http_config_provider/config_direct.toml",
- WireMockMode::Wrapped => "http_config_provider/config_wrapped.toml",
- };
-
- runtime.init(config_path, Some(envs), iggy_setup).await;
- WiremockConnectorsRuntime {
- connectors_runtime: runtime,
- wiremock_container: container,
- }
-}
-
-async fn get_base_url(container: &ContainerAsync<GenericImage>) -> String {
- let host = container
- .get_host()
- .await
- .expect("WireMock container should have a host");
- let host_port = container
- .get_host_port_ipv4(8080)
- .await
- .expect("WireMock container should have a port");
- format!("http://{host}:{host_port}")
-}
-
-async fn create_wiremock_container(mode: WireMockMode) -> ContainerAsync<GenericImage> {
- let current_dir = std::env::current_dir().unwrap();
- GenericImage::new("wiremock/wiremock", "latest")
- .with_exposed_port(8080.tcp())
- .with_wait_for(Healthcheck(HealthWaitStrategy::default()))
- .with_mount(Mount::bind_mount(
- current_dir
- .join(format!(
- "tests/connectors/http_config_provider/wiremock/mappings/{mode}"
- ))
- .to_string_lossy()
- .to_string(),
- "/home/wiremock/mappings",
- ))
- .with_mount(Mount::bind_mount(
- current_dir
- .join("tests/connectors/http_config_provider/wiremock/__files")
- .to_string_lossy()
- .to_string(),
- "/home/wiremock/__files",
- ))
- .start()
- .await
- .expect("WireMock container should be started")
-}
-
-#[derive(Display, Clone, Copy)]
-#[strum(serialize_all = "lowercase")]
-enum WireMockMode {
- Direct,
- Wrapped,
-}
-
-struct WiremockConnectorsRuntime {
- connectors_runtime: ConnectorsRuntime,
- #[allow(dead_code)]
- wiremock_container: ContainerAsync<GenericImage>,
-}
diff --git a/core/integration/tests/connectors/http_config_provider/wrapped_responses.rs b/core/integration/tests/connectors/http_config_provider/wrapped_responses.rs
index 4ec3a93e8..1ef5f99ae 100644
--- a/core/integration/tests/connectors/http_config_provider/wrapped_responses.rs
+++ b/core/integration/tests/connectors/http_config_provider/wrapped_responses.rs
@@ -16,17 +16,24 @@
* specific language governing permissions and limitations
* under the License.
*/
-use crate::connectors::http_config_provider::WireMockMode;
+
+use crate::connectors::fixtures::WireMockWrappedFixture;
+use integration::harness::seeds;
+use integration::iggy_harness;
use reqwest::StatusCode;
-#[tokio::test]
-async fn test_source_configs_list_returns_all_versions() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Wrapped).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_wrapped.toml")),
+ seed = seeds::connector_stream
+)]
+async fn source_configs_list_returns_all_versions(
+ harness: &TestHarness,
+ _wiremock: WireMockWrappedFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -37,14 +44,11 @@ async fn test_source_configs_list_returns_all_versions() {
assert_eq!(response.status(), StatusCode::OK);
- // Note: The connector runtime API always returns direct format,
- // even when it fetches configs from a wrapped-format provider
let body: serde_json::Value = response.json().await.unwrap();
assert!(body.is_array(), "Should return JSON array");
let configs = body.as_array().unwrap();
assert_eq!(configs.len(), 2, "Should have 2 config versions");
- // Validate version 0 schema
let v0 = &configs[0];
assert_eq!(v0["key"].as_str().unwrap(), "random");
assert!(v0["enabled"].as_bool().unwrap());
@@ -57,35 +61,36 @@ async fn test_source_configs_list_returns_all_versions() {
.contains("libiggy_connector_random_source")
);
- // Validate streams array (sources have StreamProducerConfig)
let streams = v0["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1, "Should have 1 stream config");
let stream = &streams[0];
assert_eq!(stream["stream"].as_str().unwrap(), "test_stream");
- assert_eq!(stream["topic"].as_str().unwrap(), "test_topic"); // String, not array!
+ assert_eq!(stream["topic"].as_str().unwrap(), "test_topic");
assert_eq!(stream["schema"].as_str().unwrap(), "json");
assert!(stream["batch_length"].as_u64().is_some());
- assert!(stream["linger_time"].as_str().is_some()); // Source-specific field
+ assert!(stream["linger_time"].as_str().is_some());
- // Validate plugin configuration
assert_eq!(v0["plugin_config_format"].as_str().unwrap(), "json");
assert!(v0["plugin_config"].is_object());
- // Validate version 1 exists and is different
let v1 = &configs[1];
assert_eq!(v1["version"].as_u64().unwrap(), 1);
assert_eq!(v1["key"].as_str().unwrap(), "random");
}
-#[tokio::test]
-async fn test_source_config_by_version_returns_specific_version() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Wrapped).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_wrapped.toml")),
+ seed = seeds::connector_stream
+)]
+async fn source_config_by_version_returns_specific_version(
+ harness: &TestHarness,
+ _wiremock: WireMockWrappedFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -99,26 +104,28 @@ async fn test_source_config_by_version_returns_specific_version() {
let body: serde_json::Value = response.json().await.unwrap();
assert!(body.is_object(), "Should return JSON object");
- // Validate it's version 1
assert_eq!(body["version"].as_u64().unwrap(), 1);
assert_eq!(body["key"].as_str().unwrap(), "random");
assert!(body["enabled"].as_bool().unwrap());
assert_eq!(body["name"].as_str().unwrap(), "Random source");
- // Validate streams structure
let streams = body["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1);
- assert_eq!(streams[0]["topic"].as_str().unwrap(), "test_topic"); // String for sources
+ assert_eq!(streams[0]["topic"].as_str().unwrap(), "test_topic");
}
-#[tokio::test]
-async fn test_source_active_config_returns_current_version() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Wrapped).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_wrapped.toml")),
+ seed = seeds::connector_stream
+)]
+async fn source_active_config_returns_current_version(
+ harness: &TestHarness,
+ _wiremock: WireMockWrappedFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -132,27 +139,29 @@ async fn test_source_active_config_returns_current_version() {
let body: serde_json::Value = response.json().await.unwrap();
assert!(body.is_object(), "Should return JSON object");
- // Validate it's the active version (version 1)
assert_eq!(body["version"].as_u64().unwrap(), 1);
assert_eq!(body["key"].as_str().unwrap(), "random");
assert!(body["enabled"].as_bool().unwrap());
assert_eq!(body["name"].as_str().unwrap(), "Random source");
- // Validate complete structure
let streams = body["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1);
assert_eq!(body["plugin_config_format"].as_str().unwrap(), "json");
assert!(body["plugin_config"].is_object());
}
-#[tokio::test]
-async fn test_sink_configs_list_returns_all_versions() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Wrapped).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_wrapped.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sink_configs_list_returns_all_versions(
+ harness: &TestHarness,
+ _wiremock: WireMockWrappedFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -168,7 +177,6 @@ async fn test_sink_configs_list_returns_all_versions() {
let configs = body.as_array().unwrap();
assert_eq!(configs.len(), 1, "Should have 1 config version");
- // Validate version 0 schema
let v0 = &configs[0];
assert_eq!(v0["key"].as_str().unwrap(), "stdout");
assert!(v0["enabled"].as_bool().unwrap());
@@ -181,34 +189,35 @@ async fn test_sink_configs_list_returns_all_versions() {
.contains("libiggy_connector_stdout_sink")
);
- // Validate streams array (sinks have StreamConsumerConfig)
let streams = v0["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1, "Should have 1 stream config");
let stream = &streams[0];
assert_eq!(stream["stream"].as_str().unwrap(), "test_stream");
- // CRITICAL: Sinks have topics as ARRAY, not string!
let topics = stream["topics"].as_array().unwrap();
assert_eq!(topics.len(), 1);
assert_eq!(topics[0].as_str().unwrap(), "test_topic");
assert_eq!(stream["schema"].as_str().unwrap(), "json");
assert!(stream["batch_length"].as_u64().is_some());
- assert!(stream["poll_interval"].as_str().is_some()); // Sink-specific
- assert!(stream["consumer_group"].as_str().is_some()); // Sink-specific
+ assert!(stream["poll_interval"].as_str().is_some());
+ assert!(stream["consumer_group"].as_str().is_some());
- // Validate plugin configuration (can be null)
assert!(v0["plugin_config_format"].is_null());
assert!(v0["plugin_config"].is_object());
}
-#[tokio::test]
-async fn test_sink_config_by_version_returns_specific_version() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Wrapped).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_wrapped.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sink_config_by_version_returns_specific_version(
+ harness: &TestHarness,
+ _wiremock: WireMockWrappedFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -222,27 +231,29 @@ async fn test_sink_config_by_version_returns_specific_version() {
let body: serde_json::Value = response.json().await.unwrap();
assert!(body.is_object(), "Should return JSON object");
- // Validate it's version 0
assert_eq!(body["version"].as_u64().unwrap(), 0);
assert_eq!(body["key"].as_str().unwrap(), "stdout");
assert!(body["enabled"].as_bool().unwrap());
assert_eq!(body["name"].as_str().unwrap(), "Stdout sink");
- // Validate streams structure (topics as array for sinks)
let streams = body["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1);
let topics = streams[0]["topics"].as_array().unwrap();
assert_eq!(topics[0].as_str().unwrap(), "test_topic");
}
-#[tokio::test]
-async fn test_sink_active_config_returns_current_version() {
- let wiremock_runtime =
- crate::connectors::http_config_provider::setup(WireMockMode::Wrapped).await;
- let http_api_address = wiremock_runtime
- .connectors_runtime
- .connectors_api_address()
- .expect("connector runtime should be available");
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/http_config_provider/config_wrapped.toml")),
+ seed = seeds::connector_stream
+)]
+async fn sink_active_config_returns_current_version(
+ harness: &TestHarness,
+ _wiremock: WireMockWrappedFixture,
+) {
+ let http_api_address = harness
+ .connectors_runtime()
+ .expect("connector runtime should be available")
+ .http_url();
let client = reqwest::Client::new();
let response = client
@@ -256,13 +267,11 @@ async fn test_sink_active_config_returns_current_version() {
let body: serde_json::Value = response.json().await.unwrap();
assert!(body.is_object(), "Should return JSON object");
- // Validate it's the active version (version 0, only one available)
assert_eq!(body["version"].as_u64().unwrap(), 0);
assert_eq!(body["key"].as_str().unwrap(), "stdout");
assert!(body["enabled"].as_bool().unwrap());
assert_eq!(body["name"].as_str().unwrap(), "Stdout sink");
- // Validate complete structure
let streams = body["streams"].as_array().unwrap();
assert_eq!(streams.len(), 1);
assert!(body["plugin_config"].is_object());
diff --git a/core/integration/tests/connectors/mod.rs b/core/integration/tests/connectors/mod.rs
index 621b36d4d..deb819797 100644
--- a/core/integration/tests/connectors/mod.rs
+++ b/core/integration/tests/connectors/mod.rs
@@ -17,31 +17,16 @@
* under the License.
*/
-use iggy::prelude::{
- Client, DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, IggyClient, IggyMessage, Partitioning,
-};
-use iggy_binary_protocol::{MessageClient, StreamClient, TopicClient, UserClient};
-use iggy_common::{
- CompressionAlgorithm, Consumer, Identifier, IggyExpiry, IggyTimestamp, MaxTopicSize,
- PolledMessages, PollingStrategy,
-};
-use integration::{
- tcp_client::TcpClientFactory,
- test_connectors_runtime::TestConnectorsRuntime,
- test_server::{ClientFactory, IpAddrKind, TestServer},
-};
-use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-
mod api;
+mod fixtures;
mod http_config_provider;
mod postgres;
mod quickwit;
mod random;
-pub const DEFAULT_TEST_STREAM: &str = "test_stream";
-pub const DEFAULT_TEST_TOPIC: &str = "test_topic";
-const TEST_CONSUMER_NAME: &str = "test_consumer";
+use iggy_common::IggyTimestamp;
+use serde::{Deserialize, Serialize};
+
const ONE_DAY_MICROS: u64 = 24 * 60 * 60 * 1_000_000;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -54,32 +39,7 @@ pub struct TestMessage {
pub timestamp: i64,
}
-pub struct IggySetup {
- pub stream: String,
- pub topic: String,
-}
-
-impl Default for IggySetup {
- fn default() -> Self {
- Self {
- stream: DEFAULT_TEST_STREAM.to_owned(),
- topic: DEFAULT_TEST_TOPIC.to_owned(),
- }
- }
-}
-
-struct ConnectorsRuntime {
- iggy_server: TestServer,
- connectors_runtime: Option<TestConnectorsRuntime>,
-}
-
-struct ConnectorsIggyClient {
- stream_id: Identifier,
- topic_id: Identifier,
- client: IggyClient,
-}
-
-fn create_test_messages(count: usize) -> Vec<TestMessage> {
+pub fn create_test_messages(count: usize) -> Vec<TestMessage> {
let base_timestamp = IggyTimestamp::now().as_micros();
(1..=count)
.map(|i| TestMessage {
@@ -92,143 +52,3 @@ fn create_test_messages(count: usize) -> Vec<TestMessage> {
})
.collect()
}
-
-fn setup_runtime() -> ConnectorsRuntime {
- let mut iggy_envs = HashMap::new();
- iggy_envs.insert("IGGY_QUIC_ENABLED".to_owned(), "false".to_owned());
- iggy_envs.insert("IGGY_WEBSOCKET_ENABLED".to_owned(), "false".to_owned());
- let mut test_server = TestServer::new(Some(iggy_envs), true, None, IpAddrKind::V4);
- test_server.start();
- ConnectorsRuntime {
- iggy_server: test_server,
- connectors_runtime: None,
- }
-}
-
-impl ConnectorsRuntime {
- async fn init(
- &mut self,
- config_path: &str,
- envs: Option<HashMap<String, String>>,
- iggy_setup: IggySetup,
- ) {
- if self.connectors_runtime.is_some() {
- return;
- }
-
- let config_path = format!("tests/connectors/{config_path}");
- let mut all_envs = HashMap::new();
- all_envs.insert(
- "IGGY_CONNECTORS_CONFIG_PATH".to_owned(),
- config_path.to_owned(),
- );
-
- if let Some(envs) = envs {
- for (k, v) in envs {
- all_envs.insert(k, v);
- }
- }
-
- let client = self.create_iggy_client().await;
- client
- .create_stream(&iggy_setup.stream)
- .await
- .expect("Failed to create stream");
- let stream_id: Identifier = iggy_setup
- .stream
- .try_into()
- .expect("Failed to parse stream identifier");
- client
- .create_topic(
- &stream_id,
- &iggy_setup.topic,
- 1,
- CompressionAlgorithm::None,
- None,
- IggyExpiry::ServerDefault,
- MaxTopicSize::ServerDefault,
- )
- .await
- .expect("Failed to create topic");
- client.shutdown().await.expect("Failed to shutdown client");
-
- let iggy_server_address = self
- .iggy_server
- .get_raw_tcp_addr()
- .expect("Failed to get Iggy TCP address");
- let mut connectors_runtime =
- TestConnectorsRuntime::with_iggy_address(&iggy_server_address, Some(all_envs));
- connectors_runtime.start();
- connectors_runtime.ensure_started().await;
- self.connectors_runtime = Some(connectors_runtime);
- }
-
- async fn create_client(&self) -> ConnectorsIggyClient {
- let stream_id: Identifier = DEFAULT_TEST_STREAM
- .try_into()
- .expect("Failed to parse stream identifier");
- let topic_id: Identifier = DEFAULT_TEST_TOPIC
- .try_into()
- .expect("Failed to parse topic identifier");
- ConnectorsIggyClient {
- stream_id,
- topic_id,
- client: self.create_iggy_client().await,
- }
- }
-
- async fn create_iggy_client(&self) -> IggyClient {
- let server_addr = self
- .iggy_server
- .get_raw_tcp_addr()
- .expect("Failed to get Iggy TCP address");
- let client = TcpClientFactory {
- server_addr,
- ..Default::default()
- }
- .create_client()
- .await;
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .expect("Failed to login as root user");
- IggyClient::create(client, None, None)
- }
-
- fn connectors_api_address(&self) -> Option<String> {
- self.connectors_runtime
- .as_ref()
- .map(|r| r.get_http_api_address())
- }
-}
-
-impl ConnectorsIggyClient {
- async fn send_messages(&self, messages: &mut [IggyMessage]) {
- self.client
- .send_messages(
- &self.stream_id,
- &self.topic_id,
- &Partitioning::partition_id(0),
- messages,
- )
- .await
- .expect("Failed to send messages to Iggy");
- }
-
- async fn poll_messages(&self) -> Result<PolledMessages, iggy_common::IggyError> {
- let consumer_id: Identifier = TEST_CONSUMER_NAME
- .try_into()
- .expect("Failed to parse consumer identifier");
- self.client
- .poll_messages(
- &self.stream_id,
- &self.topic_id,
- None,
- &Consumer::new(consumer_id),
- &PollingStrategy::next(),
- 10,
- true,
- )
- .await
- }
-}
diff --git a/core/integration/tests/connectors/postgres/mod.rs b/core/integration/tests/connectors/postgres/mod.rs
index e6e41bbb2..cb16dfbfb 100644
--- a/core/integration/tests/connectors/postgres/mod.rs
+++ b/core/integration/tests/connectors/postgres/mod.rs
@@ -20,71 +20,12 @@
mod postgres_sink;
mod postgres_source;
-use crate::connectors::{
- ConnectorsRuntime, DEFAULT_TEST_STREAM, DEFAULT_TEST_TOPIC, IggySetup, TestMessage,
- setup_runtime,
-};
+use crate::connectors::TestMessage;
use serde::Deserialize;
-use sqlx::postgres::PgPoolOptions;
-use sqlx::{Pool, Postgres};
-use std::collections::HashMap;
-use std::time::Duration;
-use testcontainers_modules::{
- postgres,
- testcontainers::{ContainerAsync, runners::AsyncRunner},
-};
-use tokio::time::sleep;
-const SOURCE_TABLE: &str = "test_messages";
-const SOURCE_TABLE_BYTEA: &str = "test_payloads";
-const SOURCE_TABLE_JSON: &str = "test_json_payloads";
-const SOURCE_TABLE_DELETE: &str = "test_delete_rows";
-const SOURCE_TABLE_MARK: &str = "test_mark_rows";
-const SINK_TABLE: &str = "iggy_messages";
-const TEST_STREAM: &str = DEFAULT_TEST_STREAM;
-const TEST_TOPIC: &str = DEFAULT_TEST_TOPIC;
const TEST_MESSAGE_COUNT: usize = 3;
-const POSTGRES_PORT: u16 = 5432;
const POLL_ATTEMPTS: usize = 100;
const POLL_INTERVAL_MS: u64 = 50;
-const FETCH_INTERVAL_MS: u64 = 10;
-const TABLE_WAIT_INTERVAL_MS: u64 = 50;
-const ENV_SINK_CONNECTION_STRING: &str =
- "IGGY_CONNECTORS_SINK_POSTGRES_PLUGIN_CONFIG_CONNECTION_STRING";
-const ENV_SINK_TARGET_TABLE: &str = "IGGY_CONNECTORS_SINK_POSTGRES_PLUGIN_CONFIG_TARGET_TABLE";
-const ENV_SINK_PAYLOAD_FORMAT: &str = "IGGY_CONNECTORS_SINK_POSTGRES_PLUGIN_CONFIG_PAYLOAD_FORMAT";
-const ENV_SINK_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_STREAM";
-const ENV_SINK_STREAMS_0_TOPICS: &str = "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_TOPICS";
-const ENV_SINK_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_SCHEMA";
-const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str =
- "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_CONSUMER_GROUP";
-const ENV_SOURCE_CONNECTION_STRING: &str =
- "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_CONNECTION_STRING";
-const ENV_SOURCE_TABLES: &str = "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_TABLES";
-const ENV_SOURCE_TRACKING_COLUMN: &str =
- "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_TRACKING_COLUMN";
-const ENV_SOURCE_PAYLOAD_COLUMN: &str =
- "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_PAYLOAD_COLUMN";
-const ENV_SOURCE_PAYLOAD_FORMAT: &str =
- "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_PAYLOAD_FORMAT";
-const ENV_SOURCE_DELETE_AFTER_READ: &str =
- "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_DELETE_AFTER_READ";
-const ENV_SOURCE_PRIMARY_KEY_COLUMN: &str =
- "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_PRIMARY_KEY_COLUMN";
-const ENV_SOURCE_PROCESSED_COLUMN: &str =
- "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_PROCESSED_COLUMN";
-const ENV_SOURCE_INCLUDE_METADATA: &str =
- "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_INCLUDE_METADATA";
-const ENV_SOURCE_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SOURCE_POSTGRES_STREAMS_0_STREAM";
-const ENV_SOURCE_STREAMS_0_TOPIC: &str = "IGGY_CONNECTORS_SOURCE_POSTGRES_STREAMS_0_TOPIC";
-const ENV_SOURCE_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SOURCE_POSTGRES_STREAMS_0_SCHEMA";
-const ENV_SOURCE_POLL_INTERVAL: &str =
- "IGGY_CONNECTORS_SOURCE_POSTGRES_PLUGIN_CONFIG_POLL_INTERVAL";
-const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_POSTGRES_PATH";
-const ENV_SOURCE_PATH: &str = "IGGY_CONNECTORS_SOURCE_POSTGRES_PATH";
-
-type SinkRow = (i64, String, String, Vec<u8>);
-type SinkJsonRow = (i64, serde_json::Value);
#[derive(Debug, Deserialize)]
struct DatabaseRecord {
@@ -92,444 +33,3 @@ struct DatabaseRecord {
operation_type: String,
data: TestMessage,
}
-
-struct PostgresTestSetup {
- runtime: ConnectorsRuntime,
- connection_string: String,
- #[allow(dead_code)]
- container: ContainerAsync<postgres::Postgres>,
-}
-
-async fn start_container() -> (ContainerAsync<postgres::Postgres>, String) {
- let container = postgres::Postgres::default()
- .start()
- .await
- .expect("Failed to start Postgres container");
- let host_port = container
- .get_host_port_ipv4(POSTGRES_PORT)
- .await
- .expect("Failed to get Postgres port");
- let connection_string = format!("postgres://postgres:postgres@localhost:{host_port}");
- (container, connection_string)
-}
-
-async fn create_pool(connection_string: &str) -> Pool<Postgres> {
- PgPoolOptions::new()
- .max_connections(1)
- .connect(connection_string)
- .await
- .expect("Failed to connect to PostgreSQL")
-}
-
-async fn setup_sink() -> PostgresTestSetup {
- let mut envs = HashMap::new();
- envs.insert(ENV_SINK_STREAMS_0_SCHEMA.to_owned(), "json".to_owned());
- setup_sink_with_envs(envs).await
-}
-
-async fn setup_sink_bytea() -> PostgresTestSetup {
- let mut envs = HashMap::new();
- envs.insert(ENV_SINK_STREAMS_0_SCHEMA.to_owned(), "raw".to_owned());
- envs.insert(ENV_SINK_PAYLOAD_FORMAT.to_owned(), "bytea".to_owned());
- setup_sink_with_envs(envs).await
-}
-
-async fn setup_sink_json() -> PostgresTestSetup {
- let mut envs = HashMap::new();
- envs.insert(ENV_SINK_STREAMS_0_SCHEMA.to_owned(), "json".to_owned());
- envs.insert(ENV_SINK_PAYLOAD_FORMAT.to_owned(), "json".to_owned());
- setup_sink_with_envs(envs).await
-}
-
-async fn setup_sink_with_envs(mut extra_envs: HashMap<String, String>) -> PostgresTestSetup {
- let (container, connection_string) = start_container().await;
-
- extra_envs.insert(
- ENV_SINK_CONNECTION_STRING.to_owned(),
- connection_string.clone(),
- );
- extra_envs.insert(ENV_SINK_TARGET_TABLE.to_owned(), SINK_TABLE.to_owned());
- extra_envs.insert(ENV_SINK_STREAMS_0_STREAM.to_owned(), TEST_STREAM.to_owned());
- extra_envs.insert(
- ENV_SINK_STREAMS_0_TOPICS.to_owned(),
- format!("[{TEST_TOPIC}]"),
- );
- extra_envs.insert(
- ENV_SINK_STREAMS_0_CONSUMER_GROUP.to_owned(),
- "test".to_owned(),
- );
- extra_envs.insert(
- ENV_SINK_PATH.to_owned(),
- "../../target/debug/libiggy_connector_postgres_sink".to_owned(),
- );
-
- let mut runtime = setup_runtime();
- runtime
- .init("postgres/sink.toml", Some(extra_envs), IggySetup::default())
- .await;
-
- PostgresTestSetup {
- runtime,
- connection_string,
- container,
- }
-}
-
-async fn setup_source() -> PostgresTestSetup {
- let (container, connection_string) = start_container().await;
- let pool = create_pool(&connection_string).await;
-
- sqlx::query(&format!(
- "CREATE TABLE IF NOT EXISTS {SOURCE_TABLE} (
- id SERIAL PRIMARY KEY,
- name VARCHAR(255) NOT NULL,
- count INTEGER NOT NULL,
- amount DOUBLE PRECISION NOT NULL,
- active BOOLEAN NOT NULL,
- timestamp BIGINT NOT NULL
- )"
- ))
- .execute(&pool)
- .await
- .expect("Failed to create source table");
- pool.close().await;
-
- let mut envs = HashMap::new();
- envs.insert(ENV_SOURCE_TABLES.to_owned(), format!("[{SOURCE_TABLE}]"));
- envs.insert(ENV_SOURCE_INCLUDE_METADATA.to_owned(), "true".to_owned());
- envs.insert(ENV_SOURCE_STREAMS_0_SCHEMA.to_owned(), "json".to_owned());
-
- setup_source_with_envs(container, connection_string, envs).await
-}
-
-async fn setup_source_bytea() -> PostgresTestSetup {
- let (container, connection_string) = start_container().await;
- let pool = create_pool(&connection_string).await;
-
- sqlx::query(&format!(
- "CREATE TABLE IF NOT EXISTS {SOURCE_TABLE_BYTEA} (
- id SERIAL PRIMARY KEY,
- payload BYTEA NOT NULL
- )"
- ))
- .execute(&pool)
- .await
- .expect("Failed to create bytea source table");
- pool.close().await;
-
- let mut envs = HashMap::new();
- envs.insert(
- ENV_SOURCE_TABLES.to_owned(),
- format!("[{SOURCE_TABLE_BYTEA}]"),
- );
- envs.insert(ENV_SOURCE_PAYLOAD_COLUMN.to_owned(), "payload".to_owned());
- envs.insert(ENV_SOURCE_PAYLOAD_FORMAT.to_owned(), "bytea".to_owned());
- envs.insert(ENV_SOURCE_STREAMS_0_SCHEMA.to_owned(), "raw".to_owned());
-
- setup_source_with_envs(container, connection_string, envs).await
-}
-
-async fn setup_source_json() -> PostgresTestSetup {
- let (container, connection_string) = start_container().await;
- let pool = create_pool(&connection_string).await;
-
- sqlx::query(&format!(
- "CREATE TABLE IF NOT EXISTS {SOURCE_TABLE_JSON} (
- id SERIAL PRIMARY KEY,
- data JSONB NOT NULL
- )"
- ))
- .execute(&pool)
- .await
- .expect("Failed to create json source table");
- pool.close().await;
-
- let mut envs = HashMap::new();
- envs.insert(
- ENV_SOURCE_TABLES.to_owned(),
- format!("[{SOURCE_TABLE_JSON}]"),
- );
- envs.insert(ENV_SOURCE_PAYLOAD_COLUMN.to_owned(), "data".to_owned());
- envs.insert(
- ENV_SOURCE_PAYLOAD_FORMAT.to_owned(),
- "json_direct".to_owned(),
- );
- envs.insert(ENV_SOURCE_STREAMS_0_SCHEMA.to_owned(), "json".to_owned());
-
- setup_source_with_envs(container, connection_string, envs).await
-}
-
-async fn setup_source_delete() -> PostgresTestSetup {
- let (container, connection_string) = start_container().await;
- let pool = create_pool(&connection_string).await;
-
- sqlx::query(&format!(
- "CREATE TABLE IF NOT EXISTS {SOURCE_TABLE_DELETE} (
- id SERIAL PRIMARY KEY,
- name VARCHAR(255) NOT NULL,
- value INTEGER NOT NULL
- )"
- ))
- .execute(&pool)
- .await
- .expect("Failed to create delete source table");
- pool.close().await;
-
- let mut envs = HashMap::new();
- envs.insert(
- ENV_SOURCE_TABLES.to_owned(),
- format!("[{SOURCE_TABLE_DELETE}]"),
- );
- envs.insert(ENV_SOURCE_PRIMARY_KEY_COLUMN.to_owned(), "id".to_owned());
- envs.insert(ENV_SOURCE_DELETE_AFTER_READ.to_owned(), "true".to_owned());
- envs.insert(ENV_SOURCE_INCLUDE_METADATA.to_owned(), "true".to_owned());
- envs.insert(ENV_SOURCE_STREAMS_0_SCHEMA.to_owned(), "json".to_owned());
-
- setup_source_with_envs(container, connection_string, envs).await
-}
-
-async fn setup_source_mark() -> PostgresTestSetup {
- let (container, connection_string) = start_container().await;
- let pool = create_pool(&connection_string).await;
-
- sqlx::query(&format!(
- "CREATE TABLE IF NOT EXISTS {SOURCE_TABLE_MARK} (
- id SERIAL PRIMARY KEY,
- name VARCHAR(255) NOT NULL,
- value INTEGER NOT NULL,
- is_processed BOOLEAN NOT NULL DEFAULT FALSE
- )"
- ))
- .execute(&pool)
- .await
- .expect("Failed to create mark source table");
- pool.close().await;
-
- let mut envs = HashMap::new();
- envs.insert(
- ENV_SOURCE_TABLES.to_owned(),
- format!("[{SOURCE_TABLE_MARK}]"),
- );
- envs.insert(ENV_SOURCE_PRIMARY_KEY_COLUMN.to_owned(), "id".to_owned());
- envs.insert(
- ENV_SOURCE_PROCESSED_COLUMN.to_owned(),
- "is_processed".to_owned(),
- );
- envs.insert(ENV_SOURCE_INCLUDE_METADATA.to_owned(), "true".to_owned());
- envs.insert(ENV_SOURCE_STREAMS_0_SCHEMA.to_owned(), "json".to_owned());
-
- setup_source_with_envs(container, connection_string, envs).await
-}
-
-async fn setup_source_with_envs(
- container: ContainerAsync<postgres::Postgres>,
- connection_string: String,
- mut extra_envs: HashMap<String, String>,
-) -> PostgresTestSetup {
- extra_envs.insert(
- ENV_SOURCE_CONNECTION_STRING.to_owned(),
- connection_string.clone(),
- );
- extra_envs.insert(ENV_SOURCE_TRACKING_COLUMN.to_owned(), "id".to_owned());
- extra_envs.insert(
- ENV_SOURCE_STREAMS_0_STREAM.to_owned(),
- TEST_STREAM.to_owned(),
- );
- extra_envs.insert(ENV_SOURCE_STREAMS_0_TOPIC.to_owned(), TEST_TOPIC.to_owned());
- extra_envs.insert(ENV_SOURCE_POLL_INTERVAL.to_owned(), "10ms".to_owned());
- extra_envs.insert(
- ENV_SOURCE_PATH.to_owned(),
- "../../target/debug/libiggy_connector_postgres_source".to_owned(),
- );
-
- let mut runtime = setup_runtime();
- runtime
- .init(
- "postgres/source.toml",
- Some(extra_envs),
- IggySetup::default(),
- )
- .await;
-
- PostgresTestSetup {
- runtime,
- connection_string,
- container,
- }
-}
-
-impl PostgresTestSetup {
- async fn create_pool(&self) -> Pool<Postgres> {
- create_pool(&self.connection_string).await
- }
-
- async fn wait_for_table(&self, pool: &Pool<Postgres>, table: &str) {
- let query = format!("SELECT 1 FROM {table} LIMIT 1");
- for _ in 0..POLL_ATTEMPTS {
- if sqlx::query(&query).fetch_optional(pool).await.is_ok() {
- return;
- }
- sleep(Duration::from_millis(TABLE_WAIT_INTERVAL_MS)).await;
- }
- panic!("Table {table} was not created in time");
- }
-
- async fn insert_test_messages(&self, pool: &Pool<Postgres>, messages: &[TestMessage]) {
- let query = format!(
- "INSERT INTO {SOURCE_TABLE} (id, name, count, amount, active, timestamp) \
- VALUES ($1, $2, $3, $4, $5, $6)"
- );
- for msg in messages {
- sqlx::query(&query)
- .bind(msg.id as i32)
- .bind(&msg.name)
- .bind(msg.count as i32)
- .bind(msg.amount)
- .bind(msg.active)
- .bind(msg.timestamp)
- .execute(pool)
- .await
- .expect("Failed to insert test message");
- }
- }
-
- async fn insert_bytea_payloads(&self, pool: &Pool<Postgres>, payloads: &[Vec<u8>]) {
- let query = format!("INSERT INTO {SOURCE_TABLE_BYTEA} (id, payload) VALUES ($1, $2)");
- for (i, payload) in payloads.iter().enumerate() {
- sqlx::query(&query)
- .bind((i + 1) as i32)
- .bind(payload)
- .execute(pool)
- .await
- .expect("Failed to insert bytea payload");
- }
- }
-
- async fn insert_json_payloads(&self, pool: &Pool<Postgres>, payloads: &[serde_json::Value]) {
- let query = format!("INSERT INTO {SOURCE_TABLE_JSON} (id, data) VALUES ($1, $2)");
- for (i, payload) in payloads.iter().enumerate() {
- sqlx::query(&query)
- .bind((i + 1) as i32)
- .bind(payload)
- .execute(pool)
- .await
- .expect("Failed to insert json payload");
- }
- }
-
- async fn insert_delete_rows(&self, pool: &Pool<Postgres>, count: usize) {
- let query = format!("INSERT INTO {SOURCE_TABLE_DELETE} (name, value) VALUES ($1, $2)");
- for i in 0..count {
- sqlx::query(&query)
- .bind(format!("row_{i}"))
- .bind(i as i32 * 10)
- .execute(pool)
- .await
- .expect("Failed to insert delete row");
- }
- }
-
- async fn insert_mark_rows(&self, pool: &Pool<Postgres>, count: usize) {
- let query = format!(
- "INSERT INTO {SOURCE_TABLE_MARK} (name, value, is_processed) VALUES ($1, $2, $3)"
- );
- for i in 0..count {
- sqlx::query(&query)
- .bind(format!("row_{i}"))
- .bind(i as i32 * 10)
- .bind(false)
- .execute(pool)
- .await
- .expect("Failed to insert mark row");
- }
- }
-
- async fn count_rows(&self, pool: &Pool<Postgres>, table: &str) -> i64 {
- let query = format!("SELECT COUNT(*) FROM {table}");
- let count: (i64,) = sqlx::query_as(&query)
- .fetch_one(pool)
- .await
- .expect("Failed to count rows");
- count.0
- }
-
- async fn count_unprocessed_rows(&self, pool: &Pool<Postgres>, table: &str) -> i64 {
- let query = format!("SELECT COUNT(*) FROM {table} WHERE is_processed = FALSE");
- let count: (i64,) = sqlx::query_as(&query)
- .fetch_one(pool)
- .await
- .expect("Failed to count unprocessed rows");
- count.0
- }
-
- async fn count_processed_rows(&self, pool: &Pool<Postgres>, table: &str) -> i64 {
- let query = format!("SELECT COUNT(*) FROM {table} WHERE is_processed = TRUE");
- let count: (i64,) = sqlx::query_as(&query)
- .fetch_one(pool)
- .await
- .expect("Failed to count processed rows");
- count.0
- }
-
- async fn fetch_sink_rows(&self, pool: &Pool<Postgres>, expected_count: usize) -> Vec<SinkRow> {
- let query = format!(
- "SELECT iggy_offset, iggy_stream, iggy_topic, payload \
- FROM {SINK_TABLE} ORDER BY iggy_offset"
- );
- let mut rows = Vec::new();
- for _ in 0..POLL_ATTEMPTS {
- if let Ok(fetched) = sqlx::query_as::<_, SinkRow>(&query).fetch_all(pool).await {
- rows = fetched;
- if rows.len() >= expected_count {
- break;
- }
- }
- sleep(Duration::from_millis(FETCH_INTERVAL_MS)).await;
- }
- rows
- }
-
- async fn fetch_sink_json_rows(
- &self,
- pool: &Pool<Postgres>,
- expected_count: usize,
- ) -> Vec<SinkJsonRow> {
- let query = format!("SELECT iggy_offset, payload FROM {SINK_TABLE} ORDER BY iggy_offset");
- let mut rows = Vec::new();
- for _ in 0..POLL_ATTEMPTS {
- if let Ok(fetched) = sqlx::query_as::<_, SinkJsonRow>(&query)
- .fetch_all(pool)
- .await
- {
- rows = fetched;
- if rows.len() >= expected_count {
- break;
- }
- }
- sleep(Duration::from_millis(FETCH_INTERVAL_MS)).await;
- }
- rows
- }
-
- async fn poll_messages<T, F>(&self, expected_count: usize, transform: F) -> Vec<T>
- where
- F: Fn(&[u8]) -> Option<T>,
- {
- let client = self.runtime.create_client().await;
- let mut received = Vec::new();
- for _ in 0..POLL_ATTEMPTS {
- if let Ok(polled) = client.poll_messages().await {
- for msg in polled.messages {
- if let Some(item) = transform(&msg.payload) {
- received.push(item);
- }
- }
- if received.len() >= expected_count {
- break;
- }
- }
- sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
- }
- received
- }
-}
diff --git a/core/integration/tests/connectors/postgres/postgres_sink.rs b/core/integration/tests/connectors/postgres/postgres_sink.rs
index 5927b7c54..0ca736995 100644
--- a/core/integration/tests/connectors/postgres/postgres_sink.rs
+++ b/core/integration/tests/connectors/postgres/postgres_sink.rs
@@ -17,21 +17,35 @@
* under the License.
*/
-use super::{
- SINK_TABLE, TEST_MESSAGE_COUNT, TEST_STREAM, TEST_TOPIC, setup_sink, setup_sink_bytea,
- setup_sink_json,
+use super::TEST_MESSAGE_COUNT;
+use crate::connectors::fixtures::{
+ PostgresOps, PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture,
};
use crate::connectors::{TestMessage, create_test_messages};
use bytes::Bytes;
-use iggy::prelude::IggyMessage;
+use iggy::prelude::{IggyMessage, Partitioning};
+use iggy_binary_protocol::MessageClient;
+use iggy_common::Identifier;
+use integration::harness::seeds;
+use integration::iggy_harness;
-#[tokio::test]
-async fn given_json_messages_sink_connector_should_store_as_bytea() {
- let setup = setup_sink().await;
- let client = setup.runtime.create_client().await;
- let pool = setup.create_pool().await;
+const SINK_TABLE: &str = "iggy_messages";
- setup.wait_for_table(&pool, SINK_TABLE).await;
+type SinkRow = (i64, String, String, Vec<u8>);
+type SinkJsonRow = (i64, serde_json::Value);
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn json_messages_sink_stores_as_bytea(harness: &TestHarness, fixture: PostgresSinkFixture) {
+ let client = harness.client();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+
+ fixture.wait_for_table(&pool, SINK_TABLE).await;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
let messages_data = create_test_messages(TEST_MESSAGE_COUNT);
let mut messages: Vec<IggyMessage> = messages_data
@@ -47,9 +61,23 @@ async fn given_json_messages_sink_connector_should_store_as_bytea() {
})
.collect();
- client.send_messages(&mut messages).await;
-
- let rows = setup.fetch_sink_rows(&pool, TEST_MESSAGE_COUNT).await;
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("Failed to send messages");
+
+ let query = format!(
+ "SELECT iggy_offset, iggy_stream, iggy_topic, payload FROM {SINK_TABLE} ORDER BY iggy_offset"
+ );
+ let rows: Vec<SinkRow> = fixture
+ .fetch_rows_as(&pool, &query, TEST_MESSAGE_COUNT)
+ .await
+ .expect("Failed to fetch rows");
assert_eq!(
rows.len(),
@@ -59,8 +87,8 @@ async fn given_json_messages_sink_connector_should_store_as_bytea() {
for (i, (offset, stream, topic, payload)) in rows.iter().enumerate() {
assert_eq!(*offset, i as i64, "Offset mismatch at row {i}");
- assert_eq!(stream, TEST_STREAM, "Stream mismatch at row {i}");
- assert_eq!(topic, TEST_TOPIC, "Topic mismatch at row {i}");
+ assert_eq!(stream, seeds::names::STREAM, "Stream mismatch at row {i}");
+ assert_eq!(topic, seeds::names::TOPIC, "Topic mismatch at row {i}");
let stored: TestMessage =
serde_json::from_slice(payload).expect("Failed to deserialize stored payload");
@@ -68,13 +96,21 @@ async fn given_json_messages_sink_connector_should_store_as_bytea() {
}
}
-#[tokio::test]
-async fn given_binary_messages_sink_connector_should_store_as_bytea() {
- let setup = setup_sink_bytea().await;
- let client = setup.runtime.create_client().await;
- let pool = setup.create_pool().await;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn binary_messages_sink_stores_as_bytea(
+ harness: &TestHarness,
+ fixture: PostgresSinkByteaFixture,
+) {
+ let client = harness.client();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
- setup.wait_for_table(&pool, SINK_TABLE).await;
+ fixture.wait_for_table(&pool, SINK_TABLE).await;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
let raw_payloads: Vec<Vec<u8>> = vec![
b"plain text message".to_vec(),
@@ -94,9 +130,23 @@ async fn given_binary_messages_sink_connector_should_store_as_bytea() {
})
.collect();
- client.send_messages(&mut messages).await;
-
- let rows = setup.fetch_sink_rows(&pool, TEST_MESSAGE_COUNT).await;
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("Failed to send messages");
+
+ let query = format!(
+ "SELECT iggy_offset, iggy_stream, iggy_topic, payload FROM {SINK_TABLE} ORDER BY iggy_offset"
+ );
+ let rows: Vec<SinkRow> = fixture
+ .fetch_rows_as(&pool, &query, TEST_MESSAGE_COUNT)
+ .await
+ .expect("Failed to fetch rows");
assert_eq!(
rows.len(),
@@ -110,13 +160,21 @@ async fn given_binary_messages_sink_connector_should_store_as_bytea() {
}
}
-#[tokio::test]
-async fn given_json_messages_sink_connector_should_store_as_jsonb() {
- let setup = setup_sink_json().await;
- let client = setup.runtime.create_client().await;
- let pool = setup.create_pool().await;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn json_messages_sink_stores_as_jsonb(
+ harness: &TestHarness,
+ fixture: PostgresSinkJsonFixture,
+) {
+ let client = harness.client();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
- setup.wait_for_table(&pool, SINK_TABLE).await;
+ fixture.wait_for_table(&pool, SINK_TABLE).await;
+
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
let json_payloads: Vec<serde_json::Value> = vec![
serde_json::json!({"name": "Alice", "age": 30}),
@@ -137,9 +195,21 @@ async fn given_json_messages_sink_connector_should_store_as_jsonb() {
})
.collect();
- client.send_messages(&mut messages).await;
-
- let rows = setup.fetch_sink_json_rows(&pool, TEST_MESSAGE_COUNT).await;
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("Failed to send messages");
+
+ let query = format!("SELECT iggy_offset, payload FROM {SINK_TABLE} ORDER BY iggy_offset");
+ let rows: Vec<SinkJsonRow> = fixture
+ .fetch_rows_as(&pool, &query, TEST_MESSAGE_COUNT)
+ .await
+ .expect("Failed to fetch rows");
assert_eq!(
rows.len(),
diff --git a/core/integration/tests/connectors/postgres/postgres_source.rs b/core/integration/tests/connectors/postgres/postgres_source.rs
index 195e2f258..a1fc15b97 100644
--- a/core/integration/tests/connectors/postgres/postgres_source.rs
+++ b/core/integration/tests/connectors/postgres/postgres_source.rs
@@ -17,26 +17,77 @@
* under the License.
*/
-use super::{
- DatabaseRecord, SOURCE_TABLE, SOURCE_TABLE_DELETE, SOURCE_TABLE_MARK, TEST_MESSAGE_COUNT,
- setup_source, setup_source_bytea, setup_source_delete, setup_source_json, setup_source_mark,
-};
+use super::{DatabaseRecord, POLL_ATTEMPTS, POLL_INTERVAL_MS, TEST_MESSAGE_COUNT};
use crate::connectors::create_test_messages;
+use crate::connectors::fixtures::{
+ PostgresOps, PostgresSourceByteaFixture, PostgresSourceDeleteFixture,
+ PostgresSourceJsonFixture, PostgresSourceJsonbFixture, PostgresSourceMarkFixture,
+ PostgresSourceOps,
+};
+use iggy_binary_protocol::MessageClient;
+use iggy_common::{Consumer, Identifier, PollingStrategy};
+use integration::harness::seeds;
+use integration::iggy_harness;
+use std::time::Duration;
+use tokio::time::sleep;
+
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn json_rows_source_produces_messages_to_iggy(
+ harness: &TestHarness,
+ fixture: PostgresSourceJsonFixture,
+) {
+ let client = harness.client();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
-#[tokio::test]
-async fn given_json_rows_source_connector_should_produce_messages_to_iggy() {
- let setup = setup_source().await;
- let pool = setup.create_pool().await;
let test_messages = create_test_messages(TEST_MESSAGE_COUNT);
-
- setup.insert_test_messages(&pool, &test_messages).await;
+ for msg in &test_messages {
+ fixture
+ .insert_row(
+ &pool,
+ msg.id as i32,
+ &msg.name,
+ msg.count as i32,
+ msg.amount,
+ msg.active,
+ msg.timestamp,
+ )
+ .await;
+ }
pool.close().await;
- let received: Vec<DatabaseRecord> = setup
- .poll_messages(TEST_MESSAGE_COUNT, |payload| {
- serde_json::from_slice(payload).ok()
- })
- .await;
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "test_consumer".try_into().unwrap();
+
+ let mut received: Vec<DatabaseRecord> = Vec::new();
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ for msg in polled.messages {
+ if let Ok(record) = serde_json::from_slice(&msg.payload) {
+ received.push(record);
+ }
+ }
+ if received.len() >= TEST_MESSAGE_COUNT {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
assert!(
received.len() >= TEST_MESSAGE_COUNT,
@@ -46,7 +97,8 @@ async fn given_json_rows_source_connector_should_produce_messages_to_iggy() {
for (i, record) in received.iter().enumerate() {
assert_eq!(
- record.table_name, SOURCE_TABLE,
+ record.table_name,
+ fixture.table_name(),
"Table name mismatch at record {i}"
);
assert_eq!(
@@ -60,10 +112,17 @@ async fn given_json_rows_source_connector_should_produce_messages_to_iggy() {
}
}
-#[tokio::test]
-async fn given_bytea_rows_source_connector_should_produce_raw_messages_to_iggy() {
- let setup = setup_source_bytea().await;
- let pool = setup.create_pool().await;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn bytea_rows_source_produces_raw_messages_to_iggy(
+ harness: &TestHarness,
+ fixture: PostgresSourceByteaFixture,
+) {
+ let client = harness.client();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
let payloads: Vec<Vec<u8>> = vec![
b"hello world".to_vec(),
@@ -72,12 +131,38 @@ async fn given_bytea_rows_source_connector_should_produce_raw_messages_to_iggy()
.expect("Failed to serialize json"),
];
- setup.insert_bytea_payloads(&pool, &payloads).await;
+ for (i, payload) in payloads.iter().enumerate() {
+ fixture.insert_payload(&pool, (i + 1) as i32, payload).await;
+ }
pool.close().await;
- let received: Vec<Vec<u8>> = setup
- .poll_messages(TEST_MESSAGE_COUNT, |payload| Some(payload.to_vec()))
- .await;
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "test_consumer".try_into().unwrap();
+
+ let mut received: Vec<Vec<u8>> = Vec::new();
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ for msg in polled.messages {
+ received.push(msg.payload.to_vec());
+ }
+ if received.len() >= TEST_MESSAGE_COUNT {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
assert!(
received.len() >= TEST_MESSAGE_COUNT,
@@ -90,10 +175,17 @@ async fn given_bytea_rows_source_connector_should_produce_raw_messages_to_iggy()
}
}
-#[tokio::test]
-async fn given_jsonb_rows_source_connector_should_produce_json_messages_to_iggy() {
- let setup = setup_source_json().await;
- let pool = setup.create_pool().await;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn jsonb_rows_source_produces_json_messages_to_iggy(
+ harness: &TestHarness,
+ fixture: PostgresSourceJsonbFixture,
+) {
+ let client = harness.client();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
let json_payloads: Vec<serde_json::Value> = vec![
serde_json::json!({"name": "Alice", "score": 100}),
@@ -101,14 +193,40 @@ async fn given_jsonb_rows_source_connector_should_produce_json_messages_to_iggy(
serde_json::json!({"nested": {"deep": {"value": 42}}}),
];
- setup.insert_json_payloads(&pool, &json_payloads).await;
+ for (i, payload) in json_payloads.iter().enumerate() {
+ fixture.insert_json(&pool, (i + 1) as i32, payload).await;
+ }
pool.close().await;
- let received: Vec<serde_json::Value> = setup
- .poll_messages(TEST_MESSAGE_COUNT, |payload| {
- serde_json::from_slice(payload).ok()
- })
- .await;
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "test_consumer".try_into().unwrap();
+
+ let mut received: Vec<serde_json::Value> = Vec::new();
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ for msg in polled.messages {
+ if let Ok(json) = serde_json::from_slice(&msg.payload) {
+ received.push(json);
+ }
+ }
+ if received.len() >= TEST_MESSAGE_COUNT {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
assert!(
received.len() >= TEST_MESSAGE_COUNT,
@@ -124,24 +242,59 @@ async fn given_jsonb_rows_source_connector_should_produce_json_messages_to_iggy(
}
}
-#[tokio::test]
-async fn given_delete_after_read_source_connector_should_remove_rows_after_producing() {
- let setup = setup_source_delete().await;
- let pool = setup.create_pool().await;
-
- setup.insert_delete_rows(&pool, TEST_MESSAGE_COUNT).await;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn delete_after_read_source_removes_rows_after_producing(
+ harness: &TestHarness,
+ fixture: PostgresSourceDeleteFixture,
+) {
+ let client = harness.client();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+
+ for i in 0..TEST_MESSAGE_COUNT {
+ fixture
+ .insert_row(&pool, &format!("row_{i}"), (i * 10) as i32)
+ .await;
+ }
- let initial_count = setup.count_rows(&pool, SOURCE_TABLE_DELETE).await;
+ let initial_count = fixture.count_rows(&pool).await;
assert_eq!(
initial_count, TEST_MESSAGE_COUNT as i64,
"Expected {TEST_MESSAGE_COUNT} rows before processing"
);
- let received: Vec<serde_json::Value> = setup
- .poll_messages(TEST_MESSAGE_COUNT, |payload| {
- serde_json::from_slice(payload).ok()
- })
- .await;
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "test_consumer".try_into().unwrap();
+
+ let mut received: Vec<serde_json::Value> = Vec::new();
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ for msg in polled.messages {
+ if let Ok(json) = serde_json::from_slice(&msg.payload) {
+ received.push(json);
+ }
+ }
+ if received.len() >= TEST_MESSAGE_COUNT {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
assert!(
received.len() >= TEST_MESSAGE_COUNT,
@@ -149,7 +302,7 @@ async fn given_delete_after_read_source_connector_should_remove_rows_after_produ
received.len()
);
- let final_count = setup.count_rows(&pool, SOURCE_TABLE_DELETE).await;
+ let final_count = fixture.count_rows(&pool).await;
assert_eq!(
final_count, 0,
"Expected 0 rows after delete_after_read, got {final_count}"
@@ -158,15 +311,26 @@ async fn given_delete_after_read_source_connector_should_remove_rows_after_produ
pool.close().await;
}
-#[tokio::test]
-async fn given_processed_column_source_connector_should_mark_rows_after_producing() {
- let setup = setup_source_mark().await;
- let pool = setup.create_pool().await;
-
- setup.insert_mark_rows(&pool, TEST_MESSAGE_COUNT).await;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/postgres/source.toml")),
+ seed = seeds::connector_stream
+)]
+async fn processed_column_source_marks_rows_after_producing(
+ harness: &TestHarness,
+ fixture: PostgresSourceMarkFixture,
+) {
+ let client = harness.client();
+ let pool = fixture.create_pool().await.expect("Failed to create pool");
+ fixture.create_table(&pool).await;
+
+ for i in 0..TEST_MESSAGE_COUNT {
+ fixture
+ .insert_row(&pool, &format!("row_{i}"), (i * 10) as i32)
+ .await;
+ }
- let initial_unprocessed = setup.count_unprocessed_rows(&pool, SOURCE_TABLE_MARK).await;
- let initial_processed = setup.count_processed_rows(&pool, SOURCE_TABLE_MARK).await;
+ let initial_unprocessed = fixture.count_unprocessed(&pool).await;
+ let initial_processed = fixture.count_processed(&pool).await;
assert_eq!(
initial_unprocessed, TEST_MESSAGE_COUNT as i64,
"Expected {TEST_MESSAGE_COUNT} unprocessed rows before processing"
@@ -176,11 +340,35 @@ async fn given_processed_column_source_connector_should_mark_rows_after_producin
"Expected 0 processed rows before processing"
);
- let received: Vec<serde_json::Value> = setup
- .poll_messages(TEST_MESSAGE_COUNT, |payload| {
- serde_json::from_slice(payload).ok()
- })
- .await;
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "test_consumer".try_into().unwrap();
+
+ let mut received: Vec<serde_json::Value> = Vec::new();
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(polled) = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
+ .await
+ {
+ for msg in polled.messages {
+ if let Ok(json) = serde_json::from_slice(&msg.payload) {
+ received.push(json);
+ }
+ }
+ if received.len() >= TEST_MESSAGE_COUNT {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
assert!(
received.len() >= TEST_MESSAGE_COUNT,
@@ -188,8 +376,8 @@ async fn given_processed_column_source_connector_should_mark_rows_after_producin
received.len()
);
- let final_unprocessed = setup.count_unprocessed_rows(&pool, SOURCE_TABLE_MARK).await;
- let final_processed = setup.count_processed_rows(&pool, SOURCE_TABLE_MARK).await;
+ let final_unprocessed = fixture.count_unprocessed(&pool).await;
+ let final_processed = fixture.count_processed(&pool).await;
assert_eq!(
final_unprocessed, 0,
"Expected 0 unprocessed rows after processing, got {final_unprocessed}"
@@ -199,7 +387,7 @@ async fn given_processed_column_source_connector_should_mark_rows_after_producin
"Expected {TEST_MESSAGE_COUNT} processed rows after processing, got {final_processed}"
);
- let total_count = setup.count_rows(&pool, SOURCE_TABLE_MARK).await;
+ let total_count = fixture.count_rows(&pool).await;
assert_eq!(
total_count, TEST_MESSAGE_COUNT as i64,
"Rows should not be deleted, expected {TEST_MESSAGE_COUNT}, got {total_count}"
diff --git a/core/integration/tests/connectors/quickwit/mod.rs b/core/integration/tests/connectors/quickwit/mod.rs
index cfc1f0696..4b031e3b3 100644
--- a/core/integration/tests/connectors/quickwit/mod.rs
+++ b/core/integration/tests/connectors/quickwit/mod.rs
@@ -18,268 +18,3 @@
*/
mod quickwit_sink;
-
-use anyhow::{Result, anyhow};
-use reqwest_middleware::ClientWithMiddleware as HttpClient;
-use reqwest_retry::RetryTransientMiddleware;
-use reqwest_retry::policies::ExponentialBackoff;
-use std::collections::HashMap;
-use tracing::info;
-use uuid::Uuid;
-
-use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor};
-use testcontainers_modules::testcontainers::runners::AsyncRunner;
-use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt};
-
-use crate::connectors::{
- ConnectorsRuntime, DEFAULT_TEST_STREAM, DEFAULT_TEST_TOPIC, IggySetup, setup_runtime,
-};
-
-mod quickwit_container {
- pub static REPOSITORY: &str = "quickwit/quickwit";
- pub static VERSION: &str = "0.8.1";
- pub static LISTENING_IPV4: &str = "0.0.0.0";
- pub static LISTENING_PORT: u16 = 7280;
- pub static READY_MESSAGE: &str = "REST server is ready";
-}
-
-mod quickwit_paths {
- pub static INDEXES: &str = "api/v1/indexes";
-
- pub fn index_ingest(index_id: &str) -> String {
- format!("api/v1/{index_id}/ingest")
- }
-
- pub fn index_search(index_id: &str) -> String {
- format!("api/v1/{index_id}/search")
- }
-}
-
-mod quickwit_responses {
- use serde::Deserialize;
-
- #[derive(Deserialize)]
- pub struct IndexSearch {
- pub hits: Vec<serde_json::Value>,
- pub num_hits: usize,
- #[allow(dead_code)]
- pub elapsed_time_micros: usize,
- }
-}
-
-struct QuickwitTestSetup {
- quickwit: ContainerAsync<GenericImage>,
- runtime: ConnectorsRuntime,
- reqwest: HttpClient,
-}
-
-impl QuickwitTestSetup {
- async fn try_new() -> Result<Self> {
- let quickwit_container = start_quickwit_container().await;
- let connectors_runtime = start_iggy_with_quickwit_connector(&quickwit_container).await?;
- Ok(Self {
- quickwit: quickwit_container,
- runtime: connectors_runtime,
- reqwest: get_http_client_with_retries(),
- })
- }
-
- async fn try_new_with_precreate_index(
- quickwit_container: ContainerAsync<GenericImage>,
- ) -> Result<Self> {
- let http_client = get_http_client_with_retries();
- create_quickwit_test_index(&quickwit_container, &http_client).await?;
- let connectors_runtime = start_iggy_with_quickwit_connector(&quickwit_container).await?;
- Ok(Self {
- quickwit: quickwit_container,
- runtime: connectors_runtime,
- reqwest: http_client,
- })
- }
-
- async fn get_mapped_quickwit_port(&self) -> Result<u16> {
- get_mapped_quickwit_port(&self.quickwit).await
- }
-
- async fn get_quickwit_test_index_all_search(&self) -> Result<quickwit_responses::IndexSearch> {
- let match_all = "";
- let descending = "-";
- let search_response = self
- .reqwest
- .get(format!(
- "http://localhost:{}/{}",
- self.get_mapped_quickwit_port().await?,
- quickwit_paths::index_search(DEFAULT_TEST_TOPIC)
- ))
- .query(&[("query", match_all)])
- .query(&[("sort_by", format!(r#"{descending}{INDEX_TIMESTAMP_FIELD}"#))])
- .send()
- .await?;
- info!("Received search index response.");
- assert!(search_response.status().is_success());
- search_response
- .json::<quickwit_responses::IndexSearch>()
- .await
- .map_err(Into::into)
- }
-
- async fn flush_quickwit_test_index(&self) -> Result<()> {
- let ingest_response = self
- .reqwest
- .post(format!(
- "http://localhost:{}/{}",
- self.get_mapped_quickwit_port().await?,
- quickwit_paths::index_ingest(DEFAULT_TEST_TOPIC)
- ))
- .query(&[("commit", "force")])
- .json("{}")
- .send()
- .await?;
- info!("Received index ingest response.");
- assert!(ingest_response.status().is_success());
- Ok(())
- }
-}
-
-fn get_http_client_with_retries() -> HttpClient {
- let max_retries = 3;
- let retry_policy = ExponentialBackoff::builder().build_with_max_retries(max_retries);
- reqwest_middleware::ClientBuilder::new(reqwest::Client::new())
- .with(RetryTransientMiddleware::new_with_policy(retry_policy))
- .build()
-}
-
-async fn get_mapped_quickwit_port(
- quickwit_container: &ContainerAsync<GenericImage>,
-) -> Result<u16> {
- let mapped_port = quickwit_container
- .ports()
- .await?
- .map_to_host_port_ipv4(quickwit_container::LISTENING_PORT)
- .ok_or(anyhow!("No mapping for Quickwit port."))?;
- info!("Got ports details from container.");
- Ok(mapped_port)
-}
-
-static INDEX_TIMESTAMP_FIELD: &str = "timestamp";
-
-fn get_quickwit_index_config() -> String {
- format!(
- r#"
- version: 0.8
- index_id: {DEFAULT_TEST_TOPIC}
- doc_mapping:
- mode: strict
- field_mappings:
- - name: id
- type: u64
- - name: name
- type: text
- tokenizer: raw
- - name: count
- type: u64
- - name: amount
- type: f64
- - name: active
- type: bool
- - name: {INDEX_TIMESTAMP_FIELD}
- type: datetime
- input_formats:
- - unix_timestamp
- output_format: unix_timestamp_micros
- fast_precision: microseconds
- fast: true
- timestamp_field: timestamp
- retention:
- period: 7 days
- schedule: daily
- "#
- )
-}
-
-const STREAMS_KEY_PREFIX: &str = "IGGY_CONNECTORS_SINK_QUICKWIT_STREAMS";
-
-fn get_stream_0_overrides() -> HashMap<String, String> {
- HashMap::from([
- (
- format!("{STREAMS_KEY_PREFIX}_0_STREAM"),
- DEFAULT_TEST_STREAM.to_string(),
- ),
- (
- format!("{STREAMS_KEY_PREFIX}_0_TOPICS"),
- format!("[{DEFAULT_TEST_TOPIC}]"),
- ),
- (format!("{STREAMS_KEY_PREFIX}_0_SCHEMA"), "json".to_string()),
- ])
-}
-
-async fn create_quickwit_test_index(
- quickwit_container: &ContainerAsync<GenericImage>,
- http_client: &HttpClient,
-) -> Result<()> {
- let create_response = http_client
- .post(format!(
- "http://localhost:{}/{}",
- get_mapped_quickwit_port(quickwit_container).await?,
- quickwit_paths::INDEXES
- ))
- .header("Content-Type", "application/yaml")
- .body(get_quickwit_index_config())
- .send()
- .await?;
- info!("Received create index response.");
- assert!(create_response.status().is_success());
- Ok(())
-}
-
-static NETWORK_NAME_PREFIX: &str = "iggy-quickwit-sink";
-
-async fn start_quickwit_container() -> ContainerAsync<GenericImage> {
- let unique_network = format!("{NETWORK_NAME_PREFIX}-{}", Uuid::new_v4());
- let random_host = 0;
- let quickwit_container =
- GenericImage::new(quickwit_container::REPOSITORY, quickwit_container::VERSION)
- .with_exposed_port(random_host.tcp())
- .with_wait_for(WaitFor::message_on_stdout(
- quickwit_container::READY_MESSAGE,
- ))
- .with_network(unique_network)
- .with_cmd(["run"])
- .with_env_var("QW_LISTEN_ADDRESS", quickwit_container::LISTENING_IPV4)
- .with_mapped_port(random_host, quickwit_container::LISTENING_PORT.tcp())
- .start()
- .await
- .expect("Quickwit started.");
- info!("Started quickwit container.");
- quickwit_container
-}
-
-const PLUGIN_KEY_PREFIX: &str = "IGGY_CONNECTORS_SINK_QUICKWIT_PLUGIN_CONFIG";
-
-async fn start_iggy_with_quickwit_connector(
- quickwit_container: &ContainerAsync<GenericImage>,
-) -> Result<ConnectorsRuntime> {
- let mapped_port = get_mapped_quickwit_port(quickwit_container).await?;
-
- let plugin_overrides = HashMap::from([
- (
- format!("{PLUGIN_KEY_PREFIX}_URL"),
- format!("http://localhost:{mapped_port}"),
- ),
- (
- format!("{PLUGIN_KEY_PREFIX}_INDEX"),
- get_quickwit_index_config(),
- ),
- ]);
- let mut extra_envs = HashMap::new();
- extra_envs.extend(plugin_overrides);
- extra_envs.extend(get_stream_0_overrides());
-
- let mut connectors_runtime = setup_runtime();
- let runtime_config = "quickwit/config.toml";
- connectors_runtime
- .init(runtime_config, Some(extra_envs), IggySetup::default())
- .await;
-
- Ok(connectors_runtime)
-}
diff --git a/core/integration/tests/connectors/quickwit/quickwit_sink.rs b/core/integration/tests/connectors/quickwit/quickwit_sink.rs
index 5c8e02baa..ffa08f6d4 100644
--- a/core/integration/tests/connectors/quickwit/quickwit_sink.rs
+++ b/core/integration/tests/connectors/quickwit/quickwit_sink.rs
@@ -17,137 +17,258 @@
* under the License.
*/
-use anyhow::Result;
+use crate::connectors::create_test_messages;
+use crate::connectors::fixtures::{QuickwitFixture, QuickwitOps, QuickwitPreCreatedFixture};
use bytes::Bytes;
-use iggy_common::IggyMessage;
+use iggy::prelude::{IggyMessage, Partitioning};
+use iggy_binary_protocol::MessageClient;
+use iggy_common::Identifier;
+use integration::harness::seeds;
+use integration::iggy_harness;
use serde::{Deserialize, Serialize};
-use crate::connectors::create_test_messages;
-use crate::connectors::quickwit::{QuickwitTestSetup, start_quickwit_container};
-
-async fn send_test_messages(
- test_setup: &QuickwitTestSetup,
- message_count: usize,
-) -> Result<Vec<Bytes>> {
- let iggy_client = test_setup.runtime.create_client().await;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/quickwit/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_existent_quickwit_index_should_store(
+ harness: &TestHarness,
+ fixture: QuickwitPreCreatedFixture,
+) {
+ let client = harness.client();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
- let message_payloads = create_test_messages(message_count)
+ let message_count = 11;
+ let test_messages = create_test_messages(message_count);
+ let payloads: Vec<Bytes> = test_messages
.iter()
- .map(|message| {
- serde_json::to_vec(message)
- .map(Bytes::from)
- .map_err(Into::into)
- })
- .collect::<Result<Vec<Bytes>>>()?;
+ .map(|m| Bytes::from(serde_json::to_vec(m).expect("serialize")))
+ .collect();
- let mut test_messages = message_payloads
+ let mut messages: Vec<IggyMessage> = payloads
.iter()
- .cloned()
.enumerate()
- .map(|(i, message)| {
+ .map(|(i, p)| {
IggyMessage::builder()
.id((i + 1) as u128)
- .payload(message)
+ .payload(p.clone())
.build()
- .map_err(Into::into)
+ .expect("build message")
})
- .collect::<Result<Vec<IggyMessage>>>()?;
- iggy_client.send_messages(&mut test_messages).await;
+ .collect();
- Ok(message_payloads)
-}
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
-async fn assert_test_index_documents_match_message_payloads(
- test_setup: &QuickwitTestSetup,
- message_payloads: &[Bytes],
-) -> Result<()> {
- test_setup.flush_quickwit_test_index().await?;
+ fixture
+ .flush_index(seeds::names::TOPIC)
+ .await
+ .expect("flush");
- let search_response = test_setup.get_quickwit_test_index_all_search().await?;
- assert_eq!(search_response.num_hits, message_payloads.len());
+ let search = fixture
+ .search_all(seeds::names::TOPIC)
+ .await
+ .expect("search");
- for (quickwit_hit, message_payload) in search_response
- .hits
- .into_iter()
- .zip(message_payloads.iter())
- {
+ assert_eq!(search.num_hits, message_count);
+ for (hit, payload) in search.hits.iter().zip(payloads.iter()) {
assert_eq!(
- quickwit_hit,
- serde_json::from_slice::<serde_json::Value>(message_payload)?
+ hit,
+ &serde_json::from_slice::<serde_json::Value>(payload).unwrap()
);
}
-
- Ok(())
}
-#[tokio::test]
-async fn given_existent_quickwit_index_should_store() -> Result<()> {
- let quickwit_container = start_quickwit_container().await;
- let test_setup = QuickwitTestSetup::try_new_with_precreate_index(quickwit_container).await?;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/quickwit/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_nonexistent_quickwit_index_should_create_and_store(
+ harness: &TestHarness,
+ fixture: QuickwitFixture,
+) {
+ let client = harness.client();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
- let message_count = 11;
- let sent_payloads = send_test_messages(&test_setup, message_count).await?;
-
- assert_test_index_documents_match_message_payloads(&test_setup, &sent_payloads).await?;
+ let message_count = 13;
+ let test_messages = create_test_messages(message_count);
+ let payloads: Vec<Bytes> = test_messages
+ .iter()
+ .map(|m| Bytes::from(serde_json::to_vec(m).expect("serialize")))
+ .collect();
- Ok(())
-}
+ let mut messages: Vec<IggyMessage> = payloads
+ .iter()
+ .enumerate()
+ .map(|(i, p)| {
+ IggyMessage::builder()
+ .id((i + 1) as u128)
+ .payload(p.clone())
+ .build()
+ .expect("build message")
+ })
+ .collect();
-#[tokio::test]
-async fn given_nonexistent_quickwit_index_should_create_and_store() -> Result<()> {
- let test_setup = QuickwitTestSetup::try_new().await?;
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
- let message_count = 13;
- let sent_payloads = send_test_messages(&test_setup, message_count).await?;
+ fixture
+ .flush_index(seeds::names::TOPIC)
+ .await
+ .expect("flush");
- assert_test_index_documents_match_message_payloads(&test_setup, &sent_payloads).await?;
+ let search = fixture
+ .search_all(seeds::names::TOPIC)
+ .await
+ .expect("search");
- Ok(())
+ assert_eq!(search.num_hits, message_count);
+ for (hit, payload) in search.hits.iter().zip(payloads.iter()) {
+ assert_eq!(
+ hit,
+ &serde_json::from_slice::<serde_json::Value>(payload).unwrap()
+ );
+ }
}
-#[tokio::test]
-async fn given_bulk_message_send_should_store() -> Result<()> {
- let test_setup = QuickwitTestSetup::try_new().await?;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/quickwit/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_bulk_message_send_should_store(harness: &TestHarness, fixture: QuickwitFixture) {
+ let client = harness.client();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
let message_count = 1000;
- let sent_payloads = send_test_messages(&test_setup, message_count).await?;
+ let test_messages = create_test_messages(message_count);
+ let payloads: Vec<Bytes> = test_messages
+ .iter()
+ .map(|m| Bytes::from(serde_json::to_vec(m).expect("serialize")))
+ .collect();
+
+ let mut messages: Vec<IggyMessage> = payloads
+ .iter()
+ .enumerate()
+ .map(|(i, p)| {
+ IggyMessage::builder()
+ .id((i + 1) as u128)
+ .payload(p.clone())
+ .build()
+ .expect("build message")
+ })
+ .collect();
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
+
+ fixture
+ .flush_index(seeds::names::TOPIC)
+ .await
+ .expect("flush");
- assert_test_index_documents_match_message_payloads(&test_setup, &sent_payloads).await?;
+ let search = fixture
+ .search_all(seeds::names::TOPIC)
+ .await
+ .expect("search");
- Ok(())
+ assert_eq!(search.num_hits, message_count);
+ for (hit, payload) in search.hits.iter().zip(payloads.iter()) {
+ assert_eq!(
+ hit,
+ &serde_json::from_slice::<serde_json::Value>(payload).unwrap()
+ );
+ }
}
-#[tokio::test]
-async fn given_invalid_messages_should_not_store() -> Result<()> {
- let test_setup = QuickwitTestSetup::try_new().await?;
- let iggy_client = test_setup.runtime.create_client().await;
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/quickwit/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_invalid_messages_should_not_store(harness: &TestHarness, fixture: QuickwitFixture) {
+ let client = harness.client();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
- let first_valid = Bytes::from(serde_json::to_vec(&create_test_messages(1)[0])?);
- let second_valid = Bytes::from(serde_json::to_vec(&create_test_messages(1)[0])?);
+ let first_valid =
+ Bytes::from(serde_json::to_vec(&create_test_messages(1)[0]).expect("serialize"));
+ let second_valid =
+ Bytes::from(serde_json::to_vec(&create_test_messages(1)[0]).expect("serialize"));
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
struct NotTestMessage {
not_a_test_message_field: f64,
}
- let first_invalid = Bytes::from(serde_json::to_vec(&NotTestMessage {
- not_a_test_message_field: 17.,
- })?);
+ let first_invalid = Bytes::from(
+ serde_json::to_vec(&NotTestMessage {
+ not_a_test_message_field: 17.,
+ })
+ .expect("serialize"),
+ );
for (message_index, message_payload) in
[first_valid.clone(), first_invalid, second_valid.clone()]
.into_iter()
.enumerate()
{
- iggy_client
- .send_messages(&mut [IggyMessage::builder()
+ let mut messages = vec![
+ IggyMessage::builder()
.id(message_index as u128 + 1)
.payload(message_payload)
- .build()?])
- .await;
+ .build()
+ .expect("build message"),
+ ];
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
}
- assert_test_index_documents_match_message_payloads(&test_setup, &[first_valid, second_valid])
- .await?;
+ fixture
+ .flush_index(seeds::names::TOPIC)
+ .await
+ .expect("flush");
+
+ let search = fixture
+ .search_all(seeds::names::TOPIC)
+ .await
+ .expect("search");
- Ok(())
+ assert_eq!(search.num_hits, 2);
+ let expected_payloads = [first_valid, second_valid];
+ for (hit, payload) in search.hits.iter().zip(expected_payloads.iter()) {
+ assert_eq!(
+ hit,
+ &serde_json::from_slice::<serde_json::Value>(payload).unwrap()
+ );
+ }
}
diff --git a/core/integration/tests/connectors/random/mod.rs b/core/integration/tests/connectors/random/mod.rs
index 627ed1285..7b8f76b4d 100644
--- a/core/integration/tests/connectors/random/mod.rs
+++ b/core/integration/tests/connectors/random/mod.rs
@@ -17,13 +17,4 @@
* under the License.
*/
-use crate::connectors::{ConnectorsRuntime, IggySetup, setup_runtime};
-
mod random_source;
-
-async fn setup() -> ConnectorsRuntime {
- let iggy_setup = IggySetup::default();
- let mut runtime = setup_runtime();
- runtime.init("random/config.toml", None, iggy_setup).await;
- runtime
-}
diff --git a/core/integration/tests/connectors/random/random_source.rs b/core/integration/tests/connectors/random/random_source.rs
index 1bfe2de9f..e0bbfe348 100644
--- a/core/integration/tests/connectors/random/random_source.rs
+++ b/core/integration/tests/connectors/random/random_source.rs
@@ -17,20 +17,38 @@
* under the License.
*/
-use crate::connectors::random::setup;
+use iggy_binary_protocol::MessageClient;
+use iggy_common::{Consumer, Identifier, PollingStrategy};
+use integration::harness::seeds;
+use integration::iggy_harness;
use std::time::Duration;
use tokio::time::sleep;
-#[tokio::test]
-async fn given_valid_configuration_random_source_connector_should_produce_messages() {
- let runtime = setup().await;
- let client = runtime.create_client().await;
- // Wait for some messages to be produced
+#[iggy_harness(
+ server(connectors_runtime(config_path = "tests/connectors/random/config.toml")),
+ seed = seeds::connector_stream
+)]
+async fn random_source_produces_messages(harness: &TestHarness) {
sleep(Duration::from_secs(1)).await;
+
+ let client = harness.client();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = "test_consumer".try_into().unwrap();
+
let messages = client
- .poll_messages()
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id),
+ &PollingStrategy::next(),
+ 10,
+ true,
+ )
.await
.expect("Failed to poll messages");
+
assert!(
!messages.messages.is_empty(),
"No messages received from random source"
diff --git a/core/integration/tests/mod.rs b/core/integration/tests/mod.rs
index 22e917fab..e05582785 100644
--- a/core/integration/tests/mod.rs
+++ b/core/integration/tests/mod.rs
@@ -17,13 +17,19 @@
*/
use ctor::{ctor, dtor};
+use integration::harness::get_test_directory;
use lazy_static::lazy_static;
use std::collections::{HashMap, HashSet};
+use std::fs::File;
use std::io::Write;
use std::sync::RwLock;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Once};
use std::{panic, thread};
+use tracing_subscriber::fmt::MakeWriter;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::{EnvFilter, fmt};
mod cli;
mod cluster;
@@ -49,38 +55,51 @@ static UNKNOWN_TEST_NAME: &str = "unknown";
fn setup() {
let log_buffer = LOGS_BUFFER.clone();
- let mut logger = env_logger::builder();
- logger.is_test(true);
- logger.filter(None, log::LevelFilter::Info);
- logger.target(env_logger::Target::Pipe(Box::new(LogWriter(log_buffer))));
- logger.format(move |buf, record| {
- let timestamp = chrono::Local::now().format("%Y-%m-%d %H:%M:%S%.6f");
- let level = record.level();
+ let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
- writeln!(buf, "{timestamp} {level:>5} - {}", record.args())
- });
- logger.init();
+ tracing_subscriber::registry()
+ .with(env_filter)
+ .with(
+ fmt::layer()
+ .with_thread_names(true)
+ .with_writer(ThreadBufferWriter(log_buffer)),
+ )
+ .init();
- // Set up a custom panic hook to catch test failures
+ let panic_buffer = LOGS_BUFFER.clone();
let default_hook = panic::take_hook();
panic::set_hook(Box::new(move |info| {
- // store failed test name
let thread = thread::current();
let thread_name = thread.name().unwrap_or(UNKNOWN_TEST_NAME);
+
+ if let Ok(mut map) = panic_buffer.write() {
+ let buffer = map.entry(thread_name.to_string()).or_default();
+ let panic_msg = format!("\nPANIC: {info}\n");
+ buffer.extend_from_slice(panic_msg.as_bytes());
+ }
+
let failed_tests = FAILED_TEST_CASES.clone();
failed_tests
.write()
.unwrap()
.insert(thread_name.to_string());
- // If a test panics, set the failure flag to true
TESTS_FAILED.store(true, Ordering::SeqCst);
- // Call the default panic hook to continue normal behavior
default_hook(info);
}));
}
+struct ThreadBufferWriter(Arc<RwLock<HashMap<String, Vec<u8>>>>);
+
+impl<'a> MakeWriter<'a> for ThreadBufferWriter {
+ type Writer = LogWriter;
+
+ fn make_writer(&'a self) -> Self::Writer {
+ LogWriter(self.0.clone())
+ }
+}
+
struct LogWriter(Arc<RwLock<HashMap<String, Vec<u8>>>>);
impl Write for LogWriter {
@@ -101,14 +120,24 @@ impl Write for LogWriter {
}
fn teardown() {
- if TESTS_FAILED.load(Ordering::SeqCst)
- && let Ok(buffer) = LOGS_BUFFER.read()
- && let Ok(failed) = FAILED_TEST_CASES.read()
- {
- for test in failed.iter() {
- if let Some(logs) = buffer.get(test) {
- eprintln!("Logs for failed test '{test}':");
- eprintln!("{}", String::from_utf8_lossy(logs));
+ if let Ok(buffer) = LOGS_BUFFER.read() {
+ for (test_name, logs) in buffer.iter() {
+ if let Some(dir) = get_test_directory(test_name) {
+ let log_path = dir.join("test_stdout.log");
+ if let Ok(mut file) = File::create(&log_path) {
+ let _ = file.write_all(logs);
+ }
+ }
+ }
+
+ if TESTS_FAILED.load(Ordering::SeqCst)
+ && let Ok(failed) = FAILED_TEST_CASES.read()
+ {
+ for test in failed.iter() {
+ if let Some(logs) = buffer.get(test) {
+ eprintln!("Logs for failed test '{test}':");
+ eprintln!("{}", String::from_utf8_lossy(logs));
+ }
}
}
}