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/01/28 09:38:21 UTC
(iggy) branch test-harness created (now ef958c5c4)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch test-harness
in repository https://gitbox.apache.org/repos/asf/iggy.git
at ef958c5c4 feat(integration): add core harness library with server and client handles
This branch includes the following new commits:
new ef958c5c4 feat(integration): add core harness library with server and client handles
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.
(iggy) 01/01: feat(integration): add core harness library with server and client handles
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch test-harness
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit ef958c5c4c2fc30d055229dbd44664a870291ae5
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Jan 27 19:22:06 2026 +0100
feat(integration): add core harness library with server and client handles
TestHarness builder API for integration test infrastructure. Core modules:
context (test directory lifecycle), error, traits, helpers, config
(server/client/mcp/connector), handle (process management), fixtures
(PostgreSQL/WireMock), and seeds (test data generation).
Orchestrator manages binary lifecycle and client creation. Adds TLS support
to WebSocketClientFactory for TLS scenarios.
---
.gitignore | 1 +
Cargo.lock | 3 +
core/integration/.gitignore | 1 +
core/integration/Cargo.toml | 3 +
core/integration/src/harness/config/client.rs | 198 +++++
core/integration/src/harness/config/common.rs | 95 +++
.../src/{lib.rs => harness/config/connector.rs} | 40 +-
core/integration/src/harness/config/mcp.rs | 62 ++
.../src/{lib.rs => harness/config/mod.rs} | 32 +-
core/integration/src/harness/config/resolve.rs | 275 ++++++
core/integration/src/harness/config/server.rs | 78 ++
core/integration/src/harness/context.rs | 191 +++++
core/integration/src/harness/error.rs | 192 +++++
core/integration/src/harness/fixtures/mod.rs | 77 ++
core/integration/src/harness/fixtures/postgres.rs | 941 +++++++++++++++++++++
.../src/{lib.rs => harness/fixtures/random.rs} | 42 +-
core/integration/src/harness/fixtures/wiremock.rs | 153 ++++
core/integration/src/harness/handle/client.rs | 230 +++++
core/integration/src/harness/handle/common.rs | 76 ++
core/integration/src/harness/handle/connector.rs | 277 ++++++
core/integration/src/harness/handle/mcp.rs | 296 +++++++
.../src/{lib.rs => harness/handle/mod.rs} | 29 +-
core/integration/src/harness/handle/server.rs | 702 +++++++++++++++
core/integration/src/harness/helpers.rs | 113 +++
core/integration/src/harness/mod.rs | 83 ++
core/integration/src/harness/orchestrator.rs | 885 +++++++++++++++++++
core/integration/src/harness/seeds.rs | 160 ++++
core/integration/src/harness/traits.rs | 53 ++
core/integration/src/lib.rs | 1 +
core/integration/src/websocket_client.rs | 32 +-
.../tests/server/concurrent_addition.rs | 5 +-
core/integration/tests/server/mod.rs | 5 +-
core/integration/tests/server/specific.rs | 1 +
33 files changed, 5255 insertions(+), 77 deletions(-)
diff --git a/.gitignore b/.gitignore
index 312bc4e76..7c8d4d1dd 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,6 +7,7 @@ target*
.gradle
local_data*
local_state*
+test_logs/
/bench_*
/sdk/errors_table/
/rust-clippy-results.sarif
diff --git a/Cargo.lock b/Cargo.lock
index 5b992c303..a602775f7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4988,6 +4988,7 @@ dependencies = [
"ahash 0.8.12",
"assert_cmd",
"async-trait",
+ "bon",
"bytes",
"chrono",
"compio",
@@ -5006,6 +5007,7 @@ dependencies = [
"lazy_static",
"libc",
"log",
+ "once_cell",
"predicates",
"rand 0.9.2",
"rcgen",
@@ -5022,6 +5024,7 @@ dependencies = [
"test-case",
"testcontainers-modules",
"tokio",
+ "tracing-subscriber",
"twox-hash",
"uuid",
"zip",
diff --git a/core/integration/.gitignore b/core/integration/.gitignore
new file mode 100644
index 000000000..fcf6a3240
--- /dev/null
+++ b/core/integration/.gitignore
@@ -0,0 +1 @@
+test_logs/
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index befafc249..a405ac2f0 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -30,6 +30,7 @@ ci-qemu = []
ahash = { workspace = true }
assert_cmd = "2.1.2"
async-trait = { workspace = true }
+bon = { workspace = true }
bytes = { workspace = true }
chrono = { workspace = true }
compio = { workspace = true }
@@ -48,6 +49,7 @@ keyring = { workspace = true }
lazy_static = { workspace = true }
libc = "0.2.180"
log = { workspace = true }
+once_cell = { workspace = true }
predicates = { workspace = true }
rand = { workspace = true }
rcgen = "0.14.7"
@@ -69,6 +71,7 @@ tempfile = { workspace = true }
test-case = { workspace = true }
testcontainers-modules = { version = "0.14.0", features = ["postgres"] }
tokio = { workspace = true }
+tracing-subscriber = { workspace = true }
twox-hash = { workspace = true }
uuid = { workspace = true }
zip = { workspace = true }
diff --git a/core/integration/src/harness/config/client.rs b/core/integration/src/harness/config/client.rs
new file mode 100644
index 000000000..b1caf5963
--- /dev/null
+++ b/core/integration/src/harness/config/client.rs
@@ -0,0 +1,198 @@
+/*
+ * 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 bon::Builder;
+use iggy::prelude::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
+use iggy_common::TransportProtocol;
+use std::path::PathBuf;
+
+#[derive(Debug, Clone)]
+pub struct AutoLoginConfig {
+ pub username: String,
+ pub password: String,
+}
+
+impl AutoLoginConfig {
+ pub fn root() -> Self {
+ Self {
+ username: DEFAULT_ROOT_USERNAME.to_string(),
+ password: DEFAULT_ROOT_PASSWORD.to_string(),
+ }
+ }
+
+ pub fn new(username: impl Into<String>, password: impl Into<String>) -> Self {
+ Self {
+ username: username.into(),
+ password: password.into(),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Builder)]
+pub struct ClientConfig {
+ #[builder(default)]
+ pub transport: TransportProtocol,
+ #[builder(default)]
+ pub tcp_nodelay: bool,
+ #[builder(default)]
+ pub tls_enabled: bool,
+ #[builder(into)]
+ pub tls_domain: Option<String>,
+ #[builder(into)]
+ pub tls_ca_file: Option<PathBuf>,
+ #[builder(default)]
+ pub tls_validate_certificate: bool,
+ pub auto_login: Option<AutoLoginConfig>,
+}
+
+impl Default for ClientConfig {
+ fn default() -> Self {
+ Self::builder().build()
+ }
+}
+
+impl ClientConfig {
+ pub fn tcp() -> Self {
+ Self::default()
+ }
+
+ /// Creates TCP client that auto-logins as root after connect.
+ pub fn root_tcp() -> Self {
+ Self::builder().auto_login(AutoLoginConfig::root()).build()
+ }
+
+ pub fn quic() -> Self {
+ Self::builder().transport(TransportProtocol::Quic).build()
+ }
+
+ /// Creates QUIC client that auto-logins as root after connect.
+ pub fn root_quic() -> Self {
+ Self::builder()
+ .transport(TransportProtocol::Quic)
+ .auto_login(AutoLoginConfig::root())
+ .build()
+ }
+
+ pub fn http() -> Self {
+ Self::builder().transport(TransportProtocol::Http).build()
+ }
+
+ /// Creates HTTP client that auto-logins as root after connect.
+ pub fn root_http() -> Self {
+ Self::builder()
+ .transport(TransportProtocol::Http)
+ .auto_login(AutoLoginConfig::root())
+ .build()
+ }
+
+ pub fn websocket() -> Self {
+ Self::builder()
+ .transport(TransportProtocol::WebSocket)
+ .build()
+ }
+
+ /// Creates WebSocket client that auto-logins as root after connect.
+ pub fn root_websocket() -> Self {
+ Self::builder()
+ .transport(TransportProtocol::WebSocket)
+ .auto_login(AutoLoginConfig::root())
+ .build()
+ }
+
+ pub fn with_nodelay(mut self) -> Self {
+ self.tcp_nodelay = true;
+ self
+ }
+
+ pub fn with_tls(
+ mut self,
+ domain: impl Into<String>,
+ ca_file: Option<PathBuf>,
+ validate: bool,
+ ) -> Self {
+ self.tls_enabled = true;
+ self.tls_domain = Some(domain.into());
+ self.tls_ca_file = ca_file;
+ self.tls_validate_certificate = validate;
+ self
+ }
+
+ /// Add auto-login to any client config.
+ pub fn with_auto_login(
+ mut self,
+ username: impl Into<String>,
+ password: impl Into<String>,
+ ) -> Self {
+ self.auto_login = Some(AutoLoginConfig::new(username, password));
+ self
+ }
+
+ /// Add root auto-login to any client config.
+ pub fn with_root_login(mut self) -> Self {
+ self.auto_login = Some(AutoLoginConfig::root());
+ self
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_client_config() {
+ let tcp = ClientConfig::tcp().with_nodelay();
+ assert_eq!(tcp.transport, TransportProtocol::Tcp);
+ assert!(tcp.tcp_nodelay);
+
+ let quic = ClientConfig::quic();
+ assert_eq!(quic.transport, TransportProtocol::Quic);
+
+ let tls = ClientConfig::tcp().with_tls("localhost", None, false);
+ assert!(tls.tls_enabled);
+ assert_eq!(tls.tls_domain, Some("localhost".to_string()));
+ }
+
+ #[test]
+ fn test_root_tcp_config() {
+ let root_tcp = ClientConfig::root_tcp();
+ assert_eq!(root_tcp.transport, TransportProtocol::Tcp);
+ assert!(root_tcp.auto_login.is_some());
+ let login = root_tcp.auto_login.unwrap();
+ assert_eq!(login.username, DEFAULT_ROOT_USERNAME);
+ assert_eq!(login.password, DEFAULT_ROOT_PASSWORD);
+ }
+
+ #[test]
+ fn test_with_auto_login() {
+ let config = ClientConfig::tcp().with_auto_login("user", "pass");
+ assert!(config.auto_login.is_some());
+ let login = config.auto_login.unwrap();
+ assert_eq!(login.username, "user");
+ assert_eq!(login.password, "pass");
+ }
+
+ #[test]
+ fn test_with_root_login() {
+ let config = ClientConfig::quic().with_root_login();
+ assert_eq!(config.transport, TransportProtocol::Quic);
+ assert!(config.auto_login.is_some());
+ let login = config.auto_login.unwrap();
+ assert_eq!(login.username, DEFAULT_ROOT_USERNAME);
+ }
+}
diff --git a/core/integration/src/harness/config/common.rs b/core/integration/src/harness/config/common.rs
new file mode 100644
index 000000000..df40aba69
--- /dev/null
+++ b/core/integration/src/harness/config/common.rs
@@ -0,0 +1,95 @@
+/*
+ * 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 std::path::PathBuf;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum IpAddrKind {
+ #[default]
+ V4,
+ V6,
+}
+
+#[derive(Debug, Clone)]
+pub struct EncryptionConfig {
+ pub key: String,
+}
+
+impl EncryptionConfig {
+ pub fn new(key: impl Into<String>) -> Self {
+ Self { key: key.into() }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct TlsConfig {
+ pub cert_dir: PathBuf,
+ pub domain: String,
+ pub validate_certificate: bool,
+ pub self_signed: bool,
+ pub generate_certs: bool,
+}
+
+impl TlsConfig {
+ /// Create a TlsConfig with custom certificates.
+ pub fn new(cert_dir: impl Into<PathBuf>, domain: impl Into<String>, validate: bool) -> Self {
+ Self {
+ cert_dir: cert_dir.into(),
+ domain: domain.into(),
+ validate_certificate: validate,
+ self_signed: false,
+ generate_certs: false,
+ }
+ }
+
+ /// Create a TlsConfig for server-generated self-signed certificates.
+ /// The server will generate the certificates automatically.
+ /// Client cannot validate these certs as no CA file is available.
+ pub fn self_signed() -> Self {
+ Self {
+ cert_dir: PathBuf::new(),
+ domain: "localhost".to_string(),
+ validate_certificate: false,
+ self_signed: true,
+ generate_certs: false,
+ }
+ }
+
+ /// Create a TlsConfig that generates test certificates during harness setup.
+ /// The generated CA cert can be used by clients for certificate validation.
+ pub fn generated() -> Self {
+ Self {
+ cert_dir: PathBuf::new(), // Will be set by ServerHandle during setup
+ domain: "localhost".to_string(),
+ validate_certificate: true,
+ self_signed: false,
+ generate_certs: true,
+ }
+ }
+
+ /// Get the CA certificate path for client use.
+ /// Returns None for self-signed mode (no CA file available).
+ pub fn ca_cert_path(&self) -> Option<PathBuf> {
+ if self.self_signed || self.cert_dir.as_os_str().is_empty() {
+ None
+ } else {
+ Some(self.cert_dir.join("test_cert.pem"))
+ }
+ }
+}
diff --git a/core/integration/src/lib.rs b/core/integration/src/harness/config/connector.rs
similarity index 57%
copy from core/integration/src/lib.rs
copy to core/integration/src/harness/config/connector.rs
index 269698b2b..dd9900e22 100644
--- a/core/integration/src/lib.rs
+++ b/core/integration/src/harness/config/connector.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * 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
@@ -16,21 +17,24 @@
* under the License.
*/
-pub mod bench_utils;
-pub mod file;
-#[allow(deprecated)]
-pub mod http_client;
-#[allow(deprecated)]
-pub mod quic_client;
-#[allow(deprecated)]
-pub mod tcp_client;
-#[allow(deprecated)]
-pub mod websocket_client;
+use bon::Builder;
+use std::collections::HashMap;
+use std::path::PathBuf;
-#[allow(deprecated)]
-pub mod test_connectors_runtime;
-#[allow(deprecated)]
-pub mod test_mcp_server;
-#[allow(deprecated)]
-pub mod test_server;
-pub mod test_tls_utils;
+#[derive(Debug, Clone, Builder)]
+pub struct ConnectorConfig {
+ #[builder(into)]
+ pub config_path: Option<PathBuf>,
+ #[builder(default = true)]
+ pub cleanup: bool,
+ #[builder(default)]
+ pub extra_envs: HashMap<String, String>,
+ #[builder(into)]
+ pub executable_path: Option<String>,
+}
+
+impl Default for ConnectorConfig {
+ fn default() -> Self {
+ Self::builder().build()
+ }
+}
diff --git a/core/integration/src/harness/config/mcp.rs b/core/integration/src/harness/config/mcp.rs
new file mode 100644
index 000000000..77aaf90f9
--- /dev/null
+++ b/core/integration/src/harness/config/mcp.rs
@@ -0,0 +1,62 @@
+/*
+ * 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 bon::Builder;
+use std::collections::HashMap;
+
+#[derive(Debug, Clone, Builder)]
+pub struct McpConfig {
+ #[builder(into, default = "mcp".to_string())]
+ pub consumer_name: String,
+ #[builder(into, default = "/mcp".to_string())]
+ pub http_path: String,
+ #[builder(default)]
+ pub extra_envs: HashMap<String, String>,
+ #[builder(into)]
+ pub executable_path: Option<String>,
+}
+
+impl Default for McpConfig {
+ fn default() -> Self {
+ Self::builder().build()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_mcp_config_builder() {
+ let config = McpConfig::builder()
+ .consumer_name("custom")
+ .http_path("/custom-mcp")
+ .build();
+
+ assert_eq!(config.consumer_name, "custom");
+ assert_eq!(config.http_path, "/custom-mcp");
+ }
+
+ #[test]
+ fn test_mcp_config_defaults() {
+ let config = McpConfig::default();
+ assert_eq!(config.consumer_name, "mcp");
+ assert_eq!(config.http_path, "/mcp");
+ }
+}
diff --git a/core/integration/src/lib.rs b/core/integration/src/harness/config/mod.rs
similarity index 63%
copy from core/integration/src/lib.rs
copy to core/integration/src/harness/config/mod.rs
index 269698b2b..d8626c545 100644
--- a/core/integration/src/lib.rs
+++ b/core/integration/src/harness/config/mod.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * 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
@@ -16,21 +17,16 @@
* under the License.
*/
-pub mod bench_utils;
-pub mod file;
-#[allow(deprecated)]
-pub mod http_client;
-#[allow(deprecated)]
-pub mod quic_client;
-#[allow(deprecated)]
-pub mod tcp_client;
-#[allow(deprecated)]
-pub mod websocket_client;
+mod client;
+mod common;
+mod connector;
+mod mcp;
+mod resolve;
+mod server;
-#[allow(deprecated)]
-pub mod test_connectors_runtime;
-#[allow(deprecated)]
-pub mod test_mcp_server;
-#[allow(deprecated)]
-pub mod test_server;
-pub mod test_tls_utils;
+pub use client::{AutoLoginConfig, ClientConfig};
+pub use common::{EncryptionConfig, IpAddrKind, TlsConfig};
+pub use connector::ConnectorConfig;
+pub use mcp::McpConfig;
+pub use resolve::resolve_config_paths;
+pub use server::TestServerConfig;
diff --git a/core/integration/src/harness/config/resolve.rs b/core/integration/src/harness/config/resolve.rs
new file mode 100644
index 000000000..24162cdc4
--- /dev/null
+++ b/core/integration/src/harness/config/resolve.rs
@@ -0,0 +1,275 @@
+/*
+ * 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.
+ */
+
+//! Runtime validation and resolution of config paths to environment variables.
+
+use configs::ConfigEnvMappings;
+use server::configs::server::ServerConfig;
+use std::collections::HashMap;
+
+/// Resolve config paths to environment variable names.
+///
+/// Takes a map of dot-notation config paths (e.g., "segment.size") and their values,
+/// validates them against the `ServerConfig` mappings, and returns the corresponding
+/// environment variable names with values.
+///
+/// # Implicit defaults
+///
+/// - Setting any `tcp.socket.*` option automatically enables `tcp.socket.override_defaults`
+/// - Setting any `quic.socket.*` option automatically enables `quic.socket.override_defaults`
+///
+/// # Errors
+///
+/// Returns an error message if any path is invalid, including suggestions for similar paths.
+pub fn resolve_config_paths(
+ overrides: &HashMap<String, String>,
+) -> Result<HashMap<String, String>, String> {
+ let mut env_vars = HashMap::new();
+ let mut needs_tcp_socket_override = false;
+ let mut needs_quic_socket_override = false;
+ let mut needs_encryption_enabled = false;
+
+ for (path, value) in overrides {
+ // Special shorthand: "encryption" maps to "system.encryption.key"
+ let resolved_path = if path == "encryption" {
+ "system.encryption.key"
+ } else {
+ path.as_str()
+ };
+
+ let mapping = ServerConfig::find_by_config_path(resolved_path)
+ .or_else(|| ServerConfig::find_by_config_path(&format!("system.{}", resolved_path)));
+
+ match mapping {
+ Some(m) => {
+ env_vars.insert(m.env_name.to_string(), value.clone());
+
+ // Track if we need to enable socket override_defaults
+ if path.starts_with("tcp.socket.") && path != "tcp.socket.override_defaults" {
+ needs_tcp_socket_override = true;
+ }
+ if path.starts_with("quic.socket.") && path != "quic.socket.override_defaults" {
+ needs_quic_socket_override = true;
+ }
+ // Track if encryption key is set (auto-enable encryption)
+ if path == "encryption"
+ || path == "encryption.key"
+ || path == "system.encryption.key"
+ {
+ needs_encryption_enabled = true;
+ }
+ }
+ None => {
+ let suggestions = find_similar_paths(path);
+
+ let suggestions_str = if suggestions.is_empty() {
+ String::new()
+ } else {
+ format!("\nDid you mean: {:?}?", suggestions)
+ };
+
+ return Err(format!(
+ "Unknown config path: '{}'{}",
+ path, suggestions_str
+ ));
+ }
+ }
+ }
+
+ // Auto-enable override_defaults when socket settings are customized
+ if needs_tcp_socket_override
+ && let Some(m) = ServerConfig::find_by_config_path("tcp.socket.override_defaults")
+ {
+ env_vars
+ .entry(m.env_name.to_string())
+ .or_insert_with(|| "true".to_string());
+ }
+ if needs_quic_socket_override
+ && let Some(m) = ServerConfig::find_by_config_path("quic.socket.override_defaults")
+ {
+ env_vars
+ .entry(m.env_name.to_string())
+ .or_insert_with(|| "true".to_string());
+ }
+ // Auto-enable encryption when key is set
+ if needs_encryption_enabled
+ && let Some(m) = ServerConfig::find_by_config_path("system.encryption.enabled")
+ {
+ env_vars
+ .entry(m.env_name.to_string())
+ .or_insert_with(|| "true".to_string());
+ }
+
+ Ok(env_vars)
+}
+
+fn levenshtein(a: &str, b: &str) -> usize {
+ let a_len = a.len();
+ let b_len = b.len();
+
+ if a_len == 0 {
+ return b_len;
+ }
+ if b_len == 0 {
+ return a_len;
+ }
+
+ let mut prev_row: Vec<usize> = (0..=b_len).collect();
+ let mut curr_row: Vec<usize> = vec![0; b_len + 1];
+
+ for (i, ca) in a.chars().enumerate() {
+ curr_row[0] = i + 1;
+
+ for (j, cb) in b.chars().enumerate() {
+ let cost = if ca == cb { 0 } else { 1 };
+ curr_row[j + 1] = (prev_row[j + 1] + 1)
+ .min(curr_row[j] + 1)
+ .min(prev_row[j] + cost);
+ }
+
+ std::mem::swap(&mut prev_row, &mut curr_row);
+ }
+
+ prev_row[b_len]
+}
+
+fn find_similar_paths(unknown: &str) -> Vec<String> {
+ // Normalize: replace underscores with dots for comparison
+ let normalized = unknown.replace('_', ".");
+
+ let mut candidates: Vec<_> = ServerConfig::env_mappings()
+ .iter()
+ .filter_map(|m| {
+ let path = m.config_path;
+
+ // Check if the path ends with the unknown value (suffix match)
+ if path.ends_with(&normalized) || path.ends_with(unknown) {
+ return Some((path, 0)); // Perfect suffix match
+ }
+
+ // Check if any segment matches
+ let last_segment = path.rsplit('.').next().unwrap_or(path);
+ let unknown_last = normalized.rsplit('.').next().unwrap_or(&normalized);
+ if last_segment == unknown_last || last_segment == unknown {
+ return Some((path, 1)); // Segment match
+ }
+
+ // Levenshtein on the last segment
+ let segment_dist = levenshtein(unknown_last, last_segment);
+ if segment_dist <= 3 {
+ return Some((path, segment_dist + 2));
+ }
+
+ // Full path levenshtein with higher threshold
+ let full_dist = levenshtein(&normalized, path);
+ if full_dist <= 8 {
+ return Some((path, full_dist + 5));
+ }
+
+ None
+ })
+ .collect();
+
+ candidates.sort_by_key(|(_, score)| *score);
+
+ candidates
+ .into_iter()
+ .take(3)
+ .map(|(path, _)| path.to_string())
+ .collect()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn resolve_valid_path() {
+ let mut overrides = HashMap::new();
+ overrides.insert("system.segment.size".to_string(), "1MiB".to_string());
+
+ let result = resolve_config_paths(&overrides);
+ assert!(result.is_ok());
+ let env_vars = result.unwrap();
+ assert!(env_vars.contains_key("IGGY_SYSTEM_SEGMENT_SIZE"));
+ assert_eq!(
+ env_vars.get("IGGY_SYSTEM_SEGMENT_SIZE"),
+ Some(&"1MiB".to_string())
+ );
+ }
+
+ #[test]
+ fn resolve_with_implicit_system_prefix() {
+ let mut overrides = HashMap::new();
+ overrides.insert("segment.size".to_string(), "1MiB".to_string());
+
+ let result = resolve_config_paths(&overrides);
+ assert!(result.is_ok());
+ let env_vars = result.unwrap();
+ assert!(env_vars.contains_key("IGGY_SYSTEM_SEGMENT_SIZE"));
+ }
+
+ #[test]
+ fn resolve_invalid_path() {
+ let mut overrides = HashMap::new();
+ overrides.insert("segmant.size".to_string(), "1MiB".to_string());
+
+ let result = resolve_config_paths(&overrides);
+ assert!(result.is_err());
+ let err = result.unwrap_err();
+ assert!(err.contains("Unknown config path: 'segmant.size'"));
+ }
+
+ #[test]
+ fn resolve_encryption_shorthand_auto_enables() {
+ let mut overrides = HashMap::new();
+ overrides.insert(
+ "encryption".to_string(),
+ "/rvT1xP4V8u1EAhk4xDdqzqM2UOPXyy9XYkl4uRShgE=".to_string(),
+ );
+
+ let result = resolve_config_paths(&overrides);
+ assert!(result.is_ok());
+ let env_vars = result.unwrap();
+ assert_eq!(
+ env_vars.get("IGGY_SYSTEM_ENCRYPTION_KEY"),
+ Some(&"/rvT1xP4V8u1EAhk4xDdqzqM2UOPXyy9XYkl4uRShgE=".to_string())
+ );
+ assert_eq!(
+ env_vars.get("IGGY_SYSTEM_ENCRYPTION_ENABLED"),
+ Some(&"true".to_string())
+ );
+ }
+
+ #[test]
+ fn levenshtein_identical() {
+ assert_eq!(levenshtein("hello", "hello"), 0);
+ }
+
+ #[test]
+ fn levenshtein_one_char_diff() {
+ assert_eq!(levenshtein("hello", "hallo"), 1);
+ }
+
+ #[test]
+ fn levenshtein_empty() {
+ assert_eq!(levenshtein("", "hello"), 5);
+ assert_eq!(levenshtein("hello", ""), 5);
+ }
+}
diff --git a/core/integration/src/harness/config/server.rs b/core/integration/src/harness/config/server.rs
new file mode 100644
index 000000000..78b58d4f5
--- /dev/null
+++ b/core/integration/src/harness/config/server.rs
@@ -0,0 +1,78 @@
+/*
+ * 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::common::{EncryptionConfig, IpAddrKind, TlsConfig};
+use bon::Builder;
+use std::collections::HashMap;
+
+#[derive(Debug, Clone, Builder)]
+pub struct TestServerConfig {
+ #[builder(default = true)]
+ pub quic_enabled: bool,
+ #[builder(default = true)]
+ pub websocket_enabled: bool,
+ #[builder(default = true)]
+ pub http_enabled: bool,
+ pub encryption: Option<EncryptionConfig>,
+ /// TLS configuration for TCP transport.
+ pub tls: Option<TlsConfig>,
+ /// TLS configuration for WebSocket transport.
+ pub websocket_tls: Option<TlsConfig>,
+ #[builder(default = true)]
+ pub cleanup: bool,
+ #[builder(default)]
+ pub ip_kind: IpAddrKind,
+ /// Environment variables passed to the server. Config paths resolved via
+ /// `resolve_config_paths` should be placed here.
+ #[builder(default)]
+ pub extra_envs: HashMap<String, String>,
+ #[builder(into)]
+ pub executable_path: Option<String>,
+}
+
+impl Default for TestServerConfig {
+ fn default() -> Self {
+ Self::builder().build()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_server_config_builder() {
+ let config = TestServerConfig::builder()
+ .quic_enabled(false)
+ .extra_envs(HashMap::from([("FOO".to_string(), "BAR".to_string())]))
+ .build();
+
+ assert!(!config.quic_enabled);
+ assert_eq!(config.extra_envs.get("FOO"), Some(&"BAR".to_string()));
+ }
+
+ #[test]
+ fn test_server_config_defaults() {
+ let config = TestServerConfig::default();
+ assert!(config.quic_enabled);
+ assert!(config.websocket_enabled);
+ assert!(config.http_enabled);
+ assert!(config.cleanup);
+ }
+}
diff --git a/core/integration/src/harness/context.rs b/core/integration/src/harness/context.rs
new file mode 100644
index 000000000..b03e81459
--- /dev/null
+++ b/core/integration/src/harness/context.rs
@@ -0,0 +1,191 @@
+/*
+ * 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 crate::harness::error::TestBinaryError;
+use once_cell::sync::Lazy;
+use std::collections::HashMap;
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::sync::RwLock;
+use std::thread;
+use uuid::Uuid;
+
+const TEST_CLEANUP_DISABLED_ENV_VAR: &str = "IGGY_TEST_CLEANUP_DISABLED";
+
+/// Global registry mapping test names to their log directories.
+static TEST_DIRECTORIES: Lazy<RwLock<HashMap<String, PathBuf>>> =
+ Lazy::new(|| RwLock::new(HashMap::new()));
+
+/// Get the log directory for a test by name.
+pub fn get_test_directory(test_name: &str) -> Option<PathBuf> {
+ TEST_DIRECTORIES
+ .read()
+ .ok()
+ .and_then(|map| map.get(test_name).cloned())
+}
+
+fn register_test_directory(test_name: &str, path: &Path) {
+ if let Ok(mut map) = TEST_DIRECTORIES.write() {
+ map.insert(test_name.to_string(), path.to_path_buf());
+ }
+}
+
+fn is_cleanup_disabled_by_env() -> bool {
+ std::env::var(TEST_CLEANUP_DISABLED_ENV_VAR)
+ .map(|v| matches!(v.to_lowercase().as_str(), "true" | "1"))
+ .unwrap_or(false)
+}
+
+static TEST_LOGS_DIR: Lazy<PathBuf> = Lazy::new(|| {
+ PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+ .parent()
+ .and_then(|p| p.parent())
+ .expect("Failed to find workspace root")
+ .join("test_logs")
+});
+
+/// Owns the test directory and provides paths for all binaries within a test.
+pub struct TestContext {
+ test_name: String,
+ base_dir: PathBuf,
+ cleanup: bool,
+ created: bool,
+}
+
+impl TestContext {
+ pub fn new(test_name: Option<String>, cleanup: bool) -> Result<Self, TestBinaryError> {
+ let test_name = test_name.unwrap_or_else(Self::derive_test_name);
+ let uuid_suffix = Uuid::new_v4().to_string()[..8].to_string();
+ let dir_name = format!("{}_{}", sanitize_path(&test_name), uuid_suffix);
+
+ let base_dir = (*TEST_LOGS_DIR).join(dir_name);
+
+ Ok(Self {
+ test_name,
+ base_dir,
+ cleanup,
+ created: false,
+ })
+ }
+
+ /// Creates the test directory structure. Called lazily on first access.
+ pub fn ensure_created(&mut self) -> Result<(), TestBinaryError> {
+ if self.created {
+ return Ok(());
+ }
+
+ fs::create_dir_all(&self.base_dir).map_err(|e| TestBinaryError::DirectoryCreation {
+ path: self.base_dir.clone(),
+ source: e,
+ })?;
+
+ register_test_directory(&self.test_name, &self.base_dir);
+ self.created = true;
+ Ok(())
+ }
+
+ fn derive_test_name() -> String {
+ thread::current()
+ .name()
+ .map(|s| s.to_string())
+ .unwrap_or_else(|| Uuid::new_v4().to_string())
+ }
+
+ pub fn test_name(&self) -> &str {
+ &self.test_name
+ }
+
+ pub fn base_dir(&self) -> &Path {
+ &self.base_dir
+ }
+
+ pub fn mcp_stdout_path(&self) -> PathBuf {
+ self.base_dir.join("mcp_stdout.log")
+ }
+
+ pub fn mcp_stderr_path(&self) -> PathBuf {
+ self.base_dir.join("mcp_stderr.log")
+ }
+
+ pub fn connector_state_path(&self) -> PathBuf {
+ self.base_dir.join("connectors_state")
+ }
+
+ pub fn connector_stdout_path(&self) -> PathBuf {
+ self.base_dir.join("connectors_stdout.log")
+ }
+
+ pub fn connector_stderr_path(&self) -> PathBuf {
+ self.base_dir.join("connectors_stderr.log")
+ }
+
+ pub fn test_stdout_path(&self) -> PathBuf {
+ self.base_dir.join("test_stdout.log")
+ }
+
+ pub fn cleanup(&self) {
+ if !self.cleanup || is_cleanup_disabled_by_env() {
+ return;
+ }
+ if self.base_dir.exists() {
+ let _ = fs::remove_dir_all(&self.base_dir);
+ }
+ }
+
+ pub fn set_cleanup(&mut self, cleanup: bool) {
+ self.cleanup = cleanup;
+ }
+}
+
+impl Drop for TestContext {
+ fn drop(&mut self) {
+ if !thread::panicking() {
+ self.cleanup();
+ }
+ }
+}
+
+fn sanitize_path(name: &str) -> String {
+ name.chars()
+ .map(|c| match c {
+ '/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' => '_',
+ c if c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '.' => c,
+ _ => '_',
+ })
+ .collect()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_sanitize_path() {
+ assert_eq!(sanitize_path("test::foo::bar"), "test__foo__bar");
+ assert_eq!(sanitize_path("my/test"), "my_test");
+ assert_eq!(sanitize_path("test<>name"), "test__name");
+ }
+
+ #[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.connector_state_path().ends_with("connectors_state"));
+ }
+}
diff --git a/core/integration/src/harness/error.rs b/core/integration/src/harness/error.rs
new file mode 100644
index 000000000..237a22a9f
--- /dev/null
+++ b/core/integration/src/harness/error.rs
@@ -0,0 +1,192 @@
+/*
+ * 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 std::io;
+use std::path::PathBuf;
+
+#[derive(Debug)]
+pub enum TestBinaryError {
+ Io(io::Error),
+ ProcessSpawn {
+ binary: String,
+ source: io::Error,
+ },
+ ProcessCrashed {
+ binary: String,
+ exit_code: Option<i32>,
+ stdout: String,
+ stderr: String,
+ },
+ StartupTimeout {
+ binary: String,
+ timeout_secs: u64,
+ },
+ HealthCheckFailed {
+ binary: String,
+ address: String,
+ retries: u32,
+ },
+ ConfigLoad {
+ path: PathBuf,
+ message: String,
+ },
+ InvalidState {
+ message: String,
+ },
+ ClientCreation {
+ transport: String,
+ address: String,
+ source: String,
+ },
+ ClientConnection {
+ transport: String,
+ address: String,
+ source: String,
+ },
+ DirectoryCreation {
+ path: PathBuf,
+ source: io::Error,
+ },
+ MissingServer,
+ MissingMcp,
+ MissingConnector,
+ NotStarted,
+ AlreadyStarted,
+ SeedFailed(String),
+ FixtureSetup {
+ fixture_type: String,
+ message: String,
+ },
+}
+
+impl std::fmt::Display for TestBinaryError {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ TestBinaryError::Io(e) => write!(f, "IO error: {}", e),
+ TestBinaryError::ProcessSpawn { binary, source } => {
+ write!(f, "Failed to spawn {}: {}", binary, source)
+ }
+ TestBinaryError::ProcessCrashed {
+ binary,
+ exit_code,
+ stdout,
+ stderr,
+ } => {
+ write!(
+ f,
+ "{} crashed with exit code {:?}\n=== STDOUT ===\n{}\n=== STDERR ===\n{}",
+ binary, exit_code, stdout, stderr
+ )
+ }
+ TestBinaryError::StartupTimeout {
+ binary,
+ timeout_secs,
+ } => {
+ write!(
+ f,
+ "{} failed to start within {} seconds",
+ binary, timeout_secs
+ )
+ }
+ TestBinaryError::HealthCheckFailed {
+ binary,
+ address,
+ retries,
+ } => {
+ write!(
+ f,
+ "Health check failed for {} at {} after {} retries",
+ binary, address, retries
+ )
+ }
+ TestBinaryError::ConfigLoad { path, message } => {
+ write!(f, "Failed to load config from {:?}: {}", path, message)
+ }
+ TestBinaryError::InvalidState { message } => {
+ write!(f, "Invalid state: {}", message)
+ }
+ TestBinaryError::ClientCreation {
+ transport,
+ address,
+ source,
+ } => {
+ write!(
+ f,
+ "Failed to create {} client for {}: {}",
+ transport, address, source
+ )
+ }
+ TestBinaryError::ClientConnection {
+ transport,
+ address,
+ source,
+ } => {
+ write!(
+ f,
+ "Failed to connect {} client to {}: {}",
+ transport, address, source
+ )
+ }
+ TestBinaryError::DirectoryCreation { path, source } => {
+ write!(f, "Failed to create directory {:?}: {}", path, source)
+ }
+ TestBinaryError::MissingServer => {
+ write!(f, "Server not configured in harness")
+ }
+ TestBinaryError::MissingMcp => {
+ write!(f, "MCP server not configured in harness")
+ }
+ TestBinaryError::MissingConnector => {
+ write!(f, "Connector runtime not configured in harness")
+ }
+ TestBinaryError::NotStarted => {
+ write!(f, "Harness not started")
+ }
+ TestBinaryError::AlreadyStarted => {
+ write!(f, "Harness already started")
+ }
+ TestBinaryError::SeedFailed(msg) => {
+ write!(f, "Seed function failed: {}", msg)
+ }
+ TestBinaryError::FixtureSetup {
+ fixture_type,
+ message,
+ } => {
+ write!(f, "Fixture setup failed for {}: {}", fixture_type, message)
+ }
+ }
+ }
+}
+
+impl std::error::Error for TestBinaryError {
+ fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+ match self {
+ TestBinaryError::Io(e) => Some(e),
+ TestBinaryError::ProcessSpawn { source, .. } => Some(source),
+ TestBinaryError::DirectoryCreation { source, .. } => Some(source),
+ _ => None,
+ }
+ }
+}
+
+impl From<io::Error> for TestBinaryError {
+ fn from(e: io::Error) -> Self {
+ TestBinaryError::Io(e)
+ }
+}
diff --git a/core/integration/src/harness/fixtures/mod.rs b/core/integration/src/harness/fixtures/mod.rs
new file mode 100644
index 000000000..e7a577b86
--- /dev/null
+++ b/core/integration/src/harness/fixtures/mod.rs
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+
+//! Test fixtures for connector and integration tests.
+//!
+//! Fixtures manage external dependencies (containers, databases) and provide
+//! environment variables to connector configuration.
+//!
+//! # Example
+//!
+//! ```ignore
+//! use integration::{iggy_harness, harness::fixtures::TestFixture};
+//!
+//! #[iggy_harness(server(connector(config_path = "postgres/sink.toml")))]
+//! async fn test_postgres_sink(client: &IggyClient, fixture: PostgresSinkFixture) {
+//! let pool = fixture.create_pool().await;
+//! // ...
+//! }
+//! ```
+
+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;
+use iggy::prelude::IggyClient;
+use std::collections::HashMap;
+
+/// Trait for test fixtures that manage external dependencies.
+///
+/// Fixtures are set up before the harness, provide environment variables
+/// for connector configuration, and optionally seed data after harness start.
+#[async_trait]
+pub trait TestFixture: Sized + Send {
+ /// Set up the fixture (e.g., start containers, create database connections).
+ async fn setup() -> Result<Self, TestBinaryError>;
+
+ /// Environment variables to pass to the connector configuration.
+ fn connector_envs(&self) -> HashMap<String, String>;
+
+ /// Optional seed function to run after harness start.
+ async fn seed(&self, _client: &IggyClient) -> Result<(), SeedError> {
+ Ok(())
+ }
+
+ /// Whether this fixture has a seed function that should be called.
+ fn has_seed(&self) -> bool {
+ false
+ }
+}
diff --git a/core/integration/src/harness/fixtures/postgres.rs b/core/integration/src/harness/fixtures/postgres.rs
new file mode 100644
index 000000000..39368e9dc
--- /dev/null
+++ b/core/integration/src/harness/fixtures/postgres.rs
@@ -0,0 +1,941 @@
+/*
+ * 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.
+ */
+
+//! PostgreSQL container fixtures for connector tests.
+
+use super::TestFixture;
+use crate::harness::error::TestBinaryError;
+use async_trait::async_trait;
+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 POSTGRES_PORT: u16 = 5432;
+const DEFAULT_POLL_ATTEMPTS: usize = 100;
+const DEFAULT_POLL_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_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_POSTGRES_PATH";
+
+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_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_SOURCE_PATH: &str = "IGGY_CONNECTORS_SOURCE_POSTGRES_PATH";
+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 DEFAULT_TEST_STREAM: &str = "test_stream";
+const DEFAULT_TEST_TOPIC: &str = "test_topic";
+const DEFAULT_SINK_TABLE: &str = "iggy_messages";
+
+/// 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 {
+ self.container().create_pool()
+ }
+}
+
+/// 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>,
+ ) -> impl std::future::Future<Output = i64> + Send + 'a {
+ async move {
+ let query = format!("SELECT COUNT(*) FROM {}", self.table_name());
+ let count: (i64,) = sqlx::query_as(&query)
+ .fetch_one(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to count rows: {e}"));
+ count.0
+ }
+ }
+}
+
+/// Base container management for PostgreSQL fixtures.
+pub struct PostgresContainer {
+ #[allow(dead_code)]
+ container: ContainerAsync<postgres::Postgres>,
+ connection_string: String,
+}
+
+impl PostgresContainer {
+ async fn start() -> Result<Self, TestBinaryError> {
+ let container = postgres::Postgres::default().start().await.map_err(|e| {
+ TestBinaryError::FixtureSetup {
+ fixture_type: "PostgresContainer".to_string(),
+ message: format!("Failed to start container: {e}"),
+ }
+ })?;
+
+ let host_port = container
+ .get_host_port_ipv4(POSTGRES_PORT)
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "PostgresContainer".to_string(),
+ message: format!("Failed to get port: {e}"),
+ })?;
+
+ let connection_string = format!("postgres://postgres:postgres@localhost:{host_port}");
+
+ Ok(Self {
+ container,
+ connection_string,
+ })
+ }
+
+ pub fn connection_string(&self) -> &str {
+ &self.connection_string
+ }
+
+ pub async fn create_pool(&self) -> Result<Pool<Postgres>, TestBinaryError> {
+ PgPoolOptions::new()
+ .max_connections(1)
+ .connect(&self.connection_string)
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "PostgresContainer".to_string(),
+ message: format!("Failed to connect: {e}"),
+ })
+ }
+}
+
+/// Payload format for sink connector.
+#[derive(Debug, Clone, Copy, Default)]
+pub enum SinkPayloadFormat {
+ /// Store as bytea (default)
+ #[default]
+ Bytea,
+ /// Store as JSONB
+ Json,
+}
+
+/// Schema format for message encoding.
+#[derive(Debug, Clone, Copy, Default)]
+pub enum SinkSchema {
+ /// JSON encoding
+ #[default]
+ Json,
+ /// Raw binary
+ Raw,
+}
+
+/// PostgreSQL sink connector fixture.
+///
+/// Starts a PostgreSQL container and provides environment variables
+/// for the sink connector to connect to it.
+pub struct PostgresSinkFixture {
+ container: PostgresContainer,
+ payload_format: SinkPayloadFormat,
+ schema: SinkSchema,
+}
+
+impl PostgresOps for PostgresSinkFixture {
+ fn container(&self) -> &PostgresContainer {
+ &self.container
+ }
+}
+
+impl PostgresSinkFixture {
+ /// Wait for a table to be created by the sink connector.
+ pub async fn wait_for_table(&self, pool: &Pool<Postgres>, table: &str) {
+ let query = format!("SELECT 1 FROM {table} LIMIT 1");
+ for _ in 0..DEFAULT_POLL_ATTEMPTS {
+ if sqlx::query(&query).fetch_optional(pool).await.is_ok() {
+ return;
+ }
+ sleep(Duration::from_millis(DEFAULT_POLL_INTERVAL_MS)).await;
+ }
+ panic!("Table {table} was not created in time");
+ }
+
+ /// Fetch rows from the sink table with polling until expected count is reached.
+ pub async fn fetch_rows_as<T>(
+ &self,
+ pool: &Pool<Postgres>,
+ query: &str,
+ expected_count: usize,
+ ) -> Vec<T>
+ where
+ T: Send + Unpin + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
+ {
+ let mut rows = Vec::new();
+ for _ in 0..DEFAULT_POLL_ATTEMPTS {
+ if let Ok(fetched) = sqlx::query_as::<_, T>(query).fetch_all(pool).await {
+ rows = fetched;
+ if rows.len() >= expected_count {
+ break;
+ }
+ }
+ sleep(Duration::from_millis(DEFAULT_POLL_INTERVAL_MS / 5)).await;
+ }
+ rows
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSinkFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = PostgresContainer::start().await?;
+ Ok(Self {
+ container,
+ payload_format: SinkPayloadFormat::default(),
+ schema: SinkSchema::default(),
+ })
+ }
+
+ fn connector_envs(&self) -> HashMap<String, String> {
+ let mut envs = HashMap::new();
+
+ envs.insert(
+ ENV_SINK_CONNECTION_STRING.to_string(),
+ self.container.connection_string.clone(),
+ );
+ envs.insert(
+ ENV_SINK_TARGET_TABLE.to_string(),
+ DEFAULT_SINK_TABLE.to_string(),
+ );
+ envs.insert(
+ ENV_SINK_STREAMS_0_STREAM.to_string(),
+ DEFAULT_TEST_STREAM.to_string(),
+ );
+ envs.insert(
+ ENV_SINK_STREAMS_0_TOPICS.to_string(),
+ format!("[{}]", DEFAULT_TEST_TOPIC),
+ );
+ envs.insert(
+ ENV_SINK_STREAMS_0_CONSUMER_GROUP.to_string(),
+ "test".to_string(),
+ );
+ envs.insert(
+ ENV_SINK_PATH.to_string(),
+ "../../target/debug/libiggy_connector_postgres_sink".to_string(),
+ );
+
+ let schema_str = match self.schema {
+ SinkSchema::Json => "json",
+ SinkSchema::Raw => "raw",
+ };
+ envs.insert(
+ ENV_SINK_STREAMS_0_SCHEMA.to_string(),
+ schema_str.to_string(),
+ );
+
+ let format_str = match self.payload_format {
+ SinkPayloadFormat::Bytea => "bytea",
+ SinkPayloadFormat::Json => "json",
+ };
+ envs.insert(ENV_SINK_PAYLOAD_FORMAT.to_string(), format_str.to_string());
+
+ envs
+ }
+}
+
+/// PostgreSQL sink fixture for bytea payload format.
+pub struct PostgresSinkByteaFixture {
+ inner: PostgresSinkFixture,
+}
+
+impl std::ops::Deref for PostgresSinkByteaFixture {
+ type Target = PostgresSinkFixture;
+ fn deref(&self) -> &Self::Target {
+ &self.inner
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSinkByteaFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = PostgresContainer::start().await?;
+ Ok(Self {
+ inner: PostgresSinkFixture {
+ container,
+ payload_format: SinkPayloadFormat::Bytea,
+ schema: SinkSchema::Raw,
+ },
+ })
+ }
+
+ fn connector_envs(&self) -> HashMap<String, String> {
+ self.inner.connector_envs()
+ }
+}
+
+/// PostgreSQL sink fixture for JSON payload format.
+pub struct PostgresSinkJsonFixture {
+ inner: PostgresSinkFixture,
+}
+
+impl std::ops::Deref for PostgresSinkJsonFixture {
+ type Target = PostgresSinkFixture;
+ fn deref(&self) -> &Self::Target {
+ &self.inner
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSinkJsonFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = PostgresContainer::start().await?;
+ Ok(Self {
+ inner: PostgresSinkFixture {
+ container,
+ payload_format: SinkPayloadFormat::Json,
+ schema: SinkSchema::Json,
+ },
+ })
+ }
+
+ fn connector_envs(&self) -> HashMap<String, String> {
+ self.inner.connector_envs()
+ }
+}
+
+/// 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
+ }
+
+ /// Execute a query (e.g., CREATE TABLE).
+ 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}"));
+ }
+
+ /// Count rows in a table.
+ 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
+ }
+
+ /// Count rows matching a condition.
+ 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 connector_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.
+pub struct PostgresSourceJsonFixture {
+ container: PostgresContainer,
+}
+
+impl PostgresOps for PostgresSourceJsonFixture {
+ fn container(&self) -> &PostgresContainer {
+ &self.container
+ }
+}
+
+impl PostgresSourceOps for PostgresSourceJsonFixture {
+ fn table_name(&self) -> &str {
+ Self::TABLE
+ }
+}
+
+impl PostgresSourceJsonFixture {
+ const TABLE: &'static str = "test_messages";
+
+ pub async fn create_table(&self, pool: &Pool<Postgres>) {
+ let query = format!(
+ "CREATE TABLE IF NOT EXISTS {} (
+ 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
+ )",
+ Self::TABLE
+ );
+ sqlx::query(&query)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to create table: {e}"));
+ }
+
+ #[allow(clippy::too_many_arguments)]
+ pub async fn insert_row(
+ &self,
+ pool: &Pool<Postgres>,
+ id: i32,
+ name: &str,
+ count: i32,
+ amount: f64,
+ active: bool,
+ timestamp: i64,
+ ) {
+ let query = format!(
+ "INSERT INTO {} (id, name, count, amount, active, timestamp) VALUES ($1, $2, $3, $4, $5, $6)",
+ Self::TABLE
+ );
+ sqlx::query(&query)
+ .bind(id)
+ .bind(name)
+ .bind(count)
+ .bind(amount)
+ .bind(active)
+ .bind(timestamp)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to insert row: {e}"));
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSourceJsonFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = PostgresContainer::start().await?;
+ Ok(Self { container })
+ }
+
+ fn connector_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));
+ envs.insert(ENV_SOURCE_TRACKING_COLUMN.to_string(), "id".to_string());
+ envs.insert(ENV_SOURCE_INCLUDE_METADATA.to_string(), "true".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_STREAMS_0_SCHEMA.to_string(), "json".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(),
+ );
+ envs
+ }
+}
+
+/// PostgreSQL source fixture for bytea payload column.
+pub struct PostgresSourceByteaFixture {
+ container: PostgresContainer,
+}
+
+impl PostgresOps for PostgresSourceByteaFixture {
+ fn container(&self) -> &PostgresContainer {
+ &self.container
+ }
+}
+
+impl PostgresSourceOps for PostgresSourceByteaFixture {
+ fn table_name(&self) -> &str {
+ Self::TABLE
+ }
+}
+
+impl PostgresSourceByteaFixture {
+ const TABLE: &'static str = "test_payloads";
+
+ pub async fn create_table(&self, pool: &Pool<Postgres>) {
+ let query = format!(
+ "CREATE TABLE IF NOT EXISTS {} (
+ id SERIAL PRIMARY KEY,
+ payload BYTEA NOT NULL
+ )",
+ Self::TABLE
+ );
+ sqlx::query(&query)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to create table: {e}"));
+ }
+
+ pub async fn insert_payload(&self, pool: &Pool<Postgres>, id: i32, payload: &[u8]) {
+ let query = format!("INSERT INTO {} (id, payload) VALUES ($1, $2)", Self::TABLE);
+ sqlx::query(&query)
+ .bind(id)
+ .bind(payload)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to insert payload: {e}"));
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSourceByteaFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = PostgresContainer::start().await?;
+ Ok(Self { container })
+ }
+
+ fn connector_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));
+ envs.insert(ENV_SOURCE_TRACKING_COLUMN.to_string(), "id".to_string());
+ envs.insert(ENV_SOURCE_PAYLOAD_COLUMN.to_string(), "payload".to_string());
+ envs.insert(ENV_SOURCE_PAYLOAD_FORMAT.to_string(), "bytea".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_STREAMS_0_SCHEMA.to_string(), "raw".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(),
+ );
+ envs
+ }
+}
+
+/// PostgreSQL source fixture for JSONB payload column.
+pub struct PostgresSourceJsonbFixture {
+ container: PostgresContainer,
+}
+
+impl PostgresOps for PostgresSourceJsonbFixture {
+ fn container(&self) -> &PostgresContainer {
+ &self.container
+ }
+}
+
+impl PostgresSourceOps for PostgresSourceJsonbFixture {
+ fn table_name(&self) -> &str {
+ Self::TABLE
+ }
+}
+
+impl PostgresSourceJsonbFixture {
+ const TABLE: &'static str = "test_json_payloads";
+
+ pub async fn create_table(&self, pool: &Pool<Postgres>) {
+ let query = format!(
+ "CREATE TABLE IF NOT EXISTS {} (
+ id SERIAL PRIMARY KEY,
+ data JSONB NOT NULL
+ )",
+ Self::TABLE
+ );
+ sqlx::query(&query)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to create table: {e}"));
+ }
+
+ pub async fn insert_json(&self, pool: &Pool<Postgres>, id: i32, data: &serde_json::Value) {
+ let query = format!("INSERT INTO {} (id, data) VALUES ($1, $2)", Self::TABLE);
+ sqlx::query(&query)
+ .bind(id)
+ .bind(data)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to insert json: {e}"));
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSourceJsonbFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = PostgresContainer::start().await?;
+ Ok(Self { container })
+ }
+
+ fn connector_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));
+ envs.insert(ENV_SOURCE_TRACKING_COLUMN.to_string(), "id".to_string());
+ envs.insert(ENV_SOURCE_PAYLOAD_COLUMN.to_string(), "data".to_string());
+ envs.insert(
+ ENV_SOURCE_PAYLOAD_FORMAT.to_string(),
+ "json_direct".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_STREAMS_0_SCHEMA.to_string(), "json".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(),
+ );
+ envs
+ }
+}
+
+/// PostgreSQL source fixture with delete_after_read enabled.
+pub struct PostgresSourceDeleteFixture {
+ container: PostgresContainer,
+}
+
+impl PostgresOps for PostgresSourceDeleteFixture {
+ fn container(&self) -> &PostgresContainer {
+ &self.container
+ }
+}
+
+impl PostgresSourceOps for PostgresSourceDeleteFixture {
+ fn table_name(&self) -> &str {
+ Self::TABLE
+ }
+}
+
+impl PostgresSourceDeleteFixture {
+ const TABLE: &'static str = "test_delete_rows";
+
+ pub async fn create_table(&self, pool: &Pool<Postgres>) {
+ let query = format!(
+ "CREATE TABLE IF NOT EXISTS {} (
+ id SERIAL PRIMARY KEY,
+ name VARCHAR(255) NOT NULL,
+ value INTEGER NOT NULL
+ )",
+ Self::TABLE
+ );
+ sqlx::query(&query)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to create table: {e}"));
+ }
+
+ pub async fn insert_row(&self, pool: &Pool<Postgres>, name: &str, value: i32) {
+ let query = format!("INSERT INTO {} (name, value) VALUES ($1, $2)", Self::TABLE);
+ sqlx::query(&query)
+ .bind(name)
+ .bind(value)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to insert row: {e}"));
+ }
+
+ pub async fn count_rows(&self, pool: &Pool<Postgres>) -> i64 {
+ PostgresSourceOps::count_rows(self, pool).await
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSourceDeleteFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = PostgresContainer::start().await?;
+ Ok(Self { container })
+ }
+
+ fn connector_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));
+ envs.insert(ENV_SOURCE_TRACKING_COLUMN.to_string(), "id".to_string());
+ envs.insert(ENV_SOURCE_PRIMARY_KEY_COLUMN.to_string(), "id".to_string());
+ envs.insert(ENV_SOURCE_DELETE_AFTER_READ.to_string(), "true".to_string());
+ envs.insert(ENV_SOURCE_INCLUDE_METADATA.to_string(), "true".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_STREAMS_0_SCHEMA.to_string(), "json".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(),
+ );
+ envs
+ }
+}
+
+/// PostgreSQL source fixture with processed_column marking.
+pub struct PostgresSourceMarkFixture {
+ container: PostgresContainer,
+}
+
+impl PostgresOps for PostgresSourceMarkFixture {
+ fn container(&self) -> &PostgresContainer {
+ &self.container
+ }
+}
+
+impl PostgresSourceOps for PostgresSourceMarkFixture {
+ fn table_name(&self) -> &str {
+ Self::TABLE
+ }
+}
+
+impl PostgresSourceMarkFixture {
+ const TABLE: &'static str = "test_mark_rows";
+
+ pub async fn create_table(&self, pool: &Pool<Postgres>) {
+ let query = format!(
+ "CREATE TABLE IF NOT EXISTS {} (
+ id SERIAL PRIMARY KEY,
+ name VARCHAR(255) NOT NULL,
+ value INTEGER NOT NULL,
+ is_processed BOOLEAN NOT NULL DEFAULT FALSE
+ )",
+ Self::TABLE
+ );
+ sqlx::query(&query)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to create table: {e}"));
+ }
+
+ pub async fn insert_row(&self, pool: &Pool<Postgres>, name: &str, value: i32) {
+ let query = format!(
+ "INSERT INTO {} (name, value, is_processed) VALUES ($1, $2, $3)",
+ Self::TABLE
+ );
+ sqlx::query(&query)
+ .bind(name)
+ .bind(value)
+ .bind(false)
+ .execute(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to insert row: {e}"));
+ }
+
+ pub async fn count_rows(&self, pool: &Pool<Postgres>) -> i64 {
+ PostgresSourceOps::count_rows(self, pool).await
+ }
+
+ pub async fn count_unprocessed(&self, pool: &Pool<Postgres>) -> i64 {
+ let query = format!(
+ "SELECT COUNT(*) FROM {} WHERE is_processed = FALSE",
+ Self::TABLE
+ );
+ let count: (i64,) = sqlx::query_as(&query)
+ .fetch_one(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to count rows: {e}"));
+ count.0
+ }
+
+ pub async fn count_processed(&self, pool: &Pool<Postgres>) -> i64 {
+ let query = format!(
+ "SELECT COUNT(*) FROM {} WHERE is_processed = TRUE",
+ Self::TABLE
+ );
+ let count: (i64,) = sqlx::query_as(&query)
+ .fetch_one(pool)
+ .await
+ .unwrap_or_else(|e| panic!("Failed to count rows: {e}"));
+ count.0
+ }
+}
+
+#[async_trait]
+impl TestFixture for PostgresSourceMarkFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = PostgresContainer::start().await?;
+ Ok(Self { container })
+ }
+
+ fn connector_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));
+ envs.insert(ENV_SOURCE_TRACKING_COLUMN.to_string(), "id".to_string());
+ envs.insert(ENV_SOURCE_PRIMARY_KEY_COLUMN.to_string(), "id".to_string());
+ envs.insert(
+ ENV_SOURCE_PROCESSED_COLUMN.to_string(),
+ "is_processed".to_string(),
+ );
+ envs.insert(ENV_SOURCE_INCLUDE_METADATA.to_string(), "true".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_STREAMS_0_SCHEMA.to_string(), "json".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(),
+ );
+ envs
+ }
+}
diff --git a/core/integration/src/lib.rs b/core/integration/src/harness/fixtures/random.rs
similarity index 53%
copy from core/integration/src/lib.rs
copy to core/integration/src/harness/fixtures/random.rs
index 269698b2b..3f072aef8 100644
--- a/core/integration/src/lib.rs
+++ b/core/integration/src/harness/fixtures/random.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * 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
@@ -16,21 +17,26 @@
* under the License.
*/
-pub mod bench_utils;
-pub mod file;
-#[allow(deprecated)]
-pub mod http_client;
-#[allow(deprecated)]
-pub mod quic_client;
-#[allow(deprecated)]
-pub mod tcp_client;
-#[allow(deprecated)]
-pub mod websocket_client;
+//! Random source connector fixture.
-#[allow(deprecated)]
-pub mod test_connectors_runtime;
-#[allow(deprecated)]
-pub mod test_mcp_server;
-#[allow(deprecated)]
-pub mod test_server;
-pub mod test_tls_utils;
+use super::TestFixture;
+use crate::harness::error::TestBinaryError;
+use async_trait::async_trait;
+use std::collections::HashMap;
+
+/// Fixture for the random source connector.
+///
+/// The random source connector doesn't require any external dependencies,
+/// so this fixture is minimal.
+pub struct RandomSourceFixture;
+
+#[async_trait]
+impl TestFixture for RandomSourceFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ Ok(Self)
+ }
+
+ fn connector_envs(&self) -> HashMap<String, String> {
+ HashMap::new()
+ }
+}
diff --git a/core/integration/src/harness/fixtures/wiremock.rs b/core/integration/src/harness/fixtures/wiremock.rs
new file mode 100644
index 000000000..ebfc81ce6
--- /dev/null
+++ b/core/integration/src/harness/fixtures/wiremock.rs
@@ -0,0 +1,153 @@
+/*
+ * 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.
+ */
+
+//! WireMock container fixtures for HTTP config provider tests.
+
+use super::TestFixture;
+use crate::harness::error::TestBinaryError;
+use async_trait::async_trait;
+use std::collections::HashMap;
+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};
+
+const WIREMOCK_PORT: u16 = 8080;
+
+/// Base WireMock container management.
+struct WireMockContainer {
+ #[allow(dead_code)]
+ container: ContainerAsync<GenericImage>,
+ base_url: String,
+}
+
+impl WireMockContainer {
+ async fn start(mode: &str) -> Result<Self, TestBinaryError> {
+ let current_dir = std::env::current_dir().map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "WireMockContainer".to_string(),
+ message: format!("Failed to get current dir: {e}"),
+ })?;
+
+ let container = GenericImage::new("wiremock/wiremock", "latest")
+ .with_exposed_port(WIREMOCK_PORT.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
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "WireMockContainer".to_string(),
+ message: format!("Failed to start container: {e}"),
+ })?;
+
+ let host = container
+ .get_host()
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "WireMockContainer".to_string(),
+ message: format!("Failed to get host: {e}"),
+ })?;
+
+ let host_port = container
+ .get_host_port_ipv4(WIREMOCK_PORT)
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "WireMockContainer".to_string(),
+ message: format!("Failed to get port: {e}"),
+ })?;
+
+ let base_url = format!("http://{host}:{host_port}");
+
+ Ok(Self {
+ container,
+ base_url,
+ })
+ }
+}
+
+/// WireMock fixture for "direct" response format tests.
+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> {
+ let container = WireMockContainer::start("direct").await?;
+ Ok(Self { container })
+ }
+
+ fn connector_envs(&self) -> HashMap<String, String> {
+ let mut envs = HashMap::new();
+ envs.insert(
+ "IGGY_CONNECTORS_CONNECTORS_BASE_URL".to_string(),
+ self.container.base_url.clone(),
+ );
+ envs
+ }
+}
+
+/// WireMock fixture for "wrapped" response format tests.
+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> {
+ let container = WireMockContainer::start("wrapped").await?;
+ Ok(Self { container })
+ }
+
+ fn connector_envs(&self) -> HashMap<String, String> {
+ let mut envs = HashMap::new();
+ envs.insert(
+ "IGGY_CONNECTORS_CONNECTORS_BASE_URL".to_string(),
+ self.container.base_url.clone(),
+ );
+ envs
+ }
+}
diff --git a/core/integration/src/harness/handle/client.rs b/core/integration/src/harness/handle/client.rs
new file mode 100644
index 000000000..edc5b6f87
--- /dev/null
+++ b/core/integration/src/harness/handle/client.rs
@@ -0,0 +1,230 @@
+/*
+ * 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 crate::harness::config::ClientConfig;
+use crate::harness::error::TestBinaryError;
+use iggy::http::http_client::HttpClient;
+use iggy::prelude::{
+ Client, ClientWrapper, HttpClientConfig, QuicClientConfig, TcpClient, TcpClientConfig,
+ UserClient, WebSocketClientConfig,
+};
+use iggy::quic::quic_client::QuicClient;
+use iggy::websocket::websocket_client::WebSocketClient;
+use iggy_common::TransportProtocol;
+use std::net::SocketAddr;
+use std::ops::Deref;
+use std::sync::Arc;
+
+pub struct ClientHandle {
+ config: ClientConfig,
+ address: SocketAddr,
+ client: Option<ClientWrapper>,
+}
+
+impl std::fmt::Debug for ClientHandle {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("ClientHandle")
+ .field("transport", &self.config.transport)
+ .field("address", &self.address)
+ .field("connected", &self.client.is_some())
+ .finish()
+ }
+}
+
+impl ClientHandle {
+ pub fn new(config: ClientConfig, address: SocketAddr) -> Self {
+ Self {
+ config,
+ address,
+ client: None,
+ }
+ }
+
+ pub async fn connect(&mut self) -> Result<(), TestBinaryError> {
+ let client = self.create_client().await?;
+
+ if let Some(ref login) = self.config.auto_login {
+ let _: iggy::prelude::IdentityInfo = client
+ .login_user(&login.username, &login.password)
+ .await
+ .map_err(|e| TestBinaryError::ClientConnection {
+ transport: format!("{:?}", self.config.transport),
+ address: self.address.to_string(),
+ source: format!("Auto-login failed: {}", e),
+ })?;
+ }
+
+ self.client = Some(client);
+ Ok(())
+ }
+
+ pub async fn reconnect(&mut self) -> Result<(), TestBinaryError> {
+ self.disconnect().await;
+ self.connect().await
+ }
+
+ pub async fn disconnect(&mut self) {
+ if let Some(ref client) = self.client {
+ let _ = client.disconnect().await;
+ }
+ self.client = None;
+ }
+
+ pub fn update_address(&mut self, address: SocketAddr) {
+ self.address = address;
+ }
+
+ pub fn transport(&self) -> TransportProtocol {
+ self.config.transport
+ }
+
+ pub fn address(&self) -> SocketAddr {
+ self.address
+ }
+
+ pub fn inner(&self) -> Option<&ClientWrapper> {
+ self.client.as_ref()
+ }
+
+ async fn create_client(&self) -> Result<ClientWrapper, TestBinaryError> {
+ match self.config.transport {
+ TransportProtocol::Tcp => self.create_tcp_client().await,
+ TransportProtocol::Quic => self.create_quic_client().await,
+ TransportProtocol::Http => self.create_http_client().await,
+ TransportProtocol::WebSocket => self.create_websocket_client().await,
+ }
+ }
+
+ async fn create_tcp_client(&self) -> Result<ClientWrapper, TestBinaryError> {
+ let config = TcpClientConfig {
+ server_address: self.address.to_string(),
+ nodelay: self.config.tcp_nodelay,
+ tls_enabled: self.config.tls_enabled,
+ tls_domain: self.config.tls_domain.clone().unwrap_or_default(),
+ tls_ca_file: self
+ .config
+ .tls_ca_file
+ .as_ref()
+ .map(|p| p.display().to_string()),
+ tls_validate_certificate: self.config.tls_validate_certificate,
+ ..TcpClientConfig::default()
+ };
+
+ let client =
+ TcpClient::create(Arc::new(config)).map_err(|e| TestBinaryError::ClientCreation {
+ transport: "TCP".to_string(),
+ address: self.address.to_string(),
+ source: e.to_string(),
+ })?;
+
+ Client::connect(&client)
+ .await
+ .map_err(|e| TestBinaryError::ClientConnection {
+ transport: "TCP".to_string(),
+ address: self.address.to_string(),
+ source: e.to_string(),
+ })?;
+
+ Ok(ClientWrapper::Tcp(client))
+ }
+
+ async fn create_quic_client(&self) -> Result<ClientWrapper, TestBinaryError> {
+ let config = QuicClientConfig {
+ server_address: self.address.to_string(),
+ max_idle_timeout: 2_000_000,
+ ..QuicClientConfig::default()
+ };
+
+ let client =
+ QuicClient::create(Arc::new(config)).map_err(|e| TestBinaryError::ClientCreation {
+ transport: "QUIC".to_string(),
+ address: self.address.to_string(),
+ source: e.to_string(),
+ })?;
+
+ Client::connect(&client)
+ .await
+ .map_err(|e| TestBinaryError::ClientConnection {
+ transport: "QUIC".to_string(),
+ address: self.address.to_string(),
+ source: e.to_string(),
+ })?;
+
+ Ok(ClientWrapper::Quic(client))
+ }
+
+ async fn create_http_client(&self) -> Result<ClientWrapper, TestBinaryError> {
+ let config = HttpClientConfig {
+ api_url: format!("http://{}", self.address),
+ ..HttpClientConfig::default()
+ };
+
+ let client =
+ HttpClient::create(Arc::new(config)).map_err(|e| TestBinaryError::ClientCreation {
+ transport: "HTTP".to_string(),
+ address: self.address.to_string(),
+ source: e.to_string(),
+ })?;
+
+ Ok(ClientWrapper::Http(client))
+ }
+
+ async fn create_websocket_client(&self) -> Result<ClientWrapper, TestBinaryError> {
+ let config = WebSocketClientConfig {
+ server_address: self.address.to_string(),
+ tls_enabled: self.config.tls_enabled,
+ tls_domain: self.config.tls_domain.clone().unwrap_or_default(),
+ tls_ca_file: self
+ .config
+ .tls_ca_file
+ .as_ref()
+ .map(|p| p.display().to_string()),
+ tls_validate_certificate: self.config.tls_validate_certificate,
+ ..WebSocketClientConfig::default()
+ };
+
+ let client = WebSocketClient::create(Arc::new(config)).map_err(|e| {
+ TestBinaryError::ClientCreation {
+ transport: "WebSocket".to_string(),
+ address: self.address.to_string(),
+ source: e.to_string(),
+ }
+ })?;
+
+ Client::connect(&client)
+ .await
+ .map_err(|e| TestBinaryError::ClientConnection {
+ transport: "WebSocket".to_string(),
+ address: self.address.to_string(),
+ source: e.to_string(),
+ })?;
+
+ Ok(ClientWrapper::WebSocket(client))
+ }
+}
+
+impl Deref for ClientHandle {
+ type Target = ClientWrapper;
+
+ fn deref(&self) -> &Self::Target {
+ self.client
+ .as_ref()
+ .expect("Client not connected. Call connect() first.")
+ }
+}
diff --git a/core/integration/src/harness/handle/common.rs b/core/integration/src/harness/handle/common.rs
new file mode 100644
index 000000000..a6e7e4b9f
--- /dev/null
+++ b/core/integration/src/harness/handle/common.rs
@@ -0,0 +1,76 @@
+/*
+ * 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.
+ */
+
+//! Shared utilities for test binary handles.
+
+use rand::Rng as _;
+use std::fs;
+use std::net::{Ipv4Addr, SocketAddr, TcpListener};
+use std::ops::RangeInclusive;
+use std::path::PathBuf;
+
+pub const DEFAULT_HEALTH_CHECK_RETRIES: u32 = 1000;
+pub const DEFAULT_HEALTH_CHECK_INTERVAL_MS: u64 = 20;
+
+/// Collect stdout and stderr logs from file paths.
+pub fn collect_logs(
+ stdout_path: &Option<PathBuf>,
+ stderr_path: &Option<PathBuf>,
+) -> (String, String) {
+ let stdout = stdout_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stdout log]".to_string());
+
+ let stderr = stderr_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stderr log]".to_string());
+
+ (stdout, stderr)
+}
+
+/// Dump logs to stderr if we're panicking (for Drop impls).
+pub fn dump_logs_on_panic(
+ binary_name: &str,
+ stdout_path: &Option<PathBuf>,
+ stderr_path: &Option<PathBuf>,
+) {
+ if std::thread::panicking() {
+ let (stdout, stderr) = collect_logs(stdout_path, stderr_path);
+ eprintln!("{} stdout:\n{}", binary_name, stdout);
+ eprintln!("{} stderr:\n{}", binary_name, stderr);
+ }
+}
+
+/// Find an available port within the given range.
+pub fn find_available_port(range: RangeInclusive<u16>) -> SocketAddr {
+ let mut rng = rand::rng();
+ const MAX_RETRIES: u32 = 100;
+
+ for _ in 0..MAX_RETRIES {
+ let port = rng.random_range(range.clone());
+ let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), port);
+ if TcpListener::bind(addr).is_ok() {
+ return addr;
+ }
+ }
+
+ panic!("Failed to find available port after {MAX_RETRIES} retries");
+}
diff --git a/core/integration/src/harness/handle/connector.rs b/core/integration/src/harness/handle/connector.rs
new file mode 100644
index 000000000..87fb0fbb9
--- /dev/null
+++ b/core/integration/src/harness/handle/connector.rs
@@ -0,0 +1,277 @@
+/*
+ * 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::common;
+use crate::harness::config::ConnectorConfig;
+use crate::harness::context::TestContext;
+use crate::harness::error::TestBinaryError;
+use crate::harness::traits::{IggyDependent, TestBinary};
+use assert_cmd::prelude::CommandCargoExt;
+use std::collections::HashMap;
+use std::fs::{self, File, OpenOptions};
+use std::io::Write;
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::time::sleep;
+
+const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
+
+pub struct ConnectorHandle {
+ config: ConnectorConfig,
+ context: Arc<TestContext>,
+ envs: HashMap<String, String>,
+ child_handle: Option<Child>,
+ server_address: SocketAddr,
+ iggy_address: Option<SocketAddr>,
+ stdout_path: Option<PathBuf>,
+ stderr_path: Option<PathBuf>,
+}
+
+impl std::fmt::Debug for ConnectorHandle {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("ConnectorHandle")
+ .field("server_address", &self.server_address)
+ .field("iggy_address", &self.iggy_address)
+ .field("is_running", &self.child_handle.is_some())
+ .finish_non_exhaustive()
+ }
+}
+
+impl ConnectorHandle {
+ pub fn http_address(&self) -> SocketAddr {
+ self.server_address
+ }
+
+ pub fn http_url(&self) -> String {
+ format!(
+ "http://{}:{}",
+ self.server_address.ip(),
+ self.server_address.port()
+ )
+ }
+
+ pub fn state_path(&self) -> PathBuf {
+ self.context.connector_state_path()
+ }
+
+ pub fn collect_logs(&self) -> (String, String) {
+ common::collect_logs(&self.stdout_path, &self.stderr_path)
+ }
+
+ fn find_available_port() -> SocketAddr {
+ common::find_available_port(20000..=29999)
+ }
+
+ fn build_envs(&mut self) {
+ let state_path = self.context.connector_state_path();
+ self.envs.insert(
+ "IGGY_CONNECTORS_STATE_PATH".to_string(),
+ state_path.display().to_string(),
+ );
+ self.envs.insert(
+ "IGGY_CONNECTORS_HTTP_ADDRESS".to_string(),
+ self.server_address.to_string(),
+ );
+
+ if let Some(addr) = self.iggy_address {
+ self.envs
+ .insert("IGGY_CONNECTORS_IGGY_ADDRESS".to_string(), addr.to_string());
+ }
+
+ if let Some(ref config_path) = self.config.config_path {
+ self.envs.insert(
+ "IGGY_CONNECTORS_CONFIG_PATH".to_string(),
+ config_path.display().to_string(),
+ );
+ }
+
+ for (k, v) in &self.config.extra_envs {
+ self.envs.insert(k.clone(), v.clone());
+ }
+ }
+}
+
+impl TestBinary for ConnectorHandle {
+ type Config = ConnectorConfig;
+
+ fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+ let server_address = Self::find_available_port();
+
+ Self {
+ config,
+ context,
+ envs: HashMap::new(),
+ child_handle: None,
+ server_address,
+ iggy_address: None,
+ stdout_path: None,
+ stderr_path: None,
+ }
+ }
+
+ fn start(&mut self) -> Result<(), TestBinaryError> {
+ self.build_envs();
+
+ #[allow(deprecated)]
+ let mut command = if let Some(ref path) = self.config.executable_path {
+ Command::new(path)
+ } else {
+ Command::cargo_bin("iggy-connectors").map_err(|e| TestBinaryError::ProcessSpawn {
+ binary: "iggy-connectors".to_string(),
+ source: std::io::Error::other(e.to_string()),
+ })?
+ };
+
+ command.envs(self.envs.clone());
+
+ let verbose = std::env::var(TEST_VERBOSITY_ENV_VAR).is_ok()
+ || self.envs.contains_key(TEST_VERBOSITY_ENV_VAR);
+
+ if verbose {
+ command.stdout(Stdio::inherit());
+ command.stderr(Stdio::inherit());
+ } else {
+ let stdout_path = self.context.connector_stdout_path();
+ let stderr_path = self.context.connector_stderr_path();
+
+ let stdout_file =
+ File::create(&stdout_path).map_err(|e| TestBinaryError::DirectoryCreation {
+ path: stdout_path.clone(),
+ source: e,
+ })?;
+ let stderr_file =
+ File::create(&stderr_path).map_err(|e| TestBinaryError::DirectoryCreation {
+ path: stderr_path.clone(),
+ source: e,
+ })?;
+
+ command.stdout(stdout_file);
+ command.stderr(stderr_file);
+
+ self.stdout_path = Some(fs::canonicalize(&stdout_path)?);
+ self.stderr_path = Some(fs::canonicalize(&stderr_path)?);
+ }
+
+ let child = command.spawn().map_err(|e| TestBinaryError::ProcessSpawn {
+ binary: "iggy-connectors".to_string(),
+ source: e,
+ })?;
+ self.child_handle = Some(child);
+
+ Ok(())
+ }
+
+ fn stop(&mut self) -> Result<(), TestBinaryError> {
+ #[allow(unused_mut)]
+ if let Some(mut child) = self.child_handle.take() {
+ #[cfg(unix)]
+ unsafe {
+ libc::kill(child.id() as libc::pid_t, libc::SIGTERM);
+ }
+
+ #[cfg(not(unix))]
+ child.kill().ok();
+
+ if let Ok(output) = child.wait_with_output() {
+ let stderr = String::from_utf8_lossy(&output.stderr);
+ let stdout = String::from_utf8_lossy(&output.stdout);
+
+ if let Some(path) = self.stderr_path.as_ref()
+ && let Ok(mut f) = OpenOptions::new().append(true).create(true).open(path)
+ {
+ let _ = f.write_all(stderr.as_bytes());
+ }
+
+ if let Some(path) = self.stdout_path.as_ref()
+ && let Ok(mut f) = OpenOptions::new().append(true).create(true).open(path)
+ {
+ let _ = f.write_all(stdout.as_bytes());
+ }
+ }
+ }
+
+ Ok(())
+ }
+
+ fn is_running(&self) -> bool {
+ self.child_handle.is_some()
+ }
+
+ fn assert_running(&self) {
+ if let Some(ref child) = self.child_handle {
+ let mut child_clone = unsafe { std::ptr::read(child as *const Child) };
+ if let Ok(Some(status)) = child_clone.try_wait() {
+ let (stdout, stderr) = self.collect_logs();
+ panic!(
+ "Connector runtime has crashed with exit status: {}\n\n\
+ === STDOUT ===\n{}\n\n\
+ === STDERR ===\n{}",
+ status, stdout, stderr
+ );
+ }
+ std::mem::forget(child_clone);
+ }
+ }
+
+ fn pid(&self) -> Option<u32> {
+ self.child_handle.as_ref().map(|c| c.id())
+ }
+}
+
+impl IggyDependent for ConnectorHandle {
+ fn set_iggy_address(&mut self, addr: SocketAddr) {
+ self.iggy_address = Some(addr);
+ }
+
+ async fn wait_ready(&self) -> Result<(), TestBinaryError> {
+ let http_address = self.http_url();
+ let client = reqwest::Client::new();
+
+ for retry in 0..common::DEFAULT_HEALTH_CHECK_RETRIES {
+ match client.get(&http_address).send().await {
+ Ok(_) => return Ok(()),
+ Err(_) => {
+ if retry == common::DEFAULT_HEALTH_CHECK_RETRIES - 1 {
+ return Err(TestBinaryError::HealthCheckFailed {
+ binary: "iggy-connectors".to_string(),
+ address: http_address,
+ retries: common::DEFAULT_HEALTH_CHECK_RETRIES,
+ });
+ }
+ sleep(Duration::from_millis(
+ common::DEFAULT_HEALTH_CHECK_INTERVAL_MS,
+ ))
+ .await;
+ }
+ }
+ }
+
+ Ok(())
+ }
+}
+
+impl Drop for ConnectorHandle {
+ fn drop(&mut self) {
+ let _ = self.stop();
+ common::dump_logs_on_panic("Iggy connectors", &self.stdout_path, &self.stderr_path);
+ }
+}
diff --git a/core/integration/src/harness/handle/mcp.rs b/core/integration/src/harness/handle/mcp.rs
new file mode 100644
index 000000000..9416bb561
--- /dev/null
+++ b/core/integration/src/harness/handle/mcp.rs
@@ -0,0 +1,296 @@
+/*
+ * 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::common;
+use crate::harness::config::McpConfig;
+use crate::harness::context::TestContext;
+use crate::harness::error::TestBinaryError;
+use crate::harness::traits::{IggyDependent, TestBinary};
+use assert_cmd::prelude::CommandCargoExt;
+use rmcp::{
+ RoleClient, ServiceExt,
+ model::{ClientCapabilities, ClientInfo, Implementation, InitializeRequestParams},
+ service::RunningService,
+ transport::StreamableHttpClientTransport,
+};
+use std::collections::HashMap;
+use std::fs::{self, File};
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::time::sleep;
+
+const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
+const MCP_HEALTH_CHECK_RETRIES: u32 = common::DEFAULT_HEALTH_CHECK_RETRIES * 3;
+
+pub type McpClient = RunningService<RoleClient, InitializeRequestParams>;
+
+pub struct McpHandle {
+ config: McpConfig,
+ context: Arc<TestContext>,
+ envs: HashMap<String, String>,
+ child_handle: Option<Child>,
+ server_address: SocketAddr,
+ iggy_address: Option<SocketAddr>,
+ stdout_path: Option<PathBuf>,
+ stderr_path: Option<PathBuf>,
+}
+
+impl std::fmt::Debug for McpHandle {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("McpHandle")
+ .field("server_address", &self.server_address)
+ .field("iggy_address", &self.iggy_address)
+ .field("is_running", &self.child_handle.is_some())
+ .finish_non_exhaustive()
+ }
+}
+
+impl McpHandle {
+ pub fn http_address(&self) -> SocketAddr {
+ self.server_address
+ }
+
+ pub fn mcp_url(&self) -> String {
+ format!(
+ "http://{}:{}{}",
+ self.server_address.ip(),
+ self.server_address.port(),
+ self.config.http_path
+ )
+ }
+
+ pub async fn create_client(&self) -> Result<McpClient, TestBinaryError> {
+ let mcp_url = self.mcp_url();
+ let transport = StreamableHttpClientTransport::from_uri(mcp_url.clone());
+ let client_info = ClientInfo {
+ protocol_version: Default::default(),
+ capabilities: ClientCapabilities::default(),
+ client_info: Implementation {
+ name: "test-mcp-client".to_string(),
+ version: "1.0.0".to_string(),
+ ..Default::default()
+ },
+ meta: None,
+ };
+
+ client_info
+ .serve(transport)
+ .await
+ .map_err(|e| TestBinaryError::ClientCreation {
+ transport: "MCP".to_string(),
+ address: mcp_url,
+ source: e.to_string(),
+ })
+ }
+
+ pub fn collect_logs(&self) -> (String, String) {
+ common::collect_logs(&self.stdout_path, &self.stderr_path)
+ }
+
+ fn find_available_port() -> SocketAddr {
+ #[cfg(target_os = "linux")]
+ const PORT_RANGE: std::ops::RangeInclusive<u16> = 20000..=29999;
+ #[cfg(not(target_os = "linux"))]
+ const PORT_RANGE: std::ops::RangeInclusive<u16> = 20000..=49151;
+
+ common::find_available_port(PORT_RANGE)
+ }
+
+ fn build_envs(&mut self) {
+ self.envs.insert(
+ "IGGY_MCP_HTTP_PATH".to_string(),
+ self.config.http_path.clone(),
+ );
+ self.envs.insert(
+ "IGGY_MCP_HTTP_ADDRESS".to_string(),
+ self.server_address.to_string(),
+ );
+ self.envs.insert(
+ "IGGY_MCP_IGGY_CONSUMER".to_string(),
+ self.config.consumer_name.clone(),
+ );
+ self.envs
+ .insert("IGGY_MCP_TRANSPORT".to_string(), "http".to_string());
+
+ if let Some(addr) = self.iggy_address {
+ self.envs
+ .insert("IGGY_MCP_IGGY_ADDRESS".to_string(), addr.to_string());
+ }
+
+ for (k, v) in &self.config.extra_envs {
+ self.envs.insert(k.clone(), v.clone());
+ }
+ }
+}
+
+impl TestBinary for McpHandle {
+ type Config = McpConfig;
+
+ fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+ let server_address = Self::find_available_port();
+
+ Self {
+ config,
+ context,
+ envs: HashMap::new(),
+ child_handle: None,
+ server_address,
+ iggy_address: None,
+ stdout_path: None,
+ stderr_path: None,
+ }
+ }
+
+ #[allow(deprecated)]
+ fn start(&mut self) -> Result<(), TestBinaryError> {
+ self.build_envs();
+
+ let mut command = if let Some(ref path) = self.config.executable_path {
+ Command::new(path)
+ } else {
+ Command::cargo_bin("iggy-mcp").map_err(|e| TestBinaryError::ProcessSpawn {
+ binary: "iggy-mcp".to_string(),
+ source: std::io::Error::other(e.to_string()),
+ })?
+ };
+
+ command.envs(self.envs.clone());
+
+ let verbose = std::env::var(TEST_VERBOSITY_ENV_VAR).is_ok()
+ || self.envs.contains_key(TEST_VERBOSITY_ENV_VAR);
+
+ if verbose {
+ 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_file =
+ File::create(&stdout_path).map_err(|e| TestBinaryError::DirectoryCreation {
+ path: stdout_path.clone(),
+ source: e,
+ })?;
+ let stderr_file =
+ File::create(&stderr_path).map_err(|e| TestBinaryError::DirectoryCreation {
+ path: stderr_path.clone(),
+ source: e,
+ })?;
+
+ command.stdout(stdout_file);
+ command.stderr(stderr_file);
+
+ self.stdout_path = Some(fs::canonicalize(&stdout_path)?);
+ self.stderr_path = Some(fs::canonicalize(&stderr_path)?);
+ }
+
+ let child = command.spawn().map_err(|e| TestBinaryError::ProcessSpawn {
+ binary: "iggy-mcp".to_string(),
+ source: e,
+ })?;
+ self.child_handle = Some(child);
+
+ Ok(())
+ }
+
+ fn stop(&mut self) -> Result<(), TestBinaryError> {
+ if let Some(mut child) = self.child_handle.take() {
+ #[cfg(unix)]
+ unsafe {
+ libc::kill(child.id() as libc::pid_t, libc::SIGTERM);
+ }
+
+ #[cfg(not(unix))]
+ child.kill().ok();
+
+ let _ = child.wait();
+ }
+ Ok(())
+ }
+
+ fn is_running(&self) -> bool {
+ self.child_handle.is_some()
+ }
+
+ fn assert_running(&self) {
+ if let Some(ref child) = self.child_handle {
+ let mut child_clone = unsafe { std::ptr::read(child as *const Child) };
+ if let Ok(Some(status)) = child_clone.try_wait() {
+ let (stdout, stderr) = self.collect_logs();
+ panic!(
+ "MCP server process has crashed with exit status: {}\n\n\
+ === STDOUT ===\n{}\n\n\
+ === STDERR ===\n{}",
+ status, stdout, stderr
+ );
+ }
+ std::mem::forget(child_clone);
+ }
+ }
+
+ fn pid(&self) -> Option<u32> {
+ self.child_handle.as_ref().map(|c| c.id())
+ }
+}
+
+impl IggyDependent for McpHandle {
+ fn set_iggy_address(&mut self, addr: SocketAddr) {
+ self.iggy_address = Some(addr);
+ }
+
+ async fn wait_ready(&self) -> Result<(), TestBinaryError> {
+ let http_address = format!(
+ "http://{}:{}",
+ self.server_address.ip(),
+ self.server_address.port()
+ );
+ let client = reqwest::Client::new();
+
+ for retry in 0..MCP_HEALTH_CHECK_RETRIES {
+ match client.get(&http_address).send().await {
+ Ok(_) => return Ok(()),
+ Err(_) => {
+ if retry == MCP_HEALTH_CHECK_RETRIES - 1 {
+ return Err(TestBinaryError::HealthCheckFailed {
+ binary: "iggy-mcp".to_string(),
+ address: http_address,
+ retries: MCP_HEALTH_CHECK_RETRIES,
+ });
+ }
+ sleep(Duration::from_millis(
+ common::DEFAULT_HEALTH_CHECK_INTERVAL_MS,
+ ))
+ .await;
+ }
+ }
+ }
+
+ Ok(())
+ }
+}
+
+impl Drop for McpHandle {
+ fn drop(&mut self) {
+ let _ = self.stop();
+ common::dump_logs_on_panic("Iggy MCP server", &self.stdout_path, &self.stderr_path);
+ }
+}
diff --git a/core/integration/src/lib.rs b/core/integration/src/harness/handle/mod.rs
similarity index 63%
copy from core/integration/src/lib.rs
copy to core/integration/src/harness/handle/mod.rs
index 269698b2b..dcdc07bf5 100644
--- a/core/integration/src/lib.rs
+++ b/core/integration/src/harness/handle/mod.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * 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
@@ -16,21 +17,13 @@
* under the License.
*/
-pub mod bench_utils;
-pub mod file;
-#[allow(deprecated)]
-pub mod http_client;
-#[allow(deprecated)]
-pub mod quic_client;
-#[allow(deprecated)]
-pub mod tcp_client;
-#[allow(deprecated)]
-pub mod websocket_client;
+mod client;
+pub mod common;
+mod connector;
+mod mcp;
+mod server;
-#[allow(deprecated)]
-pub mod test_connectors_runtime;
-#[allow(deprecated)]
-pub mod test_mcp_server;
-#[allow(deprecated)]
-pub mod test_server;
-pub mod test_tls_utils;
+pub use client::ClientHandle;
+pub use connector::ConnectorHandle;
+pub use mcp::{McpClient, McpHandle};
+pub use server::ServerHandle;
diff --git a/core/integration/src/harness/handle/server.rs b/core/integration/src/harness/handle/server.rs
new file mode 100644
index 000000000..2a1a1ff7b
--- /dev/null
+++ b/core/integration/src/harness/handle/server.rs
@@ -0,0 +1,702 @@
+/*
+ * 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 crate::harness::config::{IpAddrKind, TestServerConfig};
+use crate::harness::context::TestContext;
+use crate::harness::error::TestBinaryError;
+use crate::harness::traits::{Restartable, TestBinary};
+use assert_cmd::prelude::CommandCargoExt;
+use configs::ConfigProvider;
+use futures::executor::block_on;
+use iggy::prelude::DEFAULT_ROOT_PASSWORD;
+use iggy::prelude::DEFAULT_ROOT_USERNAME;
+use rand::Rng as _;
+use server::configs::server::ServerConfig as IggyServerConfig;
+use std::collections::HashMap;
+use std::fs::{self, File, OpenOptions};
+use std::io::Write;
+use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::thread::{self, JoinHandle, available_parallelism, sleep};
+use std::time::Duration;
+
+const SLEEP_INTERVAL_MS: u64 = 20;
+const MAX_PORT_WAIT_DURATION_S: u64 = 60;
+const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
+
+#[derive(Debug, Clone)]
+struct ServerProtocolAddr {
+ tcp: Option<SocketAddr>,
+ http: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+ websocket: Option<SocketAddr>,
+}
+
+impl ServerProtocolAddr {
+ fn empty() -> Self {
+ Self {
+ tcp: None,
+ http: None,
+ quic: None,
+ websocket: None,
+ }
+ }
+}
+
+pub struct ServerHandle {
+ server_id: u32,
+ config: TestServerConfig,
+ context: Arc<TestContext>,
+ envs: HashMap<String, String>,
+ child_handle: Option<Child>,
+ addrs: ServerProtocolAddr,
+ stdout_path: Option<PathBuf>,
+ stderr_path: Option<PathBuf>,
+ watchdog_handle: Option<JoinHandle<()>>,
+ watchdog_stop: Arc<AtomicBool>,
+ generated_cert_dir: Option<PathBuf>,
+}
+
+impl std::fmt::Debug for ServerHandle {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("ServerHandle")
+ .field("addrs", &self.addrs)
+ .field("cleanup", &self.config.cleanup)
+ .field("is_running", &self.child_handle.is_some())
+ .finish_non_exhaustive()
+ }
+}
+
+impl ServerHandle {
+ pub fn tcp_addr(&self) -> Option<SocketAddr> {
+ self.addrs.tcp
+ }
+
+ pub fn raw_tcp_addr(&self) -> Option<String> {
+ self.addrs.tcp.map(|addr| addr.to_string())
+ }
+
+ pub fn http_addr(&self) -> Option<SocketAddr> {
+ self.addrs.http
+ }
+
+ pub fn quic_addr(&self) -> Option<SocketAddr> {
+ self.addrs.quic
+ }
+
+ pub fn websocket_addr(&self) -> Option<SocketAddr> {
+ self.addrs.websocket
+ }
+
+ pub fn data_path(&self) -> PathBuf {
+ self.context
+ .base_dir()
+ .join(format!("server_{}_local_data", self.server_id))
+ }
+
+ fn stdout_log_path(&self) -> PathBuf {
+ self.context
+ .base_dir()
+ .join(format!("server_{}_stdout.log", self.server_id))
+ }
+
+ fn stderr_log_path(&self) -> PathBuf {
+ self.context
+ .base_dir()
+ .join(format!("server_{}_stderr.log", self.server_id))
+ }
+
+ /// Get the TLS CA certificate path for client use.
+ /// Returns Some only when TLS is configured with generated certs.
+ pub fn tls_ca_cert_path(&self) -> Option<PathBuf> {
+ self.generated_cert_dir
+ .as_ref()
+ .map(|dir| dir.join("test_cert.pem"))
+ }
+
+ /// Check if server has TLS enabled.
+ pub fn has_tls(&self) -> bool {
+ self.config.tls.is_some()
+ }
+
+ /// Check if server TLS uses self-signed certificates (no CA available).
+ pub fn has_self_signed_tls(&self) -> bool {
+ self.config.tls.as_ref().is_some_and(|tls| tls.self_signed)
+ }
+
+ pub fn collect_logs(&self) -> (String, String) {
+ super::common::collect_logs(&self.stdout_path, &self.stderr_path)
+ }
+
+ fn build_envs(&mut self) {
+ // Pass through IGGY_* env vars from parent process, except those critical for test isolation.
+ const PROTECTED_PREFIXES: &[&str] = &[
+ "IGGY_SYSTEM_PATH",
+ "IGGY_TCP_ADDRESS",
+ "IGGY_HTTP_ADDRESS",
+ "IGGY_QUIC_ADDRESS",
+ "IGGY_WEBSOCKET_ADDRESS",
+ ];
+
+ for (key, value) in std::env::vars() {
+ if key.starts_with("IGGY_") && !PROTECTED_PREFIXES.iter().any(|p| key.starts_with(p)) {
+ self.envs.insert(key, value);
+ }
+ }
+
+ let cpu_allocation = match available_parallelism() {
+ Ok(parallelism) => {
+ let available_cpus = parallelism.get();
+ if available_cpus >= 4 {
+ let mut rng = rand::rng();
+ let max_start = available_cpus - 4;
+ let start = rng.random_range(0..=max_start);
+ format!("{}..{}", start, start + 4)
+ } else {
+ "all".to_string()
+ }
+ }
+ Err(_) => "0..4".to_string(),
+ };
+ self.envs
+ .entry("IGGY_SYSTEM_SHARDING_CPU_ALLOCATION".to_string())
+ .or_insert(cpu_allocation);
+
+ if self.config.ip_kind == IpAddrKind::V6 {
+ self.envs
+ .entry("IGGY_TCP_IPV6".to_string())
+ .or_insert_with(|| "true".to_string());
+ }
+
+ self.envs
+ .entry("IGGY_ROOT_USERNAME".to_string())
+ .or_insert_with(|| DEFAULT_ROOT_USERNAME.to_string());
+ self.envs
+ .entry("IGGY_ROOT_PASSWORD".to_string())
+ .or_insert_with(|| DEFAULT_ROOT_PASSWORD.to_string());
+
+ let data_path = self.data_path();
+ self.envs.insert(
+ "IGGY_SYSTEM_PATH".to_string(),
+ data_path.display().to_string(),
+ );
+
+ // Protocol enablement (special handling for defaults)
+ if !self.config.quic_enabled {
+ self.envs
+ .entry("IGGY_QUIC_ENABLED".to_string())
+ .or_insert_with(|| "false".to_string());
+ }
+ if !self.config.websocket_enabled {
+ self.envs
+ .entry("IGGY_WEBSOCKET_ENABLED".to_string())
+ .or_insert_with(|| "false".to_string());
+ }
+ if !self.config.http_enabled {
+ self.envs
+ .entry("IGGY_HTTP_ENABLED".to_string())
+ .or_insert_with(|| "false".to_string());
+ }
+
+ // Encryption (special handling for key injection)
+ if let Some(ref enc) = self.config.encryption {
+ self.envs
+ .entry("IGGY_SYSTEM_ENCRYPTION_ENABLED".to_string())
+ .or_insert_with(|| "true".to_string());
+ self.envs
+ .entry("IGGY_SYSTEM_ENCRYPTION_KEY".to_string())
+ .or_insert_with(|| enc.key.clone());
+ }
+
+ if let Some(tls) = self.config.tls.clone() {
+ self.set_tls_envs("TCP", &tls);
+ }
+ if let Some(tls) = self.config.websocket_tls.clone() {
+ self.set_tls_envs("WEBSOCKET", &tls);
+ }
+
+ // Extra envs from config (includes resolved config paths from macro)
+ for (k, v) in &self.config.extra_envs {
+ self.envs.insert(k.clone(), v.clone());
+ }
+
+ self.set_protocol_addresses();
+ }
+
+ fn set_tls_envs(&mut self, protocol: &str, tls: &crate::harness::config::TlsConfig) {
+ self.envs
+ .entry(format!("IGGY_{protocol}_TLS_ENABLED"))
+ .or_insert_with(|| "true".to_string());
+
+ let self_signed_str = if tls.self_signed { "true" } else { "false" };
+ self.envs
+ .entry(format!("IGGY_{protocol}_TLS_SELF_SIGNED"))
+ .or_insert_with(|| self_signed_str.to_string());
+
+ if !tls.self_signed {
+ let cert_dir = self.generated_cert_dir.as_ref().unwrap_or(&tls.cert_dir);
+ self.envs
+ .entry(format!("IGGY_{protocol}_TLS_CERT_FILE"))
+ .or_insert_with(|| cert_dir.join("test_cert.pem").display().to_string());
+ self.envs
+ .entry(format!("IGGY_{protocol}_TLS_KEY_FILE"))
+ .or_insert_with(|| cert_dir.join("test_key.pem").display().to_string());
+ }
+ }
+
+ fn set_protocol_addresses(&mut self) {
+ let default_addr = match self.config.ip_kind {
+ IpAddrKind::V4 => SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0),
+ IpAddrKind::V6 => SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 0),
+ };
+
+ if !self.envs.contains_key("IGGY_TCP_ADDRESS") {
+ self.envs
+ .insert("IGGY_TCP_ADDRESS".to_string(), default_addr.to_string());
+ }
+
+ if self.config.http_enabled && !self.envs.contains_key("IGGY_HTTP_ADDRESS") {
+ self.envs
+ .insert("IGGY_HTTP_ADDRESS".to_string(), default_addr.to_string());
+ }
+
+ if self.config.quic_enabled && !self.envs.contains_key("IGGY_QUIC_ADDRESS") {
+ self.envs
+ .insert("IGGY_QUIC_ADDRESS".to_string(), default_addr.to_string());
+ }
+
+ if self.config.websocket_enabled && !self.envs.contains_key("IGGY_WEBSOCKET_ADDRESS") {
+ self.envs.insert(
+ "IGGY_WEBSOCKET_ADDRESS".to_string(),
+ default_addr.to_string(),
+ );
+ }
+ }
+
+ fn wait_for_server_ready(&mut self) -> Result<(), TestBinaryError> {
+ let data_path = self.data_path();
+ let config_path = data_path.join("runtime/current_config.toml");
+ let file_config_provider = IggyServerConfig::config_provider(config_path.to_str().unwrap());
+
+ let max_attempts = (MAX_PORT_WAIT_DURATION_S * 1000) / SLEEP_INTERVAL_MS;
+
+ let config = block_on(async {
+ for _ in 0..max_attempts {
+ if !config_path.exists() {
+ if let Some(child) = self.child_handle.as_mut()
+ && let Ok(Some(status)) = child.try_wait()
+ {
+ let (stdout, stderr) = self.collect_logs();
+ return Err(TestBinaryError::ProcessCrashed {
+ binary: "iggy-server".to_string(),
+ exit_code: status.code(),
+ stdout,
+ stderr,
+ });
+ }
+ sleep(Duration::from_millis(SLEEP_INTERVAL_MS));
+ continue;
+ }
+
+ let config_result: Result<IggyServerConfig, _> =
+ file_config_provider.load_config().await;
+ match config_result {
+ Ok(config) => {
+ let tcp_port: u16 = config
+ .tcp
+ .address
+ .split(':')
+ .nth(1)
+ .and_then(|s| s.parse::<u16>().ok())
+ .unwrap_or(0);
+ let http_port: u16 = config
+ .http
+ .address
+ .split(':')
+ .nth(1)
+ .and_then(|s| s.parse::<u16>().ok())
+ .unwrap_or(0);
+ let quic_port: u16 = config
+ .quic
+ .address
+ .split(':')
+ .nth(1)
+ .and_then(|s| s.parse::<u16>().ok())
+ .unwrap_or(0);
+ let websocket_port: u16 = config
+ .websocket
+ .address
+ .split(':')
+ .nth(1)
+ .and_then(|s| s.parse::<u16>().ok())
+ .unwrap_or(0);
+
+ // Wait for non-default ports only for ENABLED protocols
+ // (disabled protocols keep their default addresses)
+ let tcp_has_default = config.tcp.enabled && tcp_port == 8090;
+ let http_has_default = config.http.enabled && http_port == 3000;
+ let quic_has_default = config.quic.enabled && quic_port == 8080;
+ let ws_has_default = config.websocket.enabled && websocket_port == 8092;
+
+ if tcp_has_default || http_has_default || quic_has_default || ws_has_default
+ {
+ sleep(Duration::from_millis(SLEEP_INTERVAL_MS));
+ continue;
+ }
+
+ return Ok(config);
+ }
+ Err(_) => {
+ sleep(Duration::from_millis(SLEEP_INTERVAL_MS));
+ }
+ }
+ }
+ Err(TestBinaryError::StartupTimeout {
+ binary: "iggy-server".to_string(),
+ timeout_secs: MAX_PORT_WAIT_DURATION_S,
+ })
+ })?;
+
+ self.addrs = ServerProtocolAddr::empty();
+
+ if config.tcp.enabled {
+ let addr = config.tcp.address.parse::<SocketAddr>().map_err(|_| {
+ TestBinaryError::InvalidState {
+ message: format!("Invalid TCP address: {}", config.tcp.address),
+ }
+ })?;
+ if addr.port() == 0 {
+ return Err(TestBinaryError::InvalidState {
+ message: "TCP port is 0".to_string(),
+ });
+ }
+ self.addrs.tcp = Some(addr);
+ }
+
+ if config.http.enabled {
+ let addr = config.http.address.parse::<SocketAddr>().map_err(|_| {
+ TestBinaryError::InvalidState {
+ message: format!("Invalid HTTP address: {}", config.http.address),
+ }
+ })?;
+ if addr.port() == 0 {
+ return Err(TestBinaryError::InvalidState {
+ message: "HTTP port is 0".to_string(),
+ });
+ }
+ self.addrs.http = Some(addr);
+ }
+
+ if config.quic.enabled {
+ let addr = config.quic.address.parse::<SocketAddr>().map_err(|_| {
+ TestBinaryError::InvalidState {
+ message: format!("Invalid QUIC address: {}", config.quic.address),
+ }
+ })?;
+ if addr.port() == 0 {
+ return Err(TestBinaryError::InvalidState {
+ message: "QUIC port is 0".to_string(),
+ });
+ }
+ self.addrs.quic = Some(addr);
+ }
+
+ if config.websocket.enabled {
+ let addr = config
+ .websocket
+ .address
+ .parse::<SocketAddr>()
+ .map_err(|_| TestBinaryError::InvalidState {
+ message: format!("Invalid WebSocket address: {}", config.websocket.address),
+ })?;
+ if addr.port() == 0 {
+ return Err(TestBinaryError::InvalidState {
+ message: "WebSocket port is 0".to_string(),
+ });
+ }
+ self.addrs.websocket = Some(addr);
+ }
+
+ Ok(())
+ }
+
+ fn start_watchdog(&mut self) {
+ let Some(child) = &self.child_handle else {
+ return;
+ };
+ let pid = child.id();
+ let stop_signal = self.watchdog_stop.clone();
+ let stdout_path = self.stdout_path.clone();
+ let stderr_path = self.stderr_path.clone();
+
+ let handle = thread::Builder::new()
+ .name("test-server-watchdog".to_string())
+ .spawn(move || {
+ Self::watchdog_loop(pid, stop_signal, stdout_path, stderr_path);
+ })
+ .expect("Failed to spawn watchdog thread");
+
+ self.watchdog_handle = Some(handle);
+ }
+
+ fn watchdog_loop(
+ pid: u32,
+ stop_signal: Arc<AtomicBool>,
+ stdout_path: Option<PathBuf>,
+ stderr_path: Option<PathBuf>,
+ ) {
+ const CHECK_INTERVAL: Duration = Duration::from_millis(100);
+
+ loop {
+ if stop_signal.load(Ordering::SeqCst) {
+ return;
+ }
+
+ if !Self::is_process_alive(pid) {
+ let stdout_content = stdout_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stdout log]".to_string());
+
+ let stderr_content = stderr_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stderr log]".to_string());
+
+ eprintln!(
+ "\n\n=== SERVER CRASHED ===\n\
+ The iggy-server process (PID {}) has died unexpectedly!\n\
+ This usually indicates a bug in the server.\n\n\
+ === STDOUT ===\n{}\n\n\
+ === STDERR ===\n{}\n",
+ pid, stdout_content, stderr_content
+ );
+ std::process::abort();
+ }
+
+ thread::sleep(CHECK_INTERVAL);
+ }
+ }
+
+ #[cfg(target_os = "linux")]
+ fn is_process_alive(pid: u32) -> bool {
+ let stat_path = format!("/proc/{}/stat", pid);
+ match fs::read_to_string(&stat_path) {
+ Ok(content) => {
+ if let Some(state_start) = content.rfind(')') {
+ let state = content[state_start + 1..].trim().chars().next();
+ !matches!(state, Some('Z') | Some('X'))
+ } else {
+ false
+ }
+ }
+ Err(_) => false,
+ }
+ }
+
+ #[cfg(all(unix, not(target_os = "linux")))]
+ fn is_process_alive(pid: u32) -> bool {
+ unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
+ }
+
+ fn stop_watchdog(&mut self) {
+ self.watchdog_stop.store(true, Ordering::SeqCst);
+ if let Some(handle) = self.watchdog_handle.take() {
+ let _ = handle.join();
+ }
+ }
+}
+
+impl TestBinary for ServerHandle {
+ type Config = TestServerConfig;
+
+ fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+ Self {
+ server_id: 0,
+ config,
+ context,
+ envs: HashMap::new(),
+ child_handle: None,
+ addrs: ServerProtocolAddr::empty(),
+ stdout_path: None,
+ stderr_path: None,
+ watchdog_handle: None,
+ watchdog_stop: Arc::new(AtomicBool::new(false)),
+ generated_cert_dir: None,
+ }
+ }
+
+ fn start(&mut self) -> Result<(), TestBinaryError> {
+ // Generate TLS certificates if configured for TCP or WebSocket
+ let needs_cert_generation = self
+ .config
+ .tls
+ .as_ref()
+ .is_some_and(|tls| tls.generate_certs)
+ || self
+ .config
+ .websocket_tls
+ .as_ref()
+ .is_some_and(|tls| tls.generate_certs);
+
+ if needs_cert_generation {
+ let cert_dir = self.context.base_dir().join("certs");
+ fs::create_dir_all(&cert_dir).map_err(|e| TestBinaryError::DirectoryCreation {
+ path: cert_dir.clone(),
+ source: e,
+ })?;
+ crate::test_tls_utils::generate_test_certificates(cert_dir.to_str().unwrap()).map_err(
+ |e| TestBinaryError::InvalidState {
+ message: format!("Failed to generate TLS certificates: {e}"),
+ },
+ )?;
+ self.generated_cert_dir = Some(cert_dir);
+ }
+
+ self.build_envs();
+
+ let data_path = self.data_path();
+ let config_path = data_path.join("runtime/current_config.toml");
+ if config_path.exists() {
+ let _ = fs::remove_file(&config_path);
+ }
+
+ #[allow(deprecated)]
+ let mut command = if let Some(ref path) = self.config.executable_path {
+ Command::new(path)
+ } else {
+ Command::cargo_bin("iggy-server").map_err(|e| TestBinaryError::ProcessSpawn {
+ binary: "iggy-server".to_string(),
+ source: std::io::Error::other(e.to_string()),
+ })?
+ };
+
+ command.env("IGGY_SYSTEM_PATH", data_path.display().to_string());
+ command.envs(self.envs.clone());
+
+ let verbose = std::env::var(TEST_VERBOSITY_ENV_VAR).is_ok()
+ || self.envs.contains_key(TEST_VERBOSITY_ENV_VAR);
+
+ if verbose {
+ command.stdout(Stdio::inherit());
+ command.stderr(Stdio::inherit());
+ } else {
+ let stdout_path = self.stdout_log_path();
+ let stderr_path = self.stderr_log_path();
+
+ let stdout_file =
+ File::create(&stdout_path).map_err(|e| TestBinaryError::DirectoryCreation {
+ path: stdout_path.clone(),
+ source: e,
+ })?;
+ let stderr_file =
+ File::create(&stderr_path).map_err(|e| TestBinaryError::DirectoryCreation {
+ path: stderr_path.clone(),
+ source: e,
+ })?;
+
+ command.stdout(stdout_file);
+ command.stderr(stderr_file);
+
+ self.stdout_path = Some(fs::canonicalize(&stdout_path)?);
+ self.stderr_path = Some(fs::canonicalize(&stderr_path)?);
+ }
+
+ let child = command.spawn().map_err(|e| TestBinaryError::ProcessSpawn {
+ binary: "iggy-server".to_string(),
+ source: e,
+ })?;
+ self.child_handle = Some(child);
+ self.watchdog_stop = Arc::new(AtomicBool::new(false));
+
+ self.wait_for_server_ready()?;
+ self.start_watchdog();
+
+ Ok(())
+ }
+
+ fn stop(&mut self) -> Result<(), TestBinaryError> {
+ self.stop_watchdog();
+
+ if let Some(child) = self.child_handle.take() {
+ unsafe {
+ libc::kill(child.id() as libc::pid_t, libc::SIGTERM);
+ }
+
+ if let Ok(output) = child.wait_with_output() {
+ let stderr = String::from_utf8_lossy(&output.stderr);
+ let stdout = String::from_utf8_lossy(&output.stdout);
+
+ if let Some(path) = self.stderr_path.as_ref()
+ && let Ok(mut f) = OpenOptions::new().append(true).create(true).open(path)
+ {
+ let _ = f.write_all(stderr.as_bytes());
+ }
+
+ if let Some(path) = self.stdout_path.as_ref()
+ && let Ok(mut f) = OpenOptions::new().append(true).create(true).open(path)
+ {
+ let _ = f.write_all(stdout.as_bytes());
+ }
+ }
+ }
+
+ Ok(())
+ }
+
+ fn is_running(&self) -> bool {
+ self.child_handle.is_some()
+ }
+
+ fn assert_running(&self) {
+ if let Some(pid) = self.pid().filter(|&p| !Self::is_process_alive(p)) {
+ let (stdout, stderr) = self.collect_logs();
+ panic!(
+ "Server process (pid {}) has crashed\n\n\
+ === STDOUT ===\n{}\n\n\
+ === STDERR ===\n{}",
+ pid, stdout, stderr
+ );
+ }
+ }
+
+ fn pid(&self) -> Option<u32> {
+ self.child_handle.as_ref().map(|c| c.id())
+ }
+}
+
+impl Restartable for ServerHandle {
+ fn restart(&mut self) -> Result<(), TestBinaryError> {
+ let cleanup = self.config.cleanup;
+ self.config.cleanup = false;
+ self.stop()?;
+ self.config.cleanup = cleanup;
+ self.start()
+ }
+}
+
+impl Drop for ServerHandle {
+ fn drop(&mut self) {
+ let _ = self.stop();
+ super::common::dump_logs_on_panic("Iggy server", &self.stdout_path, &self.stderr_path);
+ }
+}
diff --git a/core/integration/src/harness/helpers.rs b/core/integration/src/harness/helpers.rs
new file mode 100644
index 000000000..45704549a
--- /dev/null
+++ b/core/integration/src/harness/helpers.rs
@@ -0,0 +1,113 @@
+/*
+ * 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.
+ */
+
+//! Helper functions for common test operations.
+
+use iggy::prelude::{
+ DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, GlobalPermissions, Identifier, IdentityInfo,
+ Permissions, StreamClient, UserClient, UserStatus,
+};
+
+pub const USER_PASSWORD: &str = "secret";
+
+/// Login as root user.
+pub async fn login_root(client: &impl UserClient) -> IdentityInfo {
+ client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .expect("Failed to login as root")
+}
+
+/// Login as a specific user with the default test password.
+pub async fn login_user(client: &impl UserClient, username: &str) -> IdentityInfo {
+ client
+ .login_user(username, USER_PASSWORD)
+ .await
+ .expect("Failed to login user")
+}
+
+/// Create a user with full permissions.
+pub async fn create_user(client: &impl UserClient, username: &str) {
+ client
+ .create_user(
+ username,
+ USER_PASSWORD,
+ UserStatus::Active,
+ Some(full_permissions()),
+ )
+ .await
+ .expect("Failed to create user");
+}
+
+/// Delete a user by username.
+pub async fn delete_user(client: &impl UserClient, username: &str) {
+ client
+ .delete_user(&Identifier::named(username).expect("Invalid username"))
+ .await
+ .expect("Failed to delete user");
+}
+
+/// Assert the system is in a clean state (no streams, only root user).
+pub async fn assert_clean_system<C: StreamClient + UserClient>(client: &C) {
+ let streams = client.get_streams().await.expect("Failed to get streams");
+ assert!(
+ streams.is_empty(),
+ "Expected no streams, found {}",
+ streams.len()
+ );
+
+ let users = client.get_users().await.expect("Failed to get users");
+ assert_eq!(
+ users.len(),
+ 1,
+ "Expected only root user, found {} users",
+ users.len()
+ );
+}
+
+fn full_permissions() -> Permissions {
+ Permissions {
+ global: GlobalPermissions {
+ manage_servers: true,
+ read_servers: true,
+ manage_users: true,
+ read_users: true,
+ manage_streams: true,
+ read_streams: true,
+ manage_topics: true,
+ read_topics: true,
+ poll_messages: true,
+ send_messages: true,
+ },
+ streams: None,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_full_permissions() {
+ let perms = full_permissions();
+ assert!(perms.global.manage_servers);
+ assert!(perms.global.poll_messages);
+ assert!(perms.global.send_messages);
+ }
+}
diff --git a/core/integration/src/harness/mod.rs b/core/integration/src/harness/mod.rs
new file mode 100644
index 000000000..f2764b624
--- /dev/null
+++ b/core/integration/src/harness/mod.rs
@@ -0,0 +1,83 @@
+/*
+ * 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.
+ */
+
+//! Ergonomic test infrastructure for Iggy integration tests.
+//!
+//! This module provides a unified, builder-based API for managing test binaries
+//! (iggy-server, iggy-mcp, iggy-connectors) and their clients.
+//!
+//! # Module Structure
+//!
+//! - `config/` - Configuration types and builders for all components
+//! - `handle/` - Binary handle implementations (actual process management)
+//! - `context` - Test directory and path management
+//! - `error` - Error types
+//! - `traits` - Common traits for binary lifecycle
+//! - `orchestrator` - TestHarness that ties everything together
+//! - `helpers` - Common test helper functions
+//!
+//! # Example
+//!
+//! ```ignore
+//! use integration::harness::TestHarness;
+//!
+//! #[tokio::test]
+//! async fn test_basic_operations() {
+//! let mut harness = TestHarness::builder()
+//! .default_server()
+//! .root_tcp_client() // Auto-logins as root
+//! .build()
+//! .unwrap();
+//!
+//! harness.start().await.unwrap();
+//!
+//! // Client is already logged in as root
+//! let client = harness.client();
+//! // client.create_stream(...).await.unwrap();
+//!
+//! harness.stop().await.unwrap();
+//! }
+//! ```
+
+pub mod config;
+mod context;
+mod error;
+pub mod fixtures;
+pub mod handle;
+mod helpers;
+mod orchestrator;
+pub mod seeds;
+mod traits;
+
+pub use config::{
+ AutoLoginConfig, ClientConfig, ConnectorConfig, EncryptionConfig, IpAddrKind, McpConfig,
+ TestServerConfig, TlsConfig, resolve_config_paths,
+};
+
+pub use context::{TestContext, get_test_directory};
+pub use error::TestBinaryError;
+pub use handle::{ClientHandle, ConnectorHandle, McpClient, McpHandle, ServerHandle};
+pub use orchestrator::{TestHarness, TestHarnessBuilder, TestLogs};
+pub use traits::{IggyDependent, Restartable, TestBinary};
+
+pub use helpers::{
+ USER_PASSWORD, assert_clean_system, create_user, delete_user, login_root, login_user,
+};
+
+pub use fixtures::TestFixture;
diff --git a/core/integration/src/harness/orchestrator.rs b/core/integration/src/harness/orchestrator.rs
new file mode 100644
index 000000000..6fbe4415c
--- /dev/null
+++ b/core/integration/src/harness/orchestrator.rs
@@ -0,0 +1,885 @@
+/*
+ * 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 crate::harness::config::{ClientConfig, ConnectorConfig, McpConfig, TestServerConfig};
+use crate::harness::context::TestContext;
+use crate::harness::error::TestBinaryError;
+use crate::harness::handle::{ClientHandle, ConnectorHandle, McpClient, McpHandle, ServerHandle};
+use crate::harness::traits::{IggyDependent, Restartable, TestBinary};
+use crate::http_client::HttpClientFactory;
+use crate::quic_client::QuicClientFactory;
+use crate::tcp_client::TcpClientFactory;
+use crate::test_server::ClientFactory;
+use crate::websocket_client::WebSocketClientFactory;
+use iggy::prelude::{
+ ClientWrapper, DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, IggyClient, UserClient,
+};
+use iggy_common::TransportProtocol;
+use std::path::Path;
+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 connector: Option<(String, String)>,
+}
+
+#[derive(Default)]
+struct TlsSettings {
+ enabled: bool,
+ domain: String,
+ ca_file: Option<String>,
+ validate_certificate: bool,
+}
+
+/// Orchestrates test binaries and clients for integration tests.
+pub struct TestHarness {
+ context: Arc<TestContext>,
+ server: Option<ServerHandle>,
+ mcp: Option<McpHandle>,
+ connector: Option<ConnectorHandle>,
+ clients: Vec<ClientHandle>,
+ client_configs: Vec<ClientConfig>,
+ primary_transport: Option<TransportProtocol>,
+ primary_client_config: Option<ClientConfig>,
+ started: bool,
+}
+
+impl std::fmt::Debug for TestHarness {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("TestHarness")
+ .field("test_name", &self.context.test_name())
+ .field("started", &self.started)
+ .field("has_server", &self.server.is_some())
+ .field("has_mcp", &self.mcp.is_some())
+ .field("has_connector", &self.connector.is_some())
+ .field("client_count", &self.clients.len())
+ .finish()
+ }
+}
+
+impl TestHarness {
+ pub fn builder() -> TestHarnessBuilder {
+ TestHarnessBuilder::default()
+ }
+
+ /// Start all configured binaries and create clients.
+ pub async fn start(&mut self) -> Result<(), TestBinaryError> {
+ self.start_internal(
+ None::<
+ fn(
+ IggyClient,
+ )
+ -> std::future::Ready<Result<(), Box<dyn std::error::Error + Send + Sync>>>,
+ >,
+ )
+ .await
+ }
+
+ /// Start all configured binaries with a seed function that runs before dependent binaries.
+ ///
+ /// The seed function is called after the server starts but before MCP and connector,
+ /// allowing streams/topics to be created that dependent binaries may need.
+ ///
+ /// # Example
+ /// ```ignore
+ /// harness.start_with_seed(|client| async move {
+ /// client.create_stream("test").await?;
+ /// Ok(())
+ /// }).await?;
+ /// ```
+ pub async fn start_with_seed<F, Fut>(&mut self, seed: F) -> Result<(), TestBinaryError>
+ where
+ F: FnOnce(IggyClient) -> Fut,
+ Fut: std::future::Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>>,
+ {
+ self.start_internal(Some(seed)).await
+ }
+
+ async fn start_internal<F, Fut>(&mut self, seed: Option<F>) -> Result<(), TestBinaryError>
+ where
+ F: FnOnce(IggyClient) -> Fut,
+ Fut: std::future::Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>>,
+ {
+ if self.started {
+ return Err(TestBinaryError::AlreadyStarted);
+ }
+
+ if let Some(ref mut server) = self.server {
+ server.start()?;
+ }
+
+ if let Some(seed_fn) = seed {
+ let client = self.tcp_root_client().await?;
+ seed_fn(client)
+ .await
+ .map_err(|e| TestBinaryError::SeedFailed(e.to_string()))?;
+ }
+
+ self.start_dependents().await?;
+ self.create_clients().await?;
+
+ self.started = true;
+ Ok(())
+ }
+
+ async fn start_dependents(&mut self) -> Result<(), TestBinaryError> {
+ let tcp_addr = self.server.as_ref().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 connector) = self.connector {
+ if let Some(addr) = tcp_addr {
+ connector.set_iggy_address(addr);
+ }
+ connector.start()?;
+ connector.wait_ready().await?;
+ }
+
+ Ok(())
+ }
+
+ /// Stop all binaries and disconnect clients.
+ pub async fn stop(&mut self) -> Result<(), TestBinaryError> {
+ // Disconnect clients first
+ for client in &mut self.clients {
+ client.disconnect().await;
+ }
+ self.clients.clear();
+
+ // Stop binaries in reverse order
+ if let Some(ref mut connector) = self.connector {
+ connector.stop()?;
+ }
+
+ if let Some(ref mut mcp) = self.mcp {
+ mcp.stop()?;
+ }
+
+ if let Some(ref mut server) = self.server {
+ server.stop()?;
+ }
+
+ self.started = false;
+ Ok(())
+ }
+
+ /// Restart the server and reconnect all clients.
+ pub async fn restart_server(&mut self) -> Result<(), TestBinaryError> {
+ let Some(ref mut server) = self.server else {
+ return Err(TestBinaryError::MissingServer);
+ };
+
+ // Disconnect all clients
+ for client in &mut self.clients {
+ client.disconnect().await;
+ }
+
+ // Restart server
+ server.restart()?;
+
+ // Update client addresses and reconnect
+ self.update_client_addresses();
+ for client in &mut self.clients {
+ client.connect().await?;
+ }
+
+ Ok(())
+ }
+
+ /// Get reference to the server handle.
+ pub fn server(&self) -> &ServerHandle {
+ self.server.as_ref().expect("Server not configured")
+ }
+
+ /// Get mutable reference to the server handle.
+ pub fn server_mut(&mut self) -> &mut ServerHandle {
+ self.server.as_mut().expect("Server not configured")
+ }
+
+ /// Get the first client (panics if no clients configured).
+ pub fn client(&self) -> &ClientWrapper {
+ self.clients
+ .first()
+ .expect("No clients configured")
+ .inner()
+ .expect("Client not connected")
+ }
+
+ /// Get all client handles.
+ pub fn clients(&self) -> &[ClientHandle] {
+ &self.clients
+ }
+
+ /// Get mutable reference to all client handles.
+ pub fn clients_mut(&mut self) -> &mut [ClientHandle] {
+ &mut self.clients
+ }
+
+ /// Get the MCP handle if configured.
+ pub fn mcp(&self) -> Option<&McpHandle> {
+ self.mcp.as_ref()
+ }
+
+ /// Create an MCP client (convenience method).
+ pub async fn mcp_client(&self) -> Result<McpClient, TestBinaryError> {
+ self.mcp
+ .as_ref()
+ .ok_or(TestBinaryError::MissingMcp)?
+ .create_client()
+ .await
+ }
+
+ /// Get the connector handle if configured.
+ pub fn connector(&self) -> Option<&ConnectorHandle> {
+ self.connector.as_ref()
+ }
+
+ /// Get the test directory path.
+ pub fn test_dir(&self) -> &Path {
+ self.context.base_dir()
+ }
+
+ /// Collect logs from all binaries.
+ pub fn collect_logs(&self) -> TestLogs {
+ TestLogs {
+ server: self.server.as_ref().map(|s| s.collect_logs()),
+ mcp: self.mcp.as_ref().map(|m| m.collect_logs()),
+ connector: self.connector.as_ref().map(|c| c.collect_logs()),
+ }
+ }
+
+ /// Get a TCP client factory for creating additional clients.
+ pub fn tcp_client_factory(&self) -> Option<TcpClientFactory> {
+ let server = self.server.as_ref()?;
+ let addr = server.tcp_addr()?;
+ let config = self.find_client_config(TransportProtocol::Tcp);
+ let tls = self.extract_tls_settings(config, server);
+
+ Some(TcpClientFactory {
+ server_addr: addr.to_string(),
+ nodelay: config.map(|c| c.tcp_nodelay).unwrap_or_default(),
+ tls_enabled: tls.enabled,
+ tls_domain: tls.domain,
+ tls_ca_file: tls.ca_file,
+ tls_validate_certificate: tls.validate_certificate,
+ })
+ }
+
+ /// Get an HTTP client factory for creating additional clients.
+ pub fn http_client_factory(&self) -> Option<HttpClientFactory> {
+ self.server
+ .as_ref()
+ .and_then(|s| s.http_addr())
+ .map(|addr| HttpClientFactory {
+ server_addr: addr.to_string(),
+ })
+ }
+
+ /// Get a QUIC client factory for creating additional clients.
+ pub fn quic_client_factory(&self) -> Option<QuicClientFactory> {
+ self.server
+ .as_ref()
+ .and_then(|s| s.quic_addr())
+ .map(|addr| QuicClientFactory {
+ server_addr: addr.to_string(),
+ })
+ }
+
+ /// Get a WebSocket client factory for creating additional clients.
+ pub fn websocket_client_factory(&self) -> Option<WebSocketClientFactory> {
+ let server = self.server.as_ref()?;
+ let addr = server.websocket_addr()?;
+ let config = self.find_client_config(TransportProtocol::WebSocket);
+ let tls = self.extract_tls_settings(config, server);
+
+ Some(WebSocketClientFactory {
+ server_addr: addr.to_string(),
+ tls_enabled: tls.enabled,
+ tls_domain: tls.domain,
+ tls_ca_file: tls.ca_file,
+ tls_validate_certificate: tls.validate_certificate,
+ })
+ }
+
+ fn find_client_config(&self, transport: TransportProtocol) -> Option<&ClientConfig> {
+ self.client_configs
+ .iter()
+ .find(|c| c.transport == transport)
+ .or(self
+ .primary_client_config
+ .as_ref()
+ .filter(|c| c.transport == transport))
+ }
+
+ fn extract_tls_settings(
+ &self,
+ config: Option<&ClientConfig>,
+ server: &ServerHandle,
+ ) -> TlsSettings {
+ let (enabled, domain, validate) = config
+ .map(|c| {
+ (
+ c.tls_enabled,
+ c.tls_domain
+ .clone()
+ .unwrap_or_else(|| "localhost".to_string()),
+ c.tls_validate_certificate,
+ )
+ })
+ .unwrap_or_default();
+
+ let ca_file = if enabled {
+ server
+ .tls_ca_cert_path()
+ .map(|p| p.to_string_lossy().to_string())
+ } else {
+ None
+ };
+
+ TlsSettings {
+ enabled,
+ domain,
+ ca_file,
+ validate_certificate: validate,
+ }
+ }
+
+ /// Get all available client factories.
+ #[allow(clippy::vec_box)]
+ pub fn all_client_factories(&self) -> Vec<Box<dyn ClientFactory>> {
+ let mut factories: Vec<Box<dyn ClientFactory>> = Vec::new();
+ if let Some(f) = self.tcp_client_factory() {
+ factories.push(Box::new(f));
+ }
+ if let Some(f) = self.http_client_factory() {
+ factories.push(Box::new(f));
+ }
+ if let Some(f) = self.quic_client_factory() {
+ factories.push(Box::new(f));
+ }
+ if let Some(f) = self.websocket_client_factory() {
+ factories.push(Box::new(f));
+ }
+ factories
+ }
+
+ fn client_factory_for(
+ &self,
+ transport: TransportProtocol,
+ ) -> Result<Box<dyn ClientFactory>, TestBinaryError> {
+ let factory: Box<dyn ClientFactory> = match transport {
+ TransportProtocol::Tcp => Box::new(self.tcp_client_factory().ok_or_else(|| {
+ TestBinaryError::InvalidState {
+ message: "TCP transport not available".to_string(),
+ }
+ })?),
+ TransportProtocol::Http => Box::new(self.http_client_factory().ok_or_else(|| {
+ TestBinaryError::InvalidState {
+ message: "HTTP transport not available".to_string(),
+ }
+ })?),
+ TransportProtocol::Quic => Box::new(self.quic_client_factory().ok_or_else(|| {
+ TestBinaryError::InvalidState {
+ message: "QUIC transport not available".to_string(),
+ }
+ })?),
+ TransportProtocol::WebSocket => {
+ Box::new(self.websocket_client_factory().ok_or_else(|| {
+ TestBinaryError::InvalidState {
+ message: "WebSocket transport not available".to_string(),
+ }
+ })?)
+ }
+ };
+ Ok(factory)
+ }
+
+ /// Create a new client logged in as root for the specified transport.
+ pub async fn root_client_for(
+ &self,
+ transport: TransportProtocol,
+ ) -> Result<IggyClient, TestBinaryError> {
+ let factory = self.client_factory_for(transport)?;
+ self.create_root_client(&*factory).await
+ }
+
+ /// Create multiple root clients for the specified transport.
+ pub async fn root_clients_for(
+ &self,
+ transport: TransportProtocol,
+ count: usize,
+ ) -> Result<Vec<IggyClient>, TestBinaryError> {
+ let mut clients = Vec::with_capacity(count);
+ for _ in 0..count {
+ clients.push(self.root_client_for(transport).await?);
+ }
+ Ok(clients)
+ }
+
+ /// Create a new unauthenticated client for the specified transport.
+ pub async fn new_client_for(
+ &self,
+ transport: TransportProtocol,
+ ) -> Result<IggyClient, TestBinaryError> {
+ let factory = self.client_factory_for(transport)?;
+ let client = factory.create_client().await;
+ Ok(IggyClient::create(client, None, None))
+ }
+
+ pub async fn tcp_root_client(&self) -> Result<IggyClient, TestBinaryError> {
+ self.root_client_for(TransportProtocol::Tcp).await
+ }
+
+ pub async fn http_root_client(&self) -> Result<IggyClient, TestBinaryError> {
+ self.root_client_for(TransportProtocol::Http).await
+ }
+
+ pub async fn quic_root_client(&self) -> Result<IggyClient, TestBinaryError> {
+ self.root_client_for(TransportProtocol::Quic).await
+ }
+
+ pub async fn websocket_root_client(&self) -> Result<IggyClient, TestBinaryError> {
+ self.root_client_for(TransportProtocol::WebSocket).await
+ }
+
+ pub fn transport(&self) -> Result<TransportProtocol, TestBinaryError> {
+ self.client_configs
+ .first()
+ .map(|c| c.transport)
+ .or(self.primary_transport)
+ .ok_or_else(|| TestBinaryError::InvalidState {
+ message: "No client transport configured".to_string(),
+ })
+ }
+
+ pub async fn root_client(&self) -> Result<IggyClient, TestBinaryError> {
+ self.root_client_for(self.transport()?).await
+ }
+
+ pub async fn root_clients(&self, count: usize) -> Result<Vec<IggyClient>, TestBinaryError> {
+ self.root_clients_for(self.transport()?, count).await
+ }
+
+ pub async fn new_client(&self) -> Result<IggyClient, TestBinaryError> {
+ let transport = self
+ .client_configs
+ .first()
+ .map(|c| c.transport)
+ .or(self.primary_client_config.as_ref().map(|c| c.transport))
+ .ok_or_else(|| TestBinaryError::InvalidState {
+ message: "No client transport configured".to_string(),
+ })?;
+ self.new_client_for(transport).await
+ }
+
+ pub async fn new_clients(&self, count: usize) -> Result<Vec<IggyClient>, TestBinaryError> {
+ let mut clients = Vec::with_capacity(count);
+ for _ in 0..count {
+ clients.push(self.new_client().await?);
+ }
+ Ok(clients)
+ }
+
+ pub async fn tcp_new_client(&self) -> Result<IggyClient, TestBinaryError> {
+ self.new_client_for(TransportProtocol::Tcp).await
+ }
+
+ pub async fn tcp_root_clients(&self, count: usize) -> Result<Vec<IggyClient>, TestBinaryError> {
+ self.root_clients_for(TransportProtocol::Tcp, count).await
+ }
+
+ pub async fn http_new_client(&self) -> Result<IggyClient, TestBinaryError> {
+ self.new_client_for(TransportProtocol::Http).await
+ }
+
+ pub async fn http_root_clients(
+ &self,
+ count: usize,
+ ) -> Result<Vec<IggyClient>, TestBinaryError> {
+ self.root_clients_for(TransportProtocol::Http, count).await
+ }
+
+ pub async fn quic_new_client(&self) -> Result<IggyClient, TestBinaryError> {
+ self.new_client_for(TransportProtocol::Quic).await
+ }
+
+ pub async fn quic_root_clients(
+ &self,
+ count: usize,
+ ) -> Result<Vec<IggyClient>, TestBinaryError> {
+ self.root_clients_for(TransportProtocol::Quic, count).await
+ }
+
+ pub async fn websocket_new_client(&self) -> Result<IggyClient, TestBinaryError> {
+ self.new_client_for(TransportProtocol::WebSocket).await
+ }
+
+ pub async fn websocket_root_clients(
+ &self,
+ count: usize,
+ ) -> Result<Vec<IggyClient>, TestBinaryError> {
+ self.root_clients_for(TransportProtocol::WebSocket, count)
+ .await
+ }
+
+ async fn create_root_client(
+ &self,
+ factory: &dyn ClientFactory,
+ ) -> Result<IggyClient, TestBinaryError> {
+ let client = factory.create_client().await;
+ let iggy_client = IggyClient::create(client, None, None);
+ iggy_client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to login as root: {e}"),
+ })?;
+ Ok(iggy_client)
+ }
+
+ async fn create_clients(&mut self) -> Result<(), TestBinaryError> {
+ let Some(ref server) = self.server else {
+ return Ok(());
+ };
+
+ for config in &self.client_configs {
+ let address = match config.transport {
+ TransportProtocol::Tcp => server.tcp_addr(),
+ TransportProtocol::Http => server.http_addr(),
+ TransportProtocol::Quic => server.quic_addr(),
+ TransportProtocol::WebSocket => server.websocket_addr(),
+ };
+
+ let Some(address) = address else {
+ return Err(TestBinaryError::InvalidState {
+ message: format!("{:?} transport not available on server", config.transport),
+ });
+ };
+
+ // If server has generated TLS certs and client has TLS enabled,
+ // inject the CA cert path into the client config
+ let mut config = config.clone();
+ if config.tls_enabled
+ && let Some(ca_cert_path) = server.tls_ca_cert_path()
+ {
+ config.tls_ca_file = Some(ca_cert_path);
+ }
+
+ let mut client = ClientHandle::new(config, address);
+ client.connect().await?;
+ self.clients.push(client);
+ }
+
+ Ok(())
+ }
+
+ fn update_client_addresses(&mut self) {
+ let Some(ref server) = self.server else {
+ return;
+ };
+
+ for client in &mut self.clients {
+ let address = match client.transport() {
+ TransportProtocol::Tcp => server.tcp_addr(),
+ TransportProtocol::Http => server.http_addr(),
+ TransportProtocol::Quic => server.quic_addr(),
+ TransportProtocol::WebSocket => server.websocket_addr(),
+ };
+
+ if let Some(addr) = address {
+ client.update_address(addr);
+ }
+ }
+ }
+}
+
+/// Builder for TestHarness with fluent configuration API.
+pub struct TestHarnessBuilder {
+ test_name: Option<String>,
+ server_config: Option<TestServerConfig>,
+ mcp_config: Option<McpConfig>,
+ connector_config: Option<ConnectorConfig>,
+ primary_transport: Option<TransportProtocol>,
+ primary_client_config: Option<ClientConfig>,
+ clients: Vec<ClientConfig>,
+ cleanup: bool,
+}
+
+impl Default for TestHarnessBuilder {
+ fn default() -> Self {
+ Self {
+ test_name: None,
+ server_config: None,
+ mcp_config: None,
+ connector_config: None,
+ primary_transport: None,
+ primary_client_config: None,
+ clients: Vec::new(),
+ cleanup: true,
+ }
+ }
+}
+
+impl TestHarnessBuilder {
+ /// Override the test name (defaults to thread name or UUID).
+ pub fn test_name(mut self, name: impl Into<String>) -> Self {
+ self.test_name = Some(name.into());
+ self
+ }
+
+ /// Configure the iggy-server.
+ pub fn server(mut self, config: TestServerConfig) -> Self {
+ self.server_config = Some(config);
+ self
+ }
+
+ /// Configure the iggy-server with default settings.
+ pub fn default_server(mut self) -> Self {
+ self.server_config = Some(TestServerConfig::default());
+ self
+ }
+
+ /// Configure the MCP server.
+ pub fn mcp(mut self, config: McpConfig) -> Self {
+ self.mcp_config = Some(config);
+ self
+ }
+
+ /// Configure the MCP server with default settings.
+ pub fn default_mcp(mut self) -> Self {
+ self.mcp_config = Some(McpConfig::default());
+ self
+ }
+
+ /// Configure the connector runtime.
+ pub fn connector(mut self, config: ConnectorConfig) -> Self {
+ self.connector_config = Some(config);
+ self
+ }
+
+ /// Configure the connector runtime with default settings.
+ pub fn default_connector(mut self) -> Self {
+ self.connector_config = Some(ConnectorConfig::default());
+ self
+ }
+
+ /// Add a TCP client.
+ pub fn tcp_client(mut self) -> Self {
+ self.clients.push(ClientConfig::tcp());
+ self
+ }
+
+ /// Add a TCP client with TCP_NODELAY.
+ pub fn tcp_client_nodelay(mut self) -> Self {
+ self.clients.push(ClientConfig::tcp().with_nodelay());
+ self
+ }
+
+ /// Add a TCP client that auto-logins as root.
+ pub fn root_tcp_client(mut self) -> Self {
+ self.clients.push(ClientConfig::root_tcp());
+ self
+ }
+
+ /// Add a TCP client with TCP_NODELAY that auto-logins as root.
+ pub fn root_tcp_client_nodelay(mut self) -> Self {
+ self.clients.push(ClientConfig::root_tcp().with_nodelay());
+ self
+ }
+
+ /// Add a QUIC client.
+ pub fn quic_client(mut self) -> Self {
+ self.clients.push(ClientConfig::quic());
+ self
+ }
+
+ /// Add a QUIC client that auto-logins as root.
+ pub fn root_quic_client(mut self) -> Self {
+ self.clients.push(ClientConfig::root_quic());
+ self
+ }
+
+ /// Add an HTTP client.
+ pub fn http_client(mut self) -> Self {
+ self.clients.push(ClientConfig::http());
+ self
+ }
+
+ /// Add an HTTP client that auto-logins as root.
+ pub fn root_http_client(mut self) -> Self {
+ self.clients.push(ClientConfig::root_http());
+ self
+ }
+
+ /// Add a WebSocket client.
+ pub fn websocket_client(mut self) -> Self {
+ self.clients.push(ClientConfig::websocket());
+ self
+ }
+
+ /// Add a WebSocket client that auto-logins as root.
+ pub fn root_websocket_client(mut self) -> Self {
+ self.clients.push(ClientConfig::root_websocket());
+ self
+ }
+
+ /// Add a custom client configuration.
+ pub fn client(mut self, config: ClientConfig) -> Self {
+ self.clients.push(config);
+ self
+ }
+
+ /// Set the primary transport and client config without auto-connecting.
+ /// Used by the macro when the test doesn't use `client: &IggyClient` parameter
+ /// but still needs transport info for `new_client()` etc.
+ pub fn primary_client(mut self, config: ClientConfig) -> Self {
+ self.primary_transport = Some(config.transport);
+ self.primary_client_config = Some(config);
+ self
+ }
+
+ /// Set whether to cleanup test data on successful completion.
+ pub fn cleanup(mut self, cleanup: bool) -> Self {
+ self.cleanup = cleanup;
+ self
+ }
+
+ /// Build the TestHarness. Does NOT start any binaries.
+ pub fn build(self) -> Result<TestHarness, TestBinaryError> {
+ let mut context = TestContext::new(self.test_name, self.cleanup)?;
+ context.ensure_created()?;
+ let context = Arc::new(context);
+
+ let server = self
+ .server_config
+ .map(|config| ServerHandle::with_config(config, context.clone()));
+
+ let mcp = self
+ .mcp_config
+ .map(|config| McpHandle::with_config(config, context.clone()));
+
+ let connector = self
+ .connector_config
+ .map(|config| ConnectorHandle::with_config(config, context.clone()));
+
+ Ok(TestHarness {
+ context,
+ server,
+ mcp,
+ connector,
+ clients: Vec::new(),
+ client_configs: self.clients,
+ primary_transport: self.primary_transport,
+ primary_client_config: self.primary_client_config,
+ started: false,
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::collections::HashMap;
+
+ #[test]
+ fn test_builder_with_server() {
+ let harness = TestHarness::builder()
+ .server(
+ TestServerConfig::builder()
+ .extra_envs(HashMap::from([(
+ "IGGY_SYSTEM_SEGMENT_SIZE".to_string(),
+ "1MiB".to_string(),
+ )]))
+ .build(),
+ )
+ .tcp_client()
+ .build()
+ .unwrap();
+
+ assert!(harness.server.is_some());
+ assert!(!harness.started);
+ assert_eq!(harness.client_configs.len(), 1);
+ }
+
+ #[test]
+ fn test_builder_with_mcp() {
+ let harness = TestHarness::builder()
+ .default_server()
+ .default_mcp()
+ .tcp_client()
+ .build()
+ .unwrap();
+
+ assert!(harness.server.is_some());
+ assert!(harness.mcp.is_some());
+ }
+
+ #[test]
+ fn test_builder_multiple_clients() {
+ let harness = TestHarness::builder()
+ .default_server()
+ .tcp_client()
+ .http_client()
+ .quic_client()
+ .build()
+ .unwrap();
+
+ assert_eq!(harness.client_configs.len(), 3);
+ }
+
+ #[test]
+ fn test_builder_root_tcp_client() {
+ let harness = TestHarness::builder()
+ .default_server()
+ .root_tcp_client()
+ .build()
+ .unwrap();
+
+ assert_eq!(harness.client_configs.len(), 1);
+ assert!(harness.client_configs[0].auto_login.is_some());
+ }
+
+ #[test]
+ fn test_builder_with_custom_config() {
+ let harness = TestHarness::builder()
+ .server(
+ TestServerConfig::builder()
+ .quic_enabled(false)
+ .websocket_enabled(false)
+ .extra_envs(HashMap::from([
+ ("IGGY_SYSTEM_SEGMENT_SIZE".to_string(), "2MiB".to_string()),
+ ("TEST".to_string(), "value".to_string()),
+ ]))
+ .build(),
+ )
+ .build()
+ .unwrap();
+
+ assert!(harness.server.is_some());
+ }
+}
diff --git a/core/integration/src/harness/seeds.rs b/core/integration/src/harness/seeds.rs
new file mode 100644
index 000000000..ef4e62c85
--- /dev/null
+++ b/core/integration/src/harness/seeds.rs
@@ -0,0 +1,160 @@
+/*
+ * 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.
+ */
+
+//! Standard seed functions for MCP and integration tests.
+//!
+//! Seed functions populate the test server with data before the test body runs.
+//! They accept an `IggyClient` reference and return a `Result`.
+//!
+//! # Example
+//!
+//! ```ignore
+//! use integration::{iggy_harness, harness::seeds};
+//!
+//! #[iggy_harness(mcp, seed = seeds::stream_only)]
+//! async fn test_get_streams(mcp_client: McpClient) {
+//! // Stream already created by seed function
+//! }
+//! ```
+
+use iggy::prelude::{IggyClient, StreamClient, TopicClient};
+use iggy_binary_protocol::{
+ ConsumerGroupClient, ConsumerOffsetClient, MessageClient, PersonalAccessTokenClient, UserClient,
+};
+use iggy_common::{
+ CompressionAlgorithm, Consumer, Identifier, IggyExpiry, IggyMessage, MaxTopicSize,
+ Partitioning, PersonalAccessTokenExpiry, UserStatus,
+};
+use std::error::Error;
+
+/// Seed error type for compatibility.
+pub type SeedError = Box<dyn Error + Send + Sync>;
+
+/// Standard names used by MCP seed functions.
+pub mod names {
+ pub const STREAM: &str = "test_stream";
+ pub const TOPIC: &str = "test_topic";
+ pub const MESSAGE_PAYLOAD: &str = "test_message";
+ pub const CONSUMER_GROUP: &str = "test_consumer_group";
+ pub const CONSUMER: &str = "mcp";
+ pub const USER: &str = "test_user";
+ pub const USER_PASSWORD: &str = "secret";
+ pub const PERSONAL_ACCESS_TOKEN: &str = "test_personal_access_token";
+}
+
+/// Creates a single stream named "test".
+pub async fn stream_only(client: &IggyClient) -> Result<(), SeedError> {
+ client.create_stream("test").await?;
+ Ok(())
+}
+
+/// Creates a stream with a topic.
+pub async fn stream_with_topic(client: &IggyClient) -> Result<(), SeedError> {
+ client.create_stream("test_stream").await?;
+ client
+ .create_topic(
+ &"test_stream".try_into()?,
+ "test_topic",
+ 1,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::Unlimited,
+ )
+ .await?;
+ Ok(())
+}
+
+/// Seed for connector tests: creates stream and topic that connector will subscribe to.
+pub async fn connector_stream(client: &IggyClient) -> Result<(), SeedError> {
+ let stream_id: Identifier = names::STREAM.try_into()?;
+
+ client.create_stream(names::STREAM).await?;
+
+ client
+ .create_topic(
+ &stream_id,
+ names::TOPIC,
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::ServerDefault,
+ MaxTopicSize::ServerDefault,
+ )
+ .await?;
+
+ Ok(())
+}
+
+/// Standard MCP test data: stream, topic, message, consumer group, consumer offset, user, PAT.
+pub async fn mcp_standard(client: &IggyClient) -> Result<(), SeedError> {
+ let stream_id: Identifier = names::STREAM.try_into()?;
+ let topic_id: Identifier = names::TOPIC.try_into()?;
+
+ client.create_stream(names::STREAM).await?;
+
+ client
+ .create_topic(
+ &stream_id,
+ names::TOPIC,
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::ServerDefault,
+ MaxTopicSize::ServerDefault,
+ )
+ .await?;
+
+ let mut messages = vec![
+ IggyMessage::builder()
+ .payload(names::MESSAGE_PAYLOAD.into())
+ .build()?,
+ ];
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await?;
+
+ let consumer = Consumer::new(Identifier::named(names::CONSUMER)?);
+ client
+ .store_consumer_offset(&consumer, &stream_id, &topic_id, Some(0), 0)
+ .await?;
+
+ client
+ .create_consumer_group(&stream_id, &topic_id, names::CONSUMER_GROUP)
+ .await?;
+
+ client
+ .create_user(names::USER, names::USER_PASSWORD, UserStatus::Active, None)
+ .await?;
+
+ client
+ .create_personal_access_token(
+ names::PERSONAL_ACCESS_TOKEN,
+ PersonalAccessTokenExpiry::NeverExpire,
+ )
+ .await?;
+
+ Ok(())
+}
diff --git a/core/integration/src/harness/traits.rs b/core/integration/src/harness/traits.rs
new file mode 100644
index 000000000..bcf7bff85
--- /dev/null
+++ b/core/integration/src/harness/traits.rs
@@ -0,0 +1,53 @@
+/*
+ * 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 crate::harness::context::TestContext;
+use crate::harness::error::TestBinaryError;
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+/// Common lifecycle for all test binaries.
+pub trait TestBinary: Send {
+ type Config: Default + Clone;
+
+ fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self;
+
+ fn start(&mut self) -> Result<(), TestBinaryError>;
+
+ fn stop(&mut self) -> Result<(), TestBinaryError>;
+
+ fn is_running(&self) -> bool;
+
+ fn assert_running(&self);
+
+ fn pid(&self) -> Option<u32>;
+}
+
+/// Marks a binary that can be restarted while preserving state.
+pub trait Restartable: TestBinary {
+ fn restart(&mut self) -> Result<(), TestBinaryError>;
+}
+
+/// Marks a binary that depends on iggy-server (needs its address to connect).
+pub trait IggyDependent: TestBinary {
+ fn set_iggy_address(&mut self, addr: SocketAddr);
+
+ /// Wait until the dependent service is ready to accept requests.
+ fn wait_ready(&self) -> impl std::future::Future<Output = Result<(), TestBinaryError>> + Send;
+}
diff --git a/core/integration/src/lib.rs b/core/integration/src/lib.rs
index 269698b2b..0359462aa 100644
--- a/core/integration/src/lib.rs
+++ b/core/integration/src/lib.rs
@@ -18,6 +18,7 @@
pub mod bench_utils;
pub mod file;
+pub mod harness;
#[allow(deprecated)]
pub mod http_client;
#[allow(deprecated)]
diff --git a/core/integration/src/websocket_client.rs b/core/integration/src/websocket_client.rs
index cc0cc0826..14d0e31e1 100644
--- a/core/integration/src/websocket_client.rs
+++ b/core/integration/src/websocket_client.rs
@@ -23,9 +23,13 @@ use iggy::websocket::websocket_client::WebSocketClient;
use iggy_common::TransportProtocol;
use std::sync::Arc;
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
pub struct WebSocketClientFactory {
pub server_addr: String,
+ pub tls_enabled: bool,
+ pub tls_domain: String,
+ pub tls_ca_file: Option<String>,
+ pub tls_validate_certificate: bool,
}
#[async_trait]
@@ -33,10 +37,32 @@ impl ClientFactory for WebSocketClientFactory {
async fn create_client(&self) -> ClientWrapper {
let config = WebSocketClientConfig {
server_address: self.server_addr.clone(),
+ tls_enabled: self.tls_enabled,
+ tls_domain: self.tls_domain.clone(),
+ tls_ca_file: self.tls_ca_file.clone(),
+ tls_validate_certificate: self.tls_validate_certificate,
..WebSocketClientConfig::default()
};
- let client = WebSocketClient::create(Arc::new(config)).unwrap();
- Client::connect(&client).await.unwrap();
+ let client = WebSocketClient::create(Arc::new(config)).unwrap_or_else(|e| {
+ panic!(
+ "Failed to create WebSocketClient, iggy-server has address {}, error: {:?}",
+ self.server_addr, e
+ )
+ });
+ Client::connect(&client).await.unwrap_or_else(|e| {
+ if self.tls_enabled {
+ panic!(
+ "Failed to connect to iggy-server at {} with TLS enabled, error: {:?}\n\
+ Hint: Make sure the server is started with WebSocket TLS enabled.",
+ self.server_addr, e
+ )
+ } else {
+ panic!(
+ "Failed to connect to iggy-server at {}, error: {:?}",
+ self.server_addr, e
+ )
+ }
+ });
ClientWrapper::WebSocket(client)
}
diff --git a/core/integration/tests/server/concurrent_addition.rs b/core/integration/tests/server/concurrent_addition.rs
index a1618f6b8..25259f702 100644
--- a/core/integration/tests/server/concurrent_addition.rs
+++ b/core/integration/tests/server/concurrent_addition.rs
@@ -85,7 +85,10 @@ async fn matrix(
}
TransportProtocol::WebSocket => {
let server_addr = test_server.get_websocket_addr().unwrap();
- Box::new(WebSocketClientFactory { server_addr })
+ Box::new(WebSocketClientFactory {
+ server_addr,
+ ..Default::default()
+ })
}
};
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 65ee5d2c4..de58d27cc 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -144,7 +144,10 @@ async fn run_scenario(transport: TransportProtocol, scenario: ScenarioFn) {
}
TransportProtocol::WebSocket => {
let server_addr = test_server.get_websocket_addr().unwrap();
- Box::new(WebSocketClientFactory { server_addr })
+ Box::new(WebSocketClientFactory {
+ server_addr,
+ ..Default::default()
+ })
}
};
diff --git a/core/integration/tests/server/specific.rs b/core/integration/tests/server/specific.rs
index 17cc6b05e..aaa9bf79d 100644
--- a/core/integration/tests/server/specific.rs
+++ b/core/integration/tests/server/specific.rs
@@ -282,6 +282,7 @@ async fn segment_rotation_scenario() {
let websocket_factory = WebSocketClientFactory {
server_addr: test_server.get_websocket_addr().unwrap(),
+ ..Default::default()
};
let factories: Vec<&dyn integration::test_server::ClientFactory> = vec![