Update tests and context passing

Signed-off-by: Heinz N. Gies <heinz@licenser.net>
tremor-rs · May 8, 2023 · 9b82ae4 · 9b82ae4
1 parent aea52d4
commit 9b82ae4
Show file tree

Hide file tree

Showing 17 changed files with 108 additions and 94 deletions.
diff --git a/src/config.rs b/src/config.rs
@@ -385,7 +385,7 @@ mod tests {
         let id = Alias::new("my_id");
         let res = Connector::from_config(&id, "fancy_schmancy".into(), &config);
         assert!(res.is_err());
-        assert_eq!(String::from("Invalid Definition for connector \"app/flow::my_id\": Expected type I64 for key metrics_interval_s but got String"), res.err().map(|e| e.to_string()).unwrap_or_default());
+        assert_eq!(String::from("Invalid Definition for connector \"my_id\": Expected type I64 for key metrics_interval_s but got String"), res.err().map(|e| e.to_string()).unwrap_or_default());
     }
 
     #[test]

diff --git a/src/connectors.rs b/src/connectors.rs
@@ -48,7 +48,6 @@ use crate::{
     log_error, pipeline, qsize, raft,
     system::{flow::AppContext, KillSwitch, Runtime},
 };
-use beef::Cow;
 use futures::Future;
 use halfbrown::HashMap;
 use simd_json::{Builder, Mutable, ValueAccess};
@@ -315,7 +314,7 @@ pub(crate) trait Context: Display + Clone {
         'ct: 'event,
     {
         let t: &str = self.connector_type().into();
-        event_meta.get(&Cow::borrowed(t))
+        event_meta.get(t)
     }
 }
 

diff --git a/src/connectors/impls/elastic.rs b/src/connectors/impls/elastic.rs
@@ -971,9 +971,7 @@ mod tests {
         let connector_config =
             ConnectorConfig::from_config(&alias, builder.connector_type(), &config)?;
         assert_eq!(
-            String::from(
-                "Invalid Definition for connector \"app/flow::my_elastic\": empty nodes provided"
-            ),
+            String::from("Invalid Definition for connector \"my_elastic\": empty nodes provided"),
             builder
                 .build(&alias, &connector_config,)
                 .await

diff --git a/src/connectors/impls/gbq/writer/sink.rs b/src/connectors/impls/gbq/writer/sink.rs
@@ -781,7 +781,7 @@ mod test {
                 precision: 0,
                 scale: 0,
             }],
-            &SinkContext::dummy(),
+            &SinkContext::dummy("gbq_writer"),
         );
 
         assert_eq!(result.0.field.len(), 0);
@@ -802,7 +802,7 @@ mod test {
                 precision: 0,
                 scale: 0,
             }],
-            &SinkContext::dummy(),
+            &SinkContext::dummy("gbq_writer"),
         );
 
         assert_eq!(result.0.field.len(), 0);
@@ -832,7 +832,7 @@ mod test {
                     precision: 0,
                     scale: 0,
                 }],
-                &SinkContext::dummy(),
+                &SinkContext::dummy("gbq_writer"),
             );
 
             assert_eq!(result.1.len(), 1);
@@ -864,7 +864,7 @@ mod test {
                 precision: 0,
                 scale: 0,
             }],
-            &SinkContext::dummy(),
+            &SinkContext::dummy("gbq_writer"),
         );
 
         assert_eq!(result.1.len(), 1);
@@ -1040,9 +1040,8 @@ mod test {
 
         let mut result = Vec::new();
         assert!(encode_field(&value, &field, &mut result).is_ok());
-
         // json is currently not supported, so we expect the field to be skipped
-        assert_eq!([] as [u8; 0], result[..]);
+        assert_eq!([10, 2, 123, 125] as [u8; 4], result[..]);
     }
 
     #[test]
