refactor(integration): migrate connectors tests to iggy_harness proc macro

Eliminates boilerplate in connector tests by leveraging the same
proc macro infrastructure used for MCP tests. Connector-specific
fixtures (postgres, wiremock, random) moved to tests/ since
they're not part of core harness.

MCP and connectors runtime are now owned by ServerHandle rather
than TestHarness, enabling per-node dependent binaries in cluster
setups.

Also switches test output capture from env_logger to
tracing-subscriber for consistency with server logging.

Author: hubcio

Cargo.lock

@@ -255,8 +255,6 @@ embedded-io: 0.6.1, "Apache-2.0 OR MIT",
 encode_unicode: 1.0.0, "Apache-2.0 OR MIT",
 encoding_rs: 0.8.35, "(Apache-2.0 OR MIT) AND BSD-3-Clause",
 enum_dispatch: 0.3.13, "Apache-2.0 OR MIT",
-env_filter: 0.1.4, "Apache-2.0 OR MIT",
-env_logger: 0.11.8, "Apache-2.0 OR MIT",
 equivalent: 1.0.2, "Apache-2.0 OR MIT",
 err_trail: 0.11.0, "Apache-2.0",
 errno: 0.3.14, "Apache-2.0 OR MIT",

DEPENDENCIES.md

@@ -476,7 +476,7 @@ fn generate_harness_setup(
         #config_resolution
         let mut __harness = ::integration::harness::TestHarness::builder()
             .server(#server_config)
-            .primary_client(#client_config)
+            .client(#client_config)
             #mcp_builder_call
             #connectors_runtime_builder_call
             #cluster_builder_call
@@ -526,6 +526,12 @@ fn generate_start_and_seed(attrs: &IggyTestAttrs, fixture_seed: TokenStream) ->
     }
 }
 
+fn fixture_var_ident(name: &syn::Ident) -> syn::Ident {
+    let name_str = name.to_string();
+    let clean_name = name_str.trim_start_matches('_');
+    format_ident!("__fixture_{}", clean_name)
+}
+
 /// Generate fixture setup calls (before harness setup).
 fn generate_fixture_setup(params: &[DetectedParam]) -> TokenStream {
     let fixtures = fixture_params(params);
@@ -537,7 +543,7 @@ fn generate_fixture_setup(params: &[DetectedParam]) -> TokenStream {
         .iter()
         .filter_map(|p| {
             if let DetectedParam::Fixture { name, ty } = p {
-                let var_name = format_ident!("__fixture_{}", name);
+                let var_name = fixture_var_ident(name);
                 Some(quote! {
                     let #var_name = <#ty as ::integration::harness::TestFixture>::setup()
                         .await
@@ -563,7 +569,7 @@ fn generate_fixture_envs_collection(params: &[DetectedParam]) -> TokenStream {
         .iter()
         .filter_map(|p| {
             if let DetectedParam::Fixture { name, .. } = p {
-                let var_name = format_ident!("__fixture_{}", name);
+                let var_name = fixture_var_ident(name);
                 Some(quote! {
                     __fixture_envs.extend(
                         ::integration::harness::TestFixture::connectors_runtime_envs(&#var_name)
@@ -622,7 +628,7 @@ fn generate_param_bindings(params: &[DetectedParam]) -> TokenStream {
                 });
             }
             DetectedParam::Fixture { name, .. } => {
-                let fixture_var = format_ident!("__fixture_{}", name);
+                let fixture_var = fixture_var_ident(name);
                 bindings.push(quote! {
                     let #name = #fixture_var;
                 });

core/harness_derive/src/codegen.rs

@@ -28,18 +28,15 @@ ci-qemu = []
 
 [dependencies]
 ahash = { workspace = true }
-anyhow = { workspace = true }
 assert_cmd = { workspace = true }
 async-trait = { workspace = true }
 bon = { workspace = true }
 bytes = { workspace = true }
-chrono = { workspace = true }
 compio = { workspace = true }
 configs = { workspace = true }
 configs_derive = { workspace = true }
 ctor = { workspace = true }
 derive_more = { workspace = true }
-env_logger = { workspace = true }
 figment = { workspace = true }
 futures = { workspace = true }
 harness_derive = { workspace = true }
@@ -50,7 +47,6 @@ iggy_common = { workspace = true }
 keyring = { workspace = true }
 lazy_static = { workspace = true }
 libc = { workspace = true }
-log = { workspace = true }
 once_cell = { workspace = true }
 predicates = { workspace = true }
 rand = { workspace = true }
@@ -71,12 +67,12 @@ server = { workspace = true }
 socket2 = { workspace = true }
 sqlx = { workspace = true }
 strum = { workspace = true }
-strum_macros = { workspace = true }
 tempfile = { workspace = true }
 test-case = { workspace = true }
 testcontainers-modules = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
 twox-hash = { workspace = true }
 uuid = { workspace = true }
 zip = { workspace = true }

core/integration/Cargo.toml

@@ -115,24 +115,27 @@ impl TestContext {
         &self.base_dir
     }
 
-    pub fn mcp_stdout_path(&self) -> PathBuf {
-        self.base_dir.join("mcp_stdout.log")
+    pub fn mcp_stdout_path(&self, server_id: u32) -> PathBuf {
+        self.base_dir.join(format!("mcp_{server_id}_stdout.log"))
     }
 
-    pub fn mcp_stderr_path(&self) -> PathBuf {
-        self.base_dir.join("mcp_stderr.log")
+    pub fn mcp_stderr_path(&self, server_id: u32) -> PathBuf {
+        self.base_dir.join(format!("mcp_{server_id}_stderr.log"))
     }
 
-    pub fn connectors_runtime_state_path(&self) -> PathBuf {
-        self.base_dir.join("connectors_state")
+    pub fn connectors_runtime_state_path(&self, server_id: u32) -> PathBuf {
+        self.base_dir
+            .join(format!("connectors_runtime_{server_id}_state"))
     }
 
-    pub fn connectors_runtime_stdout_path(&self) -> PathBuf {
-        self.base_dir.join("connectors_stdout.log")
+    pub fn connectors_runtime_stdout_path(&self, server_id: u32) -> PathBuf {
+        self.base_dir
+            .join(format!("connectors_runtime_{server_id}_stdout.log"))
     }
 
-    pub fn connectors_runtime_stderr_path(&self) -> PathBuf {
-        self.base_dir.join("connectors_stderr.log")
+    pub fn connectors_runtime_stderr_path(&self, server_id: u32) -> PathBuf {
+        self.base_dir
+            .join(format!("connectors_runtime_{server_id}_stderr.log"))
     }
 
     pub fn test_stdout_path(&self) -> PathBuf {
@@ -185,10 +188,11 @@ mod tests {
     #[test]
     fn test_context_paths() {
         let ctx = TestContext::new(Some("test_context_paths".to_string()), true).unwrap();
-        assert!(ctx.mcp_stdout_path().ends_with("mcp_stdout.log"));
+        assert!(ctx.mcp_stdout_path(0).ends_with("mcp_0_stdout.log"));
         assert!(
-            ctx.connectors_runtime_state_path()
-                .ends_with("connectors_state")
+            ctx.connectors_runtime_state_path(0)
+                .ends_with("connectors_runtime_0_state")
         );
+        assert!(ctx.mcp_stdout_path(1).ends_with("mcp_1_stdout.log"));
     }
 }

core/integration/src/harness/context.rs

@@ -17,19 +17,6 @@
  * under the License.
  */
 
-mod postgres;
-mod random;
-mod wiremock;
-
-pub use postgres::{
-    PostgresContainer, PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture,
-    PostgresSourceByteaFixture, PostgresSourceDeleteFixture, PostgresSourceFixture,
-    PostgresSourceJsonFixture, PostgresSourceJsonbFixture, PostgresSourceMarkFixture,
-    SinkPayloadFormat, SinkSchema,
-};
-pub use random::RandomSourceFixture;
-pub use wiremock::{WireMockDirectFixture, WireMockWrappedFixture};
-
 use crate::harness::error::TestBinaryError;
 use crate::harness::seeds::SeedError;
 use async_trait::async_trait;

…/integration/src/harness/fixtures/mod.rs core/integration/src/harness/fixture.rscore/integration/src/harness/fixtures/mod.rs renamed to core/integration/src/harness/fixture.rs core/integration/src/harness/fixtures/mod.rs renamed to core/integration/src/harness/fixture.rs

@@ -37,6 +37,7 @@ use tokio::time::sleep;
 const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
 
 pub struct ConnectorsRuntimeHandle {
+    server_id: u32,
     config: ConnectorsRuntimeConfig,
     context: Arc<TestContext>,
     envs: HashMap<String, String>,
@@ -72,15 +73,15 @@ impl ConnectorsRuntimeHandle {
     }
 
     pub fn state_path(&self) -> PathBuf {
-        self.context.connectors_runtime_state_path()
+        self.context.connectors_runtime_state_path(self.server_id)
     }
 
     pub fn collect_logs(&self) -> (String, String) {
         common::collect_logs(&self.stdout_path, &self.stderr_path)
     }
 
     fn build_envs(&mut self) {
-        let state_path = self.context.connectors_runtime_state_path();
+        let state_path = self.context.connectors_runtime_state_path(self.server_id);
         self.envs.insert(
             "IGGY_CONNECTORS_STATE_PATH".to_string(),
             state_path.display().to_string(),
@@ -108,15 +109,18 @@ impl ConnectorsRuntimeHandle {
     }
 }
 
-impl TestBinary for ConnectorsRuntimeHandle {
-    type Config = ConnectorsRuntimeConfig;
-
-    fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+impl ConnectorsRuntimeHandle {
+    pub fn with_server_id(
+        config: ConnectorsRuntimeConfig,
+        context: Arc<TestContext>,
+        server_id: u32,
+    ) -> Self {
         let reserver =
             SinglePortReserver::new().expect("Failed to reserve port for connectors runtime");
         let server_address = reserver.address();
 
         Self {
+            server_id,
             config,
             context,
             envs: HashMap::new(),
@@ -128,6 +132,14 @@ impl TestBinary for ConnectorsRuntimeHandle {
             port_reserver: Some(reserver),
         }
     }
+}
+
+impl TestBinary for ConnectorsRuntimeHandle {
+    type Config = ConnectorsRuntimeConfig;
+
+    fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+        Self::with_server_id(config, context, 0)
+    }
 
     fn start(&mut self) -> Result<(), TestBinaryError> {
         self.build_envs();
@@ -151,8 +163,8 @@ impl TestBinary for ConnectorsRuntimeHandle {
             command.stdout(Stdio::inherit());
             command.stderr(Stdio::inherit());
         } else {
-            let stdout_path = self.context.connectors_runtime_stdout_path();
-            let stderr_path = self.context.connectors_runtime_stderr_path();
+            let stdout_path = self.context.connectors_runtime_stdout_path(self.server_id);
+            let stderr_path = self.context.connectors_runtime_stderr_path(self.server_id);
 
             let stdout_file =
                 File::create(&stdout_path).map_err(|e| TestBinaryError::FileSystemError {

core/integration/src/harness/handle/connectors_runtime.rs

@@ -45,6 +45,7 @@ const MCP_HEALTH_CHECK_RETRIES: u32 = common::DEFAULT_HEALTH_CHECK_RETRIES * 3;
 pub type McpClient = RunningService<RoleClient, InitializeRequestParams>;
 
 pub struct McpHandle {
+    server_id: u32,
     config: McpConfig,
     context: Arc<TestContext>,
     envs: HashMap<String, String>,
@@ -135,14 +136,13 @@ impl McpHandle {
     }
 }
 
-impl TestBinary for McpHandle {
-    type Config = McpConfig;
-
-    fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+impl McpHandle {
+    pub fn with_server_id(config: McpConfig, context: Arc<TestContext>, server_id: u32) -> Self {
         let reserver = SinglePortReserver::new().expect("Failed to reserve port for MCP server");
         let server_address = reserver.address();
 
         Self {
+            server_id,
             config,
             context,
             envs: HashMap::new(),
@@ -154,6 +154,14 @@ impl TestBinary for McpHandle {
             port_reserver: Some(reserver),
         }
     }
+}
+
+impl TestBinary for McpHandle {
+    type Config = McpConfig;
+
+    fn with_config(config: Self::Config, context: Arc<TestContext>) -> Self {
+        Self::with_server_id(config, context, 0)
+    }
 
     #[allow(deprecated)]
     fn start(&mut self) -> Result<(), TestBinaryError> {
@@ -177,8 +185,8 @@ impl TestBinary for McpHandle {
             command.stdout(Stdio::inherit());
             command.stderr(Stdio::inherit());
         } else {
-            let stdout_path = self.context.mcp_stdout_path();
-            let stderr_path = self.context.mcp_stderr_path();
+            let stdout_path = self.context.mcp_stdout_path(self.server_id);
+            let stderr_path = self.context.mcp_stderr_path(self.server_id);
 
             let stdout_file =
                 File::create(&stdout_path).map_err(|e| TestBinaryError::FileSystemError {