feat(component validation): add sink error path validation + multi config

src/components/validation/mod.rs

@@ -57,6 +57,51 @@ pub enum ComponentConfiguration {
     Sink(BoxedSink),
 }
 
+/// Component configuration for a test case.
+#[derive(Clone)]
+pub struct ComponentTestCaseConfig {
+    config: ComponentConfiguration,
+    /// If specified, this name must match the `config_name` field of at least one of the test case events.
+    test_case: Option<String>,
+    external_resource: Option<ExternalResource>,
+}
+
+impl ComponentTestCaseConfig {
+    pub fn from_source<C: Into<BoxedSource>>(
+        config: C,
+        test_case: Option<String>,
+        external_resource: Option<ExternalResource>,
+    ) -> Self {
+        Self {
+            config: ComponentConfiguration::Source(config.into()),
+            test_case,
+            external_resource,
+        }
+    }
+    pub fn from_transform<C: Into<BoxedTransform>>(
+        config: C,
+        test_case: Option<String>,
+        external_resource: Option<ExternalResource>,
+    ) -> Self {
+        Self {
+            config: ComponentConfiguration::Transform(config.into()),
+            test_case,
+            external_resource,
+        }
+    }
+    pub fn from_sink<C: Into<BoxedSink>>(
+        config: C,
+        test_case: Option<String>,
+        external_resource: Option<ExternalResource>,
+    ) -> Self {
+        Self {
+            config: ComponentConfiguration::Sink(config.into()),
+            test_case,
+            external_resource,
+        }
+    }
+}
+
 /// Configuration for validating a component.
 ///
 /// This type encompasses all of the required information for configuring and validating a
