You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/02/05 14:41:16 UTC
(iggy) branch harness-final updated (7100c5af3 -> 178d2787f)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch harness-final
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit 7100c5af3 Merge branch 'master' into harness-final
omit d2f728011 refactor(integration): remove legacy test infrastructure
add d98176a36 fix(server): memory leak in segment rotation (#2686)
new 178d2787f refactor(integration): remove legacy test infrastructure
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 (7100c5af3)
\
N -- N -- N refs/heads/harness-final (178d2787f)
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:
Cargo.lock | 1 +
core/integration/Cargo.toml | 1 +
.../scenarios/segment_rotation_race_scenario.rs | 24 ++++++++++++++++++++++
core/server/src/shard/system/segments.rs | 20 ++++++++++++++++--
4 files changed, 44 insertions(+), 2 deletions(-)
(iggy) 01/01: refactor(integration): remove legacy test infrastructure
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch harness-final
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 178d2787f2d3a24338e2d1feda1fac5ea1d72d82
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Wed Feb 4 13:54:53 2026 +0100
refactor(integration): remove legacy test infrastructure
TestServer, ClientFactory, and protocol-specific client factories
duplicated functionality now provided by TestHarness and the
iggy_harness proc macro.
Changes:
- Delete ~2000 lines of deprecated test utilities
- Migrate all tests to unified TestHarness API
- Move TLS cert generation into ServerHandle
- Add port binding verification on server startup
- Update BDD, CLI, and server scenario tests
---
Cargo.lock | 3 +-
bdd/rust/Cargo.toml | 1 -
bdd/rust/tests/helpers/cluster.rs | 19 +-
bdd/rust/tests/steps/auth.rs | 26 +-
bdd/rust/tests/steps/leader_redirection.rs | 6 +-
core/harness_derive/src/codegen.rs | 2 +-
core/integration/Cargo.toml | 2 +-
core/integration/src/bench_utils.rs | 20 +-
core/integration/src/file.rs | 32 -
core/integration/src/harness/handle/server.rs | 99 ++-
core/integration/src/harness/helpers.rs | 2 -
.../src/harness/orchestrator/builder.rs | 10 +-
.../src/harness/orchestrator/harness.rs | 199 +-----
core/integration/src/http_client.rs | 52 --
core/integration/src/lib.rs | 17 -
core/integration/src/quic_client.rs | 55 --
core/integration/src/tcp_client.rs | 82 ---
core/integration/src/test_connectors_runtime.rs | 266 -------
core/integration/src/test_mcp_server.rs | 274 --------
core/integration/src/test_server.rs | 769 ---------------------
core/integration/src/test_tls_utils.rs | 52 --
core/integration/src/websocket_client.rs | 76 --
core/integration/tests/cli/common/mod.rs | 72 +-
.../tests/cli/context/test_context_applied.rs | 8 +-
.../tests/cli/system/test_me_command.rs | 10 +-
core/integration/tests/config_provider/mod.rs | 7 +-
.../connectors/fixtures/quickwit/container.rs | 33 +
.../tests/connectors/postgres/postgres_sink.rs | 6 +-
.../tests/connectors/postgres/postgres_source.rs | 10 +-
.../tests/connectors/quickwit/quickwit_sink.rs | 36 +-
.../tests/connectors/random/random_source.rs | 2 +-
.../data_integrity/verify_after_server_restart.rs | 171 ++---
core/integration/tests/mod.rs | 72 +-
core/integration/tests/sdk/producer/background.rs | 195 +-----
core/integration/tests/server/cg.rs | 67 +-
.../tests/server/concurrent_addition.rs | 55 +-
core/integration/tests/server/general.rs | 157 ++++-
core/integration/tests/server/message_retrieval.rs | 112 ++-
core/integration/tests/server/mod.rs | 131 ----
.../server/scenarios/authentication_scenario.rs | 10 +-
.../tests/server/scenarios/bench_scenario.rs | 25 +-
.../tests/server/scenarios/concurrent_scenario.rs | 101 +--
...umer_group_auto_commit_reconnection_scenario.rs | 14 +-
.../scenarios/consumer_group_join_scenario.rs | 19 +-
.../consumer_group_offset_cleanup_scenario.rs | 12 +-
...h_multiple_clients_polling_messages_scenario.rs | 18 +-
...with_single_client_polling_messages_scenario.rs | 12 +-
.../consumer_timestamp_polling_scenario.rs | 12 +-
.../server/scenarios/create_message_payload.rs | 12 +-
.../scenarios/cross_protocol_pat_scenario.rs | 133 +---
.../tests/server/scenarios/encryption_scenario.rs | 91 +--
.../server/scenarios/log_rotation_scenario.rs | 71 +-
.../server/scenarios/message_headers_scenario.rs | 12 +-
.../server/scenarios/message_size_scenario.rs | 12 +-
core/integration/tests/server/scenarios/mod.rs | 10 +-
.../tests/server/scenarios/offset_scenario.rs | 11 +-
.../tests/server/scenarios/permissions_scenario.rs | 159 ++---
.../scenarios/read_during_persistence_scenario.rs | 65 +-
.../scenarios/segment_rotation_race_scenario.rs | 60 +-
.../scenarios/single_message_per_batch_scenario.rs | 11 +-
.../tests/server/scenarios/snapshot_scenario.rs | 13 +-
.../stale_client_consumer_group_scenario.rs | 50 +-
.../scenarios/stream_size_validation_scenario.rs | 13 +-
.../tests/server/scenarios/system_scenario.rs | 16 +-
.../tests/server/scenarios/tcp_tls_scenario.rs | 4 +-
.../tests/server/scenarios/timestamp_scenario.rs | 11 +-
.../tests/server/scenarios/user_scenario.rs | 10 +-
.../server/scenarios/websocket_tls_scenario.rs | 4 +-
core/integration/tests/server/specific.rs | 297 ++------
.../server/src/websocket/websocket_tls_listener.rs | 2 +-
70 files changed, 1096 insertions(+), 3402 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 70ad39614..19ff17599 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1041,7 +1041,6 @@ dependencies = [
"bytes",
"cucumber",
"iggy",
- "integration",
"tokio",
]
@@ -4985,7 +4984,6 @@ dependencies = [
"configs",
"configs_derive",
"ctor",
- "derive_more",
"figment",
"futures",
"harness_derive",
@@ -5016,6 +5014,7 @@ dependencies = [
"test-case",
"testcontainers-modules",
"tokio",
+ "toml 0.9.11+spec-1.1.0",
"tracing",
"tracing-subscriber",
"twox-hash",
diff --git a/bdd/rust/Cargo.toml b/bdd/rust/Cargo.toml
index e33087470..ec16e484f 100644
--- a/bdd/rust/Cargo.toml
+++ b/bdd/rust/Cargo.toml
@@ -29,7 +29,6 @@ bdd = []
bytes = { workspace = true }
cucumber = { workspace = true }
iggy = { workspace = true }
-integration = { workspace = true }
tokio = { workspace = true }
# `test` is set to false because cucumber uses its own test harness, which
diff --git a/bdd/rust/tests/helpers/cluster.rs b/bdd/rust/tests/helpers/cluster.rs
index b58c05be9..7d0b94672 100644
--- a/bdd/rust/tests/helpers/cluster.rs
+++ b/bdd/rust/tests/helpers/cluster.rs
@@ -16,11 +16,9 @@
* under the License.
*/
-use iggy::clients::client::IggyClient;
use iggy::prelude::*;
-use integration::tcp_client::TcpClientFactory;
-use integration::test_server::ClientFactory;
use std::env;
+use std::sync::Arc;
/// Resolves server address based on role and port, checking environment variables first
pub fn resolve_server_address(role: &str, port: u16) -> String {
@@ -39,16 +37,17 @@ pub fn resolve_server_address(role: &str, port: u16) -> String {
/// Creates and connects a client to the specified address
pub async fn create_and_connect_client(addr: &str) -> IggyClient {
- let client_factory = TcpClientFactory {
- server_addr: addr.to_string(),
- ..Default::default()
+ let config = TcpClientConfig {
+ server_address: addr.to_string(),
+ ..TcpClientConfig::default()
};
- let client = client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
+ let client = TcpClient::create(Arc::new(config)).expect("Failed to create TCP client");
+ Client::connect(&client)
+ .await
+ .expect("Client should connect");
- client.connect().await.expect("Client should connect");
- client
+ IggyClient::create(ClientWrapper::Tcp(client), None, None)
}
/// Verifies that a client is connected to the expected port
diff --git a/bdd/rust/tests/steps/auth.rs b/bdd/rust/tests/steps/auth.rs
index 5b2e0f47d..187a5308a 100644
--- a/bdd/rust/tests/steps/auth.rs
+++ b/bdd/rust/tests/steps/auth.rs
@@ -18,10 +18,8 @@
use crate::common::global_context::GlobalContext;
use cucumber::given;
-use iggy::clients::client::IggyClient;
-use iggy::prelude::SystemClient;
-use integration::tcp_client::TcpClientFactory;
-use integration::test_server::{ClientFactory, login_root};
+use iggy::prelude::*;
+use std::sync::Arc;
#[given("I am authenticated as the root user")]
pub async fn given_authenticated_as_root(world: &mut GlobalContext) {
@@ -30,16 +28,24 @@ pub async fn given_authenticated_as_root(world: &mut GlobalContext) {
.as_ref()
.expect("Server should be running")
.clone();
- let client_factory = TcpClientFactory {
- server_addr,
- ..Default::default()
+
+ let config = TcpClientConfig {
+ server_address: server_addr,
+ ..TcpClientConfig::default()
};
- let client = client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
+ let tcp_client = TcpClient::create(Arc::new(config)).expect("Failed to create TCP client");
+ Client::connect(&tcp_client)
+ .await
+ .expect("Client should connect");
+
+ let client = IggyClient::create(ClientWrapper::Tcp(tcp_client), None, None);
client.ping().await.expect("Server should respond to ping");
- login_root(&client).await;
+ client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .expect("Failed to login as root");
world.client = Some(client);
}
diff --git a/bdd/rust/tests/steps/leader_redirection.rs b/bdd/rust/tests/steps/leader_redirection.rs
index 20c36af7e..6090ef2a3 100644
--- a/bdd/rust/tests/steps/leader_redirection.rs
+++ b/bdd/rust/tests/steps/leader_redirection.rs
@@ -20,7 +20,6 @@ use crate::common::leader_context::LeaderContext;
use crate::helpers::cluster;
use cucumber::{given, then, when};
use iggy::prelude::*;
-use integration::test_server::login_root;
use std::time::Duration;
use tokio::time::sleep;
@@ -161,7 +160,10 @@ async fn when_authenticate_root(world: &mut LeaderContext) {
.get_client(&client_name)
.unwrap_or_else(|| panic!("Client {} should be created", client_name));
- login_root(client).await;
+ client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .expect("Failed to login as root");
// Small delay between multiple authentications to avoid race conditions
if world.clients.len() > 1 {
diff --git a/core/harness_derive/src/codegen.rs b/core/harness_derive/src/codegen.rs
index eb06e655b..52b383e3a 100644
--- a/core/harness_derive/src/codegen.rs
+++ b/core/harness_derive/src/codegen.rs
@@ -476,7 +476,7 @@ fn generate_harness_setup(
#config_resolution
let mut __harness = ::integration::harness::TestHarness::builder()
.server(#server_config)
- .client(#client_config)
+ .primary_client(#client_config)
#mcp_builder_call
#connectors_runtime_builder_call
#cluster_builder_call
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index e3c05ba82..f97fbdb95 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -36,7 +36,6 @@ compio = { workspace = true }
configs = { workspace = true }
configs_derive = { workspace = true }
ctor = { workspace = true }
-derive_more = { workspace = true }
figment = { workspace = true }
futures = { workspace = true }
harness_derive = { workspace = true }
@@ -72,6 +71,7 @@ tempfile = { workspace = true }
test-case = { workspace = true }
testcontainers-modules = { workspace = true }
tokio = { workspace = true }
+toml = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
twox-hash = { workspace = true }
diff --git a/core/integration/src/bench_utils.rs b/core/integration/src/bench_utils.rs
index fbcc834d2..d1d670c26 100644
--- a/core/integration/src/bench_utils.rs
+++ b/core/integration/src/bench_utils.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::harness::get_test_directory;
use assert_cmd::prelude::CommandCargoExt;
use iggy::prelude::*;
use iggy_common::TransportProtocol;
@@ -22,7 +23,7 @@ use std::{
fs::{self, File, OpenOptions},
io::Write,
process::{Command, Stdio},
- thread::panicking,
+ thread::{self, panicking},
};
use uuid::Uuid;
@@ -151,5 +152,20 @@ pub fn run_bench_and_wait_for_finish(
}
pub fn get_random_path() -> String {
- format!("{}{}", BENCH_FILES_PREFIX, Uuid::now_v7().to_u128_le())
+ let test_name = thread::current()
+ .name()
+ .map(|s| s.to_string())
+ .unwrap_or_default();
+
+ let dir = get_test_directory(&test_name)
+ .unwrap_or_else(|| std::env::current_dir().expect("Failed to get current directory"));
+
+ let _ = fs::create_dir_all(&dir);
+ dir.join(format!(
+ "{}{}",
+ BENCH_FILES_PREFIX,
+ Uuid::now_v7().to_u128_le()
+ ))
+ .display()
+ .to_string()
}
diff --git a/core/integration/src/file.rs b/core/integration/src/file.rs
deleted file mode 100644
index b47983d3b..000000000
--- a/core/integration/src/file.rs
+++ /dev/null
@@ -1,32 +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 std::ffi::OsStr;
-use std::path::{Path, PathBuf};
-
-pub fn get_root_path() -> PathBuf {
- let manifest_dir = std::env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR not set!");
- PathBuf::from(manifest_dir)
-}
-
-pub fn file_exists<S>(file_path: &S) -> bool
-where
- S: AsRef<OsStr> + ?Sized,
-{
- Path::new(file_path).is_file()
-}
diff --git a/core/integration/src/harness/handle/server.rs b/core/integration/src/harness/handle/server.rs
index e602b561c..c900eb5c2 100644
--- a/core/integration/src/harness/handle/server.rs
+++ b/core/integration/src/harness/handle/server.rs
@@ -33,12 +33,13 @@ use std::collections::HashMap;
use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::net::SocketAddr;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread::{self, JoinHandle, available_parallelism, sleep};
use std::time::{Duration, Instant};
+use toml::Value;
const SLEEP_INTERVAL_MS: u64 = 20;
const MAX_PORT_WAIT_DURATION_S: u64 = 60;
@@ -148,6 +149,12 @@ impl ServerHandle {
self.config.tls.as_ref().is_some_and(|tls| tls.self_signed)
}
+ /// Add an environment variable to the server config.
+ /// Must be called before starting the server.
+ pub fn add_env(&mut self, key: impl Into<String>, value: impl Into<String>) {
+ self.config.extra_envs.insert(key.into(), value.into());
+ }
+
pub fn collect_logs(&self) -> (String, String) {
super::common::collect_logs(&self.stdout_path, &self.stderr_path)
}
@@ -403,7 +410,7 @@ impl ServerHandle {
}
if config_path.exists() {
- return Ok(());
+ return self.verify_bound_ports(&config_path);
}
if Instant::now() >= deadline {
@@ -419,6 +426,64 @@ impl ServerHandle {
})
}
+ fn verify_bound_ports(&self, config_path: &Path) -> Result<(), TestBinaryError> {
+ let content =
+ fs::read_to_string(config_path).map_err(|e| TestBinaryError::InvalidState {
+ message: format!(
+ "Failed to read server config at {}: {e}",
+ config_path.display()
+ ),
+ })?;
+
+ let config: Value =
+ toml::from_str(&content).map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to parse server config: {e}"),
+ })?;
+
+ let bound_tcp = Self::extract_address(&config, "tcp");
+ let bound_http = Self::extract_address(&config, "http");
+ let bound_quic = Self::extract_address(&config, "quic");
+ let bound_websocket = Self::extract_address(&config, "websocket");
+
+ let mut mismatches = Vec::new();
+
+ if let (Some(expected), Some(bound)) = (self.addrs.tcp, bound_tcp)
+ && expected != bound
+ {
+ mismatches.push(format!("TCP: expected {expected}, got {bound}"));
+ }
+ if let (Some(expected), Some(bound)) = (self.addrs.http, bound_http)
+ && expected != bound
+ {
+ mismatches.push(format!("HTTP: expected {expected}, got {bound}"));
+ }
+ if let (Some(expected), Some(bound)) = (self.addrs.quic, bound_quic)
+ && expected != bound
+ {
+ mismatches.push(format!("QUIC: expected {expected}, got {bound}"));
+ }
+ if let (Some(expected), Some(bound)) = (self.addrs.websocket, bound_websocket)
+ && expected != bound
+ {
+ mismatches.push(format!("WebSocket: expected {expected}, got {bound}"));
+ }
+
+ if !mismatches.is_empty() {
+ return Err(TestBinaryError::InvalidState {
+ message: format!(
+ "Server bound to unexpected ports:\n {}",
+ mismatches.join("\n ")
+ ),
+ });
+ }
+
+ Ok(())
+ }
+
+ fn extract_address(config: &Value, protocol: &str) -> Option<SocketAddr> {
+ config.get(protocol)?.get("address")?.as_str()?.parse().ok()
+ }
+
fn start_watchdog(&mut self) {
let Some(child) = &self.child_handle else {
return;
@@ -688,11 +753,11 @@ impl TestBinary for ServerHandle {
path: cert_dir.clone(),
source: e,
})?;
- crate::test_tls_utils::generate_test_certificates(cert_dir.to_str().unwrap()).map_err(
- |e| TestBinaryError::InvalidState {
+ 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);
}
@@ -839,3 +904,25 @@ impl Drop for ServerHandle {
super::common::dump_logs_on_panic("Iggy server", &self.stdout_path, &self.stderr_path);
}
}
+
+fn generate_test_certificates(cert_dir: &str) -> Result<(), Box<dyn std::error::Error>> {
+ fs::create_dir_all(cert_dir)?;
+
+ let subject_alt_names = vec![
+ "localhost".to_string(),
+ "127.0.0.1".to_string(),
+ "::1".to_string(),
+ ];
+
+ let cert = rcgen::generate_simple_self_signed(subject_alt_names)?;
+
+ let cert_path = Path::new(cert_dir).join("test_cert.pem");
+ let mut cert_file = File::create(&cert_path)?;
+ cert_file.write_all(cert.cert.pem().as_bytes())?;
+
+ let key_path = Path::new(cert_dir).join("test_key.pem");
+ let mut key_file = File::create(&key_path)?;
+ key_file.write_all(cert.signing_key.serialize_pem().as_bytes())?;
+
+ Ok(())
+}
diff --git a/core/integration/src/harness/helpers.rs b/core/integration/src/harness/helpers.rs
index 0b712644b..4839bb6a7 100644
--- a/core/integration/src/harness/helpers.rs
+++ b/core/integration/src/harness/helpers.rs
@@ -17,8 +17,6 @@
* under the License.
*/
-//! Helper functions for common test operations.
-
use iggy::prelude::{
DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, GlobalPermissions, Identifier, IdentityInfo,
Permissions, StreamClient, UserClient, UserStatus,
diff --git a/core/integration/src/harness/orchestrator/builder.rs b/core/integration/src/harness/orchestrator/builder.rs
index 3b503f920..46afddd3b 100644
--- a/core/integration/src/harness/orchestrator/builder.rs
+++ b/core/integration/src/harness/orchestrator/builder.rs
@@ -210,12 +210,20 @@ impl TestHarnessBuilder {
}
}
+ // Default to TCP transport if server is configured but no client transport specified
+ let primary_transport =
+ if self.primary_transport.is_none() && !servers.is_empty() && self.clients.is_empty() {
+ Some(iggy_common::TransportProtocol::Tcp)
+ } else {
+ self.primary_transport
+ };
+
Ok(TestHarness {
context,
servers,
clients: Vec::new(),
client_configs: self.clients,
- primary_transport: self.primary_transport,
+ primary_transport,
primary_client_config: self.primary_client_config,
started: false,
})
diff --git a/core/integration/src/harness/orchestrator/harness.rs b/core/integration/src/harness/orchestrator/harness.rs
index b2e4630dd..6e79058b8 100644
--- a/core/integration/src/harness/orchestrator/harness.rs
+++ b/core/integration/src/harness/orchestrator/harness.rs
@@ -25,15 +25,8 @@ use crate::harness::handle::{
ClientHandle, ConnectorsRuntimeHandle, McpClient, McpHandle, ServerHandle, ServerLogs,
};
use crate::harness::traits::{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 futures::executor::block_on;
-use iggy::prelude::{
- ClientWrapper, DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, IggyClient, UserClient,
-};
+use iggy::prelude::{ClientWrapper, IggyClient};
use iggy_common::TransportProtocol;
use std::path::Path;
use std::sync::Arc;
@@ -44,14 +37,6 @@ pub struct TestLogs {
pub servers: Vec<ServerLogs>,
}
-#[derive(Default)]
-pub(super) struct TlsSettings {
- pub enabled: bool,
- pub domain: String,
- pub ca_file: Option<String>,
- pub validate_certificate: bool,
-}
-
/// Orchestrates test binaries and clients for integration tests.
pub struct TestHarness {
pub(super) context: Arc<TestContext>,
@@ -286,159 +271,19 @@ impl TestHarness {
}
}
- /// Get a TCP client factory for creating additional clients.
- pub fn tcp_client_factory(&self) -> Option<TcpClientFactory> {
- let server = self.servers.first()?;
- let addr = server.tcp_addr()?;
- let config = self.find_client_config(TransportProtocol::Tcp);
- let tls = self.extract_tls_settings(config, server);
-
- Some(TcpClientFactory {
- server_addr: addr.to_string(),
- nodelay: config.map(|c| c.tcp_nodelay).unwrap_or_default(),
- tls_enabled: tls.enabled,
- tls_domain: tls.domain,
- tls_ca_file: tls.ca_file,
- tls_validate_certificate: tls.validate_certificate,
- })
- }
-
- /// Get an HTTP client factory for creating additional clients.
- pub fn http_client_factory(&self) -> Option<HttpClientFactory> {
- self.servers
- .first()
- .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.servers
- .first()
- .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.servers.first()?;
- let addr = server.websocket_addr()?;
- let config = self.find_client_config(TransportProtocol::WebSocket);
- let tls = self.extract_tls_settings(config, server);
-
- Some(WebSocketClientFactory {
- server_addr: addr.to_string(),
- tls_enabled: tls.enabled,
- tls_domain: tls.domain,
- tls_ca_file: tls.ca_file,
- tls_validate_certificate: tls.validate_certificate,
- })
- }
-
- fn find_client_config(&self, transport: TransportProtocol) -> Option<&ClientConfig> {
- self.client_configs
- .iter()
- .find(|c| c.transport == transport)
- .or(self
- .primary_client_config
- .as_ref()
- .filter(|c| c.transport == transport))
- }
-
- fn extract_tls_settings(
- &self,
- config: Option<&ClientConfig>,
- server: &ServerHandle,
- ) -> TlsSettings {
- let (enabled, domain, validate) = config
- .map(|c| {
- (
- c.tls_enabled,
- c.tls_domain
- .clone()
- .unwrap_or_else(|| "localhost".to_string()),
- c.tls_validate_certificate,
- )
- })
- .unwrap_or_default();
-
- let ca_file = if enabled {
- server
- .tls_ca_cert_path()
- .map(|p| p.to_string_lossy().to_string())
- } else {
- None
- };
-
- TlsSettings {
- enabled,
- domain,
- ca_file,
- validate_certificate: validate,
- }
- }
-
- /// Get all available client factories.
- #[allow(clippy::vec_box)]
- pub fn all_client_factories(&self) -> Vec<Box<dyn ClientFactory>> {
- let mut factories: Vec<Box<dyn ClientFactory>> = Vec::new();
- if let Some(f) = self.tcp_client_factory() {
- factories.push(Box::new(f));
- }
- if let Some(f) = self.http_client_factory() {
- factories.push(Box::new(f));
- }
- if let Some(f) = self.quic_client_factory() {
- factories.push(Box::new(f));
- }
- if let Some(f) = self.websocket_client_factory() {
- factories.push(Box::new(f));
- }
- factories
- }
-
- fn client_factory_for(
- &self,
- transport: TransportProtocol,
- ) -> Result<Box<dyn ClientFactory>, TestBinaryError> {
- let factory: Box<dyn ClientFactory> = match transport {
- TransportProtocol::Tcp => Box::new(self.tcp_client_factory().ok_or_else(|| {
- TestBinaryError::InvalidState {
- message: "TCP transport not available".to_string(),
- }
- })?),
- TransportProtocol::Http => Box::new(self.http_client_factory().ok_or_else(|| {
- TestBinaryError::InvalidState {
- message: "HTTP transport not available".to_string(),
- }
- })?),
- TransportProtocol::Quic => Box::new(self.quic_client_factory().ok_or_else(|| {
- TestBinaryError::InvalidState {
- message: "QUIC transport not available".to_string(),
- }
- })?),
- TransportProtocol::WebSocket => {
- Box::new(self.websocket_client_factory().ok_or_else(|| {
- TestBinaryError::InvalidState {
- message: "WebSocket transport not available".to_string(),
- }
- })?)
- }
- };
- Ok(factory)
- }
-
/// Create a new client logged in as root for the specified transport.
pub async fn root_client_for(
&self,
transport: TransportProtocol,
) -> Result<IggyClient, TestBinaryError> {
- let factory = self.client_factory_for(transport)?;
- self.create_root_client(&*factory).await
+ let server = self.servers.first().ok_or(TestBinaryError::MissingServer)?;
+ let builder = match transport {
+ TransportProtocol::Tcp => server.tcp_client()?,
+ TransportProtocol::Http => server.http_client()?,
+ TransportProtocol::Quic => server.quic_client()?,
+ TransportProtocol::WebSocket => server.websocket_client()?,
+ };
+ builder.with_root_login().connect().await
}
/// Create multiple root clients for the specified transport.
@@ -459,9 +304,14 @@ impl TestHarness {
&self,
transport: TransportProtocol,
) -> Result<IggyClient, TestBinaryError> {
- let factory = self.client_factory_for(transport)?;
- let client = factory.create_client().await;
- Ok(IggyClient::create(client, None, None))
+ let server = self.servers.first().ok_or(TestBinaryError::MissingServer)?;
+ let builder = match transport {
+ TransportProtocol::Tcp => server.tcp_client()?,
+ TransportProtocol::Http => server.http_client()?,
+ TransportProtocol::Quic => server.quic_client()?,
+ TransportProtocol::WebSocket => server.websocket_client()?,
+ };
+ builder.connect().await
}
pub async fn tcp_root_client(&self) -> Result<IggyClient, TestBinaryError> {
@@ -560,21 +410,6 @@ impl TestHarness {
.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)
- }
-
pub(super) async fn create_clients(&mut self) -> Result<(), TestBinaryError> {
let Some(server) = self.servers.first() else {
return Ok(());
diff --git a/core/integration/src/http_client.rs b/core/integration/src/http_client.rs
deleted file mode 100644
index 6b5b0223e..000000000
--- a/core/integration/src/http_client.rs
+++ /dev/null
@@ -1,52 +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::test_server::ClientFactory;
-use async_trait::async_trait;
-use iggy::http::http_client::HttpClient;
-use iggy::prelude::{ClientWrapper, HttpClientConfig};
-use iggy_common::TransportProtocol;
-use std::sync::Arc;
-
-#[derive(Debug, Clone)]
-pub struct HttpClientFactory {
- pub server_addr: String,
-}
-
-#[async_trait]
-impl ClientFactory for HttpClientFactory {
- async fn create_client(&self) -> ClientWrapper {
- let config = HttpClientConfig {
- api_url: format!("http://{}", self.server_addr.clone()),
- ..HttpClientConfig::default()
- };
- let client = HttpClient::create(Arc::new(config)).unwrap();
- ClientWrapper::Http(client)
- }
-
- fn transport(&self) -> TransportProtocol {
- TransportProtocol::Http
- }
-
- fn server_addr(&self) -> String {
- self.server_addr.clone()
- }
-}
-
-unsafe impl Send for HttpClientFactory {}
-unsafe impl Sync for HttpClientFactory {}
diff --git a/core/integration/src/lib.rs b/core/integration/src/lib.rs
index 1a4b5c4bb..fc03d02d7 100644
--- a/core/integration/src/lib.rs
+++ b/core/integration/src/lib.rs
@@ -17,24 +17,7 @@
*/
pub mod bench_utils;
-pub mod file;
pub mod harness;
-#[allow(deprecated)]
-pub mod http_client;
-#[allow(deprecated)]
-pub mod quic_client;
-#[allow(deprecated)]
-pub mod tcp_client;
-#[allow(deprecated)]
-pub mod websocket_client;
-
-#[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 harness_derive::iggy_harness;
diff --git a/core/integration/src/quic_client.rs b/core/integration/src/quic_client.rs
deleted file mode 100644
index c9fa64990..000000000
--- a/core/integration/src/quic_client.rs
+++ /dev/null
@@ -1,55 +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::test_server::ClientFactory;
-use async_trait::async_trait;
-use iggy::prelude::{Client, ClientWrapper, QuicClientConfig};
-use iggy::quic::quic_client::QuicClient;
-use iggy_common::TransportProtocol;
-use std::sync::Arc;
-
-#[derive(Debug, Clone)]
-pub struct QuicClientFactory {
- pub server_addr: String,
-}
-
-#[async_trait]
-impl ClientFactory for QuicClientFactory {
- async fn create_client(&self) -> ClientWrapper {
- let config = QuicClientConfig {
- server_address: self.server_addr.clone(),
- // TODO: Need to increase this in order to not timeout during tests
- max_idle_timeout: 2_000_000,
- ..QuicClientConfig::default()
- };
- let client = QuicClient::create(Arc::new(config)).unwrap();
- Client::connect(&client).await.unwrap();
- ClientWrapper::Quic(client)
- }
-
- fn transport(&self) -> TransportProtocol {
- TransportProtocol::Quic
- }
-
- fn server_addr(&self) -> String {
- self.server_addr.clone()
- }
-}
-
-unsafe impl Send for QuicClientFactory {}
-unsafe impl Sync for QuicClientFactory {}
diff --git a/core/integration/src/tcp_client.rs b/core/integration/src/tcp_client.rs
deleted file mode 100644
index 4a45a5912..000000000
--- a/core/integration/src/tcp_client.rs
+++ /dev/null
@@ -1,82 +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::test_server::ClientFactory;
-use async_trait::async_trait;
-use iggy::prelude::{Client, ClientWrapper, TcpClient, TcpClientConfig};
-use iggy_common::TransportProtocol;
-use std::sync::Arc;
-
-#[derive(Debug, Clone, Default)]
-pub struct TcpClientFactory {
- pub server_addr: String,
- pub nodelay: bool,
- pub tls_enabled: bool,
- pub tls_domain: String,
- pub tls_ca_file: Option<String>,
- pub tls_validate_certificate: bool,
-}
-
-#[async_trait]
-impl ClientFactory for TcpClientFactory {
- async fn create_client(&self) -> ClientWrapper {
- let config = TcpClientConfig {
- server_address: self.server_addr.clone(),
- nodelay: self.nodelay,
- 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,
- ..TcpClientConfig::default()
- };
- let client = TcpClient::create(Arc::new(config)).unwrap_or_else(|e| {
- panic!(
- "Failed to create TcpClient, 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 TLS enabled and self-signed certificate:\n\
- IGGY_TCP_TLS_ENABLED=true IGGY_TCP_TLS_SELF_SIGNED=true\n
- or start iggy-bench with relevant tcp tls arguments: --tls --tls-domain <domain> --tls-ca-file <ca_file>\n",
- self.server_addr, e
- )
- } else {
- panic!(
- "Failed to connect to iggy-server at {}, error: {:?}",
- self.server_addr, e
- )
- }
- });
- ClientWrapper::Tcp(client)
- }
-
- fn transport(&self) -> TransportProtocol {
- TransportProtocol::Tcp
- }
-
- fn server_addr(&self) -> String {
- self.server_addr.clone()
- }
-}
-
-unsafe impl Send for TcpClientFactory {}
-unsafe impl Sync for TcpClientFactory {}
diff --git a/core/integration/src/test_connectors_runtime.rs b/core/integration/src/test_connectors_runtime.rs
deleted file mode 100644
index b7adb73ef..000000000
--- a/core/integration/src/test_connectors_runtime.rs
+++ /dev/null
@@ -1,266 +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 assert_cmd::prelude::CommandCargoExt;
-use rand::Rng;
-use std::fs::{self, File, OpenOptions};
-use std::io::Write;
-use std::net::{Ipv4Addr, SocketAddr};
-use std::path::PathBuf;
-use std::process::{Child, Command, Stdio};
-use std::time::Duration;
-use std::{collections::HashMap, net::TcpListener};
-use tokio::time::sleep;
-use uuid::Uuid;
-
-pub const STATE_PATH_ENV_VAR: &str = "IGGY_CONNECTORS_STATE_PATH";
-pub const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
-const LOCAL_STATE_PREFIX: &str = "local_state_";
-
-#[derive(Debug)]
-pub struct TestConnectorsRuntime {
- envs: HashMap<String, String>,
- child_handle: Option<Child>,
- server_address: SocketAddr,
- stdout_file_path: Option<PathBuf>,
- stderr_file_path: Option<PathBuf>,
- server_executable_path: Option<String>,
- local_state_path: String,
- cleanup: bool,
-}
-
-impl TestConnectorsRuntime {
- pub fn with_iggy_address(
- iggy_tcp_server_address: &str,
- envs: Option<HashMap<String, String>>,
- ) -> Self {
- Self::new(iggy_tcp_server_address, envs, None)
- }
-
- pub fn new(
- iggy_tcp_server_address: &str,
- extra_envs: Option<HashMap<String, String>>,
- server_executable_path: Option<String>,
- ) -> Self {
- let mut envs = HashMap::new();
- if let Some(extra) = extra_envs {
- for (key, value) in extra {
- envs.insert(key, value);
- }
- }
-
- envs.insert(
- "IGGY_CONNECTORS_IGGY_ADDRESS".to_string(),
- iggy_tcp_server_address.to_string(),
- );
- Self::create(envs, server_executable_path)
- }
-
- pub fn create(
- mut envs: HashMap<String, String>,
- server_executable_path: Option<String>,
- ) -> Self {
- let server_address = Self::get_random_server_address();
-
- // If IGGY_CONNECTORS_STATE_PATH is not set, use a random path starting with "local_state_"
- let local_state_path = if let Some(state_path) = envs.get(STATE_PATH_ENV_VAR) {
- state_path.to_string()
- } else {
- Self::get_random_path()
- };
-
- envs.insert(
- "IGGY_CONNECTORS_STATE_PATH".to_string(),
- local_state_path.clone(),
- );
-
- Self {
- envs,
- child_handle: None,
- server_address,
- stdout_file_path: None,
- stderr_file_path: None,
- server_executable_path,
- local_state_path,
- cleanup: true,
- }
- }
-
- pub fn start(&mut self) {
- self.cleanup();
- self.envs
- .entry("IGGY_CONNECTORS_HTTP_ADDRESS".to_string())
- .or_insert(self.server_address.to_string());
- let mut command = if let Some(server_executable_path) = &self.server_executable_path {
- Command::new(server_executable_path)
- } else {
- Command::cargo_bin("iggy-connectors").unwrap()
- };
- command.envs(self.envs.clone());
-
- if std::env::var(TEST_VERBOSITY_ENV_VAR).is_ok()
- || self.envs.contains_key(TEST_VERBOSITY_ENV_VAR)
- {
- command.stdout(Stdio::inherit());
- command.stderr(Stdio::inherit());
- } else {
- command.stdout(self.get_stdout_file());
- self.stdout_file_path = Some(fs::canonicalize(self.get_stdout_file_path()).unwrap());
- command.stderr(self.get_stderr_file());
- self.stderr_file_path = Some(fs::canonicalize(self.get_stderr_file_path()).unwrap());
- }
-
- let child = command
- .spawn()
- .expect("Failed to start Connectors Runtime process");
- self.child_handle = Some(child);
- }
-
- pub fn stop(&mut self) {
- #[allow(unused_mut)]
- if let Some(mut child_handle) = self.child_handle.take() {
- #[cfg(unix)]
- unsafe {
- use libc::SIGTERM;
- use libc::kill;
- kill(child_handle.id() as libc::pid_t, SIGTERM);
- }
-
- #[cfg(not(unix))]
- child_handle.kill().unwrap();
-
- if let Ok(output) = child_handle.wait_with_output() {
- let stderr = String::from_utf8_lossy(&output.stderr);
- let stdout = String::from_utf8_lossy(&output.stdout);
- if let Some(stderr_file_path) = &self.stderr_file_path {
- OpenOptions::new()
- .append(true)
- .create(true)
- .open(stderr_file_path)
- .unwrap()
- .write_all(stderr.as_bytes())
- .unwrap();
- }
-
- if let Some(stdout_file_path) = &self.stdout_file_path {
- OpenOptions::new()
- .append(true)
- .create(true)
- .open(stdout_file_path)
- .unwrap()
- .write_all(stdout.as_bytes())
- .unwrap();
- }
- }
- }
- self.cleanup();
- }
-
- pub fn is_started(&self) -> bool {
- self.child_handle.is_some()
- }
-
- pub fn pid(&self) -> u32 {
- self.child_handle.as_ref().unwrap().id()
- }
-
- pub fn get_random_path() -> String {
- format!("{}{}", LOCAL_STATE_PREFIX, Uuid::now_v7().to_u128_le())
- }
-
- fn get_stdout_file_path(&self) -> PathBuf {
- format!("{}_stdout.txt", self.local_state_path).into()
- }
-
- fn get_stderr_file_path(&self) -> PathBuf {
- format!("{}_stderr.txt", self.local_state_path).into()
- }
-
- fn get_stdout_file(&self) -> File {
- File::create(self.get_stdout_file_path()).unwrap()
- }
-
- fn get_stderr_file(&self) -> File {
- File::create(self.get_stderr_file_path()).unwrap()
- }
-
- pub fn get_http_api_address(&self) -> String {
- format!(
- "http://{}:{}",
- self.server_address.ip(),
- self.server_address.port()
- )
- }
-
- pub async fn ensure_started(&self) {
- let http_api_address = self.get_http_api_address();
- let client = reqwest::Client::new();
- let max_retries = 1000;
- let mut retries = 0;
- while let Err(error) = client.get(&http_api_address).send().await {
- sleep(Duration::from_millis(20)).await;
- retries += 1;
- if retries >= max_retries {
- panic!(
- "Failed to ping Connectors runtime: {http_api_address} after {max_retries} retries. {error}"
- );
- }
- }
- println!("Connectors runtime address started at: {http_api_address}");
- }
-
- fn get_random_server_address() -> SocketAddr {
- let mut rng = rand::thread_rng();
- let max_retries = 100;
-
- for _ in 0..max_retries {
- #[cfg(target_os = "linux")]
- let port = rng.gen_range(20000..=29999);
-
- #[cfg(target_os = "macos")]
- let port = rng.gen_range(20000..=49151);
-
- #[cfg(target_os = "windows")]
- let port = rng.gen_range(20000..=49151);
-
- let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), port);
- if TcpListener::bind(addr).is_ok() {
- return addr;
- }
- }
-
- panic!("Failed to find a free port after {max_retries} retries");
- }
-
- fn cleanup(&self) {
- if !self.cleanup {
- return;
- }
-
- if fs::metadata(&self.local_state_path).is_ok() {
- fs::remove_dir_all(&self.local_state_path).unwrap();
- }
- }
-}
-
-impl Drop for TestConnectorsRuntime {
- fn drop(&mut self) {
- self.stop();
- }
-}
diff --git a/core/integration/src/test_mcp_server.rs b/core/integration/src/test_mcp_server.rs
deleted file mode 100644
index f4179a698..000000000
--- a/core/integration/src/test_mcp_server.rs
+++ /dev/null
@@ -1,274 +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 assert_cmd::prelude::CommandCargoExt;
-use rand::Rng;
-use rmcp::{
- RoleClient, ServiceExt,
- model::{ClientCapabilities, ClientInfo, Implementation, InitializeRequestParam},
- service::RunningService,
- transport::StreamableHttpClientTransport,
-};
-use std::fs::{self, File};
-use std::net::{Ipv4Addr, SocketAddr};
-use std::path::PathBuf;
-use std::process::{Child, Command, Stdio};
-use std::thread::panicking;
-use std::time::Duration;
-use std::{collections::HashMap, net::TcpListener};
-use tokio::time::sleep;
-
-pub const CONSUMER_NAME: &str = "mcp";
-const MCP_PATH: &str = "/mcp";
-const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
-
-pub type McpClient = RunningService<RoleClient, InitializeRequestParam>;
-
-#[derive(Debug)]
-pub struct TestMcpServer {
- envs: HashMap<String, String>,
- child_handle: Option<Child>,
- server_address: SocketAddr,
- stdout_file_path: Option<PathBuf>,
- stderr_file_path: Option<PathBuf>,
- server_executable_path: Option<String>,
-}
-
-impl TestMcpServer {
- pub fn with_iggy_address(iggy_tcp_server_address: &str) -> Self {
- Self::new(iggy_tcp_server_address, None, None)
- }
-
- pub fn new(
- iggy_tcp_server_address: &str,
- extra_envs: Option<HashMap<String, String>>,
- server_executable_path: Option<String>,
- ) -> Self {
- let mut envs = HashMap::new();
- if let Some(extra) = extra_envs {
- for (key, value) in extra {
- envs.insert(key, value);
- }
- }
-
- envs.insert("IGGY_MCP_HTTP_PATH".to_string(), MCP_PATH.to_string());
- envs.insert(
- "IGGY_MCP_IGGY_ADDRESS".to_string(),
- iggy_tcp_server_address.to_string(),
- );
- envs.insert(
- "IGGY_MCP_IGGY_CONSUMER".to_string(),
- CONSUMER_NAME.to_string(),
- );
- envs.insert("IGGY_MCP_TRANSPORT".to_string(), "http".to_string());
- Self::create(envs, server_executable_path)
- }
-
- pub fn create(envs: HashMap<String, String>, server_executable_path: Option<String>) -> Self {
- let server_address = Self::get_random_server_address();
-
- Self {
- envs,
- child_handle: None,
- server_address,
- stdout_file_path: None,
- stderr_file_path: None,
- server_executable_path,
- }
- }
-
- pub fn start(&mut self) {
- self.envs
- .entry("IGGY_MCP_HTTP_ADDRESS".to_string())
- .or_insert(self.server_address.to_string());
- let mut command = if let Some(server_executable_path) = &self.server_executable_path {
- Command::new(server_executable_path)
- } else {
- Command::cargo_bin("iggy-mcp").unwrap()
- };
- command.envs(self.envs.clone());
-
- // By default, MCP server logs are redirected to files,
- // and dumped to stderr when test fails. With IGGY_TEST_VERBOSE=1
- // logs are dumped to stdout during test execution.
- if std::env::var(TEST_VERBOSITY_ENV_VAR).is_ok()
- || self.envs.contains_key(TEST_VERBOSITY_ENV_VAR)
- {
- command.stdout(Stdio::inherit());
- command.stderr(Stdio::inherit());
- } else {
- command.stdout(self.get_stdout_file());
- self.stdout_file_path = Some(fs::canonicalize(self.get_stdout_file_path()).unwrap());
- command.stderr(self.get_stderr_file());
- self.stderr_file_path = Some(fs::canonicalize(self.get_stderr_file_path()).unwrap());
- }
-
- let child = command.spawn().unwrap();
- self.child_handle = Some(child);
- }
-
- pub fn stop(&mut self) {
- #[allow(unused_mut)]
- if let Some(mut child_handle) = self.child_handle.take() {
- #[cfg(unix)]
- unsafe {
- use libc::SIGTERM;
- use libc::kill;
- kill(child_handle.id() as libc::pid_t, SIGTERM);
- }
-
- #[cfg(not(unix))]
- child_handle.kill().unwrap();
-
- let _ = child_handle.wait();
- }
- }
-
- pub fn is_started(&self) -> bool {
- self.child_handle.is_some()
- }
-
- pub fn pid(&self) -> u32 {
- self.child_handle.as_ref().unwrap().id()
- }
-
- fn get_http_mcp_api_address(&self) -> String {
- format!("{}{MCP_PATH}", self.get_http_api_address())
- }
-
- fn get_http_api_address(&self) -> String {
- format!(
- "http://{}:{}",
- self.server_address.ip(),
- self.server_address.port()
- )
- }
-
- pub async fn ensure_started(&self) {
- let http_api_address = self.get_http_api_address();
- let client = reqwest::Client::new();
- let max_retries = 3000;
- let mut retries = 0;
- while let Err(error) = client.get(&http_api_address).send().await {
- sleep(Duration::from_millis(20)).await;
- retries += 1;
- if retries >= max_retries {
- panic!(
- "Failed to ping MCP server: {http_api_address} after {max_retries} retries. {error}"
- );
- }
- }
- println!("MCP server address started at: {http_api_address}");
- }
-
- pub async fn get_client(&self) -> McpClient {
- let mcp_http_api_address = self.get_http_mcp_api_address();
- let transport = StreamableHttpClientTransport::from_uri(mcp_http_api_address);
- let client_info = ClientInfo {
- protocol_version: Default::default(),
- capabilities: ClientCapabilities::default(),
- meta: None,
- client_info: Implementation {
- name: "test-mcp-client".to_string(),
- version: "1.0.0".to_string(),
- ..Default::default()
- },
- };
- client_info
- .serve(transport)
- .await
- .inspect_err(|error| {
- eprintln!("MCP client error: {error}");
- })
- .expect("Failed to create MCP client")
- }
-
- fn get_random_server_address() -> SocketAddr {
- let mut rng = rand::thread_rng();
- let max_retries = 100;
-
- for _ in 0..max_retries {
- #[cfg(target_os = "linux")]
- let port = rng.gen_range(20000..=29999);
-
- #[cfg(target_os = "macos")]
- let port = rng.gen_range(20000..=49151);
-
- #[cfg(target_os = "windows")]
- let port = rng.gen_range(20000..=49151);
-
- let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), port);
- if TcpListener::bind(addr).is_ok() {
- return addr;
- }
- }
-
- panic!("Failed to find a free port after {max_retries} retries");
- }
-
- fn get_stdout_file_path(&self) -> String {
- format!("/tmp/iggy-mcp-{}.stdout", self.server_address.port())
- }
-
- fn get_stderr_file_path(&self) -> String {
- format!("/tmp/iggy-mcp-{}.stderr", self.server_address.port())
- }
-
- fn get_stdout_file(&self) -> Stdio {
- Stdio::from(File::create(self.get_stdout_file_path()).unwrap())
- }
-
- fn get_stderr_file(&self) -> Stdio {
- Stdio::from(File::create(self.get_stderr_file_path()).unwrap())
- }
-
- fn read_file_to_string(path: &str) -> String {
- fs::read_to_string(path).unwrap_or_else(|_| format!("Failed to read file: {}", path))
- }
-}
-
-impl Drop for TestMcpServer {
- fn drop(&mut self) {
- self.stop();
-
- if panicking() {
- if let Some(stdout_file_path) = &self.stdout_file_path {
- eprintln!(
- "Iggy MCP server stdout:\n{}",
- Self::read_file_to_string(stdout_file_path.to_str().unwrap())
- );
- }
-
- if let Some(stderr_file_path) = &self.stderr_file_path {
- eprintln!(
- "Iggy MCP server stderr:\n{}",
- Self::read_file_to_string(stderr_file_path.to_str().unwrap())
- );
- }
- }
-
- // Clean up log files
- if let Some(stdout_file_path) = &self.stdout_file_path {
- fs::remove_file(stdout_file_path).unwrap();
- }
-
- if let Some(stderr_file_path) = &self.stderr_file_path {
- fs::remove_file(stderr_file_path).unwrap();
- }
- }
-}
diff --git a/core/integration/src/test_server.rs b/core/integration/src/test_server.rs
deleted file mode 100644
index 81c68b619..000000000
--- a/core/integration/src/test_server.rs
+++ /dev/null
@@ -1,769 +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 assert_cmd::prelude::CommandCargoExt;
-use async_trait::async_trait;
-use configs::ConfigProvider;
-use derive_more::Display;
-use futures::executor::block_on;
-use iggy::prelude::UserStatus::Active;
-use iggy::prelude::*;
-use iggy_common::TransportProtocol;
-use rand::Rng;
-use server::configs::server::ServerConfig;
-use std::collections::HashMap;
-use std::fs;
-use std::fs::{File, OpenOptions};
-use std::io::Write;
-use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
-use std::path::{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;
-use uuid::Uuid;
-
-pub const SYSTEM_PATH_ENV_VAR: &str = "IGGY_SYSTEM_PATH";
-pub const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
-pub const IPV6_ENV_VAR: &str = "IGGY_TCP_IPV6";
-pub const IGGY_ROOT_USERNAME_VAR: &str = "IGGY_ROOT_USERNAME";
-pub const IGGY_ROOT_PASSWORD_VAR: &str = "IGGY_ROOT_PASSWORD";
-const USER_PASSWORD: &str = "secret";
-const SLEEP_INTERVAL_MS: u64 = 20;
-const LOCAL_DATA_PREFIX: &str = "local_data_";
-
-const MAX_PORT_WAIT_DURATION_S: u64 = 60;
-
-#[derive(PartialEq)]
-pub enum IpAddrKind {
- V4,
- V6,
-}
-
-#[async_trait]
-pub trait ClientFactory: Sync + Send {
- async fn create_client(&self) -> ClientWrapper;
- fn transport(&self) -> TransportProtocol;
- fn server_addr(&self) -> String;
-}
-#[derive(Display, Debug)]
-enum ServerProtocolAddr {
- #[display("RAW_TCP:{_0}")]
- RawTcp(SocketAddr),
-
- #[display("HTTP_TCP:{_0}")]
- HttpTcp(SocketAddr),
-
- #[display("QUIC_UDP:{_0}")]
- QuicUdp(SocketAddr),
-
- #[display("WEBSOCKET:{_0}")]
- WebSocket(SocketAddr),
-}
-
-pub struct TestServer {
- local_data_path: String,
- envs: HashMap<String, String>,
- child_handle: Option<Child>,
- server_addrs: Vec<ServerProtocolAddr>,
- stdout_file_path: Option<PathBuf>,
- stderr_file_path: Option<PathBuf>,
- cleanup: bool,
- server_executable_path: Option<String>,
- watchdog_handle: Option<JoinHandle<()>>,
- watchdog_stop: Arc<AtomicBool>,
-}
-
-impl std::fmt::Debug for TestServer {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- f.debug_struct("TestServer")
- .field("local_data_path", &self.local_data_path)
- .field("server_addrs", &self.server_addrs)
- .field("cleanup", &self.cleanup)
- .finish_non_exhaustive()
- }
-}
-
-impl TestServer {
- pub fn new(
- extra_envs: Option<HashMap<String, String>>,
- cleanup: bool,
- server_executable_path: Option<String>,
- ip_kind: IpAddrKind,
- ) -> Self {
- let mut envs = HashMap::new();
- if let Some(extra) = extra_envs {
- for (key, value) in extra {
- envs.insert(key, value);
- }
- }
-
- // Randomly select 4 CPU cores to reduce interference between parallel tests
- 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);
- let end = start + 4;
- format!("{}..{}", start, end)
- } else {
- "all".to_string()
- }
- }
- Err(_) => "0..4".to_string(),
- };
-
- envs.insert(
- "IGGY_SYSTEM_SHARDING_CPU_ALLOCATION".to_string(),
- cpu_allocation,
- );
-
- if ip_kind == IpAddrKind::V6 {
- envs.insert(IPV6_ENV_VAR.to_string(), "true".to_string());
- }
-
- if !envs.contains_key(IGGY_ROOT_USERNAME_VAR) {
- envs.insert(
- IGGY_ROOT_USERNAME_VAR.to_string(),
- DEFAULT_ROOT_USERNAME.to_string(),
- );
- }
-
- if !envs.contains_key(IGGY_ROOT_PASSWORD_VAR) {
- envs.insert(
- IGGY_ROOT_PASSWORD_VAR.to_string(),
- DEFAULT_ROOT_PASSWORD.to_string(),
- );
- }
-
- // If IGGY_SYSTEM_PATH is not set, use a random path starting with "local_data_"
- let local_data_path = if let Some(system_path) = envs.get(SYSTEM_PATH_ENV_VAR) {
- system_path.to_string()
- } else {
- TestServer::get_random_path()
- };
-
- Self::create(
- local_data_path,
- envs,
- cleanup,
- server_executable_path,
- ip_kind,
- )
- }
-
- pub fn create(
- local_data_path: String,
- envs: HashMap<String, String>,
- cleanup: bool,
- server_executable_path: Option<String>,
- ip_kind: IpAddrKind,
- ) -> Self {
- let mut server_addrs = Vec::new();
-
- if let Some(tcp_addr) = envs.get("IGGY_TCP_ADDRESS") {
- server_addrs.push(ServerProtocolAddr::RawTcp(tcp_addr.parse().unwrap()));
- }
-
- if let Some(http_addr) = envs.get("IGGY_HTTP_ADDRESS") {
- server_addrs.push(ServerProtocolAddr::HttpTcp(http_addr.parse().unwrap()));
- }
-
- if let Some(quic_addr) = envs.get("IGGY_QUIC_ADDRESS") {
- server_addrs.push(ServerProtocolAddr::QuicUdp(quic_addr.parse().unwrap()));
- }
-
- if server_addrs.is_empty() {
- server_addrs = match ip_kind {
- IpAddrKind::V6 => Self::get_server_ipv6_addrs_with_random_port(),
- _ => Self::get_server_ipv4_addrs_with_random_port(),
- }
- }
-
- Self {
- local_data_path,
- envs,
- child_handle: None,
- server_addrs,
- stdout_file_path: None,
- stderr_file_path: None,
- cleanup,
- server_executable_path,
- watchdog_handle: None,
- watchdog_stop: Arc::new(AtomicBool::new(false)),
- }
- }
-
- pub fn start(&mut self) {
- self.set_server_addrs_from_env();
- self.cleanup();
-
- // Remove the config file if it exists from a previous run.
- // Without this, starting the server on existing data will not work, because
- // port detection mechanism will use port from previous runtime.
- let config_path = format!("{}/runtime/current_config.toml", self.local_data_path);
- if Path::new(&config_path).exists() {
- fs::remove_file(&config_path).ok();
- }
-
- let files_path = self.local_data_path.clone();
- let mut command = if let Some(server_executable_path) = &self.server_executable_path {
- Command::new(server_executable_path)
- } else {
- Command::cargo_bin("iggy-server").unwrap()
- };
- command.env(SYSTEM_PATH_ENV_VAR, files_path);
- command.envs(self.envs.clone());
-
- // By default, server all logs are redirected to files,
- // and dumped to stderr when test fails. With IGGY_TEST_VERBOSE=1
- // logs are dumped to stdout during test execution.
- if std::env::var(TEST_VERBOSITY_ENV_VAR).is_ok()
- || self.envs.contains_key(TEST_VERBOSITY_ENV_VAR)
- {
- command.stdout(Stdio::inherit());
- command.stderr(Stdio::inherit());
- } else {
- command.stdout(self.get_stdout_file());
- self.stdout_file_path = Some(fs::canonicalize(self.get_stdout_file_path()).unwrap());
- command.stderr(self.get_stderr_file());
- self.stderr_file_path = Some(fs::canonicalize(self.get_stderr_file_path()).unwrap());
- }
-
- let child = command.spawn().unwrap();
- let pid = child.id();
- self.child_handle = Some(child);
- self.wait_until_server_has_bound();
-
- let watchdog_stop = self.watchdog_stop.clone();
- let stdout_path = self.stdout_file_path.clone();
- let stderr_path = self.stderr_file_path.clone();
- let watchdog_handle = thread::Builder::new()
- .name("test-server-watchdog".to_string())
- .spawn(move || {
- Self::watchdog_loop(pid, watchdog_stop, stdout_path, stderr_path);
- })
- .expect("Failed to spawn watchdog thread");
- self.watchdog_handle = Some(watchdog_handle);
- }
-
- /// Watchdog loop that monitors the server process.
- /// Panics if the server exits while the watchdog is still running (i.e., not gracefully stopped).
- 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;
- }
-
- // Check if process is alive AND not a zombie via /proc/{pid}/stat
- // A zombie process still exists in the process table (kill returns 0)
- // but has state 'Z' in /proc/pid/stat
- 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 {
- // /proc is the only reliable way to distinguish zombies from live processes.
- // A zombie still responds to kill(pid, 0) but we need to detect it as dead.
- let stat_path = format!("/proc/{}/stat", pid);
- match fs::read_to_string(&stat_path) {
- Ok(content) => {
- // State char is after the command name in parentheses: "pid (comm) S ..."
- 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 {
- // macOS lacks /proc, so we use signal 0 which checks process existence
- // without actually sending anything. Zombies are less of a concern here
- // since our child will be reaped when we call wait() in stop().
- unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
- }
-
- pub fn stop(&mut self) {
- self.watchdog_stop.store(true, Ordering::SeqCst);
- if let Some(watchdog) = self.watchdog_handle.take() {
- let _ = watchdog.join();
- }
-
- #[allow(unused_mut)]
- if let Some(mut child_handle) = self.child_handle.take() {
- #[cfg(unix)]
- unsafe {
- use libc::SIGTERM;
- use libc::kill;
- kill(child_handle.id() as libc::pid_t, SIGTERM);
- }
-
- #[cfg(not(unix))]
- child_handle.kill().unwrap();
-
- if let Ok(output) = child_handle.wait_with_output() {
- let stderr = String::from_utf8_lossy(&output.stderr);
- let stdout = String::from_utf8_lossy(&output.stdout);
- if let Some(stderr_file_path) = &self.stderr_file_path {
- OpenOptions::new()
- .append(true)
- .create(true)
- .open(stderr_file_path)
- .unwrap()
- .write_all(stderr.as_bytes())
- .unwrap();
- }
-
- if let Some(stdout_file_path) = &self.stdout_file_path {
- OpenOptions::new()
- .append(true)
- .create(true)
- .open(stdout_file_path)
- .unwrap()
- .write_all(stdout.as_bytes())
- .unwrap();
- }
- }
- }
- self.cleanup();
- }
-
- pub fn is_started(&self) -> bool {
- self.child_handle.is_some()
- }
-
- /// Check if the server process is still running.
- /// Returns false if the process has exited (crashed or terminated).
- pub fn is_running(&mut self) -> bool {
- if let Some(child) = self.child_handle.as_mut() {
- match child.try_wait() {
- Ok(Some(_)) => false, // Process has exited
- Ok(None) => true, // Still running
- Err(_) => false, // Error checking, assume dead
- }
- } else {
- false
- }
- }
-
- /// Assert that the server is still running. If it has crashed,
- /// panic with server logs included for debugging.
- pub fn assert_running(&mut self) {
- if let Some(Ok(Some(exit_status))) = self.child_handle.as_mut().map(|c| c.try_wait()) {
- let (stdout_content, stderr_content) = self.collect_logs();
- panic!(
- "Server process has crashed with exit status: {}\n\n\
- === STDOUT ===\n{}\n\n\
- === STDERR ===\n{}",
- exit_status, stdout_content, stderr_content
- );
- }
- }
-
- /// Collect server stdout and stderr logs from files.
- pub fn collect_logs(&self) -> (String, String) {
- let stdout = self
- .stdout_file_path
- .as_ref()
- .and_then(|path| fs::read_to_string(path).ok())
- .unwrap_or_else(|| "[No stdout log]".to_string());
-
- let stderr = self
- .stderr_file_path
- .as_ref()
- .and_then(|path| fs::read_to_string(path).ok())
- .unwrap_or_else(|| "[No stderr log]".to_string());
-
- (stdout, stderr)
- }
-
- pub fn pid(&self) -> u32 {
- self.child_handle.as_ref().unwrap().id()
- }
-
- fn cleanup(&self) {
- if !self.cleanup {
- return;
- }
-
- if fs::metadata(&self.local_data_path).is_ok() {
- fs::remove_dir_all(&self.local_data_path).unwrap();
- }
- }
-
- fn get_server_ipv4_addrs_with_random_port() -> Vec<ServerProtocolAddr> {
- let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0);
- vec![
- ServerProtocolAddr::QuicUdp(addr),
- ServerProtocolAddr::RawTcp(addr),
- ServerProtocolAddr::HttpTcp(addr),
- ServerProtocolAddr::WebSocket(addr),
- ]
- }
-
- fn get_server_ipv6_addrs_with_random_port() -> Vec<ServerProtocolAddr> {
- let addr = SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 0);
- vec![
- ServerProtocolAddr::QuicUdp(addr),
- ServerProtocolAddr::RawTcp(addr),
- ServerProtocolAddr::HttpTcp(addr),
- ServerProtocolAddr::WebSocket(addr),
- ]
- }
-
- fn set_server_addrs_from_env(&mut self) {
- for server_protocol_addr in &self.server_addrs {
- let key = match server_protocol_addr {
- ServerProtocolAddr::RawTcp(addr) => {
- ("IGGY_TCP_ADDRESS".to_string(), addr.to_string())
- }
- ServerProtocolAddr::HttpTcp(addr) => {
- ("IGGY_HTTP_ADDRESS".to_string(), addr.to_string())
- }
- ServerProtocolAddr::QuicUdp(addr) => {
- ("IGGY_QUIC_ADDRESS".to_string(), addr.to_string())
- }
- ServerProtocolAddr::WebSocket(addr) => {
- ("IGGY_WEBSOCKET_ADDRESS".to_string(), addr.to_string())
- }
- };
-
- self.envs.entry(key.0).or_insert(key.1);
- }
- }
-
- fn wait_until_server_has_bound(&mut self) {
- let config_path = format!("{}/runtime/current_config.toml", self.local_data_path);
- let file_config_provider = ServerConfig::config_provider(&config_path);
-
- let max_attempts = (MAX_PORT_WAIT_DURATION_S * 1000) / SLEEP_INTERVAL_MS;
- self.server_addrs.clear();
-
- let config = block_on(async {
- let mut loaded_config = None;
-
- for _ in 0..max_attempts {
- if !Path::new(&config_path).exists() {
- if let Some(exit_status) =
- self.child_handle.as_mut().unwrap().try_wait().unwrap()
- {
- panic!("Server process has exited with status {exit_status}!");
- }
- sleep(Duration::from_millis(SLEEP_INTERVAL_MS));
- continue;
- }
- let config: Result<ServerConfig, _> = file_config_provider.load_config().await;
- match config {
- Ok(config) => {
- // Verify config contains fresh addresses, not stale defaults
- // Default ports: TCP=8090, HTTP=3000, QUIC=8080, WebSocket=8092
- let tcp_port: u16 = config
- .tcp
- .address
- .split(':')
- .nth(1)
- .and_then(|s| s.parse().ok())
- .unwrap_or(0);
- let http_port: u16 = config
- .http
- .address
- .split(':')
- .nth(1)
- .and_then(|s| s.parse().ok())
- .unwrap_or(0);
- let quic_port: u16 = config
- .quic
- .address
- .split(':')
- .nth(1)
- .and_then(|s| s.parse().ok())
- .unwrap_or(0);
- let websocket_port: u16 = config
- .websocket
- .address
- .split(':')
- .nth(1)
- .and_then(|s| s.parse().ok())
- .unwrap_or(0);
-
- if tcp_port == 8090
- || http_port == 3000
- || quic_port == 8080
- || websocket_port == 8092
- {
- sleep(Duration::from_millis(SLEEP_INTERVAL_MS));
- continue;
- }
-
- loaded_config = Some(config);
- break;
- }
- Err(_) => sleep(Duration::from_millis(SLEEP_INTERVAL_MS)),
- }
- }
- loaded_config
- });
-
- if let Some(config) = config {
- // Only validate and add enabled protocols
- if config.quic.enabled {
- let quic_addr: SocketAddr = config.quic.address.parse().unwrap();
- if quic_addr.port() == 0 {
- panic!("Quic address port is 0!");
- }
- self.server_addrs
- .push(ServerProtocolAddr::QuicUdp(quic_addr));
- }
-
- if config.tcp.enabled {
- let tcp_addr: SocketAddr = config.tcp.address.parse().unwrap();
- if tcp_addr.port() == 0 {
- panic!("Tcp address port is 0!");
- }
- self.server_addrs.push(ServerProtocolAddr::RawTcp(tcp_addr));
- }
-
- if config.http.enabled {
- let http_addr: SocketAddr = config.http.address.parse().unwrap();
- if http_addr.port() == 0 {
- panic!("Http address port is 0!");
- }
- self.server_addrs
- .push(ServerProtocolAddr::HttpTcp(http_addr));
- }
-
- if config.websocket.enabled {
- let websocket_addr: SocketAddr = config.websocket.address.parse().unwrap();
- if websocket_addr.port() == 0 {
- panic!("WebSocket address port is 0!");
- }
- self.server_addrs
- .push(ServerProtocolAddr::WebSocket(websocket_addr));
- }
- } else {
- panic!(
- "Failed to load config from file {config_path} in {MAX_PORT_WAIT_DURATION_S} s!"
- );
- }
- }
-
- fn get_stdout_file_path(&self) -> PathBuf {
- format!("{}_stdout.txt", self.local_data_path).into()
- }
-
- fn get_stderr_file_path(&self) -> PathBuf {
- format!("{}_stderr.txt", self.local_data_path).into()
- }
-
- fn get_stdout_file(&self) -> File {
- File::create(self.get_stdout_file_path()).unwrap()
- }
-
- fn get_stderr_file(&self) -> File {
- File::create(self.get_stderr_file_path()).unwrap()
- }
-
- fn read_file_to_string(path: &str) -> String {
- fs::read_to_string(path).unwrap()
- }
-
- pub fn get_local_data_path(&self) -> &str {
- &self.local_data_path
- }
-
- pub fn get_random_path() -> String {
- format!("{}{}", LOCAL_DATA_PREFIX, Uuid::now_v7().to_u128_le())
- }
-
- pub fn get_http_api_addr(&self) -> Option<String> {
- for server_protocol_addr in &self.server_addrs {
- if let ServerProtocolAddr::HttpTcp(a) = server_protocol_addr {
- return Some(a.to_string());
- }
- }
- None
- }
-
- pub fn get_raw_tcp_addr(&self) -> Option<String> {
- for server_protocol_addr in &self.server_addrs {
- if let ServerProtocolAddr::RawTcp(a) = server_protocol_addr {
- return Some(a.to_string());
- }
- }
- None
- }
-
- pub fn get_quic_udp_addr(&self) -> Option<String> {
- for server_protocol_addr in &self.server_addrs {
- if let ServerProtocolAddr::QuicUdp(a) = server_protocol_addr {
- return Some(a.to_string());
- }
- }
- None
- }
-
- pub fn get_server_ip_addr(&self) -> Option<String> {
- if let Some(server_address) = self
- .get_raw_tcp_addr()
- .or_else(|| self.get_http_api_addr())
- .or_else(|| self.get_quic_udp_addr())
- {
- server_address
- .split(':')
- .map(|s| s.to_string())
- .collect::<Vec<_>>()
- .first()
- .cloned()
- } else {
- None
- }
- }
-
- pub fn get_websocket_addr(&self) -> Option<String> {
- for server_protocol_addr in &self.server_addrs {
- if let ServerProtocolAddr::WebSocket(a) = server_protocol_addr {
- return Some(a.to_string());
- }
- }
- None
- }
-}
-
-impl Drop for TestServer {
- fn drop(&mut self) {
- self.stop();
- if panicking() {
- if let Some(stdout_file_path) = &self.stdout_file_path {
- eprintln!(
- "Iggy server stdout:\n{}",
- Self::read_file_to_string(stdout_file_path.to_str().unwrap())
- );
- }
-
- if let Some(stderr_file_path) = &self.stderr_file_path {
- eprintln!(
- "Iggy server stderr:\n{}",
- Self::read_file_to_string(stderr_file_path.to_str().unwrap())
- );
- }
- }
- if let Some(stdout_file_path) = &self.stdout_file_path {
- fs::remove_file(stdout_file_path).unwrap();
- }
- if let Some(stderr_file_path) = &self.stderr_file_path {
- fs::remove_file(stderr_file_path).unwrap();
- }
- }
-}
-
-impl Default for TestServer {
- fn default() -> Self {
- TestServer::new(None, true, None, IpAddrKind::V4)
- }
-}
-
-pub async fn create_user(client: &IggyClient, username: &str) {
- client
- .create_user(
- username,
- USER_PASSWORD,
- Active,
- Some(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,
- }),
- )
- .await
- .unwrap();
-}
-
-pub async fn delete_user(client: &IggyClient, username: &str) {
- client
- .delete_user(&Identifier::named(username).unwrap())
- .await
- .unwrap();
-}
-
-pub async fn login_root(client: &IggyClient) -> IdentityInfo {
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap()
-}
-
-pub async fn login_user(client: &IggyClient, username: &str) -> IdentityInfo {
- client.login_user(username, USER_PASSWORD).await.unwrap()
-}
-
-pub async fn assert_clean_system(system_client: &IggyClient) {
- let streams = system_client.get_streams().await.unwrap();
- assert!(streams.is_empty());
-
- let users = system_client.get_users().await.unwrap();
- assert_eq!(users.len(), 1);
-}
diff --git a/core/integration/src/test_tls_utils.rs b/core/integration/src/test_tls_utils.rs
deleted file mode 100644
index 75b75abc8..000000000
--- a/core/integration/src/test_tls_utils.rs
+++ /dev/null
@@ -1,52 +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 std::fs::{self, File};
-use std::io::Write;
-use std::path::Path;
-
-/// Generate self-signed certificates for TLS testing
-pub fn generate_test_certificates(cert_dir: &str) -> Result<(), Box<dyn std::error::Error>> {
- fs::create_dir_all(cert_dir)?;
-
- let subject_alt_names = vec![
- "localhost".to_string(),
- "127.0.0.1".to_string(),
- "::1".to_string(),
- ];
-
- let cert = rcgen::generate_simple_self_signed(subject_alt_names)?;
-
- let cert_path = Path::new(cert_dir).join("test_cert.pem");
- let mut cert_file = File::create(&cert_path)?;
- cert_file.write_all(cert.cert.pem().as_bytes())?;
-
- let key_path = Path::new(cert_dir).join("test_key.pem");
- let mut key_file = File::create(&key_path)?;
- key_file.write_all(cert.signing_key.serialize_pem().as_bytes())?;
-
- Ok(())
-}
-
-/// Clean up test certificates
-pub fn cleanup_test_certificates(cert_dir: &str) -> Result<(), Box<dyn std::error::Error>> {
- if Path::new(cert_dir).exists() {
- fs::remove_dir_all(cert_dir)?;
- }
- Ok(())
-}
diff --git a/core/integration/src/websocket_client.rs b/core/integration/src/websocket_client.rs
deleted file mode 100644
index c1d037669..000000000
--- a/core/integration/src/websocket_client.rs
+++ /dev/null
@@ -1,76 +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::test_server::ClientFactory;
-use async_trait::async_trait;
-use iggy::prelude::{Client, ClientWrapper, WebSocketClientConfig};
-use iggy::websocket::websocket_client::WebSocketClient;
-use iggy_common::TransportProtocol;
-use std::sync::Arc;
-
-#[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]
-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_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)
- }
-
- fn transport(&self) -> TransportProtocol {
- TransportProtocol::WebSocket
- }
-
- fn server_addr(&self) -> String {
- self.server_addr.clone()
- }
-}
diff --git a/core/integration/tests/cli/common/mod.rs b/core/integration/tests/cli/common/mod.rs
index 81545a480..c5f228e05 100644
--- a/core/integration/tests/cli/common/mod.rs
+++ b/core/integration/tests/cli/common/mod.rs
@@ -23,14 +23,13 @@ pub(crate) use crate::cli::common::help::{CLAP_INDENT, TestHelpCmd, USAGE_PREFIX
use assert_cmd::assert::{Assert, OutputAssertExt};
use assert_cmd::prelude::CommandCargoExt;
use async_trait::async_trait;
-use iggy::clients::client::IggyClient;
-use iggy::prelude::defaults::*;
-use iggy::prelude::{Client, ClientWrapper, SystemClient, TcpClient, TcpClientConfig, UserClient};
-use integration::test_server::TestServer;
+use iggy::prelude::Client;
+use integration::harness::{
+ ServerHandle, TestBinary, TestHarness, TestHarnessBuilder, TestServerConfig,
+};
use std::fmt::{Display, Formatter, Result};
use std::io::Write;
use std::process::{Command, Stdio};
-use std::sync::Arc;
pub(crate) enum TestIdentifier {
Numeric,
@@ -82,33 +81,30 @@ pub(crate) trait IggyCmdTestCase {
}
fn verify_command(&self, command_state: Assert);
async fn verify_server_state(&self, client: &dyn Client);
- fn protocol(&self, server: &TestServer) -> Vec<String> {
+ fn protocol(&self, server: &ServerHandle) -> Vec<String> {
vec![
"--tcp-server-address".into(),
- server.get_raw_tcp_addr().unwrap(),
+ server.raw_tcp_addr().unwrap(),
]
}
}
pub(crate) struct IggyCmdTest {
- server: TestServer,
- client: IggyClient,
+ harness: TestHarness,
+ start_server: bool,
}
impl IggyCmdTest {
pub(crate) fn new(start_server: bool) -> Self {
- let mut server = TestServer::default();
- if start_server {
- server.start();
- }
- let tcp_client_config = TcpClientConfig {
- server_address: server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = IggyClient::create(client, None, None);
+ let harness = TestHarnessBuilder::default()
+ .server(TestServerConfig::default())
+ .build()
+ .unwrap();
- Self { server, client }
+ Self {
+ harness,
+ start_server,
+ }
}
pub(crate) fn help_message() -> Self {
@@ -116,29 +112,29 @@ impl IggyCmdTest {
}
pub(crate) async fn setup(&mut self) {
- self.client.connect().await.unwrap();
-
- let ping_result = self.client.ping().await;
-
- assert!(ping_result.is_ok());
-
- let identity_info = self
- .client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap();
-
- assert_eq!(identity_info.user_id, 0);
+ if self.start_server && !self.harness.server().is_running() {
+ self.harness
+ .start()
+ .await
+ .expect("Failed to start test harness");
+ }
}
pub(crate) async fn execute_test(&mut self, mut test_case: impl IggyCmdTestCase) {
// Make sure server is started
assert!(
- self.server.is_started(),
+ self.harness.server().is_running(),
"Server is not running, make sure it has been started with IggyCmdTest::setup()"
);
+
+ let client = self
+ .harness
+ .tcp_root_client()
+ .await
+ .expect("Failed to create TCP root client");
+
// Prepare iggy server state before test
- test_case.prepare_server_state(&self.client).await;
+ test_case.prepare_server_state(&client).await;
// Get iggy tool
#[allow(deprecated)]
let mut command = Command::cargo_bin("iggy").unwrap();
@@ -149,7 +145,7 @@ impl IggyCmdTest {
// Set a fixed terminal width for consistent help output formatting
command.env("COLUMNS", "500");
// Set server address for the command - it's randomized for each test
- command.args(test_case.protocol(&self.server));
+ command.args(test_case.protocol(self.harness.server()));
// Set COLUMNS environment variable to 200 to avoid truncation of output
command.env("COLUMNS", "200");
@@ -200,7 +196,7 @@ impl IggyCmdTest {
// Verify command output, exit code, etc in the test (if needed)
test_case.verify_command(assert);
// Verify iggy server state after the test
- test_case.verify_server_state(&self.client).await;
+ test_case.verify_server_state(&client).await;
}
pub(crate) async fn execute_test_for_help_command(&mut self, test_case: TestHelpCmd) {
@@ -241,7 +237,7 @@ impl IggyCmdTest {
#[cfg(not(target_os = "macos"))]
pub(crate) fn get_tcp_server_address(&self) -> Option<String> {
- self.server.get_raw_tcp_addr()
+ self.harness.server().raw_tcp_addr()
}
}
diff --git a/core/integration/tests/cli/context/test_context_applied.rs b/core/integration/tests/cli/context/test_context_applied.rs
index 0bbbc1dd5..59f42a2c8 100644
--- a/core/integration/tests/cli/context/test_context_applied.rs
+++ b/core/integration/tests/cli/context/test_context_applied.rs
@@ -23,7 +23,7 @@ use async_trait::async_trait;
use iggy::prelude::ArgsOptional;
use iggy::prelude::Client;
use iggy_binary_protocol::cli::binary_context::common::ContextConfig;
-use integration::test_server::TestServer;
+use integration::harness::ServerHandle;
use predicates::str::{contains, starts_with};
use serial_test::parallel;
use std::collections::HashMap;
@@ -113,7 +113,7 @@ impl IggyCmdTestCase for TestContextApplied {
async fn verify_server_state(&self, _client: &dyn Client) {}
- fn protocol(&self, server: &TestServer) -> Vec<String> {
+ fn protocol(&self, server: &ServerHandle) -> Vec<String> {
let transport = self
.set_transport_arg
.as_ref()
@@ -123,11 +123,11 @@ impl IggyCmdTestCase for TestContextApplied {
Some(protocol) => match protocol.as_str() {
"quic" => vec![
"--quic-server-address".into(),
- server.get_quic_udp_addr().unwrap(),
+ server.quic_addr().unwrap().to_string(),
],
_ => vec![
"--tcp-server-address".into(),
- server.get_raw_tcp_addr().unwrap(),
+ server.raw_tcp_addr().unwrap(),
],
},
None => panic!("either set_transport_arg or set_transport_context must be set"),
diff --git a/core/integration/tests/cli/system/test_me_command.rs b/core/integration/tests/cli/system/test_me_command.rs
index 19dfeb628..d09700c22 100644
--- a/core/integration/tests/cli/system/test_me_command.rs
+++ b/core/integration/tests/cli/system/test_me_command.rs
@@ -21,7 +21,7 @@ use assert_cmd::assert::Assert;
use async_trait::async_trait;
use iggy::prelude::Client;
use iggy_common::TransportProtocol;
-use integration::test_server::TestServer;
+use integration::harness::ServerHandle;
use predicates::str::{contains, diff, starts_with};
use serial_test::parallel;
@@ -108,19 +108,19 @@ impl IggyCmdTestCase for TestMeCmd {
async fn verify_server_state(&self, _client: &dyn Client) {}
- fn protocol(&self, server: &TestServer) -> Vec<String> {
+ fn protocol(&self, server: &ServerHandle) -> Vec<String> {
match &self.protocol {
TransportProtocol::Tcp => vec![
"--tcp-server-address".into(),
- server.get_raw_tcp_addr().unwrap(),
+ server.raw_tcp_addr().unwrap(),
],
TransportProtocol::Quic => vec![
"--quic-server-address".into(),
- server.get_quic_udp_addr().unwrap(),
+ server.quic_addr().unwrap().to_string(),
],
TransportProtocol::WebSocket => vec![
"--websocket-server-address".into(),
- server.get_websocket_addr().unwrap(),
+ server.websocket_addr().unwrap().to_string(),
],
TransportProtocol::Http => {
panic!("HTTP transport is not supported for the 'me' command")
diff --git a/core/integration/tests/config_provider/mod.rs b/core/integration/tests/config_provider/mod.rs
index adf67eac5..3904b7f7a 100644
--- a/core/integration/tests/config_provider/mod.rs
+++ b/core/integration/tests/config_provider/mod.rs
@@ -22,11 +22,11 @@ use configs_derive::ConfigEnv;
use figment::providers::{Format, Toml};
use figment::value::Dict;
use figment::{Figment, Provider};
-use integration::file::get_root_path;
use serde::{Deserialize, Serialize};
use serial_test::serial;
use server::configs::server::ServerConfig;
use std::env;
+use std::path::PathBuf;
#[serial]
#[tokio::test]
@@ -679,3 +679,8 @@ fn debug_print_test_root_config_mappings() {
"Missing TEST_NESTED_CONFIG_TYPE -> nested.config_type mapping"
);
}
+
+fn get_root_path() -> PathBuf {
+ let manifest_dir = std::env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR not set!");
+ PathBuf::from(manifest_dir)
+}
diff --git a/core/integration/tests/connectors/fixtures/quickwit/container.rs b/core/integration/tests/connectors/fixtures/quickwit/container.rs
index ba01df3e4..5c79b9bc6 100644
--- a/core/integration/tests/connectors/fixtures/quickwit/container.rs
+++ b/core/integration/tests/connectors/fixtures/quickwit/container.rs
@@ -24,12 +24,17 @@ use reqwest_retry::RetryTransientMiddleware;
use reqwest_retry::policies::ExponentialBackoff;
use serde::Deserialize;
use std::collections::HashMap;
+use std::time::Duration;
use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor};
use testcontainers_modules::testcontainers::runners::AsyncRunner;
use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt};
+use tokio::time::sleep;
use tracing::info;
use uuid::Uuid;
+const DEFAULT_POLL_ATTEMPTS: usize = 100;
+const DEFAULT_POLL_INTERVAL_MS: u64 = 50;
+
const QUICKWIT_IMAGE: &str = "quickwit/quickwit";
const QUICKWIT_TAG: &str = "0.8.2";
const QUICKWIT_PORT: u16 = 7280;
@@ -226,6 +231,34 @@ pub trait QuickwitOps: Sync {
})
}
}
+
+ fn wait_for_documents(
+ &self,
+ index_id: &str,
+ expected_count: usize,
+ ) -> impl std::future::Future<Output = Result<QuickwitSearchResponse, TestBinaryError>> + Send
+ {
+ async move {
+ let mut last_count = 0;
+ for _ in 0..DEFAULT_POLL_ATTEMPTS {
+ if self.flush_index(index_id).await.is_ok()
+ && let Ok(search) = self.search_all(index_id).await
+ {
+ last_count = search.num_hits;
+ if search.num_hits >= expected_count {
+ return Ok(search);
+ }
+ }
+ sleep(Duration::from_millis(DEFAULT_POLL_INTERVAL_MS / 5)).await;
+ }
+ Err(TestBinaryError::InvalidState {
+ message: format!(
+ "Expected {} documents but got {} after {} poll attempts",
+ expected_count, last_count, DEFAULT_POLL_ATTEMPTS
+ ),
+ })
+ }
+ }
}
fn get_index_config(index_id: &str) -> String {
diff --git a/core/integration/tests/connectors/postgres/postgres_sink.rs b/core/integration/tests/connectors/postgres/postgres_sink.rs
index 0ca736995..bbe9c52b2 100644
--- a/core/integration/tests/connectors/postgres/postgres_sink.rs
+++ b/core/integration/tests/connectors/postgres/postgres_sink.rs
@@ -39,7 +39,7 @@ type SinkJsonRow = (i64, serde_json::Value);
seed = seeds::connector_stream
)]
async fn json_messages_sink_stores_as_bytea(harness: &TestHarness, fixture: PostgresSinkFixture) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let pool = fixture.create_pool().await.expect("Failed to create pool");
fixture.wait_for_table(&pool, SINK_TABLE).await;
@@ -104,7 +104,7 @@ async fn binary_messages_sink_stores_as_bytea(
harness: &TestHarness,
fixture: PostgresSinkByteaFixture,
) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let pool = fixture.create_pool().await.expect("Failed to create pool");
fixture.wait_for_table(&pool, SINK_TABLE).await;
@@ -168,7 +168,7 @@ async fn json_messages_sink_stores_as_jsonb(
harness: &TestHarness,
fixture: PostgresSinkJsonFixture,
) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let pool = fixture.create_pool().await.expect("Failed to create pool");
fixture.wait_for_table(&pool, SINK_TABLE).await;
diff --git a/core/integration/tests/connectors/postgres/postgres_source.rs b/core/integration/tests/connectors/postgres/postgres_source.rs
index a1fc15b97..02820e985 100644
--- a/core/integration/tests/connectors/postgres/postgres_source.rs
+++ b/core/integration/tests/connectors/postgres/postgres_source.rs
@@ -39,7 +39,7 @@ async fn json_rows_source_produces_messages_to_iggy(
harness: &TestHarness,
fixture: PostgresSourceJsonFixture,
) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let pool = fixture.create_pool().await.expect("Failed to create pool");
fixture.create_table(&pool).await;
@@ -120,7 +120,7 @@ async fn bytea_rows_source_produces_raw_messages_to_iggy(
harness: &TestHarness,
fixture: PostgresSourceByteaFixture,
) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let pool = fixture.create_pool().await.expect("Failed to create pool");
fixture.create_table(&pool).await;
@@ -183,7 +183,7 @@ async fn jsonb_rows_source_produces_json_messages_to_iggy(
harness: &TestHarness,
fixture: PostgresSourceJsonbFixture,
) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let pool = fixture.create_pool().await.expect("Failed to create pool");
fixture.create_table(&pool).await;
@@ -250,7 +250,7 @@ async fn delete_after_read_source_removes_rows_after_producing(
harness: &TestHarness,
fixture: PostgresSourceDeleteFixture,
) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let pool = fixture.create_pool().await.expect("Failed to create pool");
fixture.create_table(&pool).await;
@@ -319,7 +319,7 @@ async fn processed_column_source_marks_rows_after_producing(
harness: &TestHarness,
fixture: PostgresSourceMarkFixture,
) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let pool = fixture.create_pool().await.expect("Failed to create pool");
fixture.create_table(&pool).await;
diff --git a/core/integration/tests/connectors/quickwit/quickwit_sink.rs b/core/integration/tests/connectors/quickwit/quickwit_sink.rs
index ffa08f6d4..a888f7bba 100644
--- a/core/integration/tests/connectors/quickwit/quickwit_sink.rs
+++ b/core/integration/tests/connectors/quickwit/quickwit_sink.rs
@@ -35,7 +35,7 @@ async fn given_existent_quickwit_index_should_store(
harness: &TestHarness,
fixture: QuickwitPreCreatedFixture,
) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
@@ -68,13 +68,8 @@ async fn given_existent_quickwit_index_should_store(
.await
.expect("send messages");
- fixture
- .flush_index(seeds::names::TOPIC)
- .await
- .expect("flush");
-
let search = fixture
- .search_all(seeds::names::TOPIC)
+ .wait_for_documents(seeds::names::TOPIC, message_count)
.await
.expect("search");
@@ -95,7 +90,7 @@ async fn given_nonexistent_quickwit_index_should_create_and_store(
harness: &TestHarness,
fixture: QuickwitFixture,
) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
@@ -128,13 +123,8 @@ async fn given_nonexistent_quickwit_index_should_create_and_store(
.await
.expect("send messages");
- fixture
- .flush_index(seeds::names::TOPIC)
- .await
- .expect("flush");
-
let search = fixture
- .search_all(seeds::names::TOPIC)
+ .wait_for_documents(seeds::names::TOPIC, message_count)
.await
.expect("search");
@@ -152,7 +142,7 @@ async fn given_nonexistent_quickwit_index_should_create_and_store(
seed = seeds::connector_stream
)]
async fn given_bulk_message_send_should_store(harness: &TestHarness, fixture: QuickwitFixture) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
@@ -185,13 +175,8 @@ async fn given_bulk_message_send_should_store(harness: &TestHarness, fixture: Qu
.await
.expect("send messages");
- fixture
- .flush_index(seeds::names::TOPIC)
- .await
- .expect("flush");
-
let search = fixture
- .search_all(seeds::names::TOPIC)
+ .wait_for_documents(seeds::names::TOPIC, message_count)
.await
.expect("search");
@@ -209,7 +194,7 @@ async fn given_bulk_message_send_should_store(harness: &TestHarness, fixture: Qu
seed = seeds::connector_stream
)]
async fn given_invalid_messages_should_not_store(harness: &TestHarness, fixture: QuickwitFixture) {
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
@@ -253,13 +238,8 @@ async fn given_invalid_messages_should_not_store(harness: &TestHarness, fixture:
.expect("send messages");
}
- fixture
- .flush_index(seeds::names::TOPIC)
- .await
- .expect("flush");
-
let search = fixture
- .search_all(seeds::names::TOPIC)
+ .wait_for_documents(seeds::names::TOPIC, 2)
.await
.expect("search");
diff --git a/core/integration/tests/connectors/random/random_source.rs b/core/integration/tests/connectors/random/random_source.rs
index e0bbfe348..32309cf52 100644
--- a/core/integration/tests/connectors/random/random_source.rs
+++ b/core/integration/tests/connectors/random/random_source.rs
@@ -31,7 +31,7 @@ use tokio::time::sleep;
async fn random_source_produces_messages(harness: &TestHarness) {
sleep(Duration::from_secs(1)).await;
- let client = harness.client();
+ let client = harness.root_client().await.unwrap();
let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
let consumer_id: Identifier = "test_consumer".try_into().unwrap();
diff --git a/core/integration/tests/data_integrity/verify_after_server_restart.rs b/core/integration/tests/data_integrity/verify_after_server_restart.rs
index 655fc8246..865153a19 100644
--- a/core/integration/tests/data_integrity/verify_after_server_restart.rs
+++ b/core/integration/tests/data_integrity/verify_after_server_restart.rs
@@ -18,10 +18,7 @@
use iggy::prelude::*;
use integration::bench_utils::run_bench_and_wait_for_finish;
-use integration::{
- tcp_client::TcpClientFactory,
- test_server::{ClientFactory, IpAddrKind, SYSTEM_PATH_ENV_VAR, TestServer, login_root},
-};
+use integration::harness::{TestHarness, TestServerConfig};
use serial_test::parallel;
use std::{collections::HashMap, str::FromStr};
use test_case::test_matrix;
@@ -38,6 +35,15 @@ fn cache_none() -> &'static str {
"none"
}
+fn build_server_config(cache_setting: &str) -> TestServerConfig {
+ let mut extra_envs = HashMap::new();
+ extra_envs.insert(
+ "IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string(),
+ cache_setting.to_string(),
+ );
+ TestServerConfig::builder().extra_envs(extra_envs).build()
+}
+
// TODO(numminex) - Move the message generation method from benchmark run to a special method.
#[test_matrix(
[cache_all(), cache_open_segment(), cache_none()]
@@ -45,24 +51,16 @@ fn cache_none() -> &'static str {
#[tokio::test]
#[parallel]
async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str) {
- // 1. Start server with cache configuration
- let env_vars = HashMap::from([
- (
- SYSTEM_PATH_ENV_VAR.to_owned(),
- TestServer::get_random_path(),
- ),
- (
- "IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string(),
- cache_setting.to_string(),
- ),
- ]);
-
- let mut test_server = TestServer::new(Some(env_vars.clone()), false, None, IpAddrKind::V4);
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let local_data_path = test_server.get_local_data_path().to_owned();
-
- // 2. Run send bench to fill 5 MB of data
+ let mut harness = TestHarness::builder()
+ .server(build_server_config(cache_setting))
+ .build()
+ .unwrap();
+
+ harness.start().await.unwrap();
+
+ let server_addr = harness.server().raw_tcp_addr().unwrap();
+
+ // Run send bench to fill 5 MB of data
let amount_of_data_to_process = IggyByteSize::from_str("5 MB").unwrap();
run_bench_and_wait_for_finish(
&server_addr,
@@ -71,7 +69,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
amount_of_data_to_process,
);
- // 3. Run poll bench to check if everything's OK
+ // Run poll bench to check if everything's OK
run_bench_and_wait_for_finish(
&server_addr,
&TransportProtocol::Tcp,
@@ -79,16 +77,8 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
amount_of_data_to_process,
);
- // 4. Connect and login to newly started server
- let client = TcpClientFactory {
- server_addr,
- ..Default::default()
- }
- .create_client()
- .await;
-
- let client = IggyClient::create(client, None, None);
- login_root(&client).await;
+ // Connect and login to server
+ let client = harness.tcp_root_client().await.unwrap();
let topic_id = Identifier::numeric(0).unwrap();
for i in 0..7 {
@@ -99,7 +89,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
.unwrap();
}
- // 4b. Create consumer groups to test persistence
+ // Create consumer groups to test persistence
let consumer_group_names = ["test-cg-1", "test-cg-2", "test-cg-3"];
for (idx, cg_name) in consumer_group_names.iter().enumerate() {
let stream_id = Identifier::numeric(idx as u32).unwrap();
@@ -109,7 +99,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
.unwrap();
}
- // 5. Save stats from the first server
+ // Save stats from the first server
let stats = client.get_stats().await.unwrap();
let expected_messages_size_bytes = stats.messages_size_bytes;
let expected_streams_count = stats.streams_count;
@@ -120,27 +110,12 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
let expected_clients_count = stats.clients_count;
let expected_consumer_groups_count = stats.consumer_groups_count;
- // 6. Stop server
- test_server.stop();
- drop(test_server);
-
- // 7. Restart server with same settings
- let mut test_server = TestServer::new(Some(env_vars.clone()), false, None, IpAddrKind::V4);
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
+ // Restart server
+ harness.restart_server().await.unwrap();
+ let server_addr = harness.server().raw_tcp_addr().unwrap();
- // 8. Verify stats are preserved after restart (before adding more data)
- let client_after_restart = IggyClient::create(
- TcpClientFactory {
- server_addr: server_addr.clone(),
- ..Default::default()
- }
- .create_client()
- .await,
- None,
- None,
- );
- login_root(&client_after_restart).await;
+ // Verify stats are preserved after restart (before adding more data)
+ let client_after_restart = harness.tcp_root_client().await.unwrap();
let stats_after_restart = client_after_restart.get_stats().await.unwrap();
assert_eq!(
@@ -176,7 +151,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
"Consumer groups count should be preserved after restart"
);
- // 8b. Verify consumer groups exist after restart
+ // Verify consumer groups exist after restart
for (idx, cg_name) in consumer_group_names.iter().enumerate() {
let stream_id = Identifier::numeric(idx as u32).unwrap();
let consumer_group = client_after_restart
@@ -199,7 +174,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
);
}
- // 9. Run send bench again to add more data
+ // Run send bench again to add more data
run_bench_and_wait_for_finish(
&server_addr,
&TransportProtocol::Tcp,
@@ -207,7 +182,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
amount_of_data_to_process,
);
- // 10. Run poll bench again to check if all data is still there
+ // Run poll bench again to check if all data is still there
run_bench_and_wait_for_finish(
&server_addr,
&TransportProtocol::Tcp,
@@ -216,20 +191,10 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
);
drop(client_after_restart);
- // 11. Connect and login to newly started server
- let client = IggyClient::create(
- TcpClientFactory {
- server_addr: server_addr.clone(),
- ..Default::default()
- }
- .create_client()
- .await,
- None,
- None,
- );
- login_root(&client).await;
+ // Connect and login to server
+ let client = harness.tcp_root_client().await.unwrap();
- // 12. Flush unsaved buffer
+ // Flush unsaved buffer
let topic_id = Identifier::numeric(0).unwrap();
for i in 0..7 {
let stream_id = Identifier::numeric(i).unwrap();
@@ -239,7 +204,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
.unwrap();
}
- // 13. Save stats from the second server (should have double the data)
+ // Save stats from the second server (should have double the data)
let stats = client.get_stats().await.unwrap();
let actual_messages_size_bytes = stats.messages_size_bytes;
let actual_streams_count = stats.streams_count;
@@ -250,7 +215,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
let actual_clients_count = stats.clients_count;
let actual_consumer_groups_count = stats.consumer_groups_count;
- // 14. Compare stats (expecting double the messages/size after second bench run)
+ // Compare stats (expecting double the messages/size after second bench run)
assert_eq!(
expected_messages_size_bytes.as_bytes_usize() * 2,
actual_messages_size_bytes.as_bytes_usize(),
@@ -283,7 +248,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
"Consumer groups count"
);
- // 14b. Verify consumer groups still exist after second benchmark run
+ // Verify consumer groups still exist after second benchmark run
for (idx, cg_name) in consumer_group_names.iter().enumerate() {
let stream_id = Identifier::numeric(idx as u32).unwrap();
let consumer_group = client
@@ -306,16 +271,13 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
);
}
- // 15. Run poll bench to check if all data (10MB total) is still there
+ // Run poll bench to check if all data (10MB total) is still there
run_bench_and_wait_for_finish(
&server_addr,
&TransportProtocol::Tcp,
"pinned-consumer",
IggyByteSize::from(amount_of_data_to_process.as_bytes_u64() * 2),
);
-
- // 16. Manual cleanup
- std::fs::remove_dir_all(local_data_path).unwrap();
}
/// Test that verifies server correctly handles ID gaps after deletions at all levels:
@@ -333,27 +295,14 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
#[tokio::test]
#[parallel]
async fn should_handle_resource_deletion_and_restart() {
- let env_vars = HashMap::from([(
- SYSTEM_PATH_ENV_VAR.to_owned(),
- TestServer::get_random_path(),
- )]);
-
- let mut test_server = TestServer::new(Some(env_vars.clone()), false, None, IpAddrKind::V4);
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let local_data_path = test_server.get_local_data_path().to_owned();
-
- let client = IggyClient::create(
- TcpClientFactory {
- server_addr: server_addr.clone(),
- ..Default::default()
- }
- .create_client()
- .await,
- None,
- None,
- );
- login_root(&client).await;
+ let mut harness = TestHarness::builder()
+ .server(TestServerConfig::default())
+ .build()
+ .unwrap();
+
+ harness.start().await.unwrap();
+
+ let client = harness.tcp_root_client().await.unwrap();
// Create 3 streams
let stream_0 = client.create_stream("stream-0").await.unwrap();
@@ -479,25 +428,11 @@ async fn should_handle_resource_deletion_and_restart() {
);
drop(client);
- test_server.stop();
- drop(test_server);
// Restart server
- let mut test_server = TestServer::new(Some(env_vars.clone()), false, None, IpAddrKind::V4);
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
-
- let client = IggyClient::create(
- TcpClientFactory {
- server_addr,
- ..Default::default()
- }
- .create_client()
- .await,
- None,
- None,
- );
- login_root(&client).await;
+ harness.restart_server().await.unwrap();
+
+ let client = harness.tcp_root_client().await.unwrap();
// Verify streams after restart - should have 3 streams: 0, 1 (reused), 2
let streams = client.get_streams().await.unwrap();
@@ -611,8 +546,4 @@ async fn should_handle_resource_deletion_and_restart() {
cg_ids
);
}
-
- drop(client);
- test_server.stop();
- std::fs::remove_dir_all(local_data_path).unwrap();
}
diff --git a/core/integration/tests/mod.rs b/core/integration/tests/mod.rs
index 3c84c0f84..38058104d 100644
--- a/core/integration/tests/mod.rs
+++ b/core/integration/tests/mod.rs
@@ -20,11 +20,10 @@ use ctor::{ctor, dtor};
use integration::harness::get_test_directory;
use lazy_static::lazy_static;
use std::collections::{HashMap, HashSet};
-use std::fs::File;
+use std::fs::{File, OpenOptions};
use std::io::Write;
-use std::sync::RwLock;
use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, Once};
+use std::sync::{Arc, Mutex, Once, RwLock};
use std::{panic, thread};
use tracing_subscriber::fmt::MakeWriter;
use tracing_subscriber::layer::SubscriberExt;
@@ -47,6 +46,9 @@ lazy_static! {
Arc::new(RwLock::new(HashMap::new()));
static ref FAILED_TEST_CASES: Arc<RwLock<HashSet<String>>> =
Arc::new(RwLock::new(HashSet::new()));
+ /// File handles for real-time log writing, keyed by test name.
+ static ref LOG_FILES: Arc<Mutex<HashMap<String, File>>> =
+ Arc::new(Mutex::new(HashMap::new()));
}
static INIT: Once = Once::new();
@@ -105,12 +107,15 @@ impl Write for LogWriter {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let thread = thread::current();
let thread_name = thread.name().unwrap_or(UNKNOWN_TEST_NAME);
- let Ok(mut map) = self.0.write() else {
- return Ok(buf.len());
- };
- let buffer = map.entry(thread_name.to_string()).or_default();
- buffer.extend_from_slice(buf);
+ // Write to memory buffer (for failed test output at end)
+ if let Ok(mut map) = self.0.write() {
+ let buffer = map.entry(thread_name.to_string()).or_default();
+ buffer.extend_from_slice(buf);
+ }
+
+ // Write to file in real-time
+ write_to_log_file(thread_name, buf);
Ok(buf.len())
}
@@ -120,25 +125,44 @@ impl Write for LogWriter {
}
}
+fn write_to_log_file(test_name: &str, buf: &[u8]) {
+ let Ok(mut files) = LOG_FILES.lock() else {
+ return;
+ };
+
+ // Try to get existing file handle or create one
+ if let std::collections::hash_map::Entry::Vacant(entry) = files.entry(test_name.to_string())
+ && let Some(dir) = get_test_directory(test_name)
+ {
+ let log_path = dir.join("test_stdout.log");
+ if let Ok(file) = OpenOptions::new().create(true).append(true).open(&log_path) {
+ entry.insert(file);
+ }
+ }
+
+ // Write to file if we have a handle
+ if let Some(file) = files.get_mut(test_name) {
+ let _ = file.write_all(buf);
+ }
+}
+
fn teardown() {
- if let Ok(buffer) = LOGS_BUFFER.read() {
- for (test_name, logs) in buffer.iter() {
- if let Some(dir) = get_test_directory(test_name) {
- let log_path = dir.join("test_stdout.log");
- if let Ok(mut file) = File::create(&log_path) {
- let _ = file.write_all(logs);
- }
- }
+ // Flush and close all log file handles
+ if let Ok(mut files) = LOG_FILES.lock() {
+ for (_, file) in files.iter_mut() {
+ let _ = file.flush();
}
+ files.clear();
+ }
- if TESTS_FAILED.load(Ordering::SeqCst)
- && let Ok(failed) = FAILED_TEST_CASES.read()
- {
- for test in failed.iter() {
- if let Some(logs) = buffer.get(test) {
- eprintln!("Logs for failed test '{test}':");
- eprintln!("{}", String::from_utf8_lossy(logs));
- }
+ // Print logs for failed tests to stderr
+ if TESTS_FAILED.load(Ordering::SeqCst)
+ && let (Ok(buffer), Ok(failed)) = (LOGS_BUFFER.read(), FAILED_TEST_CASES.read())
+ {
+ for test in failed.iter() {
+ if let Some(logs) = buffer.get(test) {
+ eprintln!("Logs for failed test '{test}':");
+ eprintln!("{}", String::from_utf8_lossy(logs));
}
}
}
diff --git a/core/integration/tests/sdk/producer/background.rs b/core/integration/tests/sdk/producer/background.rs
index 8acb05de2..8129bdc84 100644
--- a/core/integration/tests/sdk/producer/background.rs
+++ b/core/integration/tests/sdk/producer/background.rs
@@ -22,36 +22,18 @@ use crate::sdk::producer::{
use bytes::Bytes;
use iggy::clients::producer_config::BackpressureMode;
use iggy::prelude::*;
-use iggy::{clients::client::IggyClient, prelude::TcpClient};
-use iggy_common::TcpClientConfig;
-use integration::test_server::{TestServer, login_root};
-use serial_test::parallel;
+use integration::iggy_harness;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::{Instant, sleep};
-#[tokio::test]
-#[parallel]
-async fn background_send_receive_ok() {
- let mut test_server = TestServer::default();
- test_server.start();
+#[iggy_harness]
+async fn background_send_receive_ok(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
+ client.ping().await.unwrap();
- let tcp_client_config = TcpClientConfig {
- server_address: test_server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = IggyClient::create(client, None, None);
-
- client.connect().await.unwrap();
- assert!(client.ping().await.is_ok(), "Failed to ping server");
-
- login_root(&client).await;
init_system(&client).await;
- client.connect().await.unwrap();
- assert!(client.ping().await.is_ok(), "Failed to ping server");
-
let messages_count = 1000;
let mut messages = Vec::new();
@@ -96,28 +78,13 @@ async fn background_send_receive_ok() {
cleanup(&client).await;
}
-#[tokio::test]
-#[parallel]
-async fn background_buffer_overflow_immediate() {
- let mut test_server = TestServer::default();
- test_server.start();
-
- let tcp_client_config = TcpClientConfig {
- server_address: test_server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = IggyClient::create(client, None, None);
+#[iggy_harness]
+async fn background_buffer_overflow_immediate(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
+ client.ping().await.unwrap();
- client.connect().await.unwrap();
- assert!(client.ping().await.is_ok(), "Failed to ping server");
-
- login_root(&client).await;
init_system(&client).await;
- client.connect().await.unwrap();
- assert!(client.ping().await.is_ok(), "Failed to ping server");
-
let cfg = BackgroundConfig::builder()
.max_buffer_size(IggyByteSize::from(1024))
.failure_mode(BackpressureMode::FailImmediately)
@@ -147,27 +114,12 @@ async fn background_buffer_overflow_immediate() {
/// - batch_length = 0, batch_size = 0: disables trigger-based flushing;
/// - linger_time = 500ms: messages are flushed only after timeout;
/// - backpressure timeout = 100ms: sending should fail if not flushed within this window.
-#[tokio::test]
-#[parallel]
-async fn background_block_with_timeout() {
- let mut test_server = TestServer::default();
- test_server.start();
-
- let tcp_client_config = TcpClientConfig {
- server_address: test_server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = IggyClient::create(client, None, None);
-
- client.connect().await.unwrap();
- assert!(client.ping().await.is_ok(), "Failed to ping server");
-
- login_root(&client).await;
- init_system(&client).await;
+#[iggy_harness]
+async fn background_block_with_timeout(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
+ client.ping().await.unwrap();
- client.connect().await.unwrap();
- assert!(client.ping().await.is_ok(), "Failed to ping server");
+ init_system(&client).await;
let big = IggyMessage::builder()
.id(1)
@@ -205,21 +157,9 @@ async fn background_block_with_timeout() {
cleanup(&client).await;
}
-#[tokio::test]
-#[parallel]
-async fn background_block_waits_then_succeeds() {
- let mut test_server = TestServer::default();
- test_server.start();
-
- let tcp_client_config = TcpClientConfig {
- server_address: test_server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = IggyClient::create(client, None, None);
-
- client.connect().await.unwrap();
- login_root(&client).await;
+#[iggy_harness]
+async fn background_block_waits_then_succeeds(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
init_system(&client).await;
let big_msg = IggyMessage::builder()
@@ -261,33 +201,18 @@ async fn background_block_waits_then_succeeds() {
cleanup(&client).await;
}
-#[tokio::test]
-#[parallel]
-async fn background_graceful_shutdown() {
- let mut test_server = TestServer::default();
- test_server.start();
-
- let tcp_client_config = TcpClientConfig {
- server_address: test_server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = IggyClient::create(client, None, None);
-
- client.connect().await.unwrap();
- assert!(client.ping().await.is_ok(), "Failed to ping server");
+#[iggy_harness]
+async fn background_graceful_shutdown(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
+ client.ping().await.unwrap();
- login_root(&client).await;
init_system(&client).await;
- client.connect().await.unwrap();
- assert!(client.ping().await.is_ok(), "Failed to ping server");
-
let cfg = BackgroundConfig::builder()
.max_in_flight(1)
.batch_length(0)
.batch_size(0)
- .linger_time(IggyDuration::from(2_000_000)) // 2s – long enough not to flush automatically
+ .linger_time(IggyDuration::from(2_000_000)) // 2s - long enough not to flush automatically
.build();
let producer = client
.producer(STREAM_NAME, TOPIC_NAME)
@@ -337,23 +262,13 @@ async fn background_graceful_shutdown() {
cleanup(&client).await;
}
-#[tokio::test]
-#[parallel]
-async fn background_many_parallel_producers() {
+#[iggy_harness]
+async fn background_many_parallel_producers(harness: &TestHarness) {
const PARALLEL_PRODUCERS: usize = 10;
- let mut test_server = TestServer::default();
- test_server.start();
-
- let tcp_client_config = TcpClientConfig {
- server_address: test_server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = Arc::new(IggyClient::create(client, None, None));
+ let client = harness.tcp_root_client().await.unwrap();
+ let client = Arc::new(client);
- client.connect().await.unwrap();
- login_root(&client).await;
init_system(&client).await;
let mut handles = Vec::with_capacity(PARALLEL_PRODUCERS);
@@ -412,23 +327,11 @@ async fn background_many_parallel_producers() {
/// timeout fired with an empty buffer. Without the fix, the deadline would
/// always be in the past after the first idle timeout, causing immediate
/// flushes instead of waiting for the full linger_time.
-#[tokio::test]
-#[parallel]
-async fn background_linger_time_respected_after_idle() {
- let mut test_server = TestServer::default();
- test_server.start();
-
- let tcp_client_config = TcpClientConfig {
- server_address: test_server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = IggyClient::create(client, None, None);
-
- client.connect().await.unwrap();
- assert!(client.ping().await.is_ok(), "Failed to ping server");
-
- login_root(&client).await;
+#[iggy_harness]
+async fn background_linger_time_respected_after_idle(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
+ client.ping().await.unwrap();
+
init_system(&client).await;
let cfg = BackgroundConfig::builder()
@@ -494,21 +397,9 @@ async fn background_linger_time_respected_after_idle() {
cleanup(&client).await;
}
-#[tokio::test]
-#[parallel]
-async fn background_preserves_message_ordering() {
- let mut test_server = TestServer::default();
- test_server.start();
-
- let tcp_client_config = TcpClientConfig {
- server_address: test_server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = IggyClient::create(client, None, None);
-
- client.connect().await.unwrap();
- login_root(&client).await;
+#[iggy_harness]
+async fn background_preserves_message_ordering(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
init_system(&client).await;
let messages_count = 1000u32;
@@ -565,21 +456,9 @@ async fn background_preserves_message_ordering() {
cleanup(&client).await;
}
-#[tokio::test]
-#[parallel]
-async fn background_preserves_ordering_with_multiple_shards() {
- let mut test_server = TestServer::default();
- test_server.start();
-
- let tcp_client_config = TcpClientConfig {
- server_address: test_server.get_raw_tcp_addr().unwrap(),
- ..TcpClientConfig::default()
- };
- let client = ClientWrapper::Tcp(TcpClient::create(Arc::new(tcp_client_config)).unwrap());
- let client = IggyClient::create(client, None, None);
-
- client.connect().await.unwrap();
- login_root(&client).await;
+#[iggy_harness]
+async fn background_preserves_ordering_with_multiple_shards(harness: &TestHarness) {
+ let client = harness.tcp_root_client().await.unwrap();
init_system(&client).await;
let messages_count = 1000u32;
diff --git a/core/integration/tests/server/cg.rs b/core/integration/tests/server/cg.rs
index bf12c871e..dbb85f98c 100644
--- a/core/integration/tests/server/cg.rs
+++ b/core/integration/tests/server/cg.rs
@@ -15,36 +15,53 @@
// specific language governing permissions and limitations
// under the License.
-use crate::server::{
- ScenarioFn, auto_commit_reconnection_scenario, join_scenario, multiple_clients_scenario,
- offset_cleanup_scenario, run_scenario, single_client_scenario,
+use crate::server::scenarios::{
+ consumer_group_auto_commit_reconnection_scenario, consumer_group_join_scenario,
+ consumer_group_offset_cleanup_scenario,
+ consumer_group_with_multiple_clients_polling_messages_scenario,
+ consumer_group_with_single_client_polling_messages_scenario,
};
-use iggy_common::TransportProtocol;
-use serial_test::parallel;
-use test_case::test_matrix;
+use integration::iggy_harness;
-fn tcp() -> TransportProtocol {
- TransportProtocol::Tcp
+// Consumer group scenarios do not support HTTP (stateful operations).
+// TODO: Add QUIC support.
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn join(harness: &TestHarness) {
+ consumer_group_join_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn single_client(harness: &TestHarness) {
+ consumer_group_with_single_client_polling_messages_scenario::run(harness).await;
}
-fn websocket() -> TransportProtocol {
- TransportProtocol::WebSocket
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn multiple_clients(harness: &TestHarness) {
+ consumer_group_with_multiple_clients_polling_messages_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn auto_commit_reconnection(harness: &TestHarness) {
+ consumer_group_auto_commit_reconnection_scenario::run(harness).await;
}
-// TODO: Add `QUIC`.
-// Consumer group scenarios do not support HTTP
-#[test_matrix(
- [tcp(), websocket()],
- [
- join_scenario(),
- single_client_scenario(),
- multiple_clients_scenario(),
- auto_commit_reconnection_scenario(),
- offset_cleanup_scenario(),
- ]
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
)]
-#[tokio::test]
-#[parallel]
-async fn matrix(transport: TransportProtocol, scenario: ScenarioFn) {
- run_scenario(transport, scenario).await;
+async fn offset_cleanup(harness: &TestHarness) {
+ consumer_group_offset_cleanup_scenario::run(harness).await;
}
diff --git a/core/integration/tests/server/concurrent_addition.rs b/core/integration/tests/server/concurrent_addition.rs
index 25259f702..c4393999c 100644
--- a/core/integration/tests/server/concurrent_addition.rs
+++ b/core/integration/tests/server/concurrent_addition.rs
@@ -20,11 +20,7 @@ use crate::server::scenarios::concurrent_scenario::{
self, ResourceType, ScenarioType, barrier_off, barrier_on,
};
use iggy_common::TransportProtocol;
-use integration::{
- http_client::HttpClientFactory, quic_client::QuicClientFactory, tcp_client::TcpClientFactory,
- test_server::TestServer, websocket_client::WebSocketClientFactory,
-};
-use serial_test::parallel;
+use integration::iggy_harness;
use test_case::test_matrix;
// Test matrix for race condition scenarios
@@ -38,61 +34,24 @@ use test_case::test_matrix;
// TODO: Websocket fails for the `cold` type, cold means that we are creating resources with the same name.
// It fails with the error assertion, instead of `AlreadyExist`, we get generic `Error`.
+#[iggy_harness(server(
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+))]
#[test_matrix(
[tcp(), http(), quic(), websocket()],
[user(), stream(), topic(), partition(), consumer_group()],
[hot(), cold()],
[barrier_on(), barrier_off()]
)]
-#[tokio::test]
-#[parallel]
async fn matrix(
+ harness: &TestHarness,
transport: TransportProtocol,
resource_type: ResourceType,
path_type: ScenarioType,
use_barrier: bool,
) {
- // TODO: Need to do this, in order to avoid timeouts from QUIC connections during tests.
- let mut extra_envs = std::collections::HashMap::new();
- extra_envs.insert("IGGY_QUIC_MAX_IDLE_TIMEOUT".to_string(), "500s".to_string());
- extra_envs.insert(
- "IGGY_QUIC_KEEP_ALIVE_INTERVAL".to_string(),
- "15s".to_string(),
- );
- let mut test_server = TestServer::new(
- Some(extra_envs),
- true,
- None,
- integration::test_server::IpAddrKind::V4,
- );
- test_server.start();
-
- let client_factory: Box<dyn integration::test_server::ClientFactory> = match transport {
- TransportProtocol::Tcp => {
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- Box::new(TcpClientFactory {
- server_addr,
- ..Default::default()
- })
- }
- TransportProtocol::Quic => {
- let server_addr = test_server.get_quic_udp_addr().unwrap();
- Box::new(QuicClientFactory { server_addr })
- }
- TransportProtocol::Http => {
- let server_addr = test_server.get_http_api_addr().unwrap();
- Box::new(HttpClientFactory { server_addr })
- }
- TransportProtocol::WebSocket => {
- let server_addr = test_server.get_websocket_addr().unwrap();
- Box::new(WebSocketClientFactory {
- server_addr,
- ..Default::default()
- })
- }
- };
-
- concurrent_scenario::run(&*client_factory, resource_type, path_type, use_barrier).await;
+ concurrent_scenario::run(harness, transport, resource_type, path_type, use_barrier).await;
}
fn tcp() -> TransportProtocol {
diff --git a/core/integration/tests/server/general.rs b/core/integration/tests/server/general.rs
index ec88d6ffc..13b7c1dfa 100644
--- a/core/integration/tests/server/general.rs
+++ b/core/integration/tests/server/general.rs
@@ -15,48 +15,139 @@
// specific language governing permissions and limitations
// under the License.
-use crate::server::{
- ScenarioFn, authentication_scenario, bench_scenario, consumer_timestamp_polling_scenario,
- create_message_payload_scenario, message_headers_scenario, permissions_scenario, run_scenario,
- snapshot_scenario, stream_size_validation_scenario, system_scenario, user_scenario,
+use crate::server::scenarios::{
+ authentication_scenario, bench_scenario, consumer_timestamp_polling_scenario,
+ create_message_payload, message_headers_scenario, permissions_scenario, snapshot_scenario,
+ stream_size_validation_scenario, system_scenario, user_scenario,
};
-use iggy_common::TransportProtocol;
-use serial_test::parallel;
-use test_case::test_matrix;
+use integration::iggy_harness;
-#[test_matrix(
- [quic(), tcp(), http(), websocket()],
- [
- authentication_scenario(),
- system_scenario(),
- user_scenario(),
- permissions_scenario(),
- message_headers_scenario(),
- create_message_payload_scenario(),
- stream_size_validation_scenario(),
- bench_scenario(),
- consumer_timestamp_polling_scenario(),
- snapshot_scenario(),
- ]
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
)]
-#[tokio::test]
-#[parallel]
-async fn matrix(transport: TransportProtocol, scenario: ScenarioFn) {
- run_scenario(transport, scenario).await;
+async fn authentication(harness: &TestHarness) {
+ authentication_scenario::run(harness).await;
}
-fn quic() -> TransportProtocol {
- TransportProtocol::Quic
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
+)]
+async fn system(harness: &TestHarness) {
+ system_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
+)]
+async fn user(harness: &TestHarness) {
+ user_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
+)]
+async fn permissions(harness: &TestHarness) {
+ permissions_scenario::run(harness).await;
}
-fn tcp() -> TransportProtocol {
- TransportProtocol::Tcp
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
+)]
+async fn message_headers(harness: &TestHarness) {
+ message_headers_scenario::run(harness).await;
}
-fn http() -> TransportProtocol {
- TransportProtocol::Http
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
+)]
+async fn create_message_payload_scenario(harness: &TestHarness) {
+ create_message_payload::run(harness).await;
}
-fn websocket() -> TransportProtocol {
- TransportProtocol::WebSocket
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
+)]
+async fn stream_size_validation(harness: &TestHarness) {
+ stream_size_validation_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
+)]
+async fn bench(harness: &TestHarness) {
+ bench_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
+)]
+async fn consumer_timestamp_polling(harness: &TestHarness) {
+ consumer_timestamp_polling_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp, Http, Quic, WebSocket],
+ server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true,
+ quic.max_idle_timeout = "500s",
+ quic.keep_alive_interval = "15s"
+ )
+)]
+async fn snapshot(harness: &TestHarness) {
+ snapshot_scenario::run(harness).await;
}
diff --git a/core/integration/tests/server/message_retrieval.rs b/core/integration/tests/server/message_retrieval.rs
index 24a4ff0d0..00798f38f 100644
--- a/core/integration/tests/server/message_retrieval.rs
+++ b/core/integration/tests/server/message_retrieval.rs
@@ -17,10 +17,7 @@
*/
use crate::server::scenarios::{offset_scenario, timestamp_scenario};
-use integration::{
- tcp_client::TcpClientFactory,
- test_server::{IpAddrKind, TestServer},
-};
+use integration::harness::{TestHarness, TestServerConfig};
use serial_test::parallel;
use std::collections::HashMap;
use test_case::test_matrix;
@@ -49,34 +46,27 @@ fn cache_open_segment() -> &'static str {
"open_segment"
}
-fn msgs_req_32() -> u32 {
- 32
+fn msgs_req_32() -> &'static str {
+ "32"
}
-fn msgs_req_64() -> u32 {
- 64
+fn msgs_req_64() -> &'static str {
+ "64"
}
-fn msgs_req_1024() -> u32 {
- 1024
+fn msgs_req_1024() -> &'static str {
+ "1024"
}
-fn msgs_req_9984() -> u32 {
- 9984
+fn msgs_req_9984() -> &'static str {
+ "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(
+fn build_server_config(
segment_size: &str,
cache_indexes: &str,
- messages_required_to_save: u32,
-) {
+ messages_required_to_save: &str,
+) -> TestServerConfig {
let mut extra_envs = HashMap::new();
extra_envs.insert(
"IGGY_SYSTEM_SEGMENT_SIZE".to_string(),
@@ -96,16 +86,33 @@ async fn get_by_offset_scenario(
);
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()
- };
+ TestServerConfig::builder().extra_envs(extra_envs).build()
+}
- 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_offset_scenario(
+ segment_size: &str,
+ cache_indexes: &str,
+ messages_required_to_save: &str,
+) {
+ let mut harness = TestHarness::builder()
+ .server(build_server_config(
+ segment_size,
+ cache_indexes,
+ messages_required_to_save,
+ ))
+ .build()
+ .unwrap();
+
+ harness.start().await.unwrap();
+
+ offset_scenario::run(&harness).await;
}
#[test_matrix(
@@ -118,35 +125,18 @@ async fn get_by_offset_scenario(
async fn get_by_timestamp_scenario(
segment_size: &str,
cache_indexes: &str,
- messages_required_to_save: u32,
+ messages_required_to_save: &str,
) {
- 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;
+ let mut harness = TestHarness::builder()
+ .server(build_server_config(
+ segment_size,
+ cache_indexes,
+ messages_required_to_save,
+ ))
+ .build()
+ .unwrap();
+
+ harness.start().await.unwrap();
+
+ timestamp_scenario::run(&harness).await;
}
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 376746436..f3b7a584a 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -23,134 +23,3 @@ mod message_cleanup;
mod message_retrieval;
mod scenarios;
mod specific;
-
-use iggy_common::TransportProtocol;
-use integration::{
- http_client::HttpClientFactory,
- quic_client::QuicClientFactory,
- tcp_client::TcpClientFactory,
- test_server::{ClientFactory, IpAddrKind, TestServer},
- websocket_client::WebSocketClientFactory,
-};
-use scenarios::{
- authentication_scenario, bench_scenario, consumer_group_auto_commit_reconnection_scenario,
- consumer_group_join_scenario, consumer_group_offset_cleanup_scenario,
- consumer_group_with_multiple_clients_polling_messages_scenario,
- consumer_group_with_single_client_polling_messages_scenario,
- consumer_timestamp_polling_scenario, create_message_payload, message_headers_scenario,
- permissions_scenario, snapshot_scenario, stream_size_validation_scenario, system_scenario,
- user_scenario,
-};
-use std::pin::Pin;
-use std::{collections::HashMap, future::Future};
-
-type ScenarioFn = fn(&dyn ClientFactory) -> Pin<Box<dyn Future<Output = ()> + '_>>;
-
-fn authentication_scenario() -> ScenarioFn {
- |factory| Box::pin(authentication_scenario::run(factory))
-}
-
-fn system_scenario() -> ScenarioFn {
- |factory| Box::pin(system_scenario::run(factory))
-}
-
-fn user_scenario() -> ScenarioFn {
- |factory| Box::pin(user_scenario::run(factory))
-}
-
-fn message_headers_scenario() -> ScenarioFn {
- |factory| Box::pin(message_headers_scenario::run(factory))
-}
-
-fn create_message_payload_scenario() -> ScenarioFn {
- |factory| Box::pin(create_message_payload::run(factory))
-}
-
-fn join_scenario() -> ScenarioFn {
- |factory| Box::pin(consumer_group_join_scenario::run(factory))
-}
-
-fn stream_size_validation_scenario() -> ScenarioFn {
- |factory| Box::pin(stream_size_validation_scenario::run(factory))
-}
-
-fn single_client_scenario() -> ScenarioFn {
- |factory| Box::pin(consumer_group_with_single_client_polling_messages_scenario::run(factory))
-}
-
-fn multiple_clients_scenario() -> ScenarioFn {
- |factory| Box::pin(consumer_group_with_multiple_clients_polling_messages_scenario::run(factory))
-}
-
-fn auto_commit_reconnection_scenario() -> ScenarioFn {
- |factory| {
- Box::pin(consumer_group_auto_commit_reconnection_scenario::run(
- factory,
- ))
- }
-}
-
-fn offset_cleanup_scenario() -> ScenarioFn {
- |factory| Box::pin(consumer_group_offset_cleanup_scenario::run(factory))
-}
-
-fn bench_scenario() -> ScenarioFn {
- |factory| Box::pin(bench_scenario::run(factory))
-}
-
-fn permissions_scenario() -> ScenarioFn {
- |factory| Box::pin(permissions_scenario::run(factory))
-}
-
-fn snapshot_scenario() -> ScenarioFn {
- |factory| Box::pin(snapshot_scenario::run(factory))
-}
-
-fn consumer_timestamp_polling_scenario() -> ScenarioFn {
- |factory| Box::pin(consumer_timestamp_polling_scenario::run(factory))
-}
-
-async fn run_scenario(transport: TransportProtocol, scenario: ScenarioFn) {
- // TODO: Need to enable `TCP_NODELAY` flag for TCP transports, due to small messages being used in the test.
- // For some reason TCP in compio can't deal with it, but in tokio it works fine.
- 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());
- extra_envs.insert("IGGY_QUIC_MAX_IDLE_TIMEOUT".to_string(), "500s".to_string());
- extra_envs.insert(
- "IGGY_QUIC_KEEP_ALIVE_INTERVAL".to_string(),
- "15s".to_string(),
- );
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
-
- let client_factory: Box<dyn ClientFactory> = match transport {
- TransportProtocol::Tcp => {
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- Box::new(TcpClientFactory {
- server_addr,
- ..Default::default()
- })
- }
- TransportProtocol::Quic => {
- let server_addr = test_server.get_quic_udp_addr().unwrap();
- Box::new(QuicClientFactory { server_addr })
- }
- TransportProtocol::Http => {
- let server_addr = test_server.get_http_api_addr().unwrap();
- Box::new(HttpClientFactory { server_addr })
- }
- TransportProtocol::WebSocket => {
- let server_addr = test_server.get_websocket_addr().unwrap();
- Box::new(WebSocketClientFactory {
- server_addr,
- ..Default::default()
- })
- }
- };
-
- scenario(&*client_factory).await;
-}
diff --git a/core/integration/tests/server/scenarios/authentication_scenario.rs b/core/integration/tests/server/scenarios/authentication_scenario.rs
index f793e49a4..4dc440c4f 100644
--- a/core/integration/tests/server/scenarios/authentication_scenario.rs
+++ b/core/integration/tests/server/scenarios/authentication_scenario.rs
@@ -29,7 +29,7 @@
use crate::server::scenarios::create_client;
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::harness::{TestHarness, login_root};
use server::binary::command::ServerCommand;
use strum::IntoEnumIterator;
@@ -57,8 +57,8 @@ impl TestContext {
}
}
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = create_client(client_factory).await;
+pub async fn run(harness: &TestHarness) {
+ let client = create_client(harness).await;
// Phase 1: Verify ping works without auth
client.ping().await.expect("ping should work without auth");
@@ -67,7 +67,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
test_all_commands_require_auth(&client).await;
// Phase 3: Login and verify commands work
- let identity = login_root(&client).await;
+ let identity = login_root(&client).await.expect("login failed");
assert_eq!(identity.user_id, 0, "root user should have id 0");
setup_test_resources(&client).await;
verify_auth_works(&client).await;
@@ -77,7 +77,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
test_all_commands_require_auth(&client).await;
// Phase 5: Test PAT authentication
- login_root(&client).await;
+ login_root(&client).await.expect("login failed");
let raw_pat = client
.create_personal_access_token("auth-test-pat", PersonalAccessTokenExpiry::NeverExpire)
.await
diff --git a/core/integration/tests/server/scenarios/bench_scenario.rs b/core/integration/tests/server/scenarios/bench_scenario.rs
index 990253f2b..3f282f699 100644
--- a/core/integration/tests/server/scenarios/bench_scenario.rs
+++ b/core/integration/tests/server/scenarios/bench_scenario.rs
@@ -17,11 +17,28 @@
*/
use iggy::prelude::*;
-use integration::{bench_utils::run_bench_and_wait_for_finish, test_server::ClientFactory};
+use iggy_common::TransportProtocol;
+use integration::bench_utils::run_bench_and_wait_for_finish;
+use integration::harness::TestHarness;
-pub async fn run(client_factory: &dyn ClientFactory) {
- let server_addr = client_factory.server_addr();
- let transport = client_factory.transport();
+pub async fn run(harness: &TestHarness) {
+ let transport = harness.transport().expect("Transport not set");
+ let server = harness.server();
+ let server_addr = match transport {
+ TransportProtocol::Tcp => server.raw_tcp_addr().expect("TCP address not available"),
+ TransportProtocol::Http => server
+ .http_addr()
+ .expect("HTTP address not available")
+ .to_string(),
+ TransportProtocol::Quic => server
+ .quic_addr()
+ .expect("QUIC address not available")
+ .to_string(),
+ TransportProtocol::WebSocket => server
+ .websocket_addr()
+ .expect("WebSocket address not available")
+ .to_string(),
+ };
let data_size = IggyByteSize::from(8 * 1024 * 1024);
run_bench_and_wait_for_finish(&server_addr, &transport, "pinned-producer", data_size);
diff --git a/core/integration/tests/server/scenarios/concurrent_scenario.rs b/core/integration/tests/server/scenarios/concurrent_scenario.rs
index d42296ba2..b6c86bd4c 100644
--- a/core/integration/tests/server/scenarios/concurrent_scenario.rs
+++ b/core/integration/tests/server/scenarios/concurrent_scenario.rs
@@ -16,11 +16,10 @@
* under the License.
*/
-use crate::server::scenarios::create_client;
use futures::future::join_all;
use iggy::prelude::*;
-use iggy_common::{ConsumerGroup, UserInfo};
-use integration::test_server::{ClientFactory, login_root};
+use iggy_common::{ConsumerGroup, TransportProtocol, UserInfo};
+use integration::harness::TestHarness;
use std::sync::Arc;
use tokio::sync::Barrier;
@@ -55,14 +54,24 @@ pub fn barrier_off() -> bool {
type OperationResult = Result<(), IggyError>;
+async fn create_client_for_transport(
+ harness: &TestHarness,
+ transport: TransportProtocol,
+) -> IggyClient {
+ harness
+ .root_client_for(transport)
+ .await
+ .expect("Failed to create client")
+}
+
pub async fn run(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
resource_type: ResourceType,
scenario_type: ScenarioType,
use_barrier: bool,
) {
- let root_client = create_client(client_factory).await;
- login_root(&root_client).await;
+ let root_client = create_client_for_transport(harness, transport).await;
// For topic tests, create parent stream first
if resource_type == ResourceType::Topic {
@@ -89,53 +98,53 @@ pub async fn run(
let results = match (resource_type, scenario_type) {
(ResourceType::User, ScenarioType::Hot) => {
- execute_users_hot(client_factory, use_barrier).await
+ execute_users_hot(harness, transport, use_barrier).await
}
(ResourceType::User, ScenarioType::Cold) => {
- execute_users_cold(client_factory, use_barrier).await
+ execute_users_cold(harness, transport, use_barrier).await
}
(ResourceType::Stream, ScenarioType::Hot) => {
- execute_streams_hot(client_factory, use_barrier).await
+ execute_streams_hot(harness, transport, use_barrier).await
}
(ResourceType::Stream, ScenarioType::Cold) => {
- execute_streams_cold(client_factory, use_barrier).await
+ execute_streams_cold(harness, transport, use_barrier).await
}
(ResourceType::Topic, ScenarioType::Hot) => {
- execute_topics_hot(client_factory, use_barrier).await
+ execute_topics_hot(harness, transport, use_barrier).await
}
(ResourceType::Topic, ScenarioType::Cold) => {
- execute_topics_cold(client_factory, use_barrier).await
+ execute_topics_cold(harness, transport, use_barrier).await
}
(ResourceType::Partition, ScenarioType::Hot) => {
- execute_partitions_hot(client_factory, use_barrier).await
+ execute_partitions_hot(harness, transport, use_barrier).await
}
// Partitions don't have names, so Cold scenario doesn't apply
(ResourceType::Partition, ScenarioType::Cold) => vec![],
(ResourceType::ConsumerGroup, ScenarioType::Hot) => {
- execute_consumer_groups_hot(client_factory, use_barrier).await
+ execute_consumer_groups_hot(harness, transport, use_barrier).await
}
(ResourceType::ConsumerGroup, ScenarioType::Cold) => {
- execute_consumer_groups_cold(client_factory, use_barrier).await
+ execute_consumer_groups_cold(harness, transport, use_barrier).await
}
};
if !results.is_empty() {
validate_results(&results, scenario_type);
- validate_server_state(client_factory, resource_type, scenario_type).await;
+ validate_server_state(harness, transport, resource_type, scenario_type).await;
cleanup_resources(&root_client, resource_type).await;
}
}
async fn execute_users_hot(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
use_barrier: bool,
) -> Vec<OperationResult> {
let barrier = use_barrier.then(|| Arc::new(Barrier::new(CONCURRENT_CLIENTS)));
let mut handles = Vec::with_capacity(CONCURRENT_CLIENTS);
for client_id in 0..CONCURRENT_CLIENTS {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let barrier = barrier.clone();
handles.push(tokio::spawn(async move {
@@ -158,7 +167,8 @@ async fn execute_users_hot(
}
async fn execute_users_cold(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
use_barrier: bool,
) -> Vec<OperationResult> {
const DUPLICATE_USER: &str = "race-user-duplicate";
@@ -166,8 +176,7 @@ async fn execute_users_cold(
let mut handles = Vec::with_capacity(CONCURRENT_CLIENTS);
for _ in 0..CONCURRENT_CLIENTS {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let barrier = barrier.clone();
handles.push(tokio::spawn(async move {
@@ -189,15 +198,15 @@ async fn execute_users_cold(
}
async fn execute_streams_hot(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
use_barrier: bool,
) -> Vec<OperationResult> {
let barrier = use_barrier.then(|| Arc::new(Barrier::new(CONCURRENT_CLIENTS)));
let mut handles = Vec::with_capacity(CONCURRENT_CLIENTS);
for client_id in 0..CONCURRENT_CLIENTS {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let barrier = barrier.clone();
handles.push(tokio::spawn(async move {
@@ -217,7 +226,8 @@ async fn execute_streams_hot(
}
async fn execute_streams_cold(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
use_barrier: bool,
) -> Vec<OperationResult> {
const DUPLICATE_STREAM: &str = "race-stream-duplicate";
@@ -225,8 +235,7 @@ async fn execute_streams_cold(
let mut handles = Vec::with_capacity(CONCURRENT_CLIENTS);
for _ in 0..CONCURRENT_CLIENTS {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let barrier = barrier.clone();
handles.push(tokio::spawn(async move {
@@ -245,15 +254,15 @@ async fn execute_streams_cold(
}
async fn execute_topics_hot(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
use_barrier: bool,
) -> Vec<OperationResult> {
let barrier = use_barrier.then(|| Arc::new(Barrier::new(CONCURRENT_CLIENTS)));
let mut handles = Vec::with_capacity(CONCURRENT_CLIENTS);
for client_id in 0..CONCURRENT_CLIENTS {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let barrier = barrier.clone();
handles.push(tokio::spawn(async move {
@@ -285,7 +294,8 @@ async fn execute_topics_hot(
}
async fn execute_topics_cold(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
use_barrier: bool,
) -> Vec<OperationResult> {
const DUPLICATE_TOPIC: &str = "race-topic-duplicate";
@@ -293,8 +303,7 @@ async fn execute_topics_cold(
let mut handles = Vec::with_capacity(CONCURRENT_CLIENTS);
for _ in 0..CONCURRENT_CLIENTS {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let barrier = barrier.clone();
handles.push(tokio::spawn(async move {
@@ -325,15 +334,15 @@ async fn execute_topics_cold(
}
async fn execute_partitions_hot(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
use_barrier: bool,
) -> Vec<OperationResult> {
let barrier = use_barrier.then(|| Arc::new(Barrier::new(CONCURRENT_CLIENTS)));
let mut handles = Vec::with_capacity(CONCURRENT_CLIENTS);
for _ in 0..CONCURRENT_CLIENTS {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let barrier = barrier.clone();
handles.push(tokio::spawn(async move {
@@ -357,15 +366,15 @@ async fn execute_partitions_hot(
}
async fn execute_consumer_groups_hot(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
use_barrier: bool,
) -> Vec<OperationResult> {
let barrier = use_barrier.then(|| Arc::new(Barrier::new(CONCURRENT_CLIENTS)));
let mut handles = Vec::with_capacity(CONCURRENT_CLIENTS);
for client_id in 0..CONCURRENT_CLIENTS {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let barrier = barrier.clone();
handles.push(tokio::spawn(async move {
@@ -390,7 +399,8 @@ async fn execute_consumer_groups_hot(
}
async fn execute_consumer_groups_cold(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
use_barrier: bool,
) -> Vec<OperationResult> {
const DUPLICATE_CONSUMER_GROUP: &str = "race-consumer-group-duplicate";
@@ -398,8 +408,7 @@ async fn execute_consumer_groups_cold(
let mut handles = Vec::with_capacity(CONCURRENT_CLIENTS);
for _ in 0..CONCURRENT_CLIENTS {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let barrier = barrier.clone();
handles.push(tokio::spawn(async move {
@@ -479,7 +488,8 @@ fn validate_results(results: &[OperationResult], scenario_type: ScenarioType) {
}
async fn validate_server_state(
- client_factory: &dyn ClientFactory,
+ harness: &TestHarness,
+ transport: TransportProtocol,
resource_type: ResourceType,
scenario_type: ScenarioType,
) {
@@ -488,8 +498,7 @@ async fn validate_server_state(
// Create multiple clients for validation
for _ in 0..VALIDATION_CLIENT_COUNT {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
clients.push(client);
}
diff --git a/core/integration/tests/server/scenarios/consumer_group_auto_commit_reconnection_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_auto_commit_reconnection_scenario.rs
index 485f02e35..24e2b810d 100644
--- a/core/integration/tests/server/scenarios/consumer_group_auto_commit_reconnection_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_auto_commit_reconnection_scenario.rs
@@ -16,22 +16,22 @@
* under the License.
*/
-use crate::server::scenarios::{
- CONSUMER_GROUP_NAME, PARTITION_ID, STREAM_NAME, TOPIC_NAME, create_client,
-};
+use crate::server::scenarios::{CONSUMER_GROUP_NAME, PARTITION_ID, STREAM_NAME, TOPIC_NAME};
use futures::StreamExt;
use iggy::prelude::*;
use iggy_common::ConsumerOffsetInfo;
-use integration::test_server::{ClientFactory, login_root};
+use integration::harness::TestHarness;
use std::str::FromStr;
use tokio::time::{Duration, sleep};
const TEST_MESSAGES_COUNT: u32 = 100;
const HALF_MESSAGES_COUNT: u32 = TEST_MESSAGES_COUNT / 2;
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
init_system(&client).await;
execute_auto_commit_reconnection_scenario(&client).await;
}
diff --git a/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
index eebf4e489..2175c7478 100644
--- a/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_join_scenario.rs
@@ -28,18 +28,17 @@ use iggy::prelude::Identifier;
use iggy::prelude::IggyExpiry;
use iggy::prelude::MaxTopicSize;
use iggy::prelude::{ConsumerGroupClient, StreamClient, SystemClient, TopicClient};
-use integration::test_server::{
- ClientFactory, assert_clean_system, create_user, login_root, login_user,
-};
-
-pub async fn run(client_factory: &dyn ClientFactory) {
- let system_client = create_client(client_factory).await;
+use integration::harness::{TestHarness, assert_clean_system, create_user, login_user};
- let client1 = create_client(client_factory).await;
- let client2 = create_client(client_factory).await;
- let client3 = create_client(client_factory).await;
+pub async fn run(harness: &TestHarness) {
+ let system_client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
- login_root(&system_client).await;
+ let client1 = create_client(harness).await;
+ let client2 = create_client(harness).await;
+ let client3 = create_client(harness).await;
// 1. Create the stream
let stream = system_client.create_stream(STREAM_NAME).await.unwrap();
diff --git a/core/integration/tests/server/scenarios/consumer_group_offset_cleanup_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_offset_cleanup_scenario.rs
index dad706b60..5097e3b57 100644
--- a/core/integration/tests/server/scenarios/consumer_group_offset_cleanup_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_offset_cleanup_scenario.rs
@@ -16,11 +16,11 @@
* under the License.
*/
-use crate::server::scenarios::{PARTITION_ID, STREAM_NAME, TOPIC_NAME, cleanup, create_client};
+use crate::server::scenarios::{PARTITION_ID, STREAM_NAME, TOPIC_NAME, cleanup};
use futures::StreamExt;
use iggy::prelude::*;
use iggy_common::ConsumerOffsetInfo;
-use integration::test_server::{ClientFactory, assert_clean_system, login_root};
+use integration::harness::{TestHarness, assert_clean_system};
use std::str::FromStr;
use tokio::time::{Duration, sleep};
@@ -30,9 +30,11 @@ const CONSUMER_GROUP_1: &str = "test-consumer-group-1";
const CONSUMER_GROUP_2: &str = "test-consumer-group-2";
const CONSUMER_GROUP_3: &str = "test-consumer-group-3";
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
init_system(&client).await;
execute_consumer_group_offset_cleanup_scenario(&client).await;
cleanup(&client, false).await;
diff --git a/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
index 076ecae25..223c2d5a8 100644
--- a/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_with_multiple_clients_polling_messages_scenario.rs
@@ -21,17 +21,17 @@ use crate::server::scenarios::{
create_client, get_consumer_group, join_consumer_group,
};
use iggy::prelude::*;
-use integration::test_server::{
- ClientFactory, assert_clean_system, create_user, login_root, login_user,
-};
+use integration::harness::{TestHarness, assert_clean_system, create_user, login_user};
use std::str::{FromStr, from_utf8};
-pub async fn run(client_factory: &dyn ClientFactory) {
- let system_client = create_client(client_factory).await;
- let client1 = create_client(client_factory).await;
- let client2 = create_client(client_factory).await;
- let client3 = create_client(client_factory).await;
- login_root(&system_client).await;
+pub async fn run(harness: &TestHarness) {
+ let system_client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
+ let client1 = create_client(harness).await;
+ let client2 = create_client(harness).await;
+ let client3 = create_client(harness).await;
init_system(&system_client, &client1, &client2, &client3, true).await;
execute_using_messages_key_key(&system_client, &client1, &client2, &client3).await;
cleanup(&system_client, false).await;
diff --git a/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
index 166267390..d8622f896 100644
--- a/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
@@ -18,16 +18,18 @@
use crate::server::scenarios::{
CONSUMER_GROUP_NAME, MESSAGES_COUNT, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME, cleanup,
- create_client, get_consumer_group, join_consumer_group,
+ get_consumer_group, join_consumer_group,
};
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, assert_clean_system, login_root};
+use integration::harness::{TestHarness, assert_clean_system};
use std::collections::HashSet;
use std::str::{FromStr, from_utf8};
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
init_system(&client).await;
execute_using_messages_key_key(&client).await;
client
diff --git a/core/integration/tests/server/scenarios/consumer_timestamp_polling_scenario.rs b/core/integration/tests/server/scenarios/consumer_timestamp_polling_scenario.rs
index 4beb9d2cf..1247e4ad8 100644
--- a/core/integration/tests/server/scenarios/consumer_timestamp_polling_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_timestamp_polling_scenario.rs
@@ -16,10 +16,10 @@
* under the License.
*/
-use crate::server::scenarios::{PARTITION_ID, STREAM_NAME, TOPIC_NAME, create_client};
+use crate::server::scenarios::{PARTITION_ID, STREAM_NAME, TOPIC_NAME};
use futures::StreamExt;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::harness::TestHarness;
use std::str::FromStr;
use tokio::time::{Duration, timeout};
@@ -27,9 +27,11 @@ const MESSAGES_COUNT: u32 = 2000;
const BATCH_LENGTH: u32 = 100;
const CONSUME_TIMEOUT: Duration = Duration::from_secs(10);
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
test_offset_strategy(&client).await;
test_timestamp_strategy(&client).await;
diff --git a/core/integration/tests/server/scenarios/create_message_payload.rs b/core/integration/tests/server/scenarios/create_message_payload.rs
index 1add0b0f7..f44a67333 100644
--- a/core/integration/tests/server/scenarios/create_message_payload.rs
+++ b/core/integration/tests/server/scenarios/create_message_payload.rs
@@ -18,7 +18,7 @@
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, assert_clean_system, login_root};
+use integration::harness::{TestHarness, assert_clean_system};
use std::collections::HashMap;
const STREAM_NAME: &str = "test-stream";
@@ -27,11 +27,11 @@ const PARTITIONS_COUNT: u32 = 3;
const MESSAGES_COUNT: u32 = 1000;
const PARTITION_ID: u32 = 1;
-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;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
init_system(&client).await;
// 1. Send messages with the included headers
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..bedfcfe67 100644
--- a/core/integration/tests/server/scenarios/cross_protocol_pat_scenario.rs
+++ b/core/integration/tests/server/scenarios/cross_protocol_pat_scenario.rs
@@ -21,40 +21,27 @@
//! 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 iggy_common::TransportProtocol;
+use integration::harness::TestHarness;
+use integration::iggy_harness;
const PAT_NAME: &str = "cross-protocol-test-pat";
const TCP_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");
+async fn create_root_client(harness: &TestHarness, transport: TransportProtocol) -> IggyClient {
+ harness
+ .root_client_for(transport)
+ .await
+ .expect("Failed to create root client")
+}
- let http_factory = HttpClientFactory {
- server_addr: http_addr,
- };
- let http_client = create_client(&http_factory).await;
- login_root(&http_client).await;
+/// Create PAT via HTTP, list via multiple TCP connections.
+#[iggy_harness(server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true
+))]
+pub async fn should_see_pat_created_via_http_when_listing_via_tcp(harness: &TestHarness) {
+ let http_client = create_root_client(harness, TransportProtocol::Http).await;
let created_pat = http_client
.create_personal_access_token(PAT_NAME, IggyExpiry::NeverExpire)
@@ -70,16 +57,10 @@ pub async fn run() {
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_client = create_root_client(harness, TransportProtocol::Tcp).await;
let tcp_pats = tcp_client
.get_personal_access_tokens()
@@ -114,30 +95,12 @@ pub async fn run() {
}
/// 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;
+#[iggy_harness(server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true
+))]
+pub async fn should_see_pat_created_via_tcp_when_listing_via_http(harness: &TestHarness) {
+ let tcp_client = create_root_client(harness, TransportProtocol::Tcp).await;
let created_pat = tcp_client
.create_personal_access_token(PAT_NAME, IggyExpiry::NeverExpire)
@@ -146,11 +109,7 @@ pub async fn run_tcp_to_http() {
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_client = create_root_client(harness, TransportProtocol::Http).await;
let http_pats = http_client
.get_personal_access_tokens()
@@ -164,41 +123,19 @@ pub async fn run_tcp_to_http() {
}
/// 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;
+#[iggy_harness(server(
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true
+))]
+pub async fn should_not_see_pat_deleted_via_http_when_listing_via_tcp(harness: &TestHarness) {
+ let tcp_client = create_root_client(harness, TransportProtocol::Tcp).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;
+ let http_client = create_root_client(harness, TransportProtocol::Http).await;
http_client
.delete_personal_access_token(PAT_NAME)
@@ -207,8 +144,7 @@ pub async fn run_delete_visibility() {
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;
+ let tcp_check_client = create_root_client(harness, TransportProtocol::Tcp).await;
let tcp_pats = tcp_check_client
.get_personal_access_tokens()
@@ -226,8 +162,3 @@ pub async fn run_delete_visibility() {
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/encryption_scenario.rs b/core/integration/tests/server/scenarios/encryption_scenario.rs
index 49f608173..f03741c44 100644
--- a/core/integration/tests/server/scenarios/encryption_scenario.rs
+++ b/core/integration/tests/server/scenarios/encryption_scenario.rs
@@ -18,10 +18,7 @@
use bytes::Bytes;
use iggy::prelude::*;
-use integration::{
- tcp_client::TcpClientFactory,
- test_server::{ClientFactory, IpAddrKind, SYSTEM_PATH_ENV_VAR, TestServer, login_root},
-};
+use integration::harness::{TestHarness, TestServerConfig};
use serial_test::parallel;
use std::collections::HashMap;
use test_case::test_matrix;
@@ -34,45 +31,38 @@ fn encryption_disabled() -> bool {
false
}
-#[test_matrix(
- [encryption_enabled(), encryption_disabled()]
-)]
-#[tokio::test]
-#[parallel]
-async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryption: bool) {
- // 1. Start server
- let mut env_vars = HashMap::from([(
- SYSTEM_PATH_ENV_VAR.to_owned(),
- TestServer::get_random_path(),
- )]);
+fn build_server_config(encryption: bool) -> TestServerConfig {
+ let mut extra_envs = HashMap::new();
if encryption {
- env_vars.insert(
+ extra_envs.insert(
"IGGY_SYSTEM_ENCRYPTION_ENABLED".to_string(),
"true".to_string(),
);
- env_vars.insert(
+ extra_envs.insert(
"IGGY_SYSTEM_ENCRYPTION_KEY".to_string(),
"/rvT1xP4V8u1EAhk4xDdqzqM2UOPXyy9XYkl4uRShgE=".to_string(),
);
}
- let mut test_server = TestServer::new(Some(env_vars.clone()), false, None, IpAddrKind::V4);
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let local_data_path = test_server.get_local_data_path().to_owned();
+ TestServerConfig::builder().extra_envs(extra_envs).build()
+}
- // 2. Connect and create initial client
- let client = TcpClientFactory {
- server_addr,
- ..Default::default()
- }
- .create_client()
- .await;
- let client = IggyClient::create(client, None, None);
- login_root(&client).await;
+#[test_matrix(
+ [encryption_enabled(), encryption_disabled()]
+)]
+#[tokio::test]
+#[parallel]
+async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryption: bool) {
+ let mut harness = TestHarness::builder()
+ .server(build_server_config(encryption))
+ .build()
+ .unwrap();
+
+ harness.start().await.unwrap();
+
+ let client = harness.tcp_root_client().await.unwrap();
- // 3. Create test stream and topic
let stream_name = "test-stream-api";
let topic_name = "test-topic-api";
let partition_count = 1;
@@ -91,7 +81,6 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
.await
.unwrap();
- // 4. Send messages with headers (first batch)
let messages_per_batch = 1000;
let mut messages_batch_1 = Vec::new();
@@ -127,25 +116,22 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
.await
.unwrap();
- // 5. Flush and get initial stats
client
.flush_unsaved_buffer(
&Identifier::named(stream_name).unwrap(),
&Identifier::named(topic_name).unwrap(),
0,
- true, // Force flush
+ true,
)
.await
.unwrap();
- // Give the server a moment to process encrypted messages
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
let initial_stats = client.get_stats().await.unwrap();
let initial_messages_count = initial_stats.messages_count;
let initial_messages_size = initial_stats.messages_size_bytes;
- // 6. Poll messages to verify initial batch
let consumer = Consumer::default();
let polled = client
.poll_messages(
@@ -162,7 +148,6 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
let all_polled_messages_1 = polled.messages;
- // Verify we got all messages with correct headers
eprintln!(
"Polled {} messages, expected {}",
all_polled_messages_1.len(),
@@ -210,25 +195,10 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
);
}
- // 7. Stop and restart server
- test_server.stop();
- drop(test_server);
-
- let mut test_server = TestServer::new(Some(env_vars.clone()), false, None, IpAddrKind::V4);
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
+ harness.restart_server().await.unwrap();
- // 8. Reconnect after restart
- let client = TcpClientFactory {
- server_addr,
- ..Default::default()
- }
- .create_client()
- .await;
- let client = IggyClient::create(client, None, None);
- login_root(&client).await;
+ let client = harness.tcp_root_client().await.unwrap();
- // 9. Send second batch of messages with different headers
let mut messages_batch_2 = Vec::new();
for i in 0..messages_per_batch {
@@ -257,27 +227,24 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
.send_messages(
&Identifier::named(stream_name).unwrap(),
&Identifier::named(topic_name).unwrap(),
- &Partitioning::partition_id(0), // Use specific partition for testing
+ &Partitioning::partition_id(0),
&mut messages_batch_2,
)
.await
.unwrap();
- // Flush the buffer after sending second batch
client
.flush_unsaved_buffer(
&Identifier::named(stream_name).unwrap(),
&Identifier::named(topic_name).unwrap(),
0,
- true, // Force flush
+ true,
)
.await
.unwrap();
- // Give the server a moment to process encrypted messages
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
- // 10. Poll all messages (both batches) and verify
let polled = client
.poll_messages(
&Identifier::named(stream_name).unwrap(),
@@ -293,10 +260,8 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
let all_polled_messages = polled.messages;
- // Verify we have all messages from both batches
assert_eq!(all_polled_messages.len(), (messages_per_batch * 2) as usize);
- // Count messages by batch
let mut batch_1_count = 0;
let mut batch_2_count = 0;
@@ -343,7 +308,6 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
assert_eq!(batch_1_count, messages_per_batch as usize);
assert_eq!(batch_2_count, messages_per_batch as usize);
- // 11. Verify final stats
let final_stats = client.get_stats().await.unwrap();
assert_eq!(final_stats.messages_count, initial_messages_count * 2);
assert!(
@@ -352,7 +316,4 @@ async fn should_fill_data_with_headers_and_verify_after_restart_using_api(encryp
final_stats.messages_size_bytes.as_bytes_u64(),
initial_messages_size.as_bytes_u64()
);
-
- // 12. Cleanup
- std::fs::remove_dir_all(local_data_path).unwrap();
}
diff --git a/core/integration/tests/server/scenarios/log_rotation_scenario.rs b/core/integration/tests/server/scenarios/log_rotation_scenario.rs
index 69ccccd41..a88c74817 100644
--- a/core/integration/tests/server/scenarios/log_rotation_scenario.rs
+++ b/core/integration/tests/server/scenarios/log_rotation_scenario.rs
@@ -22,8 +22,7 @@ use iggy::prelude::*;
use iggy_common::{
CompressionAlgorithm, Identifier, IggyByteSize, IggyDuration, IggyExpiry, MaxTopicSize,
};
-use integration::tcp_client::TcpClientFactory;
-use integration::test_server::{ClientFactory, IpAddrKind, TestServer, login_root};
+use integration::harness::{TestHarness, TestServerConfig};
use once_cell::sync::Lazy;
use serial_test::parallel;
use std::collections::HashMap;
@@ -91,50 +90,47 @@ fn config_special_scenario() -> LogRotationTestConfig {
}
}
-#[test_matrix(
- [config_regular_rotation(), config_unlimited_size(), config_unlimited_archives(), config_special_scenario()]
-)]
-#[tokio::test]
-#[parallel]
-async fn log_rotation_should_be_valid(present_log_config: LogRotationTestConfig) {
+fn build_server_config(log_config: &LogRotationTestConfig) -> TestServerConfig {
let mut extra_envs = HashMap::new();
extra_envs.insert(
"IGGY_SYSTEM_LOGGING_MAX_FILE_SIZE".to_string(),
- format!("{}", present_log_config.max_single_log_size),
+ format!("{}", log_config.max_single_log_size),
);
extra_envs.insert(
"IGGY_SYSTEM_LOGGING_MAX_TOTAL_SIZE".to_string(),
- format!("{}", present_log_config.max_total_log_size),
+ format!("{}", log_config.max_total_log_size),
);
extra_envs.insert(
"IGGY_SYSTEM_LOGGING_ROTATION_CHECK_INTERVAL".to_string(),
- format!("{}", present_log_config.rotation_check_interval),
+ format!("{}", log_config.rotation_check_interval),
);
extra_envs.insert(
"IGGY_SYSTEM_LOGGING_RETENTION".to_string(),
- format!("{}", present_log_config.retention),
+ format!("{}", log_config.retention),
);
- let mut test_server = TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4);
- test_server.start();
+ TestServerConfig::builder().extra_envs(extra_envs).build()
+}
+
+#[test_matrix(
+ [config_regular_rotation(), config_unlimited_size(), config_unlimited_archives(), config_special_scenario()]
+)]
+#[tokio::test]
+#[parallel]
+async fn log_rotation_should_be_valid(present_log_config: LogRotationTestConfig) {
+ let mut harness = TestHarness::builder()
+ .server(build_server_config(&present_log_config))
+ .build()
+ .unwrap();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
- let client_factory = TcpClientFactory {
- server_addr,
- ..Default::default()
- };
+ harness.start().await.unwrap();
- let log_dir = format!("{}/logs", test_server.get_local_data_path());
+ let log_dir = format!("{}/logs", harness.server().data_path().display());
- test_server.assert_running();
- run(&client_factory, &log_dir, present_log_config).await;
+ run(&harness, &log_dir, present_log_config).await;
}
-async fn run(
- client_factory: &dyn ClientFactory,
- log_dir: &str,
- present_log_config: LogRotationTestConfig,
-) {
+async fn run(harness: &TestHarness, log_dir: &str, present_log_config: LogRotationTestConfig) {
let done_status = false;
let present_log_test_title = present_log_config.name.clone();
let log_path = Path::new(log_dir);
@@ -143,7 +139,7 @@ async fn run(
"failed::no_such_directory => {log_dir}",
);
- let client = init_valid_client(client_factory).await;
+ let client = init_valid_client(harness).await;
assert!(
client.is_ok(),
"failed::client_initialize => {:?}",
@@ -170,24 +166,13 @@ async fn run(
nocapture_observer(log_path, &present_log_test_title, !done_status).await;
}
-async fn init_valid_client(client_factory: &dyn ClientFactory) -> Result<IggyClient, String> {
+async fn init_valid_client(harness: &TestHarness) -> Result<IggyClient, String> {
let operation_timeout = IggyDuration::new(Duration::from_secs(OPERATION_TIMEOUT_SECS));
- let client_wrapper = timeout(
- operation_timeout.get_duration(),
- client_factory.create_client(),
- )
- .await
- .map_err(|_| "ClientWrapper creation timed out")?;
-
- timeout(operation_timeout.get_duration(), client_wrapper.connect())
- .await
- .map_err(|_| "Client connection timed out")?
- .map_err(|e| format!("Client connection failed: {e:?}"))?;
- let client = IggyClient::create(client_wrapper, None, None);
- timeout(operation_timeout.get_duration(), login_root(&client))
+ let client = timeout(operation_timeout.get_duration(), harness.root_client())
.await
- .map_err(|e| format!("Root user login timed out: {e:?}"))?;
+ .map_err(|_| "Root client creation timed out")?
+ .map_err(|e| format!("Root client creation failed: {e:?}"))?;
Ok(client)
}
diff --git a/core/integration/tests/server/scenarios/message_headers_scenario.rs b/core/integration/tests/server/scenarios/message_headers_scenario.rs
index 1d8b1e190..714c56247 100644
--- a/core/integration/tests/server/scenarios/message_headers_scenario.rs
+++ b/core/integration/tests/server/scenarios/message_headers_scenario.rs
@@ -17,16 +17,18 @@
*/
use crate::server::scenarios::{
- MESSAGES_COUNT, PARTITION_ID, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME, cleanup, create_client,
+ MESSAGES_COUNT, PARTITION_ID, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME, cleanup,
};
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, assert_clean_system, login_root};
+use integration::harness::{TestHarness, assert_clean_system};
use std::collections::HashMap;
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = create_client(client_factory).await;
- login_root(&client).await;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
init_system(&client).await;
// 1. Send messages with the included headers
diff --git a/core/integration/tests/server/scenarios/message_size_scenario.rs b/core/integration/tests/server/scenarios/message_size_scenario.rs
index f63b0570b..14b240d6a 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::harness::{TestHarness, assert_clean_system};
use std::collections::HashMap;
const STREAM_NAME: &str = "test-stream";
@@ -33,11 +33,11 @@ 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);
-
- login_root(&client).await;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
init_system(&client).await;
// 3. Send message and check the result
diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs
index 90d45676f..65e635afb 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -48,7 +48,7 @@ pub mod user_scenario;
pub mod websocket_tls_scenario;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, delete_user};
+use integration::harness::{TestHarness, delete_user};
const PARTITION_ID: u32 = 0;
const STREAM_NAME: &str = "test-stream";
@@ -61,9 +61,11 @@ const USERNAME_3: &str = "user3";
const CONSUMER_KIND: ConsumerKind = ConsumerKind::Consumer;
const MESSAGES_COUNT: u32 = 1337;
-async fn create_client(client_factory: &dyn ClientFactory) -> IggyClient {
- let client = client_factory.create_client().await;
- IggyClient::create(client, None, None)
+async fn create_client(harness: &TestHarness) -> IggyClient {
+ harness
+ .new_client()
+ .await
+ .expect("Failed to create new client")
}
async fn get_consumer_group(client: &IggyClient) -> ConsumerGroupDetails {
diff --git a/core/integration/tests/server/scenarios/offset_scenario.rs b/core/integration/tests/server/scenarios/offset_scenario.rs
index 608b1b3fb..c1544137b 100644
--- a/core/integration/tests/server/scenarios/offset_scenario.rs
+++ b/core/integration/tests/server/scenarios/offset_scenario.rs
@@ -19,7 +19,7 @@
use super::PARTITION_ID;
use bytes::BytesMut;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::harness::TestHarness;
use std::collections::HashMap;
fn small_batches() -> Vec<u32> {
@@ -51,10 +51,11 @@ 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;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
for msg_size in all_message_sizes() {
for (pattern_name, batch_pattern) in all_batch_patterns() {
diff --git a/core/integration/tests/server/scenarios/permissions_scenario.rs b/core/integration/tests/server/scenarios/permissions_scenario.rs
index 786acfa84..c73f91ba1 100644
--- a/core/integration/tests/server/scenarios/permissions_scenario.rs
+++ b/core/integration/tests/server/scenarios/permissions_scenario.rs
@@ -30,7 +30,7 @@ use crate::server::scenarios::create_client;
use ahash::AHashMap;
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::harness::TestHarness;
const STREAM_1: &str = "perm-stream-1";
const STREAM_2: &str = "perm-stream-2";
@@ -38,35 +38,37 @@ const TOPIC_1: &str = "perm-topic-1";
const TOPIC_2: &str = "perm-topic-2";
const PARTITIONS: u32 = 1;
-pub async fn run(client_factory: &dyn ClientFactory) {
- let root_client = create_client(client_factory).await;
- login_root(&root_client).await;
+pub async fn run(harness: &TestHarness) {
+ let root_client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
setup_test_resources(&root_client).await;
// Test categories
- test_no_permissions(client_factory, &root_client).await;
- test_system_permissions(client_factory, &root_client).await;
- test_user_permissions(client_factory, &root_client).await;
- test_stream_permissions(client_factory, &root_client).await;
- test_topic_permissions(client_factory, &root_client).await;
- test_message_permissions(client_factory, &root_client).await;
- test_stream_specific_permissions(client_factory, &root_client).await;
- test_topic_specific_permissions(client_factory, &root_client).await;
+ test_no_permissions(harness, &root_client).await;
+ test_system_permissions(harness, &root_client).await;
+ test_user_permissions(harness, &root_client).await;
+ test_stream_permissions(harness, &root_client).await;
+ test_topic_permissions(harness, &root_client).await;
+ test_message_permissions(harness, &root_client).await;
+ test_stream_specific_permissions(harness, &root_client).await;
+ test_topic_specific_permissions(harness, &root_client).await;
// Permission inheritance/implication tests
- test_global_permission_inheritance(client_factory, &root_client).await;
- test_stream_permission_inheritance(client_factory, &root_client).await;
- test_topic_permission_inheritance(client_factory, &root_client).await;
+ test_global_permission_inheritance(harness, &root_client).await;
+ test_stream_permission_inheritance(harness, &root_client).await;
+ test_topic_permission_inheritance(harness, &root_client).await;
// Consumer group operations matrix
- test_consumer_group_operations(client_factory, &root_client).await;
+ test_consumer_group_operations(harness, &root_client).await;
// Union semantics tests
- test_union_semantics(client_factory, &root_client).await;
+ test_union_semantics(harness, &root_client).await;
// Missing resource behavior tests
- test_missing_resource_behavior(client_factory, &root_client).await;
+ test_missing_resource_behavior(harness, &root_client).await;
cleanup(&root_client).await;
}
@@ -134,8 +136,8 @@ async fn delete_test_user(root_client: &IggyClient, username: &str) {
.await;
}
-async fn login_user(client_factory: &dyn ClientFactory, username: &str) -> IggyClient {
- let client = create_client(client_factory).await;
+async fn login_user(harness: &TestHarness, username: &str) -> IggyClient {
+ let client = create_client(harness).await;
client
.login_user(username, "password123")
.await
@@ -154,10 +156,10 @@ fn no_permissions() -> Permissions {
// Test: User with no permissions
// =============================================================================
-async fn test_no_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+async fn test_no_permissions(harness: &TestHarness, root_client: &IggyClient) {
const USER: &str = "no-perms-user";
create_test_user(root_client, USER, Some(no_permissions())).await;
- let client = login_user(client_factory, USER).await;
+ let client = login_user(harness, USER).await;
let stream_id = Identifier::named(STREAM_1).unwrap();
let topic_id = Identifier::named(TOPIC_1).unwrap();
@@ -220,7 +222,7 @@ async fn test_no_permissions(client_factory: &dyn ClientFactory, root_client: &I
// Test: System permissions (read_servers, manage_servers)
// =============================================================================
-async fn test_system_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+async fn test_system_permissions(harness: &TestHarness, root_client: &IggyClient) {
// User with read_servers only
const READ_USER: &str = "read-servers-user";
create_test_user(
@@ -236,7 +238,7 @@ async fn test_system_permissions(client_factory: &dyn ClientFactory, root_client
)
.await;
- let client = login_user(client_factory, READ_USER).await;
+ let client = login_user(harness, READ_USER).await;
client
.get_stats()
.await
@@ -267,7 +269,7 @@ async fn test_system_permissions(client_factory: &dyn ClientFactory, root_client
)
.await;
- let client = login_user(client_factory, MANAGE_USER).await;
+ let client = login_user(harness, MANAGE_USER).await;
client
.get_stats()
.await
@@ -284,7 +286,7 @@ async fn test_system_permissions(client_factory: &dyn ClientFactory, root_client
// Test: User permissions (read_users, manage_users)
// =============================================================================
-async fn test_user_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+async fn test_user_permissions(harness: &TestHarness, root_client: &IggyClient) {
// User with read_users only
const READ_USER: &str = "read-users-user";
create_test_user(
@@ -300,7 +302,7 @@ async fn test_user_permissions(client_factory: &dyn ClientFactory, root_client:
)
.await;
- let client = login_user(client_factory, READ_USER).await;
+ let client = login_user(harness, READ_USER).await;
client
.get_users()
.await
@@ -346,7 +348,7 @@ async fn test_user_permissions(client_factory: &dyn ClientFactory, root_client:
)
.await;
- let client = login_user(client_factory, MANAGE_USER).await;
+ let client = login_user(harness, MANAGE_USER).await;
client
.get_users()
.await
@@ -380,7 +382,7 @@ async fn test_user_permissions(client_factory: &dyn ClientFactory, root_client:
// Test: Stream permissions (read_streams, manage_streams)
// =============================================================================
-async fn test_stream_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+async fn test_stream_permissions(harness: &TestHarness, root_client: &IggyClient) {
let stream_id = Identifier::named(STREAM_1).unwrap();
// User with read_streams only
@@ -398,7 +400,7 @@ async fn test_stream_permissions(client_factory: &dyn ClientFactory, root_client
)
.await;
- let client = login_user(client_factory, READ_USER).await;
+ let client = login_user(harness, READ_USER).await;
client
.get_streams()
.await
@@ -439,7 +441,7 @@ async fn test_stream_permissions(client_factory: &dyn ClientFactory, root_client
)
.await;
- let client = login_user(client_factory, MANAGE_USER).await;
+ let client = login_user(harness, MANAGE_USER).await;
client
.get_streams()
.await
@@ -479,7 +481,7 @@ async fn test_stream_permissions(client_factory: &dyn ClientFactory, root_client
// Test: Topic permissions (read_topics, manage_topics)
// =============================================================================
-async fn test_topic_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+async fn test_topic_permissions(harness: &TestHarness, root_client: &IggyClient) {
let stream_id = Identifier::named(STREAM_1).unwrap();
let topic_id = Identifier::named(TOPIC_1).unwrap();
@@ -498,7 +500,7 @@ async fn test_topic_permissions(client_factory: &dyn ClientFactory, root_client:
)
.await;
- let client = login_user(client_factory, READ_USER).await;
+ let client = login_user(harness, READ_USER).await;
client
.get_topics(&stream_id)
.await
@@ -561,7 +563,7 @@ async fn test_topic_permissions(client_factory: &dyn ClientFactory, root_client:
)
.await;
- let client = login_user(client_factory, MANAGE_USER).await;
+ let client = login_user(harness, MANAGE_USER).await;
// Create topic
client
@@ -610,7 +612,7 @@ async fn test_topic_permissions(client_factory: &dyn ClientFactory, root_client:
// Test: Message permissions (poll_messages, send_messages)
// =============================================================================
-async fn test_message_permissions(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+async fn test_message_permissions(harness: &TestHarness, root_client: &IggyClient) {
let stream_id = Identifier::named(STREAM_1).unwrap();
let topic_id = Identifier::named(TOPIC_1).unwrap();
@@ -629,7 +631,7 @@ async fn test_message_permissions(client_factory: &dyn ClientFactory, root_clien
)
.await;
- let client = login_user(client_factory, POLL_USER).await;
+ let client = login_user(harness, POLL_USER).await;
// Can poll messages
client
@@ -679,7 +681,7 @@ async fn test_message_permissions(client_factory: &dyn ClientFactory, root_clien
)
.await;
- let client = login_user(client_factory, SEND_USER).await;
+ let client = login_user(harness, SEND_USER).await;
// Can send messages
let mut msgs = test_messages();
@@ -727,7 +729,7 @@ async fn test_message_permissions(client_factory: &dyn ClientFactory, root_clien
)
.await;
- let client = login_user(client_factory, BOTH_USER).await;
+ let client = login_user(harness, BOTH_USER).await;
let mut msgs = test_messages();
client
@@ -760,10 +762,7 @@ async fn test_message_permissions(client_factory: &dyn ClientFactory, root_clien
// Test: Stream-specific permissions
// =============================================================================
-async fn test_stream_specific_permissions(
- client_factory: &dyn ClientFactory,
- root_client: &IggyClient,
-) {
+async fn test_stream_specific_permissions(harness: &TestHarness, root_client: &IggyClient) {
let stream1_id = Identifier::named(STREAM_1).unwrap();
let stream2_id = Identifier::named(STREAM_2).unwrap();
let topic_id = Identifier::named(TOPIC_1).unwrap();
@@ -797,7 +796,7 @@ async fn test_stream_specific_permissions(
)
.await;
- let client = login_user(client_factory, USER).await;
+ let client = login_user(harness, USER).await;
// Can access stream 1
client
@@ -889,10 +888,7 @@ async fn test_stream_specific_permissions(
// Test: Topic-specific permissions
// =============================================================================
-async fn test_topic_specific_permissions(
- client_factory: &dyn ClientFactory,
- root_client: &IggyClient,
-) {
+async fn test_topic_specific_permissions(harness: &TestHarness, root_client: &IggyClient) {
let stream_id = Identifier::named(STREAM_1).unwrap();
let topic1_id = Identifier::named(TOPIC_1).unwrap();
let topic2_id = Identifier::named(TOPIC_2).unwrap();
@@ -941,7 +937,7 @@ async fn test_topic_specific_permissions(
)
.await;
- let client = login_user(client_factory, USER).await;
+ let client = login_user(harness, USER).await;
// Can access topic 1
client
@@ -1022,10 +1018,7 @@ async fn test_topic_specific_permissions(
// setting the implied flags. This locks in the inheritance rules.
// =============================================================================
-async fn test_global_permission_inheritance(
- client_factory: &dyn ClientFactory,
- root_client: &IggyClient,
-) {
+async fn test_global_permission_inheritance(harness: &TestHarness, root_client: &IggyClient) {
let stream_id = Identifier::named(STREAM_1).unwrap();
let topic_id = Identifier::named(TOPIC_1).unwrap();
@@ -1046,7 +1039,7 @@ async fn test_global_permission_inheritance(
)
.await;
- let client = login_user(client_factory, READ_STREAMS_USER).await;
+ let client = login_user(harness, READ_STREAMS_USER).await;
// read_streams → read_topics
client
@@ -1131,7 +1124,7 @@ async fn test_global_permission_inheritance(
)
.await;
- let client = login_user(client_factory, MANAGE_STREAMS_USER).await;
+ let client = login_user(harness, MANAGE_STREAMS_USER).await;
// manage_streams → read_streams
client
@@ -1222,7 +1215,7 @@ async fn test_global_permission_inheritance(
)
.await;
- let client = login_user(client_factory, MANAGE_TOPICS_USER).await;
+ let client = login_user(harness, MANAGE_TOPICS_USER).await;
// manage_topics → read_topics
client
@@ -1278,7 +1271,7 @@ async fn test_global_permission_inheritance(
)
.await;
- let client = login_user(client_factory, READ_TOPICS_USER).await;
+ let client = login_user(harness, READ_TOPICS_USER).await;
// read_topics allows reading topics
client
@@ -1328,7 +1321,7 @@ async fn test_global_permission_inheritance(
)
.await;
- let client = login_user(client_factory, MANAGE_SERVERS_ONLY_USER).await;
+ let client = login_user(harness, MANAGE_SERVERS_ONLY_USER).await;
// read_servers implied
client
@@ -1358,7 +1351,7 @@ async fn test_global_permission_inheritance(
)
.await;
- let client = login_user(client_factory, MANAGE_USERS_ONLY_USER).await;
+ let client = login_user(harness, MANAGE_USERS_ONLY_USER).await;
// read_users implied
client
@@ -1378,10 +1371,7 @@ async fn test_global_permission_inheritance(
// Tests inheritance rules at the stream level without setting implied flags.
// =============================================================================
-async fn test_stream_permission_inheritance(
- client_factory: &dyn ClientFactory,
- root_client: &IggyClient,
-) {
+async fn test_stream_permission_inheritance(harness: &TestHarness, root_client: &IggyClient) {
let stream_id = Identifier::named(STREAM_1).unwrap();
let topic_id = Identifier::named(TOPIC_1).unwrap();
@@ -1413,7 +1403,7 @@ async fn test_stream_permission_inheritance(
)
.await;
- let client = login_user(client_factory, READ_STREAM_USER).await;
+ let client = login_user(harness, READ_STREAM_USER).await;
// read_stream allows get_stream
client
@@ -1503,7 +1493,7 @@ async fn test_stream_permission_inheritance(
)
.await;
- let client = login_user(client_factory, MANAGE_STREAM_USER).await;
+ let client = login_user(harness, MANAGE_STREAM_USER).await;
// manage_stream → read_stream
client
@@ -1609,7 +1599,7 @@ async fn test_stream_permission_inheritance(
)
.await;
- let client = login_user(client_factory, MANAGE_TOPICS_STREAM_USER).await;
+ let client = login_user(harness, MANAGE_TOPICS_STREAM_USER).await;
// manage_topics allows topic CRUD
client
@@ -1663,10 +1653,7 @@ async fn test_stream_permission_inheritance(
// Tests inheritance rules at the topic level without setting implied flags.
// =============================================================================
-async fn test_topic_permission_inheritance(
- client_factory: &dyn ClientFactory,
- root_client: &IggyClient,
-) {
+async fn test_topic_permission_inheritance(harness: &TestHarness, root_client: &IggyClient) {
let stream_id = Identifier::named(STREAM_1).unwrap();
let topic_id = Identifier::named(TOPIC_1).unwrap();
@@ -1713,7 +1700,7 @@ async fn test_topic_permission_inheritance(
)
.await;
- let client = login_user(client_factory, READ_TOPIC_USER).await;
+ let client = login_user(harness, READ_TOPIC_USER).await;
// get_topic should work
client
@@ -1782,7 +1769,7 @@ async fn test_topic_permission_inheritance(
)
.await;
- let client = login_user(client_factory, MANAGE_TOPIC_USER).await;
+ let client = login_user(harness, MANAGE_TOPIC_USER).await;
// read_topic implied
client
@@ -1819,10 +1806,7 @@ async fn test_topic_permission_inheritance(
// are denied under poll_messages-only.
// =============================================================================
-async fn test_consumer_group_operations(
- client_factory: &dyn ClientFactory,
- root_client: &IggyClient,
-) {
+async fn test_consumer_group_operations(harness: &TestHarness, root_client: &IggyClient) {
let stream_id = Identifier::named(STREAM_1).unwrap();
let topic_id = Identifier::named(TOPIC_1).unwrap();
@@ -1841,7 +1825,7 @@ async fn test_consumer_group_operations(
)
.await;
- let client = login_user(client_factory, POLL_ONLY_USER).await;
+ let client = login_user(harness, POLL_ONLY_USER).await;
// poll_messages should work
client
@@ -1892,7 +1876,7 @@ async fn test_consumer_group_operations(
)
.await;
- let client = login_user(client_factory, READ_TOPICS_CG_USER).await;
+ let client = login_user(harness, READ_TOPICS_CG_USER).await;
// All CG operations should work
client
@@ -1974,7 +1958,7 @@ async fn test_consumer_group_operations(
)
.await;
- let client = login_user(client_factory, READ_TOPIC_CG_USER).await;
+ let client = login_user(harness, READ_TOPIC_CG_USER).await;
// CG operations should work on topic 1
client
@@ -2023,7 +2007,7 @@ async fn test_consumer_group_operations(
// Tests that stream permissions extend rather than restrict global permissions.
// =============================================================================
-async fn test_union_semantics(client_factory: &dyn ClientFactory, root_client: &IggyClient) {
+async fn test_union_semantics(harness: &TestHarness, root_client: &IggyClient) {
let stream1_id = Identifier::named(STREAM_1).unwrap();
let stream2_id = Identifier::named(STREAM_2).unwrap();
let topic_id = Identifier::named(TOPIC_1).unwrap();
@@ -2067,7 +2051,7 @@ async fn test_union_semantics(client_factory: &dyn ClientFactory, root_client: &
)
.await;
- let client = login_user(client_factory, UNION_USER).await;
+ let client = login_user(harness, UNION_USER).await;
// Should STILL be able to send (global overrides scoped restriction)
let mut msgs = test_messages();
@@ -2116,7 +2100,7 @@ async fn test_union_semantics(client_factory: &dyn ClientFactory, root_client: &
)
.await;
- let client = login_user(client_factory, GLOBAL_READ_SCOPED_MAP_USER).await;
+ let client = login_user(harness, GLOBAL_READ_SCOPED_MAP_USER).await;
// Should be able to read all streams (global)
client
@@ -2185,7 +2169,7 @@ async fn test_union_semantics(client_factory: &dyn ClientFactory, root_client: &
)
.await;
- let client = login_user(client_factory, TOPIC_ADDITIVE_USER).await;
+ let client = login_user(harness, TOPIC_ADDITIVE_USER).await;
// Stream read operations should work
client
@@ -2259,10 +2243,7 @@ async fn test_union_semantics(client_factory: &dyn ClientFactory, root_client: &
// unauthorized users.
// =============================================================================
-async fn test_missing_resource_behavior(
- client_factory: &dyn ClientFactory,
- root_client: &IggyClient,
-) {
+async fn test_missing_resource_behavior(harness: &TestHarness, root_client: &IggyClient) {
let nonexistent_stream = Identifier::named("nonexistent-stream").unwrap();
let nonexistent_topic = Identifier::named("nonexistent-topic").unwrap();
let existing_stream = Identifier::named(STREAM_1).unwrap();
@@ -2286,7 +2267,7 @@ async fn test_missing_resource_behavior(
)
.await;
- let client = login_user(client_factory, AUTHORIZED_USER).await;
+ let client = login_user(harness, AUTHORIZED_USER).await;
// Non-existent stream
assert_not_found_or_related(
@@ -2338,7 +2319,7 @@ async fn test_missing_resource_behavior(
const UNAUTHORIZED_USER: &str = "missing-resource-unauthorized-user";
create_test_user(root_client, UNAUTHORIZED_USER, Some(no_permissions())).await;
- let client = login_user(client_factory, UNAUTHORIZED_USER).await;
+ let client = login_user(harness, UNAUTHORIZED_USER).await;
// Non-existent stream - returns empty (no leak of existence)
assert_unauthorized(
@@ -2412,7 +2393,7 @@ async fn test_missing_resource_behavior(
)
.await;
- let client = login_user(client_factory, SCOPED_USER).await;
+ let client = login_user(harness, SCOPED_USER).await;
// Accessing non-existent stream - returns empty (no leak of existence)
assert_unauthorized(
diff --git a/core/integration/tests/server/scenarios/read_during_persistence_scenario.rs b/core/integration/tests/server/scenarios/read_during_persistence_scenario.rs
index 5153337c2..fc8e976af 100644
--- a/core/integration/tests/server/scenarios/read_during_persistence_scenario.rs
+++ b/core/integration/tests/server/scenarios/read_during_persistence_scenario.rs
@@ -18,12 +18,7 @@
use bytes::Bytes;
use iggy::prelude::*;
-use integration::{
- tcp_client::TcpClientFactory,
- test_server::{ClientFactory, IpAddrKind, SYSTEM_PATH_ENV_VAR, TestServer, login_root},
-};
-use serial_test::parallel;
-use std::collections::HashMap;
+use integration::iggy_harness;
use std::time::{Duration, Instant};
const STREAM_NAME: &str = "eventual-consistency-stream";
@@ -40,42 +35,17 @@ const TOPIC_NAME: &str = "eventual-consistency-topic";
/// 2. Background saver calls commit() - journal becomes EMPTY
/// 3. Background saver starts async disk write
/// 4. Poll arrives - sees empty journal, data not yet on disk
-#[tokio::test]
-#[parallel]
-async fn should_read_messages_during_background_saver_flush() {
- let env_vars = HashMap::from([
- (
- SYSTEM_PATH_ENV_VAR.to_owned(),
- TestServer::get_random_path(),
- ),
- (
- "IGGY_SYSTEM_PARTITION_MESSAGES_REQUIRED_TO_SAVE".to_owned(),
- "100000".to_owned(),
- ),
- (
- "IGGY_SYSTEM_PARTITION_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE".to_owned(),
- "1GB".to_owned(),
- ),
- (
- "IGGY_SYSTEM_PARTITION_ENFORCE_FSYNC".to_owned(),
- "false".to_owned(),
- ),
- ("IGGY_MESSAGE_SAVER_INTERVAL".to_owned(), "100ms".to_owned()),
- ("IGGY_MESSAGE_SAVER_ENABLED".to_owned(), "true".to_owned()),
- ]);
-
- let mut test_server = TestServer::new(Some(env_vars), true, None, IpAddrKind::V4);
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
-
- let producer_client = TcpClientFactory {
- server_addr: server_addr.clone(),
- ..Default::default()
- }
- .create_client()
- .await;
- let producer = IggyClient::create(producer_client, None, None);
- login_root(&producer).await;
+#[iggy_harness(server(
+ partition.messages_required_to_save = "100000",
+ partition.size_of_messages_required_to_save = "1GB",
+ partition.enforce_fsync = false,
+ message_saver.interval = "100ms",
+ message_saver.enabled = true
+))]
+async fn should_read_messages_during_background_saver_flush(
+ harness: &integration::harness::TestHarness,
+) {
+ let producer = harness.tcp_root_client().await.unwrap();
producer.create_stream(STREAM_NAME).await.unwrap();
producer
@@ -91,14 +61,7 @@ async fn should_read_messages_during_background_saver_flush() {
.await
.unwrap();
- let consumer_client = TcpClientFactory {
- server_addr,
- ..Default::default()
- }
- .create_client()
- .await;
- let consumer = IggyClient::create(consumer_client, None, None);
- login_root(&consumer).await;
+ let consumer_client = harness.tcp_root_client().await.unwrap();
let stream_id = Identifier::named(STREAM_NAME).unwrap();
let topic_id = Identifier::named(TOPIC_NAME).unwrap();
@@ -147,7 +110,7 @@ async fn should_read_messages_during_background_saver_flush() {
batches_sent += 1;
- let poll_result = consumer
+ let poll_result = consumer_client
.poll_messages(
&stream_id,
&topic_id,
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 2630262ee..da35c3374 100644
--- a/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
+++ b/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
@@ -29,7 +29,8 @@
//! - Short message_saver interval to add more concurrent persist operations
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use iggy_common::TransportProtocol;
+use integration::harness::TestHarness;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
@@ -45,21 +46,29 @@ const MESSAGES_PER_BATCH: usize = 5;
const MAX_ALLOWED_MEMORY_BYTES: u64 = 200 * 1024 * 1024;
/// Runs the segment rotation race condition test with multiple protocols.
-/// Each client factory represents a different protocol (TCP, HTTP, QUIC, WebSocket).
+/// Uses all available transports from the harness (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;
+pub async fn run(harness: &TestHarness) {
+ let transports = [
+ TransportProtocol::Tcp,
+ TransportProtocol::Http,
+ TransportProtocol::Quic,
+ TransportProtocol::WebSocket,
+ ];
+
+ let total_producers = transports.len() * PRODUCERS_PER_PROTOCOL;
+
+ let admin_client = harness
+ .server()
+ .tcp_client()
+ .expect("TCP transport not available")
+ .with_root_login()
+ .connect()
+ .await
+ .expect("failed to create admin client");
let stats = admin_client.get_stats().await.unwrap();
let server_pid = stats.process_id;
-
- let total_producers = client_factories.len() * PRODUCERS_PER_PROTOCOL;
init_system(&admin_client, total_producers).await;
let stop_flag = Arc::new(AtomicBool::new(false));
@@ -67,15 +76,13 @@ pub async fn run(client_factories: &[&dyn ClientFactory]) {
let mut join_set = JoinSet::new();
let mut global_producer_id = 0usize;
- for factory in client_factories {
- let protocol = factory.transport();
+ for transport in transports {
for local_id in 0..PRODUCERS_PER_PROTOCOL {
- let client = create_client(*factory).await;
- login_root(&client).await;
+ let client = create_client_for_transport(harness, transport).await;
let stop = stop_flag.clone();
let counter = total_messages.clone();
- let producer_name = format!("{:?}-{}", protocol, local_id);
+ let producer_name = format!("{:?}-{}", transport, local_id);
let producer_id = global_producer_id;
join_set.spawn(async move {
@@ -131,9 +138,22 @@ pub async fn run(client_factories: &[&dyn ClientFactory]) {
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)
+async fn create_client_for_transport(
+ harness: &TestHarness,
+ transport: TransportProtocol,
+) -> IggyClient {
+ let server = harness.server();
+ let builder = match transport {
+ TransportProtocol::Tcp => server.tcp_client().expect("TCP not available"),
+ TransportProtocol::Http => server.http_client().expect("HTTP not available"),
+ TransportProtocol::Quic => server.quic_client().expect("QUIC not available"),
+ TransportProtocol::WebSocket => server.websocket_client().expect("WebSocket not available"),
+ };
+ builder
+ .with_root_login()
+ .connect()
+ .await
+ .expect("failed to connect client")
}
async fn init_system(client: &IggyClient, total_producers: usize) {
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..b510765d5 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,7 +18,7 @@
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::harness::TestHarness;
use std::time::{Duration, Instant};
const STREAM_NAME: &str = "single-msg-batch-stream";
@@ -40,10 +40,11 @@ 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;
+pub async fn run(harness: &TestHarness, duration_secs: u64) {
+ let producer = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
producer.create_stream(STREAM_NAME).await.unwrap();
producer
diff --git a/core/integration/tests/server/scenarios/snapshot_scenario.rs b/core/integration/tests/server/scenarios/snapshot_scenario.rs
index 0780ca6c3..fb64a6ff3 100644
--- a/core/integration/tests/server/scenarios/snapshot_scenario.rs
+++ b/core/integration/tests/server/scenarios/snapshot_scenario.rs
@@ -17,18 +17,15 @@
*/
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, assert_clean_system};
+use integration::harness::{TestHarness, assert_clean_system};
use std::io::{Cursor, Read};
use zip::ZipArchive;
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
-
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
.await
- .unwrap();
+ .expect("Failed to get root client");
let snapshot = client
.snapshot(
diff --git a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
index 7784fb6ed..e6fc95650 100644
--- a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
+++ b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
@@ -19,9 +19,7 @@
use futures::StreamExt;
use iggy::prelude::*;
use iggy_common::Credentials;
-use integration::test_server::{IpAddrKind, TestServer};
-use serial_test::parallel;
-use std::collections::HashMap;
+use integration::iggy_harness;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
@@ -33,18 +31,6 @@ const CONSUMER_GROUP_NAME: &str = "stale-test-cg";
const PARTITIONS_COUNT: u32 = 1;
const TOTAL_MESSAGES: u32 = 10;
-fn create_test_server() -> TestServer {
- let mut extra_envs = HashMap::new();
- extra_envs.insert("IGGY_HEARTBEAT_ENABLED".to_string(), "true".to_string());
- extra_envs.insert("IGGY_HEARTBEAT_INTERVAL".to_string(), "2s".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());
- TestServer::new(Some(extra_envs), true, None, IpAddrKind::V4)
-}
-
async fn create_client(server_addr: &str, heartbeat_interval: &str) -> IggyClient {
let config = TcpClientConfig {
server_address: server_addr.to_string(),
@@ -125,12 +111,16 @@ async fn setup_resources(client: &IggyClient) {
}
/// Tests that a stale client receives clean errors and can manually reconnect.
-#[tokio::test]
-#[parallel]
-async fn should_handle_stale_client_with_manual_reconnection() {
- let mut test_server = create_test_server();
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
+#[iggy_harness(server(
+ heartbeat.enabled = true,
+ heartbeat.interval = "2s",
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true
+))]
+async fn should_handle_stale_client_with_manual_reconnection(
+ harness: &integration::harness::TestHarness,
+) {
+ let server_addr = harness.server().raw_tcp_addr().unwrap();
let setup_client = create_client(&server_addr, "500ms").await;
setup_resources(&setup_client).await;
@@ -239,16 +229,19 @@ async fn should_handle_stale_client_with_manual_reconnection() {
let _ = setup_client
.delete_stream(&Identifier::named(STREAM_NAME).unwrap())
.await;
- test_server.stop();
}
/// Tests that IggyConsumer automatically recovers after stale disconnect.
-#[tokio::test]
-#[parallel]
-async fn should_handle_stale_client_with_auto_reconnection() {
- let mut test_server = create_test_server();
- test_server.start();
- let server_addr = test_server.get_raw_tcp_addr().unwrap();
+#[iggy_harness(server(
+ heartbeat.enabled = true,
+ heartbeat.interval = "2s",
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true
+))]
+async fn should_handle_stale_client_with_auto_reconnection(
+ harness: &integration::harness::TestHarness,
+) {
+ let server_addr = harness.server().raw_tcp_addr().unwrap();
let setup_client = create_client(&server_addr, "500ms").await;
setup_resources(&setup_client).await;
@@ -294,7 +287,6 @@ async fn should_handle_stale_client_with_auto_reconnection() {
let _ = setup_client
.delete_stream(&Identifier::named(STREAM_NAME).unwrap())
.await;
- test_server.stop();
assert_eq!(
messages_consumed, TOTAL_MESSAGES,
diff --git a/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs b/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
index c816b2dac..97d3d137d 100644
--- a/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
+++ b/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
@@ -16,10 +16,10 @@
* under the License.
*/
-use crate::server::scenarios::{PARTITION_ID, PARTITIONS_COUNT, create_client};
+use crate::server::scenarios::{PARTITION_ID, PARTITIONS_COUNT};
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, assert_clean_system, login_root};
+use integration::harness::{TestHarness, assert_clean_system, login_root};
use std::str::FromStr;
const S1_NAME: &str = "test-stream-1";
@@ -31,8 +31,11 @@ const MSG_SIZE: u64 = IGGY_MESSAGE_HEADER_SIZE as u64 + MESSAGE_PAYLOAD_SIZE_BYT
const MSGS_COUNT: u64 = 117; // number of messages in a single topic after one pass of appending
const MSGS_SIZE: u64 = MSG_SIZE * MSGS_COUNT; // number of bytes in a single topic after one pass of appending
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = create_client(client_factory).await;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .new_client()
+ .await
+ .expect("Failed to create new client");
// 0. Ping server, login as root user and ensure that streams do not exist
ping_login_and_validate(&client).await;
@@ -121,7 +124,7 @@ async fn ping_login_and_validate(client: &IggyClient) {
client.ping().await.unwrap();
// 2. Login as root user
- login_root(client).await;
+ login_root(client).await.expect("login failed");
// 3. Ensure that streams do not exist
let streams = client.get_streams().await.unwrap();
diff --git a/core/integration/tests/server/scenarios/system_scenario.rs b/core/integration/tests/server/scenarios/system_scenario.rs
index 7efeba1f1..09264636a 100644
--- a/core/integration/tests/server/scenarios/system_scenario.rs
+++ b/core/integration/tests/server/scenarios/system_scenario.rs
@@ -22,12 +22,14 @@ use crate::server::scenarios::{
};
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, assert_clean_system};
+use integration::harness::{TestHarness, assert_clean_system};
use std::str::FromStr;
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
let consumer = Consumer {
kind: CONSUMER_KIND,
@@ -37,12 +39,6 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 0. Ping server
client.ping().await.unwrap();
- // 1. Login as root user
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap();
-
// 2. Ensure that streams do not exist
let streams = client.get_streams().await.unwrap();
assert!(streams.is_empty());
diff --git a/core/integration/tests/server/scenarios/tcp_tls_scenario.rs b/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
index b692b3b16..1cc451a5e 100644
--- a/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
+++ b/core/integration/tests/server/scenarios/tcp_tls_scenario.rs
@@ -18,11 +18,9 @@
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{assert_clean_system, login_root};
+use integration::harness::assert_clean_system;
pub async fn run(client: &IggyClient) {
- login_root(client).await;
-
let stream_name = "test-tls-stream";
client.create_stream(stream_name).await.unwrap();
diff --git a/core/integration/tests/server/scenarios/timestamp_scenario.rs b/core/integration/tests/server/scenarios/timestamp_scenario.rs
index eb5129701..af6fb7796 100644
--- a/core/integration/tests/server/scenarios/timestamp_scenario.rs
+++ b/core/integration/tests/server/scenarios/timestamp_scenario.rs
@@ -19,7 +19,7 @@
use super::PARTITION_ID;
use bytes::BytesMut;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
+use integration::harness::TestHarness;
use std::collections::HashMap;
use tokio::time::{Duration, sleep};
@@ -52,10 +52,11 @@ 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;
+pub async fn run(harness: &TestHarness) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
for msg_size in all_message_sizes() {
for (pattern_name, batch_pattern) in all_batch_patterns() {
diff --git a/core/integration/tests/server/scenarios/user_scenario.rs b/core/integration/tests/server/scenarios/user_scenario.rs
index bfcdfb1c3..f582711b3 100644
--- a/core/integration/tests/server/scenarios/user_scenario.rs
+++ b/core/integration/tests/server/scenarios/user_scenario.rs
@@ -23,10 +23,10 @@ use iggy::prelude::UserStatus;
use iggy::prelude::defaults::DEFAULT_ROOT_USERNAME;
use iggy::prelude::{GlobalPermissions, Permissions};
use iggy::prelude::{PersonalAccessTokenClient, SEC_IN_MICRO, SystemClient, UserClient};
-use integration::test_server::{ClientFactory, assert_clean_system, login_root};
+use integration::harness::{TestHarness, assert_clean_system, login_root};
-pub async fn run(client_factory: &dyn ClientFactory) {
- let client = create_client(client_factory).await;
+pub async fn run(harness: &TestHarness) {
+ let client = create_client(harness).await;
// 1. Ping should be allowed for unauthenticated users
client.ping().await.unwrap();
@@ -35,7 +35,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
assert!(get_users.is_err());
// 3. Login as root user
- let identity_info = login_root(&client).await;
+ let identity_info = login_root(&client).await.expect("login failed");
assert_eq!(identity_info.user_id, 0);
@@ -193,7 +193,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
assert!(personal_access_tokens.is_empty());
// 19. Login as root user again
- login_root(&client).await;
+ login_root(&client).await.expect("login failed");
// 20. Trying to create a new user with the same username should fail
let create_duplicated_user = client
diff --git a/core/integration/tests/server/scenarios/websocket_tls_scenario.rs b/core/integration/tests/server/scenarios/websocket_tls_scenario.rs
index f8f2f2d09..c1e75262f 100644
--- a/core/integration/tests/server/scenarios/websocket_tls_scenario.rs
+++ b/core/integration/tests/server/scenarios/websocket_tls_scenario.rs
@@ -18,11 +18,9 @@
use bytes::Bytes;
use iggy::prelude::*;
-use integration::test_server::{assert_clean_system, login_root};
+use integration::harness::assert_clean_system;
pub async fn run(client: &IggyClient) {
- login_root(client).await;
-
let stream_name = "test-websocket-tls-stream";
client.create_stream(stream_name).await.unwrap();
diff --git a/core/integration/tests/server/specific.rs b/core/integration/tests/server/specific.rs
index 501099a83..4adaaf9bf 100644
--- a/core/integration/tests/server/specific.rs
+++ b/core/integration/tests/server/specific.rs
@@ -18,208 +18,54 @@
*/
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,
+ 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::{
- harness::{TestHarness, TestServerConfig},
- 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;
-
-#[tokio::test]
-#[parallel]
-async fn should_delete_segments_and_validate_filesystem() {
- let mut harness = TestHarness::builder()
- .server(
- TestServerConfig::builder()
- .extra_envs(HashMap::from([(
- "IGGY_SYSTEM_SEGMENT_SIZE".to_string(),
- "1MiB".to_string(),
- )]))
- .build(),
- )
- .build()
- .unwrap();
-
- harness.start().await.unwrap();
+use integration::iggy_harness;
+#[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 data_path = harness.server().data_path();
delete_segments_scenario::run(&client, &data_path).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);
-
+#[iggy_harness(
+ test_client_transport = TcpTlsGenerated,
+ server(tls = generated)
+)]
+async fn tcp_tls_scenario_should_be_valid(harness: &TestHarness) {
+ let client = harness.root_client().await.unwrap();
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 = IggyClient::create(ClientWrapper::Iggy(client), None, None);
-
+#[iggy_harness(
+ test_client_transport = TcpTlsSelfSigned,
+ server(tls = self_signed)
+)]
+async fn tcp_tls_self_signed_scenario_should_be_valid(harness: &TestHarness) {
+ let client = harness.root_client().await.unwrap();
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);
-
+#[iggy_harness(
+ test_client_transport = WebSocketTlsGenerated,
+ server(websocket_tls = generated)
+)]
+async fn websocket_tls_scenario_should_be_valid(harness: &TestHarness) {
+ let client = harness.root_client().await.unwrap();
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;
+#[iggy_harness]
+async fn message_size_scenario(harness: &TestHarness) {
+ message_size_scenario::run(harness).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;
+#[iggy_harness(server(partition.messages_required_to_save = "10000"))]
+async fn should_handle_single_message_per_batch_with_delayed_persistence(harness: &TestHarness) {
+ single_message_per_batch_scenario::run(harness, 5).await;
}
/// This test configures the server to trigger frequent segment rotations and runs
@@ -236,78 +82,15 @@ async fn should_handle_single_message_per_batch_with_delayed_persistence() {
/// 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();
-
- extra_envs.insert("IGGY_SYSTEM_SEGMENT_SIZE".to_string(), "512B".to_string());
-
- extra_envs.insert("IGGY_MESSAGE_SAVER_INTERVAL".to_string(), "1s".to_string());
-
- extra_envs.insert(
- "IGGY_SYSTEM_PARTITION_MESSAGES_REQUIRED_TO_SAVE".to_string(),
- "32".to_string(),
- );
-
- extra_envs.insert(
- "IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string(),
- "none".to_string(),
- );
-
- extra_envs.insert("IGGY_TCP_SOCKET_MIGRATION".to_string(), "false".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 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(),
- ..Default::default()
- };
-
- 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;
+#[iggy_harness(server(
+ segment.size = "512B",
+ message_saver.interval = "1s",
+ partition.messages_required_to_save = "32",
+ segment.cache_indexes = "none",
+ tcp.socket_migration = false,
+ tcp.socket.override_defaults = true,
+ tcp.socket.nodelay = true
+))]
+async fn segment_rotation_scenario(harness: &TestHarness) {
+ segment_rotation_race_scenario::run(harness).await;
}
diff --git a/core/server/src/websocket/websocket_tls_listener.rs b/core/server/src/websocket/websocket_tls_listener.rs
index 56d22a3f5..7fa0796a6 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");