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/23 16:32:43 UTC
(iggy) branch test-binaries-refactor updated (4bf29fca4 -> d47551b3d)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch test-binaries-refactor
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit 4bf29fca4 feat(integration): implement `iggy_harness` macro and test utils
new d47551b3d feat(integration): implement `iggy_harness` macro and test utils
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (4bf29fca4)
\
N -- N -- N refs/heads/test-binaries-refactor (d47551b3d)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/_common.yml | 20 --------------------
.github/workflows/_detect.yml | 2 --
.github/workflows/_test.yml | 2 --
.github/workflows/_test_bdd.yml | 2 --
.github/workflows/_test_examples.yml | 2 --
.github/workflows/pre-merge.yml | 3 ---
core/integration/src/harness/handle/server.rs | 1 +
scripts/check-backwards-compat.sh | 4 ++--
8 files changed, 3 insertions(+), 33 deletions(-)
(iggy) 01/01: feat(integration): implement `iggy_harness` macro and test utils
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch test-binaries-refactor
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit d47551b3d7f8f22ff4991186dbf9148b6fa24b64
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Jan 23 16:36:05 2026 +0100
feat(integration): implement `iggy_harness` macro and test utils
Integration tests duplicated 30+ lines of TestServer/ClientFactory
setup per scenario, making maintenance painful and inconsistent
cleanup causing occasional test pollution.
The harness centralizes binary lifecycle in TestContext which
guarantees cleanup regardless of test outcome. Tests now declare
infrastructure needs via #[iggy_harness] macro instead of
imperative setup.
IGGY_* env passthrough lets CI override test defaults without
code changes - useful for testing specific configurations.
IGGY_TEST_CLEANUP_DISABLED preserves test_logs/ for debugging.
Lastly, fixed WebSocket TLS - client factory lacked TLS config
propagation, causing handshake failures after harness refactor.
---
.github/workflows/_common.yml | 20 -
.github/workflows/_detect.yml | 2 -
.github/workflows/_test.yml | 2 -
.github/workflows/_test_bdd.yml | 2 -
.github/workflows/_test_examples.yml | 2 -
.github/workflows/pre-merge.yml | 3 -
.gitignore | 1 +
Cargo.lock | 24 +
Cargo.toml | 6 +
DEPENDENCIES.md | 2 +
core/bench/src/utils/client_factory.rs | 1 +
core/integration/.gitignore | 1 +
core/integration/Cargo.toml | 4 +
core/integration/iggy-harness-derive/Cargo.toml | 30 +
core/integration/iggy-harness-derive/src/attrs.rs | 411 +++++++++++
.../integration/iggy-harness-derive/src/codegen.rs | 542 ++++++++++++++
core/integration/iggy-harness-derive/src/lib.rs | 94 +++
core/integration/iggy-harness-derive/src/params.rs | 195 ++++++
core/integration/src/harness/config/client.rs | 198 ++++++
core/integration/src/harness/config/common.rs | 95 +++
.../src/{lib.rs => harness/config/connector.rs} | 42 +-
core/integration/src/harness/config/mcp.rs | 62 ++
.../src/{lib.rs => harness/config/mod.rs} | 30 +-
core/integration/src/harness/config/server.rs | 94 +++
core/integration/src/harness/context.rs | 180 +++++
core/integration/src/harness/error.rs | 178 +++++
core/integration/src/harness/handle/client.rs | 230 ++++++
core/integration/src/harness/handle/connector.rs | 308 ++++++++
core/integration/src/harness/handle/mcp.rs | 331 +++++++++
.../src/{lib.rs => harness/handle/mod.rs} | 28 +-
core/integration/src/harness/handle/server.rs | 777 +++++++++++++++++++++
core/integration/src/harness/helpers.rs | 115 +++
core/integration/src/harness/mod.rs | 79 +++
core/integration/src/harness/orchestrator.rs | 699 ++++++++++++++++++
core/integration/src/harness/traits.rs | 53 ++
core/integration/src/lib.rs | 11 +
core/integration/src/websocket_client.rs | 32 +-
.../tests/server/concurrent_addition.rs | 5 +-
core/integration/tests/server/message_retrieval.rs | 152 ----
core/integration/tests/server/mod.rs | 7 +-
.../scenarios/cross_protocol_pat_scenario.rs | 316 ++++-----
.../server/scenarios/delete_segments_scenario.rs | 62 +-
.../server/scenarios/message_size_scenario.rs | 25 +-
core/integration/tests/server/scenarios/mod.rs | 17 +-
...et_scenario.rs => offset_retrieval_scenario.rs} | 32 +-
.../scenarios/segment_rotation_race_scenario.rs | 85 +--
.../scenarios/single_message_per_batch_scenario.rs | 26 +-
...scenario.rs => timestamp_retrieval_scenario.rs} | 32 +-
.../{tcp_tls_scenario.rs => tls_scenario.rs} | 49 +-
.../server/scenarios/websocket_tls_scenario.rs | 93 ---
core/integration/tests/server/specific.rs | 313 ---------
.../server/src/websocket/websocket_tls_listener.rs | 2 +-
scripts/check-backwards-compat.sh | 4 +-
53 files changed, 5093 insertions(+), 1011 deletions(-)
diff --git a/.github/workflows/_common.yml b/.github/workflows/_common.yml
index 9f8eccf2e..df88620aa 100644
--- a/.github/workflows/_common.yml
+++ b/.github/workflows/_common.yml
@@ -33,8 +33,6 @@ jobs:
rust-versions:
name: Check Rust versions sync
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- uses: actions/checkout@v4
@@ -45,8 +43,6 @@ jobs:
name: Check PR Title
if: github.event_name == 'pull_request' && !inputs.skip_pr_title
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- name: Validate PR Title
uses: amannn/action-semantic-pull-request@v5
@@ -106,8 +102,6 @@ jobs:
license-headers:
name: Check license headers
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- uses: actions/checkout@v4
@@ -126,8 +120,6 @@ jobs:
license-list:
name: Check licenses list
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- uses: actions/checkout@v4
@@ -144,8 +136,6 @@ jobs:
markdown:
name: Markdown lint
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- uses: actions/checkout@v4
@@ -163,8 +153,6 @@ jobs:
shellcheck:
name: Shell scripts lint
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- name: Checkout code
uses: actions/checkout@v4
@@ -181,8 +169,6 @@ jobs:
trailing-whitespace:
name: Check trailing whitespace
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- name: Checkout code
uses: actions/checkout@v4
@@ -195,8 +181,6 @@ jobs:
trailing-newline:
name: Check trailing newline
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- name: Checkout code
uses: actions/checkout@v4
@@ -209,8 +193,6 @@ jobs:
toml-format:
name: Check TOML formatting
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- name: Checkout code
uses: actions/checkout@v4
@@ -242,8 +224,6 @@ jobs:
]
if: always()
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
steps:
- name: Summary
run: |
diff --git a/.github/workflows/_detect.yml b/.github/workflows/_detect.yml
index 0d11c5e3d..8fe50c889 100644
--- a/.github/workflows/_detect.yml
+++ b/.github/workflows/_detect.yml
@@ -51,8 +51,6 @@ on:
jobs:
detect:
runs-on: ubuntu-latest
- env:
- IGGY_CI_BUILD: true
outputs:
rust_matrix: ${{ steps.mk.outputs.rust_matrix }}
python_matrix: ${{ steps.mk.outputs.python_matrix }}
diff --git a/.github/workflows/_test.yml b/.github/workflows/_test.yml
index e40429bc5..b1e3ac67e 100644
--- a/.github/workflows/_test.yml
+++ b/.github/workflows/_test.yml
@@ -41,8 +41,6 @@ jobs:
'ubuntu-latest'
}}
timeout-minutes: 60
- env:
- IGGY_CI_BUILD: true
steps:
- name: Checkout code
uses: actions/checkout@v4
diff --git a/.github/workflows/_test_bdd.yml b/.github/workflows/_test_bdd.yml
index dbba0b26d..cfa2e2188 100644
--- a/.github/workflows/_test_bdd.yml
+++ b/.github/workflows/_test_bdd.yml
@@ -36,8 +36,6 @@ jobs:
run:
runs-on: ubuntu-latest
timeout-minutes: 60
- env:
- IGGY_CI_BUILD: true
steps:
- name: Cleanup disk space
run: |
diff --git a/.github/workflows/_test_examples.yml b/.github/workflows/_test_examples.yml
index 492239cf4..0de6c5fa3 100644
--- a/.github/workflows/_test_examples.yml
+++ b/.github/workflows/_test_examples.yml
@@ -35,8 +35,6 @@ jobs:
run:
runs-on: ubuntu-latest
timeout-minutes: 60
- env:
- IGGY_CI_BUILD: true
steps:
- name: Checkout code
uses: actions/checkout@v4
diff --git a/.github/workflows/pre-merge.yml b/.github/workflows/pre-merge.yml
index 0215afe3a..98627b9dd 100644
--- a/.github/workflows/pre-merge.yml
+++ b/.github/workflows/pre-merge.yml
@@ -21,9 +21,6 @@ on:
branches: [master]
workflow_dispatch:
-env:
- IGGY_CI_BUILD: true
-
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: ${{ startsWith(github.ref, 'refs/pull/') }}
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 f88aaf4a6..e9d278dea 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3046,6 +3046,17 @@ dependencies = [
"syn 2.0.114",
]
+[[package]]
+name = "escargot"
+version = "0.5.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "11c3aea32bc97b500c9ca6a72b768a26e558264303d101d3409cf6d57a9ed0cf"
+dependencies = [
+ "log",
+ "serde",
+ "serde_json",
+]
+
[[package]]
name = "etcetera"
version = "0.8.0"
@@ -4609,6 +4620,15 @@ dependencies = [
"tracing-subscriber",
]
+[[package]]
+name = "iggy-harness-derive"
+version = "0.1.0"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.114",
+]
+
[[package]]
name = "iggy-mcp"
version = "0.2.1-edge.3"
@@ -5005,21 +5025,25 @@ dependencies = [
"ahash 0.8.12",
"assert_cmd",
"async-trait",
+ "bon",
"bytes",
"chrono",
"compio",
"ctor",
"derive_more",
"env_logger",
+ "escargot",
"futures",
"humantime",
"iggy",
+ "iggy-harness-derive",
"iggy_binary_protocol",
"iggy_common",
"keyring",
"lazy_static",
"libc",
"log",
+ "once_cell",
"predicates",
"rand 0.9.2",
"rcgen",
diff --git a/Cargo.toml b/Cargo.toml
index dbe9f6a98..6072e622b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -41,6 +41,7 @@ members = [
"core/connectors/sources/random_source",
"core/consensus",
"core/integration",
+ "core/integration/iggy-harness-derive",
"core/journal",
"core/message_bus",
"core/metadata",
@@ -114,6 +115,7 @@ console-subscriber = "0.5.0"
crossbeam = "0.8.4"
cyper = { version = "0.7.1", features = ["rustls"], default-features = false }
cyper-axum = { version = "0.7.1" }
+darling = "0.20"
dashmap = "6.1.0"
derive-new = "0.7.0"
derive_builder = "0.20.2"
@@ -136,6 +138,7 @@ hwlocality = "1.0.0-alpha.11"
iceberg = "=0.6.0"
iceberg-catalog-rest = "=0.6.0"
iggy = { path = "core/sdk", version = "0.8.1-edge.1" }
+iggy-harness-derive = { path = "core/integration/iggy-harness-derive" }
iggy_binary_protocol = { path = "core/binary_protocol", version = "0.8.1-edge.1" }
iggy_common = { path = "core/common", version = "0.8.1-edge.1" }
iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.1.1-edge.1" }
@@ -153,7 +156,9 @@ parquet = "=55.2.0"
passterm = "=2.0.1"
postcard = { version = "1.1.3", features = ["alloc"] }
predicates = "3.1.3"
+proc-macro2 = "1.0"
quinn = "0.11.9"
+quote = "1.0"
rand = "0.9.2"
rand_xoshiro = "0.7.0"
regex = "1.12.2"
@@ -179,6 +184,7 @@ server = { path = "core/server" }
simd-json = { version = "0.17.0", features = ["serde_impl"] }
strum = { version = "0.27.2", features = ["derive"] }
strum_macros = "0.27.2"
+syn = { version = "2.0", features = ["full", "parsing", "extra-traits"] }
sysinfo = "0.37.2"
tempfile = "3.24.0"
test-case = "3.3.1"
diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md
index 814a96a36..8e5f42aea 100644
--- a/DEPENDENCIES.md
+++ b/DEPENDENCIES.md
@@ -268,6 +268,7 @@ err_trail: 0.11.0, "Apache-2.0",
errno: 0.3.14, "Apache-2.0 OR MIT",
error_set: 0.9.1, "Apache-2.0",
error_set_impl: 0.9.1, "Apache-2.0",
+escargot: 0.5.15, "Apache-2.0 OR MIT",
etcetera: 0.8.0, "Apache-2.0 OR MIT",
etcetera: 0.11.0, "Apache-2.0 OR MIT",
event-listener: 5.4.1, "Apache-2.0 OR MIT",
@@ -396,6 +397,7 @@ iggy-bench: 0.3.1-edge.1, "Apache-2.0",
iggy-bench-dashboard-server: 0.5.1-edge.1, "Apache-2.0",
iggy-cli: 0.10.1-edge.1, "Apache-2.0",
iggy-connectors: 0.2.1-edge.3, "Apache-2.0",
+iggy-harness-derive: 0.1.0, "Apache-2.0",
iggy-mcp: 0.2.1-edge.3, "Apache-2.0",
iggy_binary_protocol: 0.8.1-edge.2, "Apache-2.0",
iggy_common: 0.8.1-edge.1, "Apache-2.0",
diff --git a/core/bench/src/utils/client_factory.rs b/core/bench/src/utils/client_factory.rs
index 513d22300..a4dffc821 100644
--- a/core/bench/src/utils/client_factory.rs
+++ b/core/bench/src/utils/client_factory.rs
@@ -51,6 +51,7 @@ pub fn create_client_factory(args: &IggyBenchArgs) -> Arc<dyn ClientFactory> {
}),
TransportProtocol::WebSocket => Arc::new(WebSocketClientFactory {
server_addr: args.server_address().to_owned(),
+ ..Default::default()
}),
}
}
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 6dec491f5..f56d06268 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -30,21 +30,25 @@ 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 }
ctor = "0.6.3"
derive_more = { workspace = true }
env_logger = { workspace = true }
+escargot = "0.5"
futures = { workspace = true }
humantime = { workspace = true }
iggy = { workspace = true }
+iggy-harness-derive = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
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.6"
diff --git a/core/integration/iggy-harness-derive/Cargo.toml b/core/integration/iggy-harness-derive/Cargo.toml
new file mode 100644
index 000000000..4ee3ebe16
--- /dev/null
+++ b/core/integration/iggy-harness-derive/Cargo.toml
@@ -0,0 +1,30 @@
+# 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.
+
+[package]
+name = "iggy-harness-derive"
+version = "0.1.0"
+edition = "2024"
+license = "Apache-2.0"
+
+[lib]
+proc-macro = true
+
+[dependencies]
+proc-macro2 = { workspace = true }
+quote = { workspace = true }
+syn = { workspace = true }
diff --git a/core/integration/iggy-harness-derive/src/attrs.rs b/core/integration/iggy-harness-derive/src/attrs.rs
new file mode 100644
index 000000000..419cbea58
--- /dev/null
+++ b/core/integration/iggy-harness-derive/src/attrs.rs
@@ -0,0 +1,411 @@
+/* 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.
+ */
+
+//! Attribute parsing for `#[iggy_harness(...)]`.
+//!
+//! Supports:
+//! - `transport = Tcp` (single)
+//! - `transport = [Tcp, Http]` (matrix)
+//! - `server(key = "value")` (static)
+//! - `server(key = ["v1", "v2"])` (matrix)
+
+use proc_macro2::Span;
+use syn::parse::{Parse, ParseStream};
+use syn::punctuated::Punctuated;
+use syn::{Ident, LitStr, Token, bracketed, parenthesized};
+
+/// Parsed `#[iggy_harness(...)]` attributes.
+#[derive(Debug, Default)]
+pub struct IggyTestAttrs {
+ pub transports: Vec<Transport>,
+ pub server: ServerAttrs,
+}
+
+impl IggyTestAttrs {
+ /// Create attrs with specified transports and default server config.
+ #[allow(dead_code)]
+ pub fn with_transports(transports: Vec<Transport>) -> Self {
+ Self {
+ transports,
+ server: ServerAttrs::default(),
+ }
+ }
+}
+
+/// Transport protocol variant.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Transport {
+ Tcp,
+ TcpTlsSelfSigned,
+ TcpTlsGenerated,
+ Http,
+ Quic,
+ WebSocket,
+ WebSocketTlsSelfSigned,
+ WebSocketTlsGenerated,
+}
+
+impl Transport {
+ pub fn as_str(&self) -> &'static str {
+ match self {
+ Transport::Tcp => "tcp",
+ Transport::TcpTlsSelfSigned => "tcp_tls_self_signed",
+ Transport::TcpTlsGenerated => "tcp_tls_generated",
+ Transport::Http => "http",
+ Transport::Quic => "quic",
+ Transport::WebSocket => "websocket",
+ Transport::WebSocketTlsSelfSigned => "websocket_tls_self_signed",
+ Transport::WebSocketTlsGenerated => "websocket_tls_generated",
+ }
+ }
+
+ pub fn variant_ident(&self) -> Ident {
+ let name = match self {
+ Transport::Tcp | Transport::TcpTlsSelfSigned | Transport::TcpTlsGenerated => "Tcp",
+ Transport::Http => "Http",
+ Transport::Quic => "Quic",
+ Transport::WebSocket
+ | Transport::WebSocketTlsSelfSigned
+ | Transport::WebSocketTlsGenerated => "WebSocket",
+ };
+ Ident::new(name, Span::call_site())
+ }
+
+ pub fn client_config_method(&self) -> &'static str {
+ match self {
+ Transport::Tcp | Transport::TcpTlsSelfSigned | Transport::TcpTlsGenerated => "root_tcp",
+ Transport::Http => "root_http",
+ Transport::Quic => "root_quic",
+ Transport::WebSocket
+ | Transport::WebSocketTlsSelfSigned
+ | Transport::WebSocketTlsGenerated => "root_websocket",
+ }
+ }
+
+ /// Returns the TLS mode if this transport uses TLS.
+ pub fn tls_mode(&self) -> Option<TlsMode> {
+ match self {
+ Transport::TcpTlsSelfSigned | Transport::WebSocketTlsSelfSigned => {
+ Some(TlsMode::SelfSigned)
+ }
+ Transport::TcpTlsGenerated | Transport::WebSocketTlsGenerated => {
+ Some(TlsMode::Generated)
+ }
+ _ => None,
+ }
+ }
+
+ /// Returns true if this transport uses WebSocket protocol.
+ pub fn is_websocket(&self) -> bool {
+ matches!(
+ self,
+ Transport::WebSocket
+ | Transport::WebSocketTlsSelfSigned
+ | Transport::WebSocketTlsGenerated
+ )
+ }
+}
+
+/// Server configuration attributes.
+#[derive(Debug, Default)]
+pub struct ServerAttrs {
+ pub segment_size: ConfigValue,
+ pub cache_indexes: ConfigValue,
+ pub messages_required_to_save: ConfigValue,
+ pub tcp_nodelay: Option<bool>,
+ pub message_saver_interval: ConfigValue,
+ pub socket_migration: Option<bool>,
+}
+
+/// TLS configuration mode for server.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TlsMode {
+ /// Server generates its own self-signed certs. Client cannot validate.
+ SelfSigned,
+ /// Harness generates test certs. Client can validate using CA cert.
+ Generated,
+}
+
+/// A configuration value that can be static or a matrix.
+#[derive(Debug, Default, Clone)]
+pub enum ConfigValue {
+ #[default]
+ Unset,
+ Static(String),
+ Matrix(Vec<String>),
+}
+
+impl ConfigValue {
+ pub fn variants(&self) -> Vec<Option<&str>> {
+ match self {
+ ConfigValue::Unset => vec![None],
+ ConfigValue::Static(s) => vec![Some(s.as_str())],
+ ConfigValue::Matrix(v) => v.iter().map(|s| Some(s.as_str())).collect(),
+ }
+ }
+}
+
+impl Parse for IggyTestAttrs {
+ fn parse(input: ParseStream) -> syn::Result<Self> {
+ let mut attrs = IggyTestAttrs::default();
+
+ if input.is_empty() {
+ attrs.transports.push(Transport::Tcp);
+ return Ok(attrs);
+ }
+
+ let items: Punctuated<AttrItem, Token![,]> = Punctuated::parse_terminated(input)?;
+
+ for item in items {
+ match item {
+ AttrItem::Transport(transports) => {
+ attrs.transports = transports;
+ }
+ AttrItem::Server(server) => {
+ attrs.server = server;
+ }
+ }
+ }
+
+ if attrs.transports.is_empty() {
+ attrs.transports.push(Transport::Tcp);
+ }
+
+ Ok(attrs)
+ }
+}
+
+enum AttrItem {
+ Transport(Vec<Transport>),
+ Server(ServerAttrs),
+}
+
+impl Parse for AttrItem {
+ fn parse(input: ParseStream) -> syn::Result<Self> {
+ let ident: Ident = input.parse()?;
+ let ident_str = ident.to_string();
+
+ match ident_str.as_str() {
+ "transport" => {
+ input.parse::<Token![=]>()?;
+ let transports = parse_transport_value(input)?;
+ Ok(AttrItem::Transport(transports))
+ }
+ "server" => {
+ let content;
+ parenthesized!(content in input);
+ let server = parse_server_attrs(&content)?;
+ Ok(AttrItem::Server(server))
+ }
+ _ => Err(syn::Error::new(
+ ident.span(),
+ format!("unknown attribute: {ident_str}"),
+ )),
+ }
+ }
+}
+
+fn parse_transport_value(input: ParseStream) -> syn::Result<Vec<Transport>> {
+ if input.peek(syn::token::Bracket) {
+ let content;
+ bracketed!(content in input);
+ let idents: Punctuated<Ident, Token![,]> = Punctuated::parse_terminated(&content)?;
+ idents.into_iter().map(parse_transport_ident).collect()
+ } else {
+ let ident: Ident = input.parse()?;
+ Ok(vec![parse_transport_ident(ident)?])
+ }
+}
+
+fn parse_transport_ident(ident: Ident) -> syn::Result<Transport> {
+ match ident.to_string().as_str() {
+ "Tcp" => Ok(Transport::Tcp),
+ "TcpTlsSelfSigned" => Ok(Transport::TcpTlsSelfSigned),
+ "TcpTlsGenerated" => Ok(Transport::TcpTlsGenerated),
+ "Http" => Ok(Transport::Http),
+ "Quic" => Ok(Transport::Quic),
+ "WebSocket" => Ok(Transport::WebSocket),
+ "WebSocketTlsSelfSigned" => Ok(Transport::WebSocketTlsSelfSigned),
+ "WebSocketTlsGenerated" => Ok(Transport::WebSocketTlsGenerated),
+ other => Err(syn::Error::new(
+ ident.span(),
+ format!("unknown transport: {other}"),
+ )),
+ }
+}
+
+fn parse_server_attrs(input: ParseStream) -> syn::Result<ServerAttrs> {
+ let mut server = ServerAttrs::default();
+
+ let items: Punctuated<ServerAttrItem, Token![,]> = Punctuated::parse_terminated(input)?;
+
+ for item in items {
+ match item.key.as_str() {
+ "segment_size" => server.segment_size = item.value,
+ "cache_indexes" => server.cache_indexes = item.value,
+ "messages_required_to_save" => server.messages_required_to_save = item.value,
+ "tcp_nodelay" => {
+ if let ConfigValue::Static(s) = &item.value {
+ server.tcp_nodelay = Some(s == "true");
+ }
+ }
+ "message_saver_interval" => server.message_saver_interval = item.value,
+ "socket_migration" => {
+ if let ConfigValue::Static(s) = &item.value {
+ server.socket_migration = Some(s == "true");
+ }
+ }
+ other => {
+ return Err(syn::Error::new(
+ Span::call_site(),
+ format!("unknown server attribute: {other}"),
+ ));
+ }
+ }
+ }
+
+ Ok(server)
+}
+
+struct ServerAttrItem {
+ key: String,
+ value: ConfigValue,
+}
+
+/// A literal that can be a string or an integer.
+enum ArrayLiteral {
+ Str(LitStr),
+ Int(syn::LitInt),
+}
+
+impl Parse for ArrayLiteral {
+ fn parse(input: ParseStream) -> syn::Result<Self> {
+ if input.peek(LitStr) {
+ Ok(ArrayLiteral::Str(input.parse()?))
+ } else if input.peek(syn::LitInt) {
+ Ok(ArrayLiteral::Int(input.parse()?))
+ } else {
+ Err(input.error("expected string or integer literal"))
+ }
+ }
+}
+
+impl ArrayLiteral {
+ fn to_string_value(&self) -> String {
+ match self {
+ ArrayLiteral::Str(s) => s.value(),
+ ArrayLiteral::Int(i) => i.base10_digits().to_string(),
+ }
+ }
+}
+
+impl Parse for ServerAttrItem {
+ fn parse(input: ParseStream) -> syn::Result<Self> {
+ let ident: Ident = input.parse()?;
+ input.parse::<Token![=]>()?;
+
+ let value = if input.peek(syn::token::Bracket) {
+ let content;
+ bracketed!(content in input);
+ let values: Punctuated<ArrayLiteral, Token![,]> =
+ Punctuated::parse_terminated(&content)?;
+ ConfigValue::Matrix(values.into_iter().map(|v| v.to_string_value()).collect())
+ } else if input.peek(LitStr) {
+ let lit: LitStr = input.parse()?;
+ ConfigValue::Static(lit.value())
+ } else if input.peek(syn::LitBool) {
+ let lit: syn::LitBool = input.parse()?;
+ ConfigValue::Static(lit.value.to_string())
+ } else if input.peek(syn::LitInt) {
+ let lit: syn::LitInt = input.parse()?;
+ ConfigValue::Static(lit.base10_digits().to_string())
+ } else {
+ return Err(input.error("expected string literal, bool, int, or array"));
+ };
+
+ Ok(ServerAttrItem {
+ key: ident.to_string(),
+ value,
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn parse_empty() {
+ let attrs: IggyTestAttrs = syn::parse_quote!();
+ assert_eq!(attrs.transports.len(), 1);
+ assert_eq!(attrs.transports[0], Transport::Tcp);
+ }
+
+ #[test]
+ fn parse_single_transport() {
+ let attrs: IggyTestAttrs = syn::parse_quote!(transport = Http);
+ assert_eq!(attrs.transports.len(), 1);
+ assert_eq!(attrs.transports[0], Transport::Http);
+ }
+
+ #[test]
+ fn parse_transport_array() {
+ let attrs: IggyTestAttrs = syn::parse_quote!(transport = [Tcp, Http, Quic]);
+ assert_eq!(attrs.transports.len(), 3);
+ assert_eq!(attrs.transports[0], Transport::Tcp);
+ assert_eq!(attrs.transports[1], Transport::Http);
+ assert_eq!(attrs.transports[2], Transport::Quic);
+ }
+
+ #[test]
+ fn parse_server_static() {
+ let attrs: IggyTestAttrs = syn::parse_quote!(server(segment_size = "1MiB"));
+ assert!(matches!(attrs.server.segment_size, ConfigValue::Static(s) if s == "1MiB"));
+ }
+
+ #[test]
+ fn parse_server_matrix() {
+ let attrs: IggyTestAttrs = syn::parse_quote!(server(segment_size = ["512B", "1MiB"]));
+ assert!(matches!(&attrs.server.segment_size, ConfigValue::Matrix(v) if v.len() == 2));
+ }
+
+ #[test]
+ fn parse_full() {
+ let attrs: IggyTestAttrs = syn::parse_quote!(
+ transport = [Tcp, Http],
+ server(
+ segment_size = ["512B", "1MiB"],
+ cache_indexes = "none",
+ tcp_nodelay = true
+ )
+ );
+ assert_eq!(attrs.transports.len(), 2);
+ assert!(matches!(&attrs.server.segment_size, ConfigValue::Matrix(v) if v.len() == 2));
+ assert!(matches!(&attrs.server.cache_indexes, ConfigValue::Static(s) if s == "none"));
+ assert_eq!(attrs.server.tcp_nodelay, Some(true));
+ }
+
+ #[test]
+ fn parse_tls_transports() {
+ let attrs: IggyTestAttrs =
+ syn::parse_quote!(transport = [TcpTlsSelfSigned, TcpTlsGenerated]);
+ assert_eq!(attrs.transports.len(), 2);
+ assert_eq!(attrs.transports[0], Transport::TcpTlsSelfSigned);
+ assert_eq!(attrs.transports[1], Transport::TcpTlsGenerated);
+ }
+}
diff --git a/core/integration/iggy-harness-derive/src/codegen.rs b/core/integration/iggy-harness-derive/src/codegen.rs
new file mode 100644
index 000000000..7019d31b3
--- /dev/null
+++ b/core/integration/iggy-harness-derive/src/codegen.rs
@@ -0,0 +1,542 @@
+/* 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::attrs::{IggyTestAttrs, TlsMode, Transport};
+use crate::params::{
+ DetectedParam, analyze_signature, matrix_params, needs_client, needs_harness, needs_harness_mut,
+};
+use proc_macro2::{Span, TokenStream};
+use quote::{format_ident, quote};
+use syn::{Ident, ItemFn};
+
+/// Represents a single test variant configuration.
+#[derive(Debug)]
+struct TestVariant {
+ transport: Transport,
+ segment_size: Option<String>,
+ cache_indexes: Option<String>,
+ messages_required_to_save: Option<String>,
+ message_saver_interval: Option<String>,
+ socket_migration: Option<bool>,
+}
+
+impl TestVariant {
+ fn suffix(&self) -> String {
+ let mut parts = vec![self.transport.as_str().to_string()];
+
+ if let Some(ref s) = self.segment_size {
+ parts.push(sanitize_value(s));
+ }
+ if let Some(ref s) = self.cache_indexes {
+ parts.push(sanitize_value(s));
+ }
+ if let Some(ref s) = self.messages_required_to_save {
+ parts.push(format!("msgs_{s}"));
+ }
+
+ parts.join("_")
+ }
+}
+
+fn sanitize_value(s: &str) -> String {
+ s.to_lowercase()
+ .replace("kib", "kb")
+ .replace("mib", "mb")
+ .replace("gib", "gb")
+ .chars()
+ .filter(|c| c.is_alphanumeric() || *c == '_')
+ .collect()
+}
+
+/// Generate all infrastructure variants from attributes.
+fn generate_variants(attrs: &IggyTestAttrs) -> Vec<TestVariant> {
+ let segment_sizes = attrs.server.segment_size.variants();
+ let cache_indexes_list = attrs.server.cache_indexes.variants();
+ let msgs_required = attrs.server.messages_required_to_save.variants();
+ let msg_saver_intervals = attrs.server.message_saver_interval.variants();
+
+ let mut variants = Vec::new();
+
+ for transport in &attrs.transports {
+ for segment_size in &segment_sizes {
+ for cache_indexes in &cache_indexes_list {
+ for msgs in &msgs_required {
+ for msg_interval in &msg_saver_intervals {
+ variants.push(TestVariant {
+ transport: *transport,
+ segment_size: segment_size.map(String::from),
+ cache_indexes: cache_indexes.map(String::from),
+ messages_required_to_save: msgs.map(String::from),
+ message_saver_interval: msg_interval.map(String::from),
+ socket_migration: attrs.server.socket_migration,
+ });
+ }
+ }
+ }
+ }
+ }
+
+ variants
+}
+
+/// Generate test code from attributes and input function.
+pub fn generate_tests(attrs: &IggyTestAttrs, input: &ItemFn) -> syn::Result<TokenStream> {
+ let fn_name = &input.sig.ident;
+ let fn_vis = &input.vis;
+ let fn_body = &input.block;
+
+ let params = analyze_signature(&input.sig)?;
+ let has_client = needs_client(¶ms);
+ let has_harness = needs_harness(¶ms);
+ let has_harness_mut = needs_harness_mut(¶ms);
+ let matrix_params_list = matrix_params(¶ms);
+ let has_matrix_params = !matrix_params_list.is_empty();
+
+ let variants = generate_variants(attrs);
+
+ if variants.len() == 1 && !has_matrix_params {
+ return generate_single_test(
+ fn_name,
+ fn_vis,
+ fn_body,
+ ¶ms,
+ &variants[0],
+ has_client,
+ has_harness,
+ has_harness_mut,
+ attrs,
+ );
+ }
+
+ if has_matrix_params {
+ return generate_impl_functions_for_test_matrix(
+ fn_name,
+ fn_vis,
+ fn_body,
+ input,
+ ¶ms,
+ &variants,
+ has_client,
+ has_harness,
+ has_harness_mut,
+ attrs,
+ );
+ }
+
+ generate_test_module(
+ fn_name,
+ fn_vis,
+ fn_body,
+ ¶ms,
+ &variants,
+ has_client,
+ has_harness,
+ has_harness_mut,
+ attrs,
+ )
+}
+
+#[allow(clippy::too_many_arguments)]
+fn generate_single_test(
+ fn_name: &Ident,
+ fn_vis: &syn::Visibility,
+ fn_body: &syn::Block,
+ params: &[DetectedParam],
+ variant: &TestVariant,
+ has_client: bool,
+ has_harness: bool,
+ has_harness_mut: bool,
+ attrs: &IggyTestAttrs,
+) -> syn::Result<TokenStream> {
+ let harness_setup = generate_harness_setup(variant, attrs);
+ let client_setup = if has_client {
+ generate_client_setup()
+ } else {
+ quote!()
+ };
+ let param_bindings = generate_param_bindings(params, has_harness, has_harness_mut);
+
+ Ok(quote! {
+ #[::tokio::test]
+ #[::serial_test::parallel]
+ #fn_vis async fn #fn_name() {
+ #harness_setup
+ __harness.start().await.expect("failed to start test harness");
+ #client_setup
+ #param_bindings
+ #fn_body
+ __harness.stop().await.ok();
+ }
+ })
+}
+
+#[allow(clippy::too_many_arguments)]
+fn generate_test_module(
+ fn_name: &Ident,
+ fn_vis: &syn::Visibility,
+ fn_body: &syn::Block,
+ params: &[DetectedParam],
+ variants: &[TestVariant],
+ has_client: bool,
+ has_harness: bool,
+ has_harness_mut: bool,
+ attrs: &IggyTestAttrs,
+) -> syn::Result<TokenStream> {
+ let mut test_fns = Vec::new();
+
+ for variant in variants {
+ let test_name = format_ident!("{}", variant.suffix());
+ let harness_setup = generate_harness_setup(variant, attrs);
+ let client_setup = if has_client {
+ generate_client_setup()
+ } else {
+ quote!()
+ };
+ let param_bindings = generate_param_bindings(params, has_harness, has_harness_mut);
+
+ test_fns.push(quote! {
+ #[::tokio::test]
+ #[::serial_test::parallel]
+ async fn #test_name() {
+ #harness_setup
+ __harness.start().await.expect("failed to start test harness");
+ #client_setup
+ #param_bindings
+ #fn_body
+ __harness.stop().await.ok();
+ }
+ });
+ }
+
+ Ok(quote! {
+ #fn_vis mod #fn_name {
+ use super::*;
+ #(#test_fns)*
+ }
+ })
+}
+
+#[allow(clippy::too_many_arguments)]
+fn generate_impl_functions_for_test_matrix(
+ fn_name: &Ident,
+ fn_vis: &syn::Visibility,
+ fn_body: &syn::Block,
+ input: &ItemFn,
+ params: &[DetectedParam],
+ variants: &[TestVariant],
+ has_client: bool,
+ has_harness: bool,
+ has_harness_mut: bool,
+ attrs: &IggyTestAttrs,
+) -> syn::Result<TokenStream> {
+ let matrix_params_list: Vec<_> = params
+ .iter()
+ .filter_map(|p| {
+ if let DetectedParam::MatrixParam { name, ty } = p {
+ Some((name, ty))
+ } else {
+ None
+ }
+ })
+ .collect();
+
+ let param_names: Vec<_> = matrix_params_list.iter().map(|(name, _)| *name).collect();
+ let param_types: Vec<_> = matrix_params_list.iter().map(|(_, ty)| *ty).collect();
+
+ let other_attrs: Vec<_> = input
+ .attrs
+ .iter()
+ .filter(|attr| {
+ let path = attr.path();
+ !path.is_ident("iggy_harness")
+ })
+ .collect();
+
+ let param_bindings = generate_param_bindings(params, has_harness, has_harness_mut);
+
+ if variants.len() == 1 {
+ let variant = &variants[0];
+ let harness_setup = generate_harness_setup(variant, attrs);
+ let client_setup = if has_client {
+ generate_client_setup()
+ } else {
+ quote!()
+ };
+
+ return Ok(quote! {
+ #(#other_attrs)*
+ #[::tokio::test]
+ #[::serial_test::parallel]
+ #fn_vis async fn #fn_name(#(#param_names: #param_types),*) {
+ #harness_setup
+ __harness.start().await.expect("failed to start test harness");
+ #client_setup
+ #param_bindings
+ #fn_body
+ __harness.stop().await.ok();
+ }
+ });
+ }
+
+ let mut impl_fns = Vec::new();
+ let mut test_fn_calls = Vec::new();
+
+ for variant in variants {
+ let impl_name = format_ident!("__impl_{}", variant.suffix());
+ let harness_setup = generate_harness_setup(variant, attrs);
+ let client_setup = if has_client {
+ generate_client_setup()
+ } else {
+ quote!()
+ };
+
+ impl_fns.push(quote! {
+ async fn #impl_name(#(#param_names: #param_types),*) {
+ #harness_setup
+ __harness.start().await.expect("failed to start test harness");
+ #client_setup
+ #param_bindings
+ #fn_body
+ __harness.stop().await.ok();
+ }
+ });
+
+ let test_name = format_ident!("{}", variant.suffix());
+ test_fn_calls.push(quote! {
+ #(#other_attrs)*
+ #[::tokio::test]
+ #[::serial_test::parallel]
+ async fn #test_name(#(#param_names: #param_types),*) {
+ #impl_name(#(#param_names),*).await;
+ }
+ });
+ }
+
+ Ok(quote! {
+ #fn_vis mod #fn_name {
+ use super::*;
+ #(#impl_fns)*
+ #(#test_fn_calls)*
+ }
+ })
+}
+
+fn generate_harness_setup(variant: &TestVariant, attrs: &IggyTestAttrs) -> TokenStream {
+ let transport = variant.transport.variant_ident();
+ let tls_mode = variant.transport.tls_mode();
+
+ let mut server_builder_calls = Vec::new();
+
+ if let Some(ref s) = variant.segment_size {
+ server_builder_calls.push(quote!(.segment_size(#s)));
+ }
+ if let Some(ref s) = variant.cache_indexes {
+ server_builder_calls.push(quote!(.cache_indexes(#s)));
+ }
+ if let Some(ref s) = variant.messages_required_to_save {
+ let val: u32 = s.parse().unwrap_or_else(|_| {
+ panic!("invalid messages_required_to_save: '{s}' is not a valid u32")
+ });
+ server_builder_calls.push(quote!(.messages_required_to_save(#val)));
+ }
+ if let Some(ref s) = variant.message_saver_interval {
+ server_builder_calls.push(quote!(.message_saver_interval(#s)));
+ }
+ if let Some(val) = variant.socket_migration {
+ server_builder_calls.push(quote!(.socket_migration(#val)));
+ }
+
+ if attrs.server.tcp_nodelay.unwrap_or(true) {
+ server_builder_calls.push(quote!(.tcp_nodelay(true)));
+ }
+
+ if let Some(ref mode) = tls_mode {
+ let tls_config = match mode {
+ TlsMode::SelfSigned => quote!(::integration::harness::TlsConfig::self_signed()),
+ TlsMode::Generated => quote!(::integration::harness::TlsConfig::generated()),
+ };
+ if variant.transport.is_websocket() {
+ server_builder_calls.push(quote!(.websocket_tls(#tls_config)));
+ } else {
+ server_builder_calls.push(quote!(.tls(#tls_config)));
+ }
+ }
+
+ let server_config = if server_builder_calls.is_empty() {
+ quote!(::integration::harness::TestServerConfig::default())
+ } else {
+ quote! {
+ ::integration::harness::TestServerConfig::builder()
+ #(#server_builder_calls)*
+ .build()
+ }
+ };
+
+ // Configure client with TLS based on transport
+ let client_config_method =
+ Ident::new(variant.transport.client_config_method(), Span::call_site());
+ let client_config = match tls_mode {
+ Some(TlsMode::Generated) => {
+ // For generated TLS, enable TLS on client - CA cert path will be injected by harness
+ quote! {
+ ::integration::harness::ClientConfig::#client_config_method()
+ .with_tls("localhost".to_string(), None, true)
+ }
+ }
+ Some(TlsMode::SelfSigned) => {
+ // For self-signed TLS, enable TLS on client but disable validation
+ quote! {
+ ::integration::harness::ClientConfig::#client_config_method()
+ .with_tls("localhost".to_string(), None, false)
+ }
+ }
+ None => {
+ quote!(::integration::harness::ClientConfig::#client_config_method())
+ }
+ };
+
+ quote! {
+ let mut __harness = ::integration::harness::TestHarness::builder()
+ .server(#server_config)
+ .client(#client_config)
+ .build()
+ .expect("failed to build test harness");
+ let _ = ::integration::__macro_support::TransportProtocol::#transport;
+ }
+}
+
+fn generate_client_setup() -> TokenStream {
+ quote! {
+ let __client = __harness.client();
+ }
+}
+
+fn generate_param_bindings(
+ params: &[DetectedParam],
+ _has_harness: bool,
+ _has_harness_mut: bool,
+) -> TokenStream {
+ let mut bindings = Vec::new();
+
+ for param in params {
+ match param {
+ DetectedParam::Client { name } => {
+ bindings.push(quote! {
+ let #name = &__client;
+ });
+ }
+ DetectedParam::HarnessRef { name } => {
+ bindings.push(quote! {
+ let #name = &__harness;
+ });
+ }
+ DetectedParam::HarnessMut { name } => {
+ bindings.push(quote! {
+ let #name = &mut __harness;
+ });
+ }
+ DetectedParam::MatrixParam { .. } => {}
+ }
+ }
+
+ quote!(#(#bindings)*)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::attrs::ConfigValue;
+
+ #[test]
+ fn variant_suffix_basic() {
+ let v = TestVariant {
+ transport: Transport::Tcp,
+ segment_size: None,
+ cache_indexes: None,
+ messages_required_to_save: None,
+ message_saver_interval: None,
+ socket_migration: None,
+ };
+ assert_eq!(v.suffix(), "tcp");
+ }
+
+ #[test]
+ fn variant_suffix_full() {
+ let v = TestVariant {
+ transport: Transport::Http,
+ segment_size: Some("1MiB".to_string()),
+ cache_indexes: Some("all".to_string()),
+ messages_required_to_save: Some("64".to_string()),
+ message_saver_interval: None,
+ socket_migration: None,
+ };
+ assert_eq!(v.suffix(), "http_1mb_all_msgs_64");
+ }
+
+ #[test]
+ fn variant_suffix_with_tls() {
+ let v = TestVariant {
+ transport: Transport::TcpTlsSelfSigned,
+ segment_size: None,
+ cache_indexes: None,
+ messages_required_to_save: None,
+ message_saver_interval: None,
+ socket_migration: None,
+ };
+ assert_eq!(v.suffix(), "tcp_tls_self_signed");
+
+ let v = TestVariant {
+ transport: Transport::TcpTlsGenerated,
+ segment_size: None,
+ cache_indexes: None,
+ messages_required_to_save: None,
+ message_saver_interval: None,
+ socket_migration: None,
+ };
+ assert_eq!(v.suffix(), "tcp_tls_generated");
+ }
+
+ #[test]
+ fn generate_variants_simple() {
+ let attrs = IggyTestAttrs::with_transports(vec![Transport::Tcp]);
+ let variants = generate_variants(&attrs);
+ assert_eq!(variants.len(), 1);
+ assert_eq!(variants[0].transport, Transport::Tcp);
+ }
+
+ #[test]
+ fn generate_variants_transport_matrix() {
+ let attrs = IggyTestAttrs::with_transports(vec![Transport::Tcp, Transport::Http]);
+ let variants = generate_variants(&attrs);
+ assert_eq!(variants.len(), 2);
+ }
+
+ #[test]
+ fn generate_variants_full_matrix() {
+ let attrs = IggyTestAttrs {
+ transports: vec![Transport::Tcp, Transport::Http],
+ server: crate::attrs::ServerAttrs {
+ segment_size: ConfigValue::Matrix(vec!["512B".to_string(), "1MiB".to_string()]),
+ cache_indexes: ConfigValue::Matrix(vec!["none".to_string(), "all".to_string()]),
+ ..Default::default()
+ },
+ };
+ let variants = generate_variants(&attrs);
+ assert_eq!(variants.len(), 8);
+ }
+}
diff --git a/core/integration/iggy-harness-derive/src/lib.rs b/core/integration/iggy-harness-derive/src/lib.rs
new file mode 100644
index 000000000..4f5517eba
--- /dev/null
+++ b/core/integration/iggy-harness-derive/src/lib.rs
@@ -0,0 +1,94 @@
+/* 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.
+ */
+
+//! Procedural macro for ergonomic integration test definition.
+//!
+//! The `#[iggy_harness]` attribute replaces boilerplate harness setup with a declarative DSL.
+//!
+//! # Examples
+//!
+//! Simple test with default TCP transport:
+//! ```ignore
+//! #[iggy_harness]
+//! async fn test_ping(client: &IggyClient) {
+//! client.ping().await.unwrap();
+//! }
+//! ```
+//!
+//! Test with transport matrix:
+//! ```ignore
+//! #[iggy_harness(transport = [Tcp, Http, Quic, WebSocket])]
+//! async fn test_all_transports(client: &IggyClient) {
+//! client.ping().await.unwrap();
+//! }
+//! ```
+//!
+//! Test with server config matrix:
+//! ```ignore
+//! #[iggy_harness(server(
+//! segment_size = ["512B", "1MiB"],
+//! cache_indexes = ["none", "all"],
+//! ))]
+//! async fn test_caching(client: &IggyClient) {
+//! // 2 segment sizes × 2 cache modes = 4 tests
+//! }
+//! ```
+
+mod attrs;
+mod codegen;
+mod params;
+
+use proc_macro::TokenStream;
+use syn::{ItemFn, parse_macro_input};
+
+use attrs::IggyTestAttrs;
+use codegen::generate_tests;
+
+/// Attribute macro for declaring Iggy integration tests.
+///
+/// This macro handles test harness setup, client creation, and generates test variants
+/// for infrastructure dimensions (transport, server config).
+///
+/// # Attributes
+///
+/// - `transport = [Tcp, Http, Quic, WebSocket]` - Generate variants for each transport
+/// - `server(key = value)` - Static server config
+/// - `server(key = [v1, v2])` - Generate variants for each config value
+///
+/// # Special Parameters
+///
+/// The test function can request special parameters:
+/// - `client: &IggyClient` - Injected client connected to the test server
+/// - `harness: &TestHarness` - Reference to the test harness (for data_path, etc.)
+/// - `harness: &mut TestHarness` - Mutable reference (for restart_server, etc.)
+///
+/// Any other parameters are passed through from `#[test_matrix]`.
+#[proc_macro_attribute]
+pub fn iggy_harness(attr: TokenStream, item: TokenStream) -> TokenStream {
+ let attrs = match syn::parse::<IggyTestAttrs>(attr) {
+ Ok(a) => a,
+ Err(e) => return e.to_compile_error().into(),
+ };
+
+ let input = parse_macro_input!(item as ItemFn);
+
+ match generate_tests(&attrs, &input) {
+ Ok(tokens) => tokens.into(),
+ Err(e) => e.to_compile_error().into(),
+ }
+}
diff --git a/core/integration/iggy-harness-derive/src/params.rs b/core/integration/iggy-harness-derive/src/params.rs
new file mode 100644
index 000000000..8220f0fec
--- /dev/null
+++ b/core/integration/iggy-harness-derive/src/params.rs
@@ -0,0 +1,195 @@
+/* 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.
+ */
+
+//! Parameter detection for test function signatures.
+//!
+//! Detects special parameters like `client: &IggyClient` and `harness: &TestHarness`
+//! and distinguishes them from user-defined test_matrix parameters.
+
+use proc_macro2::Span;
+use syn::{FnArg, Ident, Pat, PatIdent, PatType, Signature, Type};
+
+/// A detected parameter from the function signature.
+#[derive(Debug)]
+pub enum DetectedParam {
+ /// A client parameter: `client: &IggyClient`
+ Client { name: Ident },
+ /// A harness reference: `harness: &TestHarness`
+ HarnessRef { name: Ident },
+ /// A mutable harness reference: `harness: &mut TestHarness`
+ HarnessMut { name: Ident },
+ /// A parameter from test_matrix (passed through)
+ MatrixParam { name: Ident, ty: Box<Type> },
+}
+
+impl DetectedParam {
+ #[allow(dead_code)]
+ pub fn name(&self) -> &Ident {
+ match self {
+ DetectedParam::Client { name }
+ | DetectedParam::HarnessRef { name }
+ | DetectedParam::HarnessMut { name }
+ | DetectedParam::MatrixParam { name, .. } => name,
+ }
+ }
+}
+
+/// Analyze a function signature and detect parameter types.
+pub fn analyze_signature(sig: &Signature) -> syn::Result<Vec<DetectedParam>> {
+ let mut params = Vec::new();
+
+ for arg in &sig.inputs {
+ let FnArg::Typed(PatType { pat, ty, .. }) = arg else {
+ return Err(syn::Error::new(
+ Span::call_site(),
+ "self parameters not supported in test functions",
+ ));
+ };
+
+ let Pat::Ident(PatIdent { ident, .. }) = pat.as_ref() else {
+ return Err(syn::Error::new(
+ Span::call_site(),
+ "pattern parameters not supported, use simple identifiers",
+ ));
+ };
+
+ let detected = detect_param_type(ident.clone(), ty)?;
+ params.push(detected);
+ }
+
+ Ok(params)
+}
+
+fn detect_param_type(name: Ident, ty: &Type) -> syn::Result<DetectedParam> {
+ let type_str = quote::quote!(#ty).to_string();
+ let normalized = type_str.replace(" ", "");
+
+ if is_iggy_client_type(&normalized) {
+ return Ok(DetectedParam::Client { name });
+ }
+
+ if is_harness_mut_type(&normalized) {
+ return Ok(DetectedParam::HarnessMut { name });
+ }
+
+ if is_harness_ref_type(&normalized) {
+ return Ok(DetectedParam::HarnessRef { name });
+ }
+
+ Ok(DetectedParam::MatrixParam {
+ name,
+ ty: Box::new(ty.clone()),
+ })
+}
+
+fn is_iggy_client_type(normalized: &str) -> bool {
+ normalized.contains("IggyClient") || normalized.contains("ClientWrapper")
+}
+
+fn is_harness_ref_type(normalized: &str) -> bool {
+ normalized.contains("TestHarness") && normalized.starts_with("&")
+}
+
+fn is_harness_mut_type(normalized: &str) -> bool {
+ normalized.contains("TestHarness") && normalized.contains("&mut")
+}
+
+/// Check if any parameter requires a client.
+pub fn needs_client(params: &[DetectedParam]) -> bool {
+ params
+ .iter()
+ .any(|p| matches!(p, DetectedParam::Client { .. }))
+}
+
+/// Check if any parameter requires harness access.
+pub fn needs_harness(params: &[DetectedParam]) -> bool {
+ params.iter().any(|p| {
+ matches!(
+ p,
+ DetectedParam::HarnessRef { .. } | DetectedParam::HarnessMut { .. }
+ )
+ })
+}
+
+/// Check if any parameter requires mutable harness.
+pub fn needs_harness_mut(params: &[DetectedParam]) -> bool {
+ params
+ .iter()
+ .any(|p| matches!(p, DetectedParam::HarnessMut { .. }))
+}
+
+/// Get the matrix parameters (those from test_matrix).
+pub fn matrix_params(params: &[DetectedParam]) -> Vec<&DetectedParam> {
+ params
+ .iter()
+ .filter(|p| matches!(p, DetectedParam::MatrixParam { .. }))
+ .collect()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn parse_sig(s: &str) -> Signature {
+ let item: syn::ItemFn = syn::parse_str(&format!("{s} {{}}")).unwrap();
+ item.sig
+ }
+
+ #[test]
+ fn detect_client_param() {
+ let sig = parse_sig("async fn test(client: &IggyClient)");
+ let params = analyze_signature(&sig).unwrap();
+ assert_eq!(params.len(), 1);
+ assert!(matches!(¶ms[0], DetectedParam::Client { name } if name == "client"));
+ }
+
+ #[test]
+ fn detect_harness_ref() {
+ let sig = parse_sig("async fn test(harness: &TestHarness)");
+ let params = analyze_signature(&sig).unwrap();
+ assert_eq!(params.len(), 1);
+ assert!(matches!(¶ms[0], DetectedParam::HarnessRef { name } if name == "harness"));
+ }
+
+ #[test]
+ fn detect_harness_mut() {
+ let sig = parse_sig("async fn test(harness: &mut TestHarness)");
+ let params = analyze_signature(&sig).unwrap();
+ assert_eq!(params.len(), 1);
+ assert!(matches!(¶ms[0], DetectedParam::HarnessMut { name } if name == "harness"));
+ }
+
+ #[test]
+ fn detect_matrix_param() {
+ let sig = parse_sig("async fn test(count: u32)");
+ let params = analyze_signature(&sig).unwrap();
+ assert_eq!(params.len(), 1);
+ assert!(matches!(¶ms[0], DetectedParam::MatrixParam { name, .. } if name == "count"));
+ }
+
+ #[test]
+ fn detect_mixed_params() {
+ let sig =
+ parse_sig("async fn test(count: u32, client: &IggyClient, harness: &TestHarness)");
+ let params = analyze_signature(&sig).unwrap();
+ assert_eq!(params.len(), 3);
+ assert!(matches!(¶ms[0], DetectedParam::MatrixParam { .. }));
+ assert!(matches!(¶ms[1], DetectedParam::Client { .. }));
+ assert!(matches!(¶ms[2], DetectedParam::HarnessRef { .. }));
+ }
+}
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 53%
copy from core/integration/src/lib.rs
copy to core/integration/src/harness/config/connector.rs
index 269698b2b..0a369f1a2 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,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;
+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, default = "connectors".to_string())]
+ pub consumer_name: String,
+ #[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..7afb0ffa1 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,14 @@
* 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 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 server::TestServerConfig;
diff --git a/core/integration/src/harness/config/server.rs b/core/integration/src/harness/config/server.rs
new file mode 100644
index 000000000..c88bc5d09
--- /dev/null
+++ b/core/integration/src/harness/config/server.rs
@@ -0,0 +1,94 @@
+/*
+ * 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(into)]
+ pub segment_size: Option<String>,
+ #[builder(into)]
+ pub cache_indexes: Option<String>,
+ pub messages_required_to_save: Option<u32>,
+ #[builder(default)]
+ pub tcp_nodelay: bool,
+ #[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,
+ #[builder(default)]
+ pub extra_envs: HashMap<String, String>,
+ #[builder(into)]
+ pub executable_path: Option<String>,
+ #[builder(into)]
+ pub message_saver_interval: Option<String>,
+ pub socket_migration: Option<bool>,
+ pub cluster_enabled: Option<bool>,
+ #[builder(into)]
+ pub heartbeat_interval: 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()
+ .segment_size("1MiB")
+ .tcp_nodelay(true)
+ .quic_enabled(false)
+ .extra_envs(HashMap::from([("FOO".to_string(), "BAR".to_string())]))
+ .build();
+
+ assert_eq!(config.segment_size, Some("1MiB".to_string()));
+ assert!(config.tcp_nodelay);
+ 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);
+ assert!(!config.tcp_nodelay);
+ }
+}
diff --git a/core/integration/src/harness/context.rs b/core/integration/src/harness/context.rs
new file mode 100644
index 000000000..427a02010
--- /dev/null
+++ b/core/integration/src/harness/context.rs
@@ -0,0 +1,180 @@
+/*
+ * 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::fs;
+use std::path::{Path, PathBuf};
+use std::thread;
+use uuid::Uuid;
+
+const TEST_CLEANUP_DISABLED_ENV_VAR: &str = "IGGY_TEST_CLEANUP_DISABLED";
+
+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::now_v7().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,
+ })?;
+
+ self.created = true;
+ Ok(())
+ }
+
+ fn derive_test_name() -> String {
+ thread::current()
+ .name()
+ .map(|s| s.to_string())
+ .unwrap_or_else(|| Uuid::now_v7().to_string())
+ }
+
+ pub fn test_name(&self) -> &str {
+ &self.test_name
+ }
+
+ pub fn base_dir(&self) -> &Path {
+ &self.base_dir
+ }
+
+ pub fn server_data_path(&self) -> PathBuf {
+ self.base_dir.join("server_data")
+ }
+
+ pub fn server_stdout_path(&self) -> PathBuf {
+ self.base_dir.join("server_stdout.log")
+ }
+
+ pub fn server_stderr_path(&self) -> PathBuf {
+ self.base_dir.join("server_stderr.log")
+ }
+
+ 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 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.server_data_path().ends_with("server_data"));
+ assert!(ctx.server_stdout_path().ends_with("server_stdout.log"));
+ 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..8643766e6
--- /dev/null
+++ b/core/integration/src/harness/error.rs
@@ -0,0 +1,178 @@
+/*
+ * 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,
+}
+
+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")
+ }
+ }
+ }
+}
+
+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/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/connector.rs b/core/integration/src/harness/handle/connector.rs
new file mode 100644
index 000000000..426fe9b0d
--- /dev/null
+++ b/core/integration/src/harness/handle/connector.rs
@@ -0,0 +1,308 @@
+/*
+ * 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::ConnectorConfig;
+use crate::harness::context::TestContext;
+use crate::harness::error::TestBinaryError;
+use crate::harness::traits::{IggyDependent, TestBinary};
+use once_cell::sync::Lazy;
+use rand::Rng as _;
+use std::collections::HashMap;
+use std::fs::{self, File, OpenOptions};
+use std::io::Write;
+use std::net::{Ipv4Addr, SocketAddr, TcpListener};
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::thread::panicking;
+use std::time::Duration;
+use tokio::time::sleep;
+
+const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
+const MAX_HEALTH_CHECK_RETRIES: u32 = 1000;
+
+static IGGY_CONNECTORS_BIN: Lazy<PathBuf> = Lazy::new(|| {
+ escargot::CargoBuild::new()
+ .bin("iggy-connectors")
+ .current_release()
+ .current_target()
+ .run()
+ .expect("Failed to build iggy-connectors")
+ .path()
+ .to_path_buf()
+});
+
+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) {
+ let stdout = self
+ .stdout_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stdout log]".to_string());
+
+ let stderr = self
+ .stderr_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stderr log]".to_string());
+
+ (stdout, stderr)
+ }
+
+ fn find_available_port() -> SocketAddr {
+ let mut rng = rand::rng();
+ let max_retries = 100;
+
+ for _ in 0..max_retries {
+ let port = rng.random_range(20000..=29999);
+ 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");
+ }
+
+ 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(),
+ );
+ self.envs.insert(
+ "IGGY_CONNECTORS_IGGY_CONSUMER".to_string(),
+ self.config.consumer_name.clone(),
+ );
+
+ if let Some(addr) = self.iggy_address {
+ self.envs
+ .insert("IGGY_CONNECTORS_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 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();
+
+ let mut command = if let Some(ref path) = self.config.executable_path {
+ Command::new(path)
+ } else {
+ Command::new(IGGY_CONNECTORS_BIN.as_path())
+ };
+
+ 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..MAX_HEALTH_CHECK_RETRIES {
+ match client.get(&http_address).send().await {
+ Ok(_) => return Ok(()),
+ Err(_) => {
+ if retry == MAX_HEALTH_CHECK_RETRIES - 1 {
+ return Err(TestBinaryError::HealthCheckFailed {
+ binary: "iggy-connectors".to_string(),
+ address: http_address,
+ retries: MAX_HEALTH_CHECK_RETRIES,
+ });
+ }
+ sleep(Duration::from_millis(20)).await;
+ }
+ }
+ }
+
+ Ok(())
+ }
+}
+
+impl Drop for ConnectorHandle {
+ fn drop(&mut self) {
+ let _ = self.stop();
+
+ if panicking() {
+ let (stdout, stderr) = self.collect_logs();
+ eprintln!("Iggy connectors stdout:\n{}", stdout);
+ eprintln!("Iggy connectors stderr:\n{}", stderr);
+ }
+ }
+}
diff --git a/core/integration/src/harness/handle/mcp.rs b/core/integration/src/harness/handle/mcp.rs
new file mode 100644
index 000000000..4bc8849c2
--- /dev/null
+++ b/core/integration/src/harness/handle/mcp.rs
@@ -0,0 +1,331 @@
+/*
+ * 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::McpConfig;
+use crate::harness::context::TestContext;
+use crate::harness::error::TestBinaryError;
+use crate::harness::traits::{IggyDependent, TestBinary};
+use once_cell::sync::Lazy;
+use rand::Rng as _;
+use rmcp::{
+ RoleClient, ServiceExt,
+ model::{ClientCapabilities, ClientInfo, Implementation, InitializeRequestParam},
+ service::RunningService,
+ transport::StreamableHttpClientTransport,
+};
+use std::collections::HashMap;
+use std::fs::{self, File};
+use std::net::{Ipv4Addr, SocketAddr, TcpListener};
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::sync::Arc;
+use std::thread::panicking;
+use std::time::Duration;
+use tokio::time::sleep;
+
+const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
+const MAX_HEALTH_CHECK_RETRIES: u32 = 3000;
+
+static IGGY_MCP_BIN: Lazy<PathBuf> = Lazy::new(|| {
+ escargot::CargoBuild::new()
+ .bin("iggy-mcp")
+ .current_release()
+ .current_target()
+ .run()
+ .expect("Failed to build iggy-mcp")
+ .path()
+ .to_path_buf()
+});
+
+pub type McpClient = RunningService<RoleClient, InitializeRequestParam>;
+
+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()
+ },
+ };
+
+ 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) {
+ let stdout = self
+ .stdout_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stdout log]".to_string());
+
+ let stderr = self
+ .stderr_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stderr log]".to_string());
+
+ (stdout, stderr)
+ }
+
+ fn find_available_port() -> SocketAddr {
+ let mut rng = rand::rng();
+ let max_retries = 100;
+
+ for _ in 0..max_retries {
+ #[cfg(target_os = "linux")]
+ let port = rng.random_range(20000..=29999);
+
+ #[cfg(target_os = "macos")]
+ let port = rng.random_range(20000..=49151);
+
+ #[cfg(target_os = "windows")]
+ let port = rng.random_range(20000..=49151);
+
+ 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");
+ }
+
+ 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,
+ }
+ }
+
+ 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::new(IGGY_MCP_BIN.as_path())
+ };
+
+ 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..MAX_HEALTH_CHECK_RETRIES {
+ match client.get(&http_address).send().await {
+ Ok(_) => return Ok(()),
+ Err(_) => {
+ if retry == MAX_HEALTH_CHECK_RETRIES - 1 {
+ return Err(TestBinaryError::HealthCheckFailed {
+ binary: "iggy-mcp".to_string(),
+ address: http_address,
+ retries: MAX_HEALTH_CHECK_RETRIES,
+ });
+ }
+ sleep(Duration::from_millis(20)).await;
+ }
+ }
+ }
+
+ Ok(())
+ }
+}
+
+impl Drop for McpHandle {
+ fn drop(&mut self) {
+ let _ = self.stop();
+
+ if panicking() {
+ let (stdout, stderr) = self.collect_logs();
+ eprintln!("Iggy MCP server stdout:\n{}", stdout);
+ eprintln!("Iggy MCP server stderr:\n{}", stderr);
+ }
+ }
+}
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..bf4373a8c 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,12 @@
* 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 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..a820b24d6
--- /dev/null
+++ b/core/integration/src/harness/handle/server.rs
@@ -0,0 +1,777 @@
+/*
+ * 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 futures::executor::block_on;
+use iggy::prelude::DEFAULT_ROOT_PASSWORD;
+use iggy::prelude::DEFAULT_ROOT_USERNAME;
+use iggy_common::ConfigProvider;
+use once_cell::sync::Lazy;
+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, panicking, 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";
+
+static IGGY_SERVER_BIN: Lazy<PathBuf> = Lazy::new(|| {
+ escargot::CargoBuild::new()
+ .package("server")
+ .bin("iggy-server")
+ .current_release()
+ .current_target()
+ .run()
+ .expect("Failed to build iggy-server")
+ .path()
+ .to_path_buf()
+});
+
+#[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 {
+ 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.server_data_path()
+ }
+
+ /// 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) {
+ let stdout = self
+ .stdout_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stdout log]".to_string());
+
+ let stderr = self
+ .stderr_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())
+ .unwrap_or_else(|| "[No stderr log]".to_string());
+
+ (stdout, stderr)
+ }
+
+ fn build_envs(&mut self) {
+ // Pass through IGGY_* env vars from parent process, except those critical for test isolation.
+ // This allows external configuration to override test defaults.
+ const PROTECTED_PREFIXES: &[&str] = &[
+ "IGGY_SYSTEM_PATH", // Test data isolation
+ "IGGY_TCP_ADDRESS", // Port allocation
+ "IGGY_HTTP_ADDRESS", // Port allocation
+ "IGGY_QUIC_ADDRESS", // Port allocation
+ "IGGY_WEBSOCKET_ADDRESS", // Port allocation
+ ];
+
+ 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());
+
+ // IGGY_SYSTEM_PATH is always set by the test harness for isolation
+ let data_path = self.context.server_data_path();
+ self.envs.insert(
+ "IGGY_SYSTEM_PATH".to_string(),
+ data_path.display().to_string(),
+ );
+
+ if let Some(ref seg_size) = self.config.segment_size {
+ self.envs
+ .entry("IGGY_SYSTEM_SEGMENT_SIZE".to_string())
+ .or_insert_with(|| seg_size.clone());
+ }
+
+ if let Some(ref cache) = self.config.cache_indexes {
+ self.envs
+ .entry("IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string())
+ .or_insert_with(|| cache.clone());
+ }
+
+ if let Some(count) = self.config.messages_required_to_save {
+ self.envs
+ .entry("IGGY_SYSTEM_PARTITION_MESSAGES_REQUIRED_TO_SAVE".to_string())
+ .or_insert_with(|| count.to_string());
+ }
+
+ if self.config.tcp_nodelay {
+ self.envs
+ .entry("IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS".to_string())
+ .or_insert_with(|| "true".to_string());
+ self.envs
+ .entry("IGGY_TCP_SOCKET_NODELAY".to_string())
+ .or_insert_with(|| "true".to_string());
+ }
+
+ 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());
+ }
+
+ if let Some(ref enc) = self.config.encryption {
+ self.envs
+ .entry("IGGY_SYSTEM_ENCRYPTION_KEY".to_string())
+ .or_insert_with(|| enc.key.clone());
+ }
+
+ if let Some(ref tls) = self.config.tls {
+ self.envs
+ .entry("IGGY_TCP_TLS_ENABLED".to_string())
+ .or_insert_with(|| "true".to_string());
+ if tls.self_signed {
+ self.envs
+ .entry("IGGY_TCP_TLS_SELF_SIGNED".to_string())
+ .or_insert_with(|| "true".to_string());
+ } else {
+ self.envs
+ .entry("IGGY_TCP_TLS_SELF_SIGNED".to_string())
+ .or_insert_with(|| "false".to_string());
+ let cert_dir = self.generated_cert_dir.as_ref().unwrap_or(&tls.cert_dir);
+ self.envs
+ .entry("IGGY_TCP_TLS_CERT_FILE".to_string())
+ .or_insert_with(|| cert_dir.join("test_cert.pem").display().to_string());
+ self.envs
+ .entry("IGGY_TCP_TLS_KEY_FILE".to_string())
+ .or_insert_with(|| cert_dir.join("test_key.pem").display().to_string());
+ }
+ }
+
+ if let Some(ref tls) = self.config.websocket_tls {
+ self.envs
+ .entry("IGGY_WEBSOCKET_TLS_ENABLED".to_string())
+ .or_insert_with(|| "true".to_string());
+ if tls.self_signed {
+ self.envs
+ .entry("IGGY_WEBSOCKET_TLS_SELF_SIGNED".to_string())
+ .or_insert_with(|| "true".to_string());
+ } else {
+ self.envs
+ .entry("IGGY_WEBSOCKET_TLS_SELF_SIGNED".to_string())
+ .or_insert_with(|| "false".to_string());
+ let cert_dir = self.generated_cert_dir.as_ref().unwrap_or(&tls.cert_dir);
+ self.envs
+ .entry("IGGY_WEBSOCKET_TLS_CERT_FILE".to_string())
+ .or_insert_with(|| cert_dir.join("test_cert.pem").display().to_string());
+ self.envs
+ .entry("IGGY_WEBSOCKET_TLS_KEY_FILE".to_string())
+ .or_insert_with(|| cert_dir.join("test_key.pem").display().to_string());
+ }
+ }
+
+ if let Some(ref interval) = self.config.message_saver_interval {
+ self.envs
+ .entry("IGGY_MESSAGE_SAVER_INTERVAL".to_string())
+ .or_insert_with(|| interval.clone());
+ }
+
+ if let Some(migration) = self.config.socket_migration {
+ self.envs
+ .entry("IGGY_TCP_SOCKET_MIGRATION".to_string())
+ .or_insert_with(|| migration.to_string());
+ }
+
+ if let Some(enabled) = self.config.cluster_enabled {
+ self.envs
+ .entry("IGGY_CLUSTER_ENABLED".to_string())
+ .or_insert_with(|| enabled.to_string());
+ }
+
+ if let Some(ref interval) = self.config.heartbeat_interval {
+ self.envs
+ .entry("IGGY_HEARTBEAT_INTERVAL".to_string())
+ .or_insert_with(|| interval.clone());
+ }
+
+ // Extra envs from config can still override (explicit test config intent)
+ for (k, v) in &self.config.extra_envs {
+ self.envs.insert(k.clone(), v.clone());
+ }
+
+ self.set_protocol_addresses();
+ }
+
+ 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.context.server_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 (default: tcp=8090, http=3000, quic=8080, ws=8092)
+ if tcp_port == 8090
+ || http_port == 3000
+ || quic_port == 8080
+ || websocket_port == 8092
+ {
+ 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 {
+ 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.context.server_data_path();
+ let config_path = data_path.join("runtime/current_config.toml");
+ if config_path.exists() {
+ let _ = fs::remove_file(&config_path);
+ }
+
+ let mut command = if let Some(ref path) = self.config.executable_path {
+ Command::new(path)
+ } else {
+ Command::new(IGGY_SERVER_BIN.as_path())
+ };
+
+ 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.context.server_stdout_path();
+ let stderr_path = self.context.server_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-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(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!(
+ "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 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();
+ if panicking() {
+ let (stdout, stderr) = self.collect_logs();
+ eprintln!("Iggy server stdout:\n{}", stdout);
+ eprintln!("Iggy server stderr:\n{}", stderr);
+ }
+ }
+}
diff --git a/core/integration/src/harness/helpers.rs b/core/integration/src/harness/helpers.rs
new file mode 100644
index 000000000..f27430704
--- /dev/null
+++ b/core/integration/src/harness/helpers.rs
@@ -0,0 +1,115 @@
+/*
+ * 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::{
+ ClientWrapper, DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, GlobalPermissions, Identifier,
+ IdentityInfo, Permissions, UserClient, UserStatus,
+};
+
+pub const USER_PASSWORD: &str = "secret";
+
+/// Login as root user.
+pub async fn login_root(client: &ClientWrapper) -> 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: &ClientWrapper, 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: &ClientWrapper, 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: &ClientWrapper, 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(client: &ClientWrapper) {
+ use iggy::prelude::StreamClient;
+
+ 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..b71e43ff7
--- /dev/null
+++ b/core/integration/src/harness/mod.rs
@@ -0,0 +1,79 @@
+/*
+ * 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 handle;
+mod helpers;
+mod orchestrator;
+mod traits;
+
+pub use config::{
+ AutoLoginConfig, ClientConfig, ConnectorConfig, EncryptionConfig, IpAddrKind, McpConfig,
+ TestServerConfig, TlsConfig,
+};
+
+pub use context::TestContext;
+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,
+};
diff --git a/core/integration/src/harness/orchestrator.rs b/core/integration/src/harness/orchestrator.rs
new file mode 100644
index 000000000..0f908cc54
--- /dev/null
+++ b/core/integration/src/harness/orchestrator.rs
@@ -0,0 +1,699 @@
+/*
+ * 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)>,
+}
+
+/// 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>,
+ 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> {
+ if self.started {
+ return Err(TestBinaryError::AlreadyStarted);
+ }
+
+ // Start server first (required for other binaries)
+ if let Some(ref mut server) = self.server {
+ server.start()?;
+ }
+
+ // Start MCP if configured
+ if let Some(ref mut mcp) = self.mcp {
+ if let Some(tcp_addr) = self.server.as_ref().and_then(|s| s.tcp_addr()) {
+ mcp.set_iggy_address(tcp_addr);
+ }
+ mcp.start()?;
+ mcp.wait_ready().await?;
+ }
+
+ // Start connectors if configured
+ if let Some(ref mut connector) = self.connector {
+ if let Some(tcp_addr) = self.server.as_ref().and_then(|s| s.tcp_addr()) {
+ connector.set_iggy_address(tcp_addr);
+ }
+ connector.start()?;
+ connector.wait_ready().await?;
+ }
+
+ // Create and connect clients
+ self.create_clients().await?;
+
+ self.started = true;
+ 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()?;
+
+ // Find TCP client config to get TLS and nodelay settings
+ let tcp_config = self
+ .client_configs
+ .iter()
+ .find(|c| c.transport == TransportProtocol::Tcp);
+
+ let (nodelay, tls_enabled, tls_domain, tls_validate_certificate) = tcp_config
+ .map(|c| {
+ (
+ c.tcp_nodelay,
+ c.tls_enabled,
+ c.tls_domain
+ .clone()
+ .unwrap_or_else(|| "localhost".to_string()),
+ c.tls_validate_certificate,
+ )
+ })
+ .unwrap_or_default();
+
+ // Get CA cert from server if TLS is enabled
+ let tls_ca_file = if tls_enabled {
+ server
+ .tls_ca_cert_path()
+ .map(|p| p.to_string_lossy().to_string())
+ } else {
+ None
+ };
+
+ Some(TcpClientFactory {
+ server_addr: addr.to_string(),
+ nodelay,
+ tls_enabled,
+ tls_domain,
+ tls_ca_file,
+ 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()?;
+
+ // Find WebSocket client config to get TLS settings
+ let ws_config = self
+ .client_configs
+ .iter()
+ .find(|c| c.transport == TransportProtocol::WebSocket);
+
+ let (tls_enabled, tls_domain, tls_validate_certificate) = ws_config
+ .map(|c| {
+ (
+ c.tls_enabled,
+ c.tls_domain
+ .clone()
+ .unwrap_or_else(|| "localhost".to_string()),
+ c.tls_validate_certificate,
+ )
+ })
+ .unwrap_or_default();
+
+ // Get CA cert from server if TLS is enabled
+ let tls_ca_file = if tls_enabled {
+ server
+ .tls_ca_cert_path()
+ .map(|p| p.to_string_lossy().to_string())
+ } else {
+ None
+ };
+
+ Some(WebSocketClientFactory {
+ server_addr: addr.to_string(),
+ tls_enabled,
+ tls_domain,
+ tls_ca_file,
+ tls_validate_certificate,
+ })
+ }
+
+ /// 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
+ }
+
+ /// Create a new TCP client logged in as root.
+ pub async fn tcp_root_client(&self) -> Result<IggyClient, TestBinaryError> {
+ let factory = self
+ .tcp_client_factory()
+ .ok_or_else(|| TestBinaryError::InvalidState {
+ message: "TCP transport not available".to_string(),
+ })?;
+ self.create_root_client(&factory).await
+ }
+
+ /// Create a new HTTP client logged in as root.
+ pub async fn http_root_client(&self) -> Result<IggyClient, TestBinaryError> {
+ let factory = self
+ .http_client_factory()
+ .ok_or_else(|| TestBinaryError::InvalidState {
+ message: "HTTP transport not available".to_string(),
+ })?;
+ self.create_root_client(&factory).await
+ }
+
+ /// Create a new QUIC client logged in as root.
+ pub async fn quic_root_client(&self) -> Result<IggyClient, TestBinaryError> {
+ let factory = self
+ .quic_client_factory()
+ .ok_or_else(|| TestBinaryError::InvalidState {
+ message: "QUIC transport not available".to_string(),
+ })?;
+ self.create_root_client(&factory).await
+ }
+
+ /// Create a new WebSocket client logged in as root.
+ pub async fn websocket_root_client(&self) -> Result<IggyClient, TestBinaryError> {
+ let factory =
+ self.websocket_client_factory()
+ .ok_or_else(|| TestBinaryError::InvalidState {
+ message: "WebSocket transport not available".to_string(),
+ })?;
+ self.create_root_client(&factory).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>,
+ 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,
+ 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 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,
+ 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()
+ .segment_size("1MiB")
+ .tcp_nodelay(true)
+ .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()
+ .segment_size("2MiB")
+ .quic_enabled(false)
+ .websocket_enabled(false)
+ .extra_envs(HashMap::from([("TEST".to_string(), "value".to_string())]))
+ .build(),
+ )
+ .build()
+ .unwrap();
+
+ assert!(harness.server.is_some());
+ }
+}
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..0c19f3550 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)]
@@ -34,3 +35,13 @@ pub mod test_mcp_server;
#[allow(deprecated)]
pub mod test_server;
pub mod test_tls_utils;
+
+pub use iggy_harness_derive::iggy_harness;
+
+/// Support module for macro-generated code. Not intended for direct use.
+#[doc(hidden)]
+pub mod __macro_support {
+ pub use crate::harness::{ClientConfig, TestHarness, TestServerConfig, TlsConfig};
+ pub use iggy::prelude::ClientWrapper;
+ pub use iggy_common::TransportProtocol;
+}
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 c63681340..400147848 100644
--- a/core/integration/tests/server/concurrent_addition.rs
+++ b/core/integration/tests/server/concurrent_addition.rs
@@ -84,7 +84,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/message_retrieval.rs b/core/integration/tests/server/message_retrieval.rs
deleted file mode 100644
index 24a4ff0d0..000000000
--- a/core/integration/tests/server/message_retrieval.rs
+++ /dev/null
@@ -1,152 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::server::scenarios::{offset_scenario, timestamp_scenario};
-use integration::{
- tcp_client::TcpClientFactory,
- test_server::{IpAddrKind, TestServer},
-};
-use serial_test::parallel;
-use std::collections::HashMap;
-use test_case::test_matrix;
-
-fn segment_size_512b() -> &'static str {
- "512B"
-}
-
-fn segment_size_1kb() -> &'static str {
- "1KiB"
-}
-
-fn segment_size_10mb() -> &'static str {
- "10MiB"
-}
-
-fn cache_none() -> &'static str {
- "none"
-}
-
-fn cache_all() -> &'static str {
- "all"
-}
-
-fn cache_open_segment() -> &'static str {
- "open_segment"
-}
-
-fn msgs_req_32() -> u32 {
- 32
-}
-
-fn msgs_req_64() -> u32 {
- 64
-}
-
-fn msgs_req_1024() -> u32 {
- 1024
-}
-
-fn msgs_req_9984() -> u32 {
- 9984
-}
-
-#[test_matrix(
- [segment_size_512b(), segment_size_1kb(), segment_size_10mb()],
- [cache_none(), cache_all(), cache_open_segment()],
- [msgs_req_32(), msgs_req_64(), msgs_req_1024(), msgs_req_9984()]
-)]
-#[tokio::test]
-#[parallel]
-async fn get_by_offset_scenario(
- segment_size: &str,
- cache_indexes: &str,
- messages_required_to_save: u32,
-) {
- let mut extra_envs = HashMap::new();
- extra_envs.insert(
- "IGGY_SYSTEM_SEGMENT_SIZE".to_string(),
- segment_size.to_string(),
- );
- extra_envs.insert(
- "IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string(),
- cache_indexes.to_string(),
- );
- extra_envs.insert(
- "IGGY_SYSTEM_PARTITION_MESSAGES_REQUIRED_TO_SAVE".to_string(),
- messages_required_to_save.to_string(),
- );
- extra_envs.insert(
- "IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS".to_string(),
- "true".to_string(),
- );
- extra_envs.insert("IGGY_TCP_SOCKET_NODELAY".to_string(), "true".to_string());
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let client_factory = TcpClientFactory {
- server_addr,
- ..Default::default()
- };
-
- offset_scenario::run(&client_factory).await;
-}
-
-#[test_matrix(
- [segment_size_512b(), segment_size_1kb(), segment_size_10mb()],
- [cache_none(), cache_all(), cache_open_segment()],
- [msgs_req_32(), msgs_req_64(), msgs_req_1024(), msgs_req_9984()]
-)]
-#[tokio::test]
-#[parallel]
-async fn get_by_timestamp_scenario(
- segment_size: &str,
- cache_indexes: &str,
- messages_required_to_save: u32,
-) {
- let mut extra_envs = HashMap::new();
- extra_envs.insert(
- "IGGY_SYSTEM_SEGMENT_SIZE".to_string(),
- segment_size.to_string(),
- );
- extra_envs.insert(
- "IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string(),
- cache_indexes.to_string(),
- );
- extra_envs.insert(
- "IGGY_SYSTEM_PARTITION_MESSAGES_REQUIRED_TO_SAVE".to_string(),
- messages_required_to_save.to_string(),
- );
- extra_envs.insert(
- "IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS".to_string(),
- "true".to_string(),
- );
- extra_envs.insert("IGGY_TCP_SOCKET_NODELAY".to_string(), "true".to_string());
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let client_factory = TcpClientFactory {
- server_addr,
- ..Default::default()
- };
-
- timestamp_scenario::run(&client_factory).await;
-}
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 65ee5d2c4..964219309 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -19,9 +19,7 @@
mod cg;
mod concurrent_addition;
mod general;
-mod message_retrieval;
mod scenarios;
-mod specific;
use iggy_common::TransportProtocol;
use integration::{
@@ -144,7 +142,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/scenarios/cross_protocol_pat_scenario.rs b/core/integration/tests/server/scenarios/cross_protocol_pat_scenario.rs
index f8226daf8..506877f8a 100644
--- a/core/integration/tests/server/scenarios/cross_protocol_pat_scenario.rs
+++ b/core/integration/tests/server/scenarios/cross_protocol_pat_scenario.rs
@@ -21,213 +21,135 @@
//! they won't see PATs created via HTTP (which runs only on shard 0).
use iggy::prelude::*;
-use integration::{
- http_client::HttpClientFactory,
- tcp_client::TcpClientFactory,
- test_server::{ClientFactory, IpAddrKind, TestServer, login_root},
-};
-use std::collections::HashMap;
+use integration::iggy_harness;
+use test_case::test_matrix;
const PAT_NAME: &str = "cross-protocol-test-pat";
-const TCP_CLIENT_COUNT: usize = 20;
+const CLIENT_COUNT: usize = 20;
-/// Create PAT via HTTP, list via multiple TCP connections.
-pub async fn run() {
- let mut extra_envs = HashMap::new();
- extra_envs.insert(
- "IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS".to_string(),
- "true".to_string(),
- );
- extra_envs.insert("IGGY_TCP_SOCKET_NODELAY".to_string(), "true".to_string());
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let http_addr = test_server
- .get_http_api_addr()
- .expect("HTTP address not available");
- let tcp_addr = test_server
- .get_raw_tcp_addr()
- .expect("TCP address not available");
-
- let http_factory = HttpClientFactory {
- server_addr: http_addr,
- };
- let http_client = create_client(&http_factory).await;
- login_root(&http_client).await;
-
- let created_pat = http_client
- .create_personal_access_token(PAT_NAME, IggyExpiry::NeverExpire)
- .await
- .expect("Failed to create PAT via HTTP");
-
- assert!(!created_pat.token.is_empty());
-
- let http_pats = http_client
- .get_personal_access_tokens()
- .await
- .expect("Failed to get PATs via HTTP");
- assert_eq!(http_pats.len(), 1);
- assert_eq!(http_pats[0].name, PAT_NAME);
-
- let tcp_factory = TcpClientFactory {
- server_addr: tcp_addr,
- ..Default::default()
- };
-
- let mut failures = Vec::new();
-
- for i in 0..TCP_CLIENT_COUNT {
- let tcp_client = create_client(&tcp_factory).await;
- login_root(&tcp_client).await;
-
- let tcp_pats = tcp_client
- .get_personal_access_tokens()
- .await
- .expect("Failed to get PATs via TCP");
-
- if tcp_pats.is_empty() {
- failures.push(format!("TCP client {} saw 0 PATs (expected 1)", i));
- } else if tcp_pats.len() != 1 {
- failures.push(format!(
- "TCP client {} saw {} PATs (expected 1)",
- i,
- tcp_pats.len()
- ));
- } else if tcp_pats[0].name != PAT_NAME {
- failures.push(format!(
- "TCP client {} saw wrong PAT name: {}",
- i, tcp_pats[0].name
- ));
- }
- }
-
- let _ = http_client.delete_personal_access_token(PAT_NAME).await;
-
- assert!(
- failures.is_empty(),
- "Cross-protocol PAT visibility failures ({}/{} clients failed):\n{}",
- failures.len(),
- TCP_CLIENT_COUNT,
- failures.join("\n")
- );
+fn http_to_tcp() -> bool {
+ true
}
-
-/// Create PAT via TCP, list via HTTP. Should work since TCP handlers broadcast.
-pub async fn run_tcp_to_http() {
- let mut extra_envs = HashMap::new();
- extra_envs.insert(
- "IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS".to_string(),
- "true".to_string(),
- );
- extra_envs.insert("IGGY_TCP_SOCKET_NODELAY".to_string(), "true".to_string());
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let http_addr = test_server
- .get_http_api_addr()
- .expect("HTTP address not available");
- let tcp_addr = test_server
- .get_raw_tcp_addr()
- .expect("TCP address not available");
-
- let tcp_factory = TcpClientFactory {
- server_addr: tcp_addr,
- ..Default::default()
- };
- let tcp_client = create_client(&tcp_factory).await;
- login_root(&tcp_client).await;
-
- let created_pat = tcp_client
- .create_personal_access_token(PAT_NAME, IggyExpiry::NeverExpire)
- .await
- .expect("Failed to create PAT via TCP");
-
- assert!(!created_pat.token.is_empty());
-
- let http_factory = HttpClientFactory {
- server_addr: http_addr,
- };
- let http_client = create_client(&http_factory).await;
- login_root(&http_client).await;
-
- let http_pats = http_client
- .get_personal_access_tokens()
- .await
- .expect("Failed to get PATs via HTTP");
-
- assert_eq!(http_pats.len(), 1);
- assert_eq!(http_pats[0].name, PAT_NAME);
-
- let _ = tcp_client.delete_personal_access_token(PAT_NAME).await;
+fn tcp_to_http() -> bool {
+ false
}
-/// Create via TCP, delete via HTTP, verify deletion visible via TCP.
-pub async fn run_delete_visibility() {
- let mut extra_envs = HashMap::new();
- extra_envs.insert(
- "IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS".to_string(),
- "true".to_string(),
- );
- extra_envs.insert("IGGY_TCP_SOCKET_NODELAY".to_string(), "true".to_string());
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let http_addr = test_server
- .get_http_api_addr()
- .expect("HTTP address not available");
- let tcp_addr = test_server
- .get_raw_tcp_addr()
- .expect("TCP address not available");
-
- let tcp_factory = TcpClientFactory {
- server_addr: tcp_addr,
- ..Default::default()
- };
- let tcp_client = create_client(&tcp_factory).await;
- login_root(&tcp_client).await;
-
- tcp_client
- .create_personal_access_token(PAT_NAME, IggyExpiry::NeverExpire)
- .await
- .expect("Failed to create PAT via TCP");
-
- let http_factory = HttpClientFactory {
- server_addr: http_addr,
- };
- let http_client = create_client(&http_factory).await;
- login_root(&http_client).await;
-
- http_client
- .delete_personal_access_token(PAT_NAME)
- .await
- .expect("Failed to delete PAT via HTTP");
-
- let mut still_visible_count = 0;
- for _ in 0..TCP_CLIENT_COUNT {
- let tcp_check_client = create_client(&tcp_factory).await;
- login_root(&tcp_check_client).await;
+/// Test full PAT lifecycle with cross-protocol operations.
+/// Verifies cross-shard event broadcasting works for both create and delete.
+///
+/// - `http_to_tcp`: Create via HTTP → Verify via TCP (20 clients) → Delete via TCP → Verify via HTTP
+/// - `tcp_to_http`: Create via TCP → Verify via HTTP → Delete via HTTP → Verify via TCP (20 clients)
+#[iggy_harness(server(tcp_nodelay = true))]
+#[test_matrix([http_to_tcp(), tcp_to_http()])]
+async fn cross_protocol_pat_lifecycle(create_via_http: bool, harness: &TestHarness) {
+ if create_via_http {
+ // Create via HTTP → Verify via TCP (20 clients) → Delete via TCP → Verify via HTTP
+ let http_client = harness.http_root_client().await.unwrap();
+
+ let created_pat = http_client
+ .create_personal_access_token(PAT_NAME, IggyExpiry::NeverExpire)
+ .await
+ .expect("Failed to create PAT via HTTP");
+ assert!(!created_pat.token.is_empty());
+
+ // Verify creation visible via multiple TCP clients (they may land on different shards)
+ let mut create_failures = Vec::new();
+ for i in 0..CLIENT_COUNT {
+ let tcp_client = harness.tcp_root_client().await.unwrap();
+ let tcp_pats = tcp_client
+ .get_personal_access_tokens()
+ .await
+ .expect("Failed to get PATs via TCP");
+
+ if tcp_pats.len() != 1 {
+ create_failures.push(format!(
+ "TCP client {} saw {} PATs (expected 1)",
+ i,
+ tcp_pats.len()
+ ));
+ } else if tcp_pats[0].name != PAT_NAME {
+ create_failures.push(format!(
+ "TCP client {} saw wrong PAT name: {}",
+ i, tcp_pats[0].name
+ ));
+ }
+ }
+ assert!(
+ create_failures.is_empty(),
+ "PAT creation visibility failures ({}/{} clients):\n{}",
+ create_failures.len(),
+ CLIENT_COUNT,
+ create_failures.join("\n")
+ );
+
+ // Delete via TCP (cross-protocol)
+ let tcp_client = harness.tcp_root_client().await.unwrap();
+ tcp_client
+ .delete_personal_access_token(PAT_NAME)
+ .await
+ .expect("Failed to delete PAT via TCP");
- let tcp_pats = tcp_check_client
+ // Verify deletion visible via HTTP
+ let http_pats = http_client
.get_personal_access_tokens()
.await
- .expect("Failed to get PATs via TCP");
+ .expect("Failed to get PATs via HTTP after delete");
+ assert!(
+ http_pats.is_empty(),
+ "PAT still visible via HTTP after TCP deletion"
+ );
+ } else {
+ // Create via TCP → Verify via HTTP → Delete via HTTP → Verify via TCP (20 clients)
+ let tcp_client = harness.tcp_root_client().await.unwrap();
+
+ let created_pat = tcp_client
+ .create_personal_access_token(PAT_NAME, IggyExpiry::NeverExpire)
+ .await
+ .expect("Failed to create PAT via TCP");
+ assert!(!created_pat.token.is_empty());
- if !tcp_pats.is_empty() {
- still_visible_count += 1;
+ // Verify creation visible via HTTP
+ let http_client = harness.http_root_client().await.unwrap();
+ let http_pats = http_client
+ .get_personal_access_tokens()
+ .await
+ .expect("Failed to get PATs via HTTP");
+ assert_eq!(
+ http_pats.len(),
+ 1,
+ "PAT not visible via HTTP after TCP creation"
+ );
+ assert_eq!(http_pats[0].name, PAT_NAME);
+
+ // Delete via HTTP (cross-protocol)
+ http_client
+ .delete_personal_access_token(PAT_NAME)
+ .await
+ .expect("Failed to delete PAT via HTTP");
+
+ // Verify deletion visible via multiple TCP clients (they may land on different shards)
+ let mut delete_failures = Vec::new();
+ for i in 0..CLIENT_COUNT {
+ let tcp_check_client = harness.tcp_root_client().await.unwrap();
+ let tcp_pats = tcp_check_client
+ .get_personal_access_tokens()
+ .await
+ .expect("Failed to get PATs via TCP after delete");
+
+ if !tcp_pats.is_empty() {
+ delete_failures.push(format!(
+ "TCP client {} still sees {} PATs after deletion",
+ i,
+ tcp_pats.len()
+ ));
+ }
}
+ assert!(
+ delete_failures.is_empty(),
+ "PAT deletion visibility failures ({}/{} clients):\n{}",
+ delete_failures.len(),
+ CLIENT_COUNT,
+ delete_failures.join("\n")
+ );
}
-
- assert_eq!(
- still_visible_count, 0,
- "Deleted PAT still visible to {} TCP clients",
- still_visible_count
- );
-}
-
-async fn create_client(client_factory: &dyn ClientFactory) -> IggyClient {
- let client = client_factory.create_client().await;
- IggyClient::create(client, None, None)
}
diff --git a/core/integration/tests/server/scenarios/delete_segments_scenario.rs b/core/integration/tests/server/scenarios/delete_segments_scenario.rs
index a5959e5b5..55bdd29a6 100644
--- a/core/integration/tests/server/scenarios/delete_segments_scenario.rs
+++ b/core/integration/tests/server/scenarios/delete_segments_scenario.rs
@@ -1,23 +1,24 @@
-// 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.
+/* 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 bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, TestServer};
+use integration::iggy_harness;
use std::fs::{DirEntry, read_dir};
const STREAM_NAME: &str = "test_stream";
@@ -25,14 +26,9 @@ const TOPIC_NAME: &str = "test_topic";
const PARTITION_ID: u32 = 0;
const LOG_EXTENSION: &str = "log";
-pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
- let client = client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
-
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap();
+#[iggy_harness(server(segment_size = "1MiB"))]
+async fn should_delete_segments_and_validate_filesystem(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
let stream = client.create_stream(STREAM_NAME).await.unwrap();
let stream_id = stream.id;
@@ -51,7 +47,6 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
.unwrap();
let topic_id = topic.id;
- // Send 5 large messages to create multiple segments
let large_payload = "A".repeat(1024 * 1024);
for i in 0..5 {
@@ -73,13 +68,13 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
.unwrap();
}
- // Wait for segments to be persisted and closed
tokio::time::sleep(std::time::Duration::from_millis(2000)).await;
- // Check initial segment count on filesystem
- let data_path = test_server.get_local_data_path();
- let partition_path =
- format!("{data_path}/streams/{stream_id}/topics/{topic_id}/partitions/{PARTITION_ID}");
+ let data_path = harness.server().data_path();
+ let partition_path = format!(
+ "{}/streams/{stream_id}/topics/{topic_id}/partitions/{PARTITION_ID}",
+ data_path.display()
+ );
let initial_segments = get_segment_paths_for_partition(&partition_path);
println!(
@@ -98,7 +93,6 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
let initial_count = initial_segments.len();
- // Test delete segments command - keep only 2 segments
let segments_to_keep = 2u32;
let result = client
.delete_segments(
@@ -116,7 +110,6 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
- // Verify segments were deleted from filesystem
let remaining_segments = get_segment_paths_for_partition(&partition_path);
println!(
"Remaining segments: {:?}",
@@ -126,7 +119,6 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
.collect::<Vec<_>>()
);
- // Should have at most segments_to_keep + 1 (closed + open segments)
assert!(
remaining_segments.len() <= (segments_to_keep + 1) as usize,
"Expected at most {} segments but got {}",
@@ -141,7 +133,6 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
remaining_segments.len()
);
- // Test edge cases
let result = client
.delete_segments(
&Identifier::named(STREAM_NAME).unwrap(),
@@ -155,7 +146,6 @@ pub async fn run(client_factory: &dyn ClientFactory, test_server: &TestServer) {
"Delete segments with 0 count should succeed"
);
- // Test error cases
let result = client
.delete_segments(
&Identifier::numeric(999).unwrap(),
diff --git a/core/integration/tests/server/scenarios/message_size_scenario.rs b/core/integration/tests/server/scenarios/message_size_scenario.rs
index f2742c8fa..1b400e064 100644
--- a/core/integration/tests/server/scenarios/message_size_scenario.rs
+++ b/core/integration/tests/server/scenarios/message_size_scenario.rs
@@ -18,7 +18,7 @@
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, assert_clean_system, login_root};
+use integration::iggy_harness;
use std::collections::HashMap;
use std::str::FromStr;
@@ -34,14 +34,16 @@ enum MessageToSend {
OfSizeWithHeaders(usize, usize),
}
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
+/// Message size scenario is specific to TCP transport to test the behavior
+/// around the maximum message size. When run on other transports, it will fail
+/// because both QUIC and HTTP have different message size limits.
+#[iggy_harness(transport = Tcp)]
+async fn message_size_scenario(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
- login_root(&client).await;
init_system(&client).await;
- // 3. Send message and check the result
+ // Send message and check the result
send_message_and_check_result(
&client,
MessageToSend::NoMessage,
@@ -88,7 +90,6 @@ pub async fn run(client_factory: &dyn ClientFactory) {
}
async fn assert_message_count(client: &IggyClient, expected_count: u32) {
- // 4. Poll messages and validate the count
let polled_messages = client
.poll_messages(
&STREAM_NAME.try_into().unwrap(),
@@ -106,10 +107,8 @@ async fn assert_message_count(client: &IggyClient, expected_count: u32) {
}
async fn init_system(client: &IggyClient) {
- // 1. Create the stream
client.create_stream(STREAM_NAME).await.unwrap();
- // 2. Create the topic
client
.create_topic(
&STREAM_NAME.try_into().unwrap(),
@@ -131,6 +130,14 @@ async fn cleanup_system(client: &IggyClient) {
.unwrap();
}
+async fn assert_clean_system(client: &IggyClient) {
+ let stats = client.get_stats().await.unwrap();
+ assert_eq!(stats.streams_count, 0);
+ assert_eq!(stats.topics_count, 0);
+ assert_eq!(stats.partitions_count, 0);
+ assert_eq!(stats.segments_count, 0);
+}
+
async fn send_message_and_check_result(
client: &IggyClient,
message_params: MessageToSend,
diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs
index 3253493c4..146dbb4b7 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -26,24 +26,23 @@ pub mod consumer_group_with_multiple_clients_polling_messages_scenario;
pub mod consumer_group_with_single_client_polling_messages_scenario;
pub mod consumer_timestamp_polling_scenario;
pub mod create_message_payload;
-pub mod cross_protocol_pat_scenario;
-pub mod delete_segments_scenario;
+mod cross_protocol_pat_scenario;
+mod delete_segments_scenario;
pub mod encryption_scenario;
pub mod message_headers_scenario;
-pub mod message_size_scenario;
-pub mod offset_scenario;
+mod message_size_scenario;
+mod offset_retrieval_scenario;
pub mod permissions_scenario;
pub mod read_during_persistence_scenario;
-pub mod segment_rotation_race_scenario;
-pub mod single_message_per_batch_scenario;
+mod segment_rotation_race_scenario;
+mod single_message_per_batch_scenario;
pub mod snapshot_scenario;
pub mod stale_client_consumer_group_scenario;
pub mod stream_size_validation_scenario;
pub mod system_scenario;
-pub mod tcp_tls_scenario;
-pub mod timestamp_scenario;
+mod timestamp_retrieval_scenario;
+mod tls_scenario;
pub mod user_scenario;
-pub mod websocket_tls_scenario;
use iggy::prelude::*;
use integration::test_server::{ClientFactory, delete_user};
diff --git a/core/integration/tests/server/scenarios/offset_scenario.rs b/core/integration/tests/server/scenarios/offset_retrieval_scenario.rs
similarity index 97%
rename from core/integration/tests/server/scenarios/offset_scenario.rs
rename to core/integration/tests/server/scenarios/offset_retrieval_scenario.rs
index cc519bd63..0cfe0b321 100644
--- a/core/integration/tests/server/scenarios/offset_scenario.rs
+++ b/core/integration/tests/server/scenarios/offset_retrieval_scenario.rs
@@ -16,13 +16,29 @@
* under the License.
*/
-use super::PARTITION_ID;
use bytes::BytesMut;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::iggy_harness;
use std::collections::HashMap;
use std::str::FromStr;
+const PARTITION_ID: u32 = 0;
+
+#[iggy_harness(server(
+ segment_size = ["512B", "1KiB", "10MiB"],
+ cache_indexes = ["none", "all", "open_segment"],
+ messages_required_to_save = [32, 64, 1024, 9984],
+))]
+async fn get_by_offset(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
+
+ for msg_size in all_message_sizes() {
+ for (pattern_name, batch_pattern) in all_batch_patterns() {
+ run_offset_test(&client, msg_size, &batch_pattern, pattern_name).await;
+ }
+ }
+}
+
fn small_batches() -> Vec<u32> {
vec![3, 4, 5, 6, 7]
}
@@ -52,18 +68,6 @@ fn all_message_sizes() -> Vec<u64> {
vec![50, 1000, 20000]
}
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- login_root(&client).await;
-
- for msg_size in all_message_sizes() {
- for (pattern_name, batch_pattern) in all_batch_patterns() {
- run_offset_test(&client, msg_size, &batch_pattern, pattern_name).await;
- }
- }
-}
-
async fn run_offset_test(
client: &IggyClient,
message_size: u64,
diff --git a/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs b/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
index 9856777b2..69f4aacac 100644
--- a/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
+++ b/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
@@ -16,20 +16,20 @@
* under the License.
*/
-//! This scenario fixes the bug that occured when concurrent message sends race with segment rotation:
+//! Regression test for race condition between concurrent message sends and segment rotation:
//! 1. Task A commits journal, ensures indexes for segment N, starts async save
//! 2. Task B's send triggers segment rotation (handle_full_segment)
//! 3. Task B clears segment N's indexes or creates segment N+1 with None indexes
//! 4. Task A calls active_indexes().unwrap() - panics because indexes are None
//!
-//! This test uses:
+//! Test configuration:
//! - Very small segment size (512B) to trigger frequent rotations
//! - 8 concurrent producers (2 per protocol: TCP, HTTP, QUIC, WebSocket)
//! - All producers write to the same partition for maximum lock contention
//! - Short message_saver interval to add more concurrent persist operations
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::iggy_harness;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
@@ -42,50 +42,45 @@ const PARTITION_ID: u32 = 0;
const TEST_DURATION_SECS: u64 = 10;
const MESSAGES_PER_BATCH: usize = 5;
-/// Runs the segment rotation race condition test with multiple protocols.
-/// Each client factory represents a different protocol (TCP, HTTP, QUIC, WebSocket).
-/// 2 producers are spawned per protocol, all writing to the same partition.
-pub async fn run(client_factories: &[&dyn ClientFactory]) {
- assert!(
- !client_factories.is_empty(),
- "At least one client factory required"
- );
-
- let admin_client = create_client(client_factories[0]).await;
- login_root(&admin_client).await;
-
- let total_producers = client_factories.len() * PRODUCERS_PER_PROTOCOL;
+/// Tests segment rotation under high concurrency with all 4 protocols.
+/// 8 producers (2 per protocol) write to the same partition simultaneously.
+#[iggy_harness(server(
+ segment_size = "512B",
+ message_saver_interval = "1s",
+ messages_required_to_save = 32,
+ cache_indexes = "none",
+ tcp_nodelay = true,
+ socket_migration = false,
+))]
+async fn segment_rotation_race(harness: &TestHarness) {
+ let admin_client = harness.tcp_root_client().await.unwrap();
+
+ let total_producers = 4 * PRODUCERS_PER_PROTOCOL;
init_system(&admin_client, total_producers).await;
let stop_flag = Arc::new(AtomicBool::new(false));
let total_messages = Arc::new(AtomicU64::new(0));
let mut join_set = JoinSet::new();
- let mut global_producer_id = 0usize;
- for factory in client_factories {
- let protocol = factory.transport();
+ let protocols = ["Tcp", "Http", "Quic", "WebSocket"];
+ for (protocol_idx, protocol_name) in protocols.iter().enumerate() {
for local_id in 0..PRODUCERS_PER_PROTOCOL {
- let client = create_client(*factory).await;
- login_root(&client).await;
+ let client = match *protocol_name {
+ "Tcp" => harness.tcp_root_client().await.unwrap(),
+ "Http" => harness.http_root_client().await.unwrap(),
+ "Quic" => harness.quic_root_client().await.unwrap(),
+ "WebSocket" => harness.websocket_root_client().await.unwrap(),
+ _ => unreachable!(),
+ };
let stop = stop_flag.clone();
let counter = total_messages.clone();
- let producer_name = format!("{:?}-{}", protocol, local_id);
- let producer_id = global_producer_id;
+ let producer_name = format!("{}-{}", protocol_name, local_id);
+ let producer_id = protocol_idx * PRODUCERS_PER_PROTOCOL + local_id;
join_set.spawn(async move {
- run_producer(
- client,
- producer_id,
- &producer_name,
- PARTITION_ID,
- stop,
- counter,
- )
- .await;
+ run_producer(client, producer_id, &producer_name, stop, counter).await;
});
-
- global_producer_id += 1;
}
}
@@ -111,12 +106,10 @@ pub async fn run(client_factories: &[&dyn ClientFactory]) {
let sent = total_messages.load(Ordering::SeqCst);
println!("Test completed successfully. Total messages sent: {}", sent);
- cleanup(&admin_client).await;
-}
-
-async fn create_client(client_factory: &dyn ClientFactory) -> IggyClient {
- let client = client_factory.create_client().await;
- IggyClient::create(client, None, None)
+ admin_client
+ .delete_stream(&Identifier::named(STREAM_NAME).unwrap())
+ .await
+ .unwrap();
}
async fn init_system(client: &IggyClient, total_producers: usize) {
@@ -145,7 +138,6 @@ async fn run_producer(
client: IggyClient,
producer_id: usize,
producer_name: &str,
- partition_id: u32,
stop: Arc<AtomicBool>,
counter: Arc<AtomicU64>,
) {
@@ -167,7 +159,7 @@ async fn run_producer(
.send_messages(
&Identifier::named(STREAM_NAME).unwrap(),
&Identifier::named(TOPIC_NAME).unwrap(),
- &Partitioning::partition_id(partition_id),
+ &Partitioning::partition_id(PARTITION_ID),
&mut messages,
)
.await
@@ -184,13 +176,6 @@ async fn run_producer(
println!(
"Producer {} (partition {}) stopped after {} batches",
- producer_name, partition_id, batch_num
+ producer_name, PARTITION_ID, batch_num
);
}
-
-async fn cleanup(client: &IggyClient) {
- client
- .delete_stream(&Identifier::named(STREAM_NAME).unwrap())
- .await
- .unwrap();
-}
diff --git a/core/integration/tests/server/scenarios/single_message_per_batch_scenario.rs b/core/integration/tests/server/scenarios/single_message_per_batch_scenario.rs
index 6c34059eb..4b7c402b6 100644
--- a/core/integration/tests/server/scenarios/single_message_per_batch_scenario.rs
+++ b/core/integration/tests/server/scenarios/single_message_per_batch_scenario.rs
@@ -18,12 +18,13 @@
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::iggy_harness;
use std::time::{Duration, Instant};
const STREAM_NAME: &str = "single-msg-batch-stream";
const TOPIC_NAME: &str = "single-msg-batch-topic";
const POLL_BATCH_SIZE: u32 = 1000;
+const TEST_DURATION_SECS: u64 = 5;
/// Test that simulates the bench scenario with single message per batch.
///
@@ -40,13 +41,12 @@ const POLL_BATCH_SIZE: u32 = 1000;
/// - Journal management under high message count with small payloads
/// - Memory allocation patterns for many small messages
/// - IOV_MAX limits when flushing many small buffers
-pub async fn run(client_factory: &dyn ClientFactory, duration_secs: u64) {
- let client = client_factory.create_client().await;
- let producer = IggyClient::create(client, None, None);
- login_root(&producer).await;
+#[iggy_harness(transport = Tcp, server(messages_required_to_save = 10000))]
+async fn single_message_per_batch_scenario(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
- producer.create_stream(STREAM_NAME).await.unwrap();
- producer
+ client.create_stream(STREAM_NAME).await.unwrap();
+ client
.create_topic(
&Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
@@ -63,13 +63,13 @@ pub async fn run(client_factory: &dyn ClientFactory, duration_secs: u64) {
let topic_id = Identifier::named(TOPIC_NAME).unwrap();
let partitioning = Partitioning::partition_id(0);
- let test_duration = Duration::from_secs(duration_secs);
+ let test_duration = Duration::from_secs(TEST_DURATION_SECS);
let start = Instant::now();
let mut messages_sent = 0u64;
println!(
"Starting single message per batch test: sequential payloads, duration: {}s",
- duration_secs
+ TEST_DURATION_SECS
);
while start.elapsed() < test_duration {
@@ -81,7 +81,7 @@ pub async fn run(client_factory: &dyn ClientFactory, duration_secs: u64) {
.expect("Failed to create message"),
];
- producer
+ client
.send_messages(&stream_id, &topic_id, &partitioning, &mut messages)
.await
.unwrap();
@@ -114,7 +114,7 @@ pub async fn run(client_factory: &dyn ClientFactory, duration_secs: u64) {
let remaining = messages_sent - messages_received;
let batch_size = remaining.min(POLL_BATCH_SIZE as u64) as u32;
- let polled = producer
+ let polled = client
.poll_messages(
&stream_id,
&topic_id,
@@ -174,5 +174,7 @@ pub async fn run(client_factory: &dyn ClientFactory, duration_secs: u64) {
total_elapsed.as_secs_f64()
);
- producer.delete_stream(&stream_id).await.unwrap();
+ client.delete_stream(&stream_id).await.unwrap();
+
+ drop(client);
}
diff --git a/core/integration/tests/server/scenarios/timestamp_scenario.rs b/core/integration/tests/server/scenarios/timestamp_retrieval_scenario.rs
similarity index 97%
rename from core/integration/tests/server/scenarios/timestamp_scenario.rs
rename to core/integration/tests/server/scenarios/timestamp_retrieval_scenario.rs
index 542ca17cc..94ec54cf5 100644
--- a/core/integration/tests/server/scenarios/timestamp_scenario.rs
+++ b/core/integration/tests/server/scenarios/timestamp_retrieval_scenario.rs
@@ -16,14 +16,30 @@
* under the License.
*/
-use super::PARTITION_ID;
use bytes::BytesMut;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::iggy_harness;
use std::collections::HashMap;
use std::str::FromStr;
use tokio::time::{Duration, sleep};
+const PARTITION_ID: u32 = 0;
+
+#[iggy_harness(server(
+ segment_size = ["512B", "1KiB", "10MiB"],
+ cache_indexes = ["none", "all", "open_segment"],
+ messages_required_to_save = [32, 64, 1024, 9984],
+))]
+async fn get_by_timestamp(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
+
+ for msg_size in all_message_sizes() {
+ for (pattern_name, batch_pattern) in all_batch_patterns() {
+ run_timestamp_test(&client, msg_size, &batch_pattern, pattern_name).await;
+ }
+ }
+}
+
fn small_batches() -> Vec<u32> {
vec![3, 4, 5, 6, 7]
}
@@ -53,18 +69,6 @@ fn all_message_sizes() -> Vec<u64> {
vec![50, 1000, 20000]
}
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- login_root(&client).await;
-
- for msg_size in all_message_sizes() {
- for (pattern_name, batch_pattern) in all_batch_patterns() {
- run_timestamp_test(&client, msg_size, &batch_pattern, pattern_name).await;
- }
- }
-}
-
async fn run_timestamp_test(
client: &IggyClient,
message_size: u64,
diff --git a/core/integration/tests/server/scenarios/tcp_tls_scenario.rs b/core/integration/tests/server/scenarios/tls_scenario.rs
similarity index 54%
rename from core/integration/tests/server/scenarios/tcp_tls_scenario.rs
rename to core/integration/tests/server/scenarios/tls_scenario.rs
index b692b3b16..a153d75bf 100644
--- a/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
+++ b/core/integration/tests/server/scenarios/tls_scenario.rs
@@ -18,26 +18,43 @@
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{assert_clean_system, login_root};
+use integration::iggy_harness;
-pub async fn run(client: &IggyClient) {
- login_root(client).await;
+const STREAM_NAME: &str = "test-tls-stream";
+const TOPIC_NAME: &str = "test-tls-topic";
- let stream_name = "test-tls-stream";
- client.create_stream(stream_name).await.unwrap();
+/// Tests TLS with both self-signed and harness-generated certificates for TCP.
+/// - SelfSigned: Server generates its own certs, client disables validation
+/// - Generated: Harness generates certs, client validates using CA cert
+#[iggy_harness(transport = [TcpTlsSelfSigned, TcpTlsGenerated])]
+async fn tcp_tls_scenario(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
+ run_tls_test(&client).await;
+}
+
+/// Tests TLS with both self-signed and harness-generated certificates for WebSocket.
+/// - SelfSigned: Server generates its own certs, client disables validation
+/// - Generated: Harness generates certs, client validates using CA cert
+#[iggy_harness(transport = [WebSocketTlsSelfSigned, WebSocketTlsGenerated])]
+async fn websocket_tls_scenario(harness: &TestHarness) {
+ let client = harness.websocket_root_client().await.unwrap();
+ run_tls_test(&client).await;
+}
+
+async fn run_tls_test(client: &IggyClient) {
+ client.create_stream(STREAM_NAME).await.unwrap();
let stream = client
- .get_stream(&Identifier::named(stream_name).unwrap())
+ .get_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap()
.unwrap();
- assert_eq!(stream.name, stream_name);
+ assert_eq!(stream.name, STREAM_NAME);
- let topic_name = "test-tls-topic";
client
.create_topic(
- &Identifier::named(stream_name).unwrap(),
- topic_name,
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
1,
CompressionAlgorithm::default(),
None,
@@ -57,8 +74,8 @@ pub async fn run(client: &IggyClient) {
client
.send_messages(
- &Identifier::named(stream_name).unwrap(),
- &Identifier::named(topic_name).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
&Partitioning::partition_id(0),
&mut messages,
)
@@ -67,8 +84,8 @@ pub async fn run(client: &IggyClient) {
let polled_messages = client
.poll_messages(
- &Identifier::named(stream_name).unwrap(),
- &Identifier::named(topic_name).unwrap(),
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
Some(0),
&Consumer::default(),
&PollingStrategy::offset(0),
@@ -82,9 +99,7 @@ pub async fn run(client: &IggyClient) {
assert_eq!(polled_messages.messages[0].payload.as_ref(), b"Hello TLS!");
client
- .delete_stream(&Identifier::named(stream_name).unwrap())
+ .delete_stream(&Identifier::named(STREAM_NAME).unwrap())
.await
.unwrap();
-
- assert_clean_system(client).await;
}
diff --git a/core/integration/tests/server/scenarios/websocket_tls_scenario.rs b/core/integration/tests/server/scenarios/websocket_tls_scenario.rs
deleted file mode 100644
index f8f2f2d09..000000000
--- a/core/integration/tests/server/scenarios/websocket_tls_scenario.rs
+++ /dev/null
@@ -1,93 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use bytes::Bytes;
-use iggy::prelude::*;
-use integration::test_server::{assert_clean_system, login_root};
-
-pub async fn run(client: &IggyClient) {
- login_root(client).await;
-
- let stream_name = "test-websocket-tls-stream";
- client.create_stream(stream_name).await.unwrap();
-
- let stream = client
- .get_stream(&Identifier::named(stream_name).unwrap())
- .await
- .unwrap()
- .unwrap();
- assert_eq!(stream.name, stream_name);
-
- let topic_name = "test-websocket-tls-topic";
- client
- .create_topic(
- &Identifier::named(stream_name).unwrap(),
- topic_name,
- 1,
- CompressionAlgorithm::default(),
- None,
- IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
- )
- .await
- .unwrap();
-
- let mut messages = vec![
- IggyMessage::builder()
- .id(1)
- .payload(Bytes::from("Hello WebSocket TLS!"))
- .build()
- .unwrap(),
- ];
-
- client
- .send_messages(
- &Identifier::named(stream_name).unwrap(),
- &Identifier::named(topic_name).unwrap(),
- &Partitioning::partition_id(0),
- &mut messages,
- )
- .await
- .unwrap();
-
- let polled_messages = client
- .poll_messages(
- &Identifier::named(stream_name).unwrap(),
- &Identifier::named(topic_name).unwrap(),
- Some(0),
- &Consumer::default(),
- &PollingStrategy::offset(0),
- 1,
- true,
- )
- .await
- .unwrap();
-
- assert_eq!(polled_messages.messages.len(), 1);
- assert_eq!(
- polled_messages.messages[0].payload.as_ref(),
- b"Hello WebSocket TLS!"
- );
-
- client
- .delete_stream(&Identifier::named(stream_name).unwrap())
- .await
- .unwrap();
-
- assert_clean_system(client).await;
-}
diff --git a/core/integration/tests/server/specific.rs b/core/integration/tests/server/specific.rs
deleted file mode 100644
index 17cc6b05e..000000000
--- a/core/integration/tests/server/specific.rs
+++ /dev/null
@@ -1,313 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::server::scenarios::{
- cross_protocol_pat_scenario, delete_segments_scenario, message_size_scenario,
- segment_rotation_race_scenario, single_message_per_batch_scenario, tcp_tls_scenario,
- websocket_tls_scenario,
-};
-use iggy::prelude::*;
-use integration::{
- http_client::HttpClientFactory,
- quic_client::QuicClientFactory,
- tcp_client::TcpClientFactory,
- test_server::{IpAddrKind, TestServer},
- test_tls_utils::generate_test_certificates,
- websocket_client::WebSocketClientFactory,
-};
-use serial_test::parallel;
-use std::collections::HashMap;
-
-// This test can run on any transport, but it requires both ClientFactory and
-// TestServer parameters, which doesn't fit the unified matrix approach.
-#[tokio::test]
-#[parallel]
-async fn should_delete_segments_and_validate_filesystem() {
- let mut extra_envs = HashMap::new();
- extra_envs.insert("IGGY_SYSTEM_SEGMENT_SIZE".to_string(), "1MiB".to_string());
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let client_factory = TcpClientFactory {
- server_addr,
- ..Default::default()
- };
-
- delete_segments_scenario::run(&client_factory, &test_server).await;
-}
-
-// TCP TLS scenario is obviously specific to TCP transport, and requires special
-// setup so it's not included in the matrix.
-#[tokio::test]
-#[parallel]
-async fn tcp_tls_scenario_should_be_valid() {
- let temp_dir = tempfile::TempDir::new().expect("Failed to create temp dir");
- let cert_dir = temp_dir.path();
- let cert_dir_str = cert_dir.to_str().unwrap();
-
- generate_test_certificates(cert_dir_str).expect("Failed to generate test certificates");
-
- let mut extra_envs = HashMap::new();
- extra_envs.insert("IGGY_TCP_TLS_ENABLED".to_string(), "true".to_string());
- extra_envs.insert(
- "IGGY_TCP_TLS_CERT_FILE".to_string(),
- cert_dir.join("test_cert.pem").to_str().unwrap().to_string(),
- );
- extra_envs.insert(
- "IGGY_TCP_TLS_KEY_FILE".to_string(),
- cert_dir.join("test_key.pem").to_str().unwrap().to_string(),
- );
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let cert_path = cert_dir.join("test_cert.pem").to_str().unwrap().to_string();
-
- let client = IggyClientBuilder::new()
- .with_tcp()
- .with_server_address(server_addr)
- .with_tls_enabled(true)
- .with_tls_domain("localhost".to_string())
- .with_tls_ca_file(cert_path)
- .build()
- .expect("Failed to create TLS client");
-
- client
- .connect()
- .await
- .expect("Failed to connect TLS client");
-
- let client = IggyClient::create(ClientWrapper::Iggy(client), None, None);
-
- tcp_tls_scenario::run(&client).await;
-}
-
-#[tokio::test]
-#[parallel]
-async fn tcp_tls_self_signed_scenario_should_be_valid() {
- use iggy::clients::client_builder::IggyClientBuilder;
-
- let mut extra_envs = HashMap::new();
- extra_envs.insert("IGGY_TCP_TLS_ENABLED".to_string(), "true".to_string());
- extra_envs.insert("IGGY_TCP_TLS_SELF_SIGNED".to_string(), "true".to_string());
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
-
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
-
- let client = IggyClientBuilder::new()
- .with_tcp()
- .with_server_address(server_addr)
- .with_tls_enabled(true)
- .with_tls_domain("localhost".to_string())
- .with_tls_validate_certificate(false)
- .build()
- .expect("Failed to create TLS client");
-
- client
- .connect()
- .await
- .expect("Failed to connect TLS client with self-signed cert");
-
- let client = iggy::clients::client::IggyClient::create(ClientWrapper::Iggy(client), None, None);
-
- tcp_tls_scenario::run(&client).await;
-}
-
-#[tokio::test]
-#[parallel]
-async fn websocket_tls_scenario_should_be_valid() {
- let temp_dir = tempfile::TempDir::new().expect("Failed to create temp dir");
- let cert_dir = temp_dir.path();
- let cert_dir_str = cert_dir.to_str().unwrap();
-
- generate_test_certificates(cert_dir_str).expect("Failed to generate test certificates");
-
- let mut extra_envs = HashMap::new();
- extra_envs.insert("IGGY_WEBSOCKET_TLS_ENABLED".to_string(), "true".to_string());
- extra_envs.insert(
- "IGGY_WEBSOCKET_TLS_CERT_FILE".to_string(),
- cert_dir.join("test_cert.pem").to_str().unwrap().to_string(),
- );
- extra_envs.insert(
- "IGGY_WEBSOCKET_TLS_KEY_FILE".to_string(),
- cert_dir.join("test_key.pem").to_str().unwrap().to_string(),
- );
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let server_addr = test_server.get_websocket_addr().unwrap();
- let cert_path = cert_dir.join("test_cert.pem").to_str().unwrap().to_string();
-
- let client = IggyClientBuilder::new()
- .with_websocket()
- .with_server_address(server_addr)
- .with_tls_enabled(true)
- .with_tls_domain("localhost".to_string())
- .with_tls_ca_file(cert_path)
- .build()
- .expect("Failed to create WebSocket TLS client");
-
- client
- .connect()
- .await
- .expect("Failed to connect WebSocket TLS client");
-
- let client = IggyClient::create(ClientWrapper::Iggy(client), None, None);
-
- websocket_tls_scenario::run(&client).await;
-}
-
-// Message size scenario is specific to TCP transport to test the behavior around the maximum message size.
-// When run on other transports, it will fail because both QUIC and HTTP have different message size limits.
-#[tokio::test]
-#[parallel]
-async fn message_size_scenario() {
- let mut test_server = TestServer::default();
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let client_factory = TcpClientFactory {
- server_addr,
- ..Default::default()
- };
-
- message_size_scenario::run(&client_factory).await;
-}
-
-#[tokio::test]
-#[parallel]
-async fn should_handle_single_message_per_batch_with_delayed_persistence() {
- let mut extra_envs = HashMap::new();
- extra_envs.insert(
- "IGGY_SYSTEM_PARTITION_MESSAGES_REQUIRED_TO_SAVE".to_string(),
- "10000".to_string(),
- );
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let client_factory = TcpClientFactory {
- server_addr,
- ..Default::default()
- };
-
- single_message_per_batch_scenario::run(&client_factory, 5).await;
-}
-
-/// This test configures the server to trigger frequent segment rotations and runs
-/// multiple concurrent producers across all protocols (TCP, HTTP, QUIC, WebSocket)
-/// to maximize the chance of hitting the race condition between persist_messages_to_disk
-/// and handle_full_segment.
-///
-/// Server configuration:
-/// - Very small segment size (512B) to trigger frequent rotations
-/// - Short message_saver interval (1s) to add concurrent persist operations
-/// - Small messages_required_to_save (32) to trigger more frequent saves
-/// - cache_indexes = none to trigger clear_active_indexes path
-///
-/// Test configuration:
-/// - 8 producers total (2 per protocol: TCP, HTTP, QUIC, WebSocket)
-/// - All producers write to the same partition for maximum lock contention
-#[tokio::test]
-#[parallel]
-async fn segment_rotation_scenario() {
- let mut extra_envs = HashMap::new();
-
- // Very small segment to trigger frequent rotations
- extra_envs.insert("IGGY_SYSTEM_SEGMENT_SIZE".to_string(), "512B".to_string());
-
- // Short message saver interval to add concurrent persist operations
- extra_envs.insert("IGGY_MESSAGE_SAVER_INTERVAL".to_string(), "1s".to_string());
-
- // Small threshold to trigger more frequent saves
- extra_envs.insert(
- "IGGY_SYSTEM_PARTITION_MESSAGES_REQUIRED_TO_SAVE".to_string(),
- "32".to_string(),
- );
-
- // cache_indexes = none triggers clear_active_indexes in handle_full_segment
- extra_envs.insert(
- "IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string(),
- "none".to_string(),
- );
-
- // Disable socket migration to keep all connections on same shard
- extra_envs.insert("IGGY_TCP_SOCKET_MIGRATION".to_string(), "false".to_string());
-
- // Enable TCP nodelay for faster message throughput
- extra_envs.insert(
- "IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS".to_string(),
- "true".to_string(),
- );
- extra_envs.insert("IGGY_TCP_SOCKET_NODELAY".to_string(), "true".to_string());
-
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let tcp_factory = TcpClientFactory {
- server_addr: test_server.get_raw_tcp_addr().unwrap(),
- ..Default::default()
- };
-
- let http_factory = HttpClientFactory {
- server_addr: test_server.get_http_api_addr().unwrap(),
- };
-
- let quic_factory = QuicClientFactory {
- server_addr: test_server.get_quic_udp_addr().unwrap(),
- };
-
- let websocket_factory = WebSocketClientFactory {
- server_addr: test_server.get_websocket_addr().unwrap(),
- };
-
- let factories: Vec<&dyn integration::test_server::ClientFactory> = vec![
- &tcp_factory,
- &http_factory,
- &quic_factory,
- &websocket_factory,
- ];
-
- segment_rotation_race_scenario::run(&factories).await;
-}
-
-#[tokio::test]
-#[parallel]
-async fn should_see_pat_created_via_http_when_listing_via_tcp() {
- cross_protocol_pat_scenario::run().await;
-}
-
-#[tokio::test]
-#[parallel]
-async fn should_see_pat_created_via_tcp_when_listing_via_http() {
- cross_protocol_pat_scenario::run_tcp_to_http().await;
-}
-
-#[tokio::test]
-#[parallel]
-async fn should_not_see_pat_deleted_via_http_when_listing_via_tcp() {
- cross_protocol_pat_scenario::run_delete_visibility().await;
-}
diff --git a/core/server/src/websocket/websocket_tls_listener.rs b/core/server/src/websocket/websocket_tls_listener.rs
index 311d56b0e..c69a41c50 100644
--- a/core/server/src/websocket/websocket_tls_listener.rs
+++ b/core/server/src/websocket/websocket_tls_listener.rs
@@ -115,7 +115,7 @@ pub async fn start(
}
// Load or generate TLS certificates
- let tls_config = &shard.config.tcp.tls;
+ let tls_config = &shard.config.websocket.tls;
let (certs, key) =
if tls_config.self_signed && !std::path::Path::new(&tls_config.cert_file).exists() {
info!("Generating self-signed certificate for WebSocket TLS server");
diff --git a/scripts/check-backwards-compat.sh b/scripts/check-backwards-compat.sh
index eb1d3bafc..6adc2ffb1 100755
--- a/scripts/check-backwards-compat.sh
+++ b/scripts/check-backwards-compat.sh
@@ -163,7 +163,7 @@ ok "worktree at $MASTER_DIR"
pushd "$MASTER_DIR" >/dev/null
info "Building iggy-server & benches (baseline: $MASTER_REF)"
-IGGY_CI_BUILD=true cargo build --bins
+cargo build --bins
ok "built baseline"
info "Starting iggy-server (baseline)"
@@ -220,7 +220,7 @@ git rev-parse --verify "$PR_REF^{commit}" >/dev/null 2>&1 || die "PR_REF '$PR_RE
git checkout -q "$PR_REF"
info "Building iggy-server & benches (PR: $PR_REF)"
-IGGY_CI_BUILD=true cargo build --bins
+cargo build --bins
ok "built PR"
info "Restoring baseline local_data/ into PR workspace"