@@ -66,46 +111,45 @@ pub enum ComponentConfiguration {
 pub struct ValidationConfiguration {
     component_name: &'static str,
     component_type: ComponentType,
-    component_configuration: ComponentConfiguration,
-    external_resource: Option<ExternalResource>,
+    /// There may be only one `ComponentTestCaseConfig` necessary to execute all test cases, but some cases
+    /// require more advanced configuration in order to hit the code path desired.
+    component_configurations: Vec<ComponentTestCaseConfig>,
 }
 
 impl ValidationConfiguration {
     /// Creates a new `ValidationConfiguration` for a source.
-    pub fn from_source<C: Into<BoxedSource>>(
+    pub fn from_source(
         component_name: &'static str,
-        config: C,
-        external_resource: Option<ExternalResource>,
+        component_configurations: Vec<ComponentTestCaseConfig>,
     ) -> Self {
         Self {
             component_name,
             component_type: ComponentType::Source,
-            component_configuration: ComponentConfiguration::Source(config.into()),
-            external_resource,
+            component_configurations,
         }
     }
 
     /// Creates a new `ValidationConfiguration` for a transform.
-    pub fn from_transform(component_name: &'static str, config: impl Into<BoxedTransform>) -> Self {
+    pub fn from_transform(
+        component_name: &'static str,
+        component_configurations: Vec<ComponentTestCaseConfig>,
+    ) -> Self {
         Self {
             component_name,
             component_type: ComponentType::Transform,
-            component_configuration: ComponentConfiguration::Transform(config.into()),
-            external_resource: None,
+            component_configurations,
         }
     }
 
     /// Creates a new `ValidationConfiguration` for a sink.
-    pub fn from_sink<C: Into<BoxedSink>>(
+    pub fn from_sink(
         component_name: &'static str,
-        config: C,
-        external_resource: Option<ExternalResource>,
+        component_configurations: Vec<ComponentTestCaseConfig>,
     ) -> Self {
         Self {
             component_name,
             component_type: ComponentType::Sink,
-            component_configuration: ComponentConfiguration::Sink(config.into()),
-            external_resource,
+            component_configurations,
         }
     }
 
@@ -120,13 +164,31 @@ impl ValidationConfiguration {
     }
 
     /// Gets the configuration of the component.
-    pub fn component_configuration(&self) -> ComponentConfiguration {
-        self.component_configuration.clone()
+    pub fn component_configurations(&self) -> Vec<ComponentTestCaseConfig> {
+        self.component_configurations.clone()
+    }
+
+    fn get_comp_test_case(&self, test_case: Option<&String>) -> Option<ComponentTestCaseConfig> {
+        let empty = String::from("");
+        let test_case = test_case.unwrap_or(&empty);
+        self.component_configurations
+            .clone()
+            .into_iter()
+            .find(|c| c.test_case.as_ref().unwrap_or(&String::from("")) == test_case)
+    }
+
+    /// Gets the configuration of the component.
+    pub fn component_configuration_for_test_case(
+        &self,
+        test_case: Option<&String>,
+    ) -> Option<ComponentConfiguration> {
+        self.get_comp_test_case(test_case).map(|c| c.config)
     }
 
     /// Gets the external resource definition for validating the component, if any.
-    pub fn external_resource(&self) -> Option<ExternalResource> {
-        self.external_resource.clone()
+    pub fn external_resource(&self, test_case: Option<&String>) -> Option<ExternalResource> {
+        self.get_comp_test_case(test_case)
+            .and_then(|c| c.external_resource)
     }
 }
 

src/components/validation/resources/event.rs

@@ -1,5 +1,6 @@
 use bytes::BytesMut;
 use serde::Deserialize;
+use serde_json::Value;
 use snafu::Snafu;
 use tokio_util::codec::Encoder as _;
 
@@ -27,7 +28,19 @@ pub enum RawTestEvent {
     ///
     /// For transforms and sinks, generally, the only way to cause an error is if the event itself
     /// is malformed in some way, which can be achieved without this test event variant.
-    Modified { modified: bool, event: EventData },
+    AlternateEncoder { fail_encoding_of: EventData },
+
+    /// The event is created, and the specified field is added to it.
+    ///
+    /// This allows the ability to hit code paths where some codecs require specific fields to be of specific
+    /// types, thus allowing us to encode into the input runner without error, but encoding in the component
+    /// under test can be set up to fail.
+    WithField {
+        event: EventData,
+        name: String,
+        value: Value,
+        fail: Option<bool>,
+    },
 }
 
 #[derive(Clone, Debug, Deserialize)]
@@ -52,45 +65,52 @@ impl EventData {
 /// metrics collection is based on the same event. Namely, one issue that can arise from creating the event
 /// from the event data twice (once for the expected and once for actual), it can result in a timestamp in
 /// the event which may or may not have the same millisecond precision as it's counterpart.
+///
+/// For transforms and sinks, generally, the only way to cause an error is if the event itself
+/// is malformed in some way, which can be achieved without this test event variant.
 #[derive(Clone, Debug, Deserialize)]
 #[serde(from = "RawTestEvent")]
 #[serde(untagged)]
 pub enum TestEvent {
     /// The event is used, as-is, without modification.
     Passthrough(Event),
 
-    /// The event is potentially modified by the external resource.
-    ///
-    /// The modification made is dependent on the external resource, but this mode is made available
-    /// for when a test case wants to exercise the failure path, but cannot cause a failure simply
-    /// by constructing the event in a certain way i.e. adding an invalid field, or removing a
-    /// required field, or using an invalid field value, and so on.
+    /// The event is encoded using an encoding that differs from the component's
+    /// configured encoding, which should cause an error when the event is decoded.
+    FailWithAlternateEncoder(Event),
+
+    /// The event has an additional field injected prior to encoding, which should cause
+    /// an error when the event is decoded.
     ///
-    /// For transforms and sinks, generally, the only way to cause an error is if the event itself
-    /// is malformed in some way, which can be achieved without this test event variant.
-    Modified { modified: bool, event: Event },
+    /// This is useful for testing encodings that have strict schemas and cannot
+    /// handle arbitrary fields or differing data types for certain fields.
+    FailWithInjectedField(Event),
 }
 
 impl TestEvent {
     #[allow(clippy::missing_const_for_fn)] // const cannot run destructor
     pub fn into_event(self) -> Event {
         match self {
             Self::Passthrough(event) => event,
-            Self::Modified { event, .. } => event,
+            Self::FailWithAlternateEncoder(event) => event,
+            Self::FailWithInjectedField(event) => event,
         }
     }
 
     pub fn get_event(&mut self) -> &mut Event {
         match self {
             Self::Passthrough(event) => event,
-            Self::Modified { event, .. } => event,
+            Self::FailWithAlternateEncoder(event) => event,
+            Self::FailWithInjectedField(event) => event,
         }
     }
 
+    /// (should_fail, event)
     pub fn get(self) -> (bool, Event) {
         match self {
             Self::Passthrough(event) => (false, event),
-            Self::Modified { modified, event } => (modified, event),
+            Self::FailWithAlternateEncoder(event) => (true, event),
+            Self::FailWithInjectedField(event) => (true, event),
         }
     }
 }
@@ -104,10 +124,25 @@ impl From<RawTestEvent> for TestEvent {
             RawTestEvent::Passthrough(event_data) => {
                 TestEvent::Passthrough(event_data.into_event())
             }
-            RawTestEvent::Modified { modified, event } => TestEvent::Modified {
-                modified,
-                event: event.into_event(),
-            },
+            RawTestEvent::AlternateEncoder {
+                fail_encoding_of: event_data,
+            } => TestEvent::FailWithAlternateEncoder(event_data.into_event()),
+            RawTestEvent::WithField {
+                event,
+                name,
+                value,
+                fail,
+            } => {
+                let mut event = event.into_event();
+                let log_event = event.as_mut_log();
+                log_event.insert(name.as_str(), value);
+
+                if fail.unwrap_or_default() {
+                    TestEvent::FailWithInjectedField(event)
+                } else {
+                    TestEvent::Passthrough(event)
+                }
+            }
         }
     }
 }
@@ -118,13 +153,13 @@ pub fn encode_test_event(
     event: TestEvent,
 ) {
     match event {
-        TestEvent::Passthrough(event) => {
+        TestEvent::Passthrough(event) | TestEvent::FailWithInjectedField(event) => {
             // Encode the event normally.
             encoder
                 .encode(event, buf)
                 .expect("should not fail to encode input event");
         }
-        TestEvent::Modified { event, .. } => {
+        TestEvent::FailWithAlternateEncoder(event) => {
             // This is a little fragile, but we check what serializer this encoder uses, and based
             // on `Serializer::supports_json`, we choose an opposing codec. For example, if the
             // encoder supports JSON, we'll use a serializer that doesn't support JSON, and vise

src/components/validation/runner/config.rs

@@ -24,9 +24,17 @@ pub struct TopologyBuilder {
 
 impl TopologyBuilder {
     /// Creates a component topology for the given component configuration.
-    pub fn from_configuration(configuration: &ValidationConfiguration) -> Self {
-        let component_configuration = configuration.component_configuration();
-        match component_configuration {
+    pub fn from_configuration(
+        configuration: &ValidationConfiguration,
+        config_name: Option<&String>,
+    ) -> Result<Self, String> {
+        let component_configuration = configuration
+            .component_configuration_for_test_case(config_name)
+            .ok_or(format!(
+                "No test case name defined for configuration {:?}.",
+                config_name
+            ))?;
+        Ok(match component_configuration {
             ComponentConfiguration::Source(source) => {
                 debug_assert_eq!(configuration.component_type(), ComponentType::Source);
                 Self::from_source(source)
@@ -39,7 +47,7 @@ impl TopologyBuilder {
                 debug_assert_eq!(configuration.component_type(), ComponentType::Sink);
                 Self::from_sink(sink)
             }
-        }
+        })
     }
 
     /// Creates a component topology for validating a source.