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:53:38 UTC
(iggy) branch connect-harness updated (f89ba0da1 -> 1a5879926)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch connect-harness
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit f89ba0da1 refactor(integration): migrate connectors tests to iggy_harness proc macro
new 1a5879926 refactor(integration): migrate connectors tests to iggy_harness proc macro
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (f89ba0da1)
\
N -- N -- N refs/heads/connect-harness (1a5879926)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/integration/src/harness/handle/server.rs | 14 ++++++++++++--
.../tests/connectors/fixtures/quickwit/container.rs | 6 +++++-
core/integration/tests/mod.rs | 11 ++++++-----
3 files changed, 23 insertions(+), 8 deletions(-)
(iggy) 01/01: refactor(integration): migrate connectors tests to iggy_harness proc macro
Posted by hg...@apache.org.
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 1a58799262e06714d3c8d422c6cd6a2903c5e3e7
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 | 35 +-
DEPENDENCIES.md | 2 -
core/harness_derive/src/codegen.rs | 14 +-
core/integration/Cargo.toml | 6 +-
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 | 111 ++++-
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 | 363 +++++++++++++++
.../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 | 84 ++--
36 files changed, 1488 insertions(+), 1785 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 42b1ace77..b82bfbbc9 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",
@@ -5000,18 +4977,15 @@ name = "integration"
version = "0.0.1"
dependencies = [
"ahash 0.8.12",
- "anyhow",
"assert_cmd",
"async-trait",
"bon",
"bytes",
- "chrono",
"compio",
"configs",
"configs_derive",
"ctor",
"derive_more",
- "env_logger",
"figment",
"futures",
"harness_derive",
@@ -5022,7 +4996,6 @@ dependencies = [
"keyring",
"lazy_static",
"libc",
- "log",
"once_cell",
"predicates",
"rand 0.9.2",
@@ -5038,12 +5011,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 +7101,7 @@ dependencies = [
"quinn-udp",
"rustc-hash",
"rustls",
- "socket2 0.5.10",
+ "socket2 0.6.2",
"thiserror 2.0.18",
"tokio",
"tracing",
@@ -7167,7 +7140,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..fa57fd1be 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -28,18 +28,15 @@ ci-qemu = []
[dependencies]
ahash = { workspace = true }
-anyhow = { workspace = true }
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 +47,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 +67,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..29544e572 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,104 @@ 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()?;
+ if let Err(e) = mcp.wait_ready().await {
+ let _ = mcp.stop();
+ return Err(e);
+ }
+ }
+
+ 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()?;
+ if let Err(e) = connectors_runtime.wait_ready().await {
+ let _ = connectors_runtime.stop();
+ if let Some(ref mut mcp) = self.mcp {
+ let _ = mcp.stop();
+ }
+ return Err(e);
+ }
+ }
+
+ 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 +641,8 @@ impl TestBinary for ServerHandle {
generated_cert_dir: None,
port_reserver: None,
test_transport: None,
+ mcp: None,
+ connectors_runtime: None,
}
}
@@ -693,6 +799,7 @@ impl Restartable for ServerHandle {
fn restart(&mut self) -> Result<(), TestBinaryError> {
let cleanup = self.config.cleanup;
self.config.cleanup = false;
+ self.stop_dependents()?;
self.stop()?;
self.config.cleanup = cleanup;
self.start()
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..d5133679b
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/quickwit/container.rs
@@ -0,0 +1,363 @@
+/*
+ * 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);
+ let client = reqwest::Client::builder()
+ .timeout(std::time::Duration::from_secs(30))
+ .build()
+ .expect("Failed to build HTTP client");
+ reqwest_middleware::ClientBuilder::new(client)
+ .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..3c84c0f84 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,45 +55,59 @@ 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 let Ok(mut failed) = failed_tests.write() {
+ failed.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 {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let thread = thread::current();
let thread_name = thread.name().unwrap_or(UNKNOWN_TEST_NAME);
- let mut map = self.0.write().unwrap();
+ let Ok(mut map) = self.0.write() else {
+ return Ok(buf.len());
+ };
let buffer = map.entry(thread_name.to_string()).or_default();
buffer.extend_from_slice(buf);
@@ -101,14 +121,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));
+ }
}
}
}