fix: better behavioral test handling

Signed-off-by: Eren Atas <[email protected]>

Author: erenatas

crates/flagd/flagd-testbed

@@ -261,6 +261,9 @@ pub struct FlagdOptions {
     pub stream_deadline_ms: u32,
     /// Offline polling interval in milliseconds
     pub offline_poll_interval_ms: Option<u32>,
+    /// Provider ID for identifying this provider instance to flagd
+    /// Used in in-process resolver for sync requests
+    pub provider_id: Option<String>,
 }
 /// Type of resolver to use for flag evaluation
 #[derive(Debug, Clone, PartialEq)]
@@ -336,6 +339,7 @@ impl Default for FlagdOptions {
                     .and_then(|s| s.parse().ok())
                     .unwrap_or(5000),
             ),
+            provider_id: std::env::var("FLAGD_PROVIDER_ID").ok(),
         };
 
         if options.source_configuration.is_some() && options.resolver_type != ResolverType::Rpc {

crates/flagd/schemas

@@ -29,6 +29,7 @@ pub struct GrpcStreamConnector {
     retry_grace_period: u32,
     stream_deadline_ms: u32,
     authority: String, // desired authority, e.g. "b-features-api.service"
+    provider_id: String, // provider identifier for sync requests
 }
 
 impl GrpcStreamConnector {
@@ -52,6 +53,7 @@ impl GrpcStreamConnector {
             retry_grace_period: options.retry_grace_period,
             stream_deadline_ms: options.stream_deadline_ms,
             authority,
+            provider_id: options.provider_id.clone().unwrap_or_else(|| "rust-flagd-provider".to_string()),
         }
     }
 
@@ -119,7 +121,7 @@ impl GrpcStreamConnector {
         // Create the gRPC client with no interceptor because the endpoint already carries the desired authority.
         let mut client = FlagSyncServiceClient::new(channel);
         let request = tonic::Request::new(SyncFlagsRequest {
-            provider_id: "rust-flagd-provider".to_string(),
+            provider_id: self.provider_id.clone(),
             selector: self.selector.clone().unwrap_or_default(),
         });
         debug!("Sending sync request with selector: {:?}", self.selector);
@@ -227,24 +229,17 @@ mod tests {
         drop(listener);
 
         // Create options configured for a failing connection.
-        let options = FlagdOptions {
-            host: addr.ip().to_string(),
-            resolver_type: crate::ResolverType::InProcess,
-            port: addr.port(),
-            target_uri: None,
-            deadline_ms: 100, // Short timeout for fast failures
-            retry_backoff_ms: 100,
-            retry_backoff_max_ms: 400,
-            retry_grace_period: 3,
-            stream_deadline_ms: 500,
-            tls: false,
-            cert_path: None,
-            selector: None,
-            socket_path: None,
-            cache_settings: None,
-            source_configuration: None,
-            offline_poll_interval_ms: None,
-        };
+        let mut options = FlagdOptions::default();
+        options.host = addr.ip().to_string();
+        options.resolver_type = crate::ResolverType::InProcess;
+        options.port = addr.port();
+        options.deadline_ms = 100; // Short timeout for fast failures
+        options.retry_backoff_ms = 100;
+        options.retry_backoff_max_ms = 400;
+        options.retry_grace_period = 3;
+        options.stream_deadline_ms = 500;
+        options.tls = false;
+        options.cache_settings = None;
 
         let target = format!("{}:{}", addr.ip(), addr.port());
         let connector =

crates/flagd/src/lib.rs

@@ -106,9 +106,11 @@ async fn env_with_value(world: &mut ConfigWorld, env: String, value: String) {
 
 #[when(expr = "a config was initialized")]
 async fn initialize_config(world: &mut ConfigWorld) {
+    // Start with defaults (which reads from environment variables)
     let mut options = FlagdOptions::default();
+    let mut resolver_explicitly_set = false;
 
-    // Handle resolver type first
+    // Handle resolver type first - explicit options override env vars
     if let Some(resolver) = world.option_values.get("resolver") {
         options.resolver_type = match resolver.to_uppercase().as_str() {
             "RPC" => ResolverType::Rpc,
@@ -117,32 +119,27 @@ async fn initialize_config(world: &mut ConfigWorld) {
             "FILE" | "OFFLINE" => ResolverType::File,
             _ => ResolverType::Rpc,
         };
-    } else if let Ok(resolver) = std::env::var("FLAGD_RESOLVER") {
-        options.resolver_type = match resolver.to_uppercase().as_str() {
-            "RPC" => ResolverType::Rpc,
-            "REST" => ResolverType::Rest,
-            "IN-PROCESS" | "INPROCESS" => ResolverType::InProcess,
-            "FILE" | "OFFLINE" => ResolverType::File,
-            _ => ResolverType::Rpc,
+        resolver_explicitly_set = true;
+        // Update port based on resolver type when explicitly set
+        options.port = match options.resolver_type {
+            ResolverType::Rpc => 8013,
+            ResolverType::InProcess => 8015,
+            _ => options.port,
         };
     }
 
-    // Set default port based on resolver type
-    options.port = match options.resolver_type {
-        ResolverType::Rpc => 8013,
-        ResolverType::InProcess => 8015,
-        _ => options.port,
-    };
-
-    // Handle source configuration after resolver type
+    // Handle source configuration - may override resolver type for backwards compatibility
+    // BUT only if resolver wasn't explicitly set to "rpc"
     if let Some(source) = world.option_values.get("offlineFlagSourcePath") {
         options.source_configuration = Some(source.clone());
-        if options.resolver_type != ResolverType::Rpc {
+        // For backwards compatibility: if offline path is set, switch to File resolver
+        // UNLESS resolver was explicitly set to "rpc" (in which case keep it as "rpc")
+        if !resolver_explicitly_set || options.resolver_type != ResolverType::Rpc {
             options.resolver_type = ResolverType::File;
         }
     }
 
-    // Handle remaining explicit options
+    // Handle remaining explicit options (these override env vars)
     if let Some(host) = world.option_values.get("host") {
         options.host = host.clone();
     }
@@ -217,6 +214,9 @@ async fn initialize_config(world: &mut ConfigWorld) {
     if let Some(selector) = world.option_values.get("selector") {
         options.selector = Some(selector.clone());
     }
+    if let Some(provider_id) = world.option_values.get("providerId") {
+        options.provider_id = Some(provider_id.clone());
+    }
     if let Some(max_size) = world
         .option_values
         .get("maxCacheSize")
@@ -269,12 +269,26 @@ async fn check_option_value(
         "retryBackoffMaxMs" => Some(world.options.retry_backoff_max_ms.to_string()),
         "retryGracePeriod" => Some(world.options.retry_grace_period.to_string()),
         "selector" => world.options.selector.clone(),
+        "providerId" => world.options.provider_id.clone(),
         "socketPath" => world.options.socket_path.clone(),
         "streamDeadlineMs" => Some(world.options.stream_deadline_ms.to_string()),
         _ => None,
     };
     let expected = convert_type(&option_type, &value);
-    assert_eq!(actual, expected, "Option '{}' value mismatch", option);
+    
+    // For resolver type, do case-insensitive comparison since enum normalizes to lowercase
+    let actual_normalized = if option == "resolver" {
+        actual.as_ref().map(|v| v.to_lowercase())
+    } else {
+        actual.clone()
+    };
+    let expected_normalized = if option == "resolver" {
+        expected.as_ref().map(|v| v.to_lowercase())
+    } else {
+        expected.clone()
+    };
+    
+    assert_eq!(actual_normalized, expected_normalized, "Option '{}' value mismatch", option);
 }
 
 #[test(tokio::test)]