@@ -1079,7 +1078,7 @@ mod test {
 
     #[test]
     pub fn mapping_generates_a_correct_descriptor() {
-        let ctx = SinkContext::dummy();
+        let ctx = SinkContext::dummy("gbq_writer");
         let mapping = JsonToProtobufMapping::new(
             &vec![
                 TableFieldSchema {
@@ -1120,7 +1119,7 @@ mod test {
 
     #[test]
     pub fn can_map_json_to_protobuf() -> Result<()> {
-        let ctx = SinkContext::dummy();
+        let ctx = SinkContext::dummy("gbq_writer");
         let mapping = JsonToProtobufMapping::new(
             &vec![
                 TableFieldSchema {
@@ -1157,7 +1156,7 @@ mod test {
 
     #[test]
     fn map_field_ignores_fields_that_are_not_in_definition() -> Result<()> {
-        let ctx = SinkContext::dummy();
+        let ctx = SinkContext::dummy("gbq_writer");
         let mapping = JsonToProtobufMapping::new(
             &vec![
                 TableFieldSchema {
@@ -1195,7 +1194,7 @@ mod test {
 
     #[test]
     fn map_field_ignores_struct_fields_that_are_not_in_definition() -> Result<()> {
-        let ctx = SinkContext::dummy();
+        let ctx = SinkContext::dummy("gbq_writer");
         let mapping = JsonToProtobufMapping::new(
             &vec![TableFieldSchema {
                 name: "a".to_string(),
@@ -1231,7 +1230,7 @@ mod test {
 
     #[test]
     fn fails_on_bytes_type_mismatch() {
-        let ctx = SinkContext::dummy();
+        let ctx = SinkContext::dummy("gbq_writer");
         let mapping = JsonToProtobufMapping::new(
             &vec![TableFieldSchema {
                 name: "a".to_string(),
@@ -1258,7 +1257,7 @@ mod test {
 
     #[test]
     fn fails_if_the_event_is_not_an_object() {
-        let ctx = SinkContext::dummy();
+        let ctx = SinkContext::dummy("gbq_writer");
         let mapping = JsonToProtobufMapping::new(
             &vec![TableFieldSchema {
                 name: "a".to_string(),
@@ -1311,7 +1310,7 @@ mod test {
             .on_event(
                 "",
                 Event::signal_tick(),
-                &SinkContext::dummy(),
+                &SinkContext::dummy("gbq_writer"),
                 &mut EventSerializer::dummy(None)?,
                 0,
             )
@@ -1340,7 +1339,7 @@ mod test {
             .on_event(
                 "",
                 Event::signal_tick(),
-                &SinkContext::dummy(),
+                &SinkContext::dummy("gbq_writer"),
                 &mut EventSerializer::dummy(None)?,
                 0,
             )
@@ -1401,7 +1400,7 @@ mod test {
             }),
         );
 
-        let ctx = SinkContext::dummy();
+        let ctx = SinkContext::dummy("gbq_writer");
 
         sink.connect(&ctx, &Attempt::default()).await?;
 
@@ -1471,7 +1470,7 @@ mod test {
             }),
         );
 
-        let ctx = SinkContext::dummy();
+        let ctx = SinkContext::dummy("gbq_writer");
 
         sink.connect(&ctx, &Attempt::default()).await?;
 

diff --git a/src/connectors/impls/gcl/writer/sink.rs b/src/connectors/impls/gcl/writer/sink.rs
@@ -356,7 +356,7 @@ mod test {
             tx,
             MockChannelFactory,
         );
-        let sink_context = SinkContext::dummy();
+        let sink_context = SinkContext::dummy("gcl_writer");
 
         sink.connect(&sink_context, &Attempt::default()).await?;
 
@@ -439,7 +439,7 @@ mod test {
             .on_event(
                 "",
                 Event::signal_tick(),
-                &SinkContext::dummy(),
+                &SinkContext::dummy("gcl_writer"),
                 &mut EventSerializer::dummy(None)?,
                 0,
             )

diff --git a/src/connectors/impls/gcs/streamer.rs b/src/connectors/impls/gcs/streamer.rs
@@ -560,7 +560,7 @@ pub(crate) mod tests {
             ChunkedBuffer,
         > = YoloSink::new(sink_impl);
 
-        let context = SinkContext::dummy();
+        let context = SinkContext::dummy("gcs_streamer");
         let mut serializer = EventSerializer::dummy(Some(CodecConfig::from("json")))?;
 
         // simulate sink lifecycle
@@ -730,7 +730,7 @@ pub(crate) mod tests {
             ChunkedBuffer,
         > = YoloSink::new(sink_impl);
 
-        let context = SinkContext::dummy();
+        let context = SinkContext::dummy("gcs_streamer");
         let mut serializer = EventSerializer::dummy(Some(CodecConfig::from("json")))?;
 
         // simulate sink lifecycle
@@ -888,7 +888,7 @@ pub(crate) mod tests {
             ChunkedBuffer,
         > = YoloSink::new(sink_impl);
 
-        let context = SinkContext::dummy();
+        let context = SinkContext::dummy("gcs_streamer");
         let mut serializer = EventSerializer::dummy(Some(CodecConfig::from("json")))?;
 
         // simulate sink lifecycle
@@ -967,7 +967,7 @@ pub(crate) mod tests {
             ChunkedBuffer,
         > = YoloSink::new(sink_impl);
 
-        let context = SinkContext::dummy();
+        let context = SinkContext::dummy("gcs_streamer");
 
         // simulate sink lifecycle
         sink.on_start(&context).await?;
@@ -1011,7 +1011,7 @@ pub(crate) mod tests {
             ChunkedBuffer,
         > = ConsistentSink::new(sink_impl);
 
-        let context = SinkContext::dummy();
+        let context = SinkContext::dummy("gcs_streamer");
         let mut serializer = EventSerializer::dummy(Some(CodecConfig::from("json")))?;
 
         // simulate standard sink lifecycle
@@ -1220,7 +1220,7 @@ pub(crate) mod tests {
             ChunkedBuffer,
         > = ConsistentSink::new(sink_impl);
 
-        let context = SinkContext::dummy();
+        let context = SinkContext::dummy("gcs_streamer");
         let mut serializer = EventSerializer::dummy(Some(CodecConfig::from("json")))?;
 
         // simulate standard sink lifecycle

diff --git a/src/connectors/impls/kafka.rs b/src/connectors/impls/kafka.rs
@@ -362,7 +362,7 @@ mod tests {
             &literal!({
                 "measurement": "kafka_consumer_stats",
                 "tags": {
-                    "connector": "app/fake::fake"
+                    "connector": "fake"
                 },
                 "fields": {
                     "rx_msgs": 42,

diff --git a/src/connectors/impls/object_storage.rs b/src/connectors/impls/object_storage.rs
@@ -108,7 +108,7 @@ pub(crate) trait ObjectStorageCommon {
     }
 
     fn get_object_id(&self, meta: Option<&Value<'_>>) -> Result<ObjectId> {
-        let name = meta
+        let name = &meta
             .get(NAME)
             .ok_or_else(|| {
                 err_object_storage(format!(

diff --git a/src/connectors/sink.rs b/src/connectors/sink.rs
@@ -284,12 +284,12 @@ pub(crate) struct SinkContextInner {
 pub(crate) struct SinkContext(Arc<SinkContextInner>);
 impl SinkContext {
     #[cfg(test)]
-    pub(crate) fn dummy() -> Self {
+    pub(crate) fn dummy(ct: &str) -> Self {
         let (tx, _rx) = bounded(1024);
         Self::new(
             SinkUId::default(),
-            Alias::from("dummy"),
-            ConnectorType::from("dummy"),
+            Alias::from(ct),
+            ConnectorType::from(ct),
             QuiescenceBeacon::default(),
             ConnectionLostNotifier::new(tx),
             AppContext::default(),