connectors

Author: hubcio

core/harness-derive/src/codegen.rs

@@ -18,8 +18,8 @@
 
 use crate::attrs::{IggyTestAttrs, TlsMode, Transport};
 use crate::params::{
-    DetectedParam, analyze_signature, matrix_params, needs_client, needs_harness,
-    needs_harness_mut, needs_mcp_client,
+    DetectedParam, analyze_signature, fixture_params, matrix_params, needs_client, needs_fixtures,
+    needs_harness, needs_harness_mut, needs_mcp_client,
 };
 use proc_macro2::{Span, TokenStream};
 use quote::{format_ident, quote};
@@ -174,20 +174,27 @@ fn generate_single_test(
     has_mcp_client: bool,
     attrs: &IggyTestAttrs,
 ) -> syn::Result<TokenStream> {
-    let harness_setup = generate_harness_setup(variant, has_client, has_mcp_client, attrs);
+    let has_fixtures = needs_fixtures(params);
+    let fixture_setup = generate_fixture_setup(params);
+    let fixture_envs = generate_fixture_envs_collection(params);
+    let harness_setup =
+        generate_harness_setup(variant, has_client, has_mcp_client, has_fixtures, attrs);
     let client_setup = if has_client {
         generate_client_setup()
     } else {
         quote!()
     };
-    let start_and_seed = generate_start_and_seed(attrs);
+    let fixture_seed = generate_fixture_seed(params);
+    let start_and_seed = generate_start_and_seed(attrs, fixture_seed);
     let mcp_client_setup = generate_mcp_client_setup(has_mcp_client);
     let param_bindings = generate_param_bindings(params, has_harness, has_harness_mut);
 
     Ok(quote! {
         #[::tokio::test]
         #[::serial_test::parallel]
         #fn_vis async fn #fn_name() {
+            #fixture_setup
+            #fixture_envs
             #harness_setup
             #start_and_seed
             #client_setup
@@ -212,24 +219,32 @@ fn generate_test_module(
     has_mcp_client: bool,
     attrs: &IggyTestAttrs,
 ) -> syn::Result<TokenStream> {
+    let has_fixtures = needs_fixtures(params);
+    let fixture_setup = generate_fixture_setup(params);
+    let fixture_envs = generate_fixture_envs_collection(params);
+    let fixture_seed = generate_fixture_seed(params);
+    let param_bindings = generate_param_bindings(params, has_harness, has_harness_mut);
+
     let mut test_fns = Vec::new();
 
     for variant in variants {
         let test_name = format_ident!("{}", variant.suffix());
-        let harness_setup = generate_harness_setup(variant, has_client, has_mcp_client, attrs);
+        let harness_setup =
+            generate_harness_setup(variant, has_client, has_mcp_client, has_fixtures, attrs);
         let client_setup = if has_client {
             generate_client_setup()
         } else {
             quote!()
         };
-        let start_and_seed = generate_start_and_seed(attrs);
+        let start_and_seed = generate_start_and_seed(attrs, fixture_seed.clone());
         let mcp_client_setup = generate_mcp_client_setup(has_mcp_client);
-        let param_bindings = generate_param_bindings(params, has_harness, has_harness_mut);
 
         test_fns.push(quote! {
             #[::tokio::test]
             #[::serial_test::parallel]
             async fn #test_name() {
+                #fixture_setup
+                #fixture_envs
                 #harness_setup
                 #start_and_seed
                 #client_setup
@@ -286,13 +301,18 @@ fn generate_impl_functions_for_test_matrix(
         })
         .collect();
 
+    let has_fixtures = needs_fixtures(params);
+    let fixture_setup = generate_fixture_setup(params);
+    let fixture_envs = generate_fixture_envs_collection(params);
+    let fixture_seed = generate_fixture_seed(params);
     let param_bindings = generate_param_bindings(params, has_harness, has_harness_mut);
-    let start_and_seed = generate_start_and_seed(attrs);
+    let start_and_seed = generate_start_and_seed(attrs, fixture_seed.clone());
     let mcp_client_setup = generate_mcp_client_setup(has_mcp_client);
 
     if variants.len() == 1 {
         let variant = &variants[0];
-        let harness_setup = generate_harness_setup(variant, has_client, has_mcp_client, attrs);
+        let harness_setup =
+            generate_harness_setup(variant, has_client, has_mcp_client, has_fixtures, attrs);
         let client_setup = if has_client {
             generate_client_setup()
         } else {
@@ -304,6 +324,8 @@ fn generate_impl_functions_for_test_matrix(
             #[::tokio::test]
             #[::serial_test::parallel]
             #fn_vis async fn #fn_name(#(#param_names: #param_types),*) {
+                #fixture_setup
+                #fixture_envs
                 #harness_setup
                 #start_and_seed
                 #client_setup
@@ -320,7 +342,8 @@ fn generate_impl_functions_for_test_matrix(
 
     for variant in variants {
         let impl_name = format_ident!("__impl_{}", variant.suffix());
-        let harness_setup = generate_harness_setup(variant, has_client, has_mcp_client, attrs);
+        let harness_setup =
+            generate_harness_setup(variant, has_client, has_mcp_client, has_fixtures, attrs);
         let client_setup = if has_client {
             generate_client_setup()
         } else {
@@ -329,6 +352,8 @@ fn generate_impl_functions_for_test_matrix(
 
         impl_fns.push(quote! {
             async fn #impl_name(#(#param_names: #param_types),*) {
+                #fixture_setup
+                #fixture_envs
                 #harness_setup
                 #start_and_seed
                 #client_setup
@@ -363,6 +388,7 @@ fn generate_harness_setup(
     variant: &TestVariant,
     has_client: bool,
     has_mcp_client: bool,
+    has_fixtures: bool,
     attrs: &IggyTestAttrs,
 ) -> TokenStream {
     let transport = variant.transport.variant_ident();
@@ -500,10 +526,18 @@ fn generate_harness_setup(
             .consumer_name
             .as_deref()
             .unwrap_or("connectors");
-        quote!(.connector(::integration::harness::ConnectorConfig::builder()
-            .config_path(::std::path::PathBuf::from(#config_path))
-            .consumer_name(#consumer_name)
-            .build()))
+        if has_fixtures {
+            quote!(.connector(::integration::harness::ConnectorConfig::builder()
+                .config_path(::std::path::PathBuf::from(#config_path))
+                .consumer_name(#consumer_name)
+                .extra_envs(__fixture_envs.clone())
+                .build()))
+        } else {
+            quote!(.connector(::integration::harness::ConnectorConfig::builder()
+                .config_path(::std::path::PathBuf::from(#config_path))
+                .consumer_name(#consumer_name)
+                .build()))
+        }
     } else {
         quote!()
     };
@@ -536,18 +570,35 @@ fn generate_client_setup() -> TokenStream {
 ///
 /// When a seed function is present, uses `start_with_seed` to run seed
 /// after server but before MCP and connector (which may depend on seed data).
-fn generate_start_and_seed(attrs: &IggyTestAttrs) -> TokenStream {
-    match &attrs.seed_fn {
-        Some(seed_fn) => {
-            // Wrap seed function in closure that takes ownership of client
-            // to avoid lifetime issues with async functions
+/// Fixture seeds are combined with the global seed.
+fn generate_start_and_seed(attrs: &IggyTestAttrs, fixture_seed: TokenStream) -> TokenStream {
+    let has_fixture_seed = !fixture_seed.is_empty();
+    match (&attrs.seed_fn, has_fixture_seed) {
+        (Some(seed_fn), true) => {
+            quote! {
+                __harness.start_with_seed(|__seed_client| async move {
+                    #seed_fn(&__seed_client).await?;
+                    #fixture_seed
+                    Ok(())
+                }).await.expect("failed to start test harness");
+            }
+        }
+        (Some(seed_fn), false) => {
             quote! {
                 __harness.start_with_seed(|__seed_client| async move {
                     #seed_fn(&__seed_client).await
                 }).await.expect("failed to start test harness");
             }
         }
-        None => {
+        (None, true) => {
+            quote! {
+                __harness.start_with_seed(|__seed_client| async move {
+                    #fixture_seed
+                    Ok(())
+                }).await.expect("failed to start test harness");
+            }
+        }
+        (None, false) => {
             quote! {
                 __harness.start().await.expect("failed to start test harness");
             }
@@ -566,6 +617,76 @@ fn generate_mcp_client_setup(has_mcp_client: bool) -> TokenStream {
     }
 }
 
+/// Generate fixture setup calls (before harness setup).
+fn generate_fixture_setup(params: &[DetectedParam]) -> TokenStream {
+    let fixtures = fixture_params(params);
+    if fixtures.is_empty() {
+        return quote!();
+    }
+
+    let setup_calls: Vec<_> = fixtures
+        .iter()
+        .filter_map(|p| {
+            if let DetectedParam::Fixture { name, ty } = p {
+                let var_name = format_ident!("__fixture_{}", name);
+                Some(quote! {
+                    let #var_name = <#ty as ::integration::harness::TestFixture>::setup()
+                        .await
+                        .expect("failed to setup fixture");
+                })
+            } else {
+                None
+            }
+        })
+        .collect();
+
+    quote!(#(#setup_calls)*)
+}
+
+/// Generate fixture envs collection (after fixture setup, before harness).
+fn generate_fixture_envs_collection(params: &[DetectedParam]) -> TokenStream {
+    let fixtures = fixture_params(params);
+    if fixtures.is_empty() {
+        return quote!();
+    }
+
+    let env_calls: Vec<_> = fixtures
+        .iter()
+        .filter_map(|p| {
+            if let DetectedParam::Fixture { name, .. } = p {
+                let var_name = format_ident!("__fixture_{}", name);
+                Some(quote! {
+                    __fixture_envs.extend(
+                        ::integration::harness::TestFixture::connector_envs(&#var_name)
+                    );
+                })
+            } else {
+                None
+            }
+        })
+        .collect();
+
+    quote! {
+        let mut __fixture_envs = ::std::collections::HashMap::<String, String>::new();
+        #(#env_calls)*
+    }
+}
+
+/// Generate fixture seed calls (inside start_with_seed closure).
+///
+/// Note: Currently disabled to avoid move semantics issues with async closures.
+/// Fixtures that need to seed data should do so in the test body after harness start.
+fn generate_fixture_seed(_params: &[DetectedParam]) -> TokenStream {
+    // Fixture seeding is disabled for now because:
+    // 1. The async move closure in start_with_seed captures the fixture by value
+    // 2. This prevents using the fixture in the test body after seeding
+    // 3. Most fixtures don't need to seed data - they just provide env vars
+    //
+    // If a fixture needs to seed data, it can be done manually in the test body:
+    //   fixture.seed(&client).await.unwrap();
+    quote!()
+}
+
 fn generate_param_bindings(
     params: &[DetectedParam],
     _has_harness: bool,
@@ -595,6 +716,12 @@ fn generate_param_bindings(
                     let #name = __mcp_client;
                 });
             }
+            DetectedParam::Fixture { name, .. } => {
+                let fixture_var = format_ident!("__fixture_{}", name);
+                bindings.push(quote! {
+                    let #name = #fixture_var;
+                });
+            }
             DetectedParam::MatrixParam { .. } => {}
         }
     }

core/harness-derive/src/params.rs

@@ -35,6 +35,8 @@ pub enum DetectedParam {
     HarnessMut { name: Ident },
     /// An MCP client parameter: `mcp_client: McpClient`
     McpClient { name: Ident },
+    /// A test fixture parameter: `fixture: PostgresSinkFixture`
+    Fixture { name: Ident, ty: Box<Type> },
     /// A parameter from test_matrix (passed through)
     MatrixParam { name: Ident, ty: Box<Type> },
 }
@@ -47,6 +49,7 @@ impl DetectedParam {
             | DetectedParam::HarnessRef { name }
             | DetectedParam::HarnessMut { name }
             | DetectedParam::McpClient { name }
+            | DetectedParam::Fixture { name, .. }
             | DetectedParam::MatrixParam { name, .. } => name,
         }
     }
@@ -98,6 +101,13 @@ fn detect_param_type(name: Ident, ty: &Type) -> syn::Result<DetectedParam> {
         return Ok(DetectedParam::HarnessRef { name });
     }
 
+    if is_fixture_type(&normalized) {
+        return Ok(DetectedParam::Fixture {
+            name,
+            ty: Box::new(ty.clone()),
+        });
+    }
+
     Ok(DetectedParam::MatrixParam {
         name,
         ty: Box::new(ty.clone()),
@@ -120,6 +130,10 @@ fn is_harness_mut_type(normalized: &str) -> bool {
     normalized.contains("TestHarness") && normalized.contains("&mut")
 }
 
+fn is_fixture_type(normalized: &str) -> bool {
+    normalized.ends_with("Fixture") || normalized.ends_with("Fixture>")
+}
+
 /// Check if any parameter requires a client.
 pub fn needs_client(params: &[DetectedParam]) -> bool {
     params
@@ -159,6 +173,21 @@ pub fn matrix_params(params: &[DetectedParam]) -> Vec<&DetectedParam> {
         .collect()
 }
 
+/// Check if any parameter is a fixture.
+pub fn needs_fixtures(params: &[DetectedParam]) -> bool {
+    params
+        .iter()
+        .any(|p| matches!(p, DetectedParam::Fixture { .. }))
+}
+
+/// Get the fixture parameters.
+pub fn fixture_params(params: &[DetectedParam]) -> Vec<&DetectedParam> {
+    params
+        .iter()
+        .filter(|p| matches!(p, DetectedParam::Fixture { .. }))
+        .collect()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -247,4 +276,41 @@ mod tests {
         let params = analyze_signature(&sig).unwrap();
         assert!(!super::needs_mcp_client(&params));
     }
+
+    #[test]
+    fn detect_fixture_param() {
+        let sig = parse_sig("async fn test(fixture: PostgresSinkFixture)");
+        let params = analyze_signature(&sig).unwrap();
+        assert_eq!(params.len(), 1);
+        assert!(matches!(&params[0], DetectedParam::Fixture { name, .. } if name == "fixture"));
+    }
+
+    #[test]
+    fn detect_fixture_with_generic() {
+        let sig = parse_sig("async fn test(fixture: Box<PostgresFixture>)");
+        let params = analyze_signature(&sig).unwrap();
+        assert_eq!(params.len(), 1);
+        assert!(matches!(&params[0], DetectedParam::Fixture { name, .. } if name == "fixture"));
+    }
+
+    #[test]
+    fn needs_fixtures_works() {
+        let sig = parse_sig("async fn test(fixture: RandomSourceFixture)");
+        let params = analyze_signature(&sig).unwrap();
+        assert!(super::needs_fixtures(&params));
+
+        let sig = parse_sig("async fn test(client: &IggyClient)");
+        let params = analyze_signature(&sig).unwrap();
+        assert!(!super::needs_fixtures(&params));
+    }
+
+    #[test]
+    fn fixture_params_works() {
+        let sig = parse_sig(
+            "async fn test(f1: PostgresFixture, client: &IggyClient, f2: RandomSourceFixture)",
+        );
+        let params = analyze_signature(&sig).unwrap();
+        let fixtures = super::fixture_params(&params);
+        assert_eq!(fixtures.len(), 2);
+    }
 }