add more testing

Author: duncanista

bottlecap/src/logger.rs

@@ -49,10 +49,9 @@ where
             for span in scope.from_root() {
                 span_prefix.push_str(span.name());
                 let ext = span.extensions();
-                let fields = &ext
-                    .get::<FormattedFields<N>>()
-                    .expect("will never be `None`");
-                if !fields.is_empty() {
+                if let Some(fields) = ext.get::<FormattedFields<N>>()
+                    && !fields.is_empty()
+                {
                     span_prefix.push('{');
                     span_prefix.push_str(fields);
                     span_prefix.push('}');
@@ -72,3 +71,109 @@ where
         writeln!(writer, "{output}")
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tracing::subscriber::with_default;
+    use tracing_subscriber::fmt::Subscriber;
+
+    /// Captures all output from a tracing subscriber using our Formatter.
+    fn capture_log<F: FnOnce()>(f: F) -> String {
+        let buf = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
+        let buf_clone = buf.clone();
+
+        let subscriber = Subscriber::builder()
+            .with_writer(move || -> Box<dyn std::io::Write + Send> {
+                Box::new(WriterGuard(buf_clone.clone()))
+            })
+            .with_max_level(tracing::Level::TRACE)
+            .with_level(true)
+            .with_target(false)
+            .without_time()
+            .event_format(Formatter)
+            .finish();
+
+        with_default(subscriber, f);
+
+        let lock = buf.lock().expect("test lock poisoned");
+        String::from_utf8(lock.clone()).expect("invalid UTF-8 in log output")
+    }
+
+    /// A wrapper so Arc<Mutex<Vec<u8>>> implements std::io::Write.
+    struct WriterGuard(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
+
+    impl std::io::Write for WriterGuard {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0.lock().unwrap().extend_from_slice(buf);
+            Ok(buf.len())
+        }
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn test_formatter_outputs_valid_json_with_level() {
+        let output = capture_log(|| {
+            tracing::info!("hello world");
+        });
+
+        let parsed: serde_json::Value =
+            serde_json::from_str(output.trim()).expect("output should be valid JSON");
+
+        assert_eq!(parsed["level"], "INFO");
+        assert!(
+            parsed["message"]
+                .as_str()
+                .unwrap()
+                .contains("DD_EXTENSION | INFO | hello world")
+        );
+    }
+
+    #[test]
+    fn test_formatter_error_level() {
+        let output = capture_log(|| {
+            tracing::error!("something broke");
+        });
+
+        let parsed: serde_json::Value = serde_json::from_str(output.trim()).unwrap();
+        assert_eq!(parsed["level"], "ERROR");
+        assert!(
+            parsed["message"]
+                .as_str()
+                .unwrap()
+                .contains("DD_EXTENSION | ERROR | something broke")
+        );
+    }
+
+    #[test]
+    fn test_formatter_debug_level() {
+        let output = capture_log(|| {
+            tracing::debug!("debug details");
+        });
+
+        let parsed: serde_json::Value = serde_json::from_str(output.trim()).unwrap();
+        assert_eq!(parsed["level"], "DEBUG");
+        assert!(
+            parsed["message"]
+                .as_str()
+                .unwrap()
+                .contains("DD_EXTENSION | DEBUG | debug details")
+        );
+    }
+
+    #[test]
+    fn test_formatter_escapes_special_characters() {
+        let output = capture_log(|| {
+            tracing::info!("message with \"quotes\" and a\nnewline");
+        });
+
+        // The output should be valid JSON despite special characters
+        let parsed: serde_json::Value =
+            serde_json::from_str(output.trim()).expect("special chars should be escaped");
+        let msg = parsed["message"].as_str().unwrap();
+        assert!(msg.contains("quotes"));
+        assert!(msg.contains("newline"));
+    }
+}

bottlecap/src/logs/lambda/processor.rs

@@ -381,9 +381,11 @@ impl LambdaProcessor {
                 lambda_message.message.clone(),
             );
 
-            // Extract log level from JSON (AWS JSON log format / Powertools)
+            // Extract log level from JSON (AWS JSON log format / Powertools).
+            // Try "level" first (standard), then fall back to "status" (Datadog convention).
             let status = json_obj
                 .get("level")
+                .or_else(|| json_obj.get("status"))
                 .and_then(|v| v.as_str())
                 .and_then(map_log_level_to_status)
                 .map_or(
@@ -1956,4 +1958,168 @@ mod tests {
         let intake_log = processor.get_intake_log(lambda_message).unwrap();
         assert_eq!(intake_log.message.status, "debug");
     }
+
+    #[tokio::test]
+    async fn test_get_intake_log_non_string_level_defaults_to_info() {
+        let config = Arc::new(config::Config {
+            service: Some("test-service".to_string()),
+            tags: HashMap::from([("test".to_string(), "tags".to_string())]),
+            ..config::Config::default()
+        });
+
+        let tags_provider = Arc::new(provider::Provider::new(
+            Arc::clone(&config),
+            LAMBDA_RUNTIME_SLUG.to_string(),
+            &HashMap::from([("function_arn".to_string(), "test-arn".to_string())]),
+        ));
+
+        let (tx, _rx) = tokio::sync::mpsc::channel(2);
+        let mut processor =
+            LambdaProcessor::new(tags_provider, Arc::clone(&config), tx.clone(), false);
+
+        // Set request_id
+        let start_event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 47).unwrap(),
+            record: TelemetryRecord::PlatformStart {
+                request_id: "test-request-id".to_string(),
+                version: Some("test".to_string()),
+            },
+        };
+        let start_msg = processor.get_message(start_event).await.unwrap();
+        processor.get_intake_log(start_msg).unwrap();
+
+        // level as a number
+        let event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 48).unwrap(),
+            record: TelemetryRecord::Function(Value::String(
+                r#"{"level":42,"message":"numeric level"}"#.to_string(),
+            )),
+        };
+        let lambda_message = processor.get_message(event).await.unwrap();
+        let intake_log = processor.get_intake_log(lambda_message).unwrap();
+        assert_eq!(intake_log.message.status, "info");
+
+        // level as null
+        let event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 49).unwrap(),
+            record: TelemetryRecord::Function(Value::String(
+                r#"{"level":null,"message":"null level"}"#.to_string(),
+            )),
+        };
+        let lambda_message = processor.get_message(event).await.unwrap();
+        let intake_log = processor.get_intake_log(lambda_message).unwrap();
+        assert_eq!(intake_log.message.status, "info");
+
+        // level as a boolean
+        let event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 50).unwrap(),
+            record: TelemetryRecord::Function(Value::String(
+                r#"{"level":true,"message":"bool level"}"#.to_string(),
+            )),
+        };
+        let lambda_message = processor.get_message(event).await.unwrap();
+        let intake_log = processor.get_intake_log(lambda_message).unwrap();
+        assert_eq!(intake_log.message.status, "info");
+    }
+
+    #[tokio::test]
+    async fn test_get_intake_log_ddtags_and_level_combined() {
+        let config = Arc::new(config::Config {
+            service: Some("test-service".to_string()),
+            tags: HashMap::from([("test".to_string(), "tags".to_string())]),
+            ..config::Config::default()
+        });
+
+        let tags_provider = Arc::new(provider::Provider::new(
+            Arc::clone(&config),
+            LAMBDA_RUNTIME_SLUG.to_string(),
+            &HashMap::from([("function_arn".to_string(), "test-arn".to_string())]),
+        ));
+
+        let (tx, _rx) = tokio::sync::mpsc::channel(2);
+        let mut processor =
+            LambdaProcessor::new(tags_provider, Arc::clone(&config), tx.clone(), false);
+
+        // Set request_id
+        let start_event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 47).unwrap(),
+            record: TelemetryRecord::PlatformStart {
+                request_id: "test-request-id".to_string(),
+                version: Some("test".to_string()),
+            },
+        };
+        let start_msg = processor.get_message(start_event).await.unwrap();
+        processor.get_intake_log(start_msg).unwrap();
+
+        // JSON log with both ddtags and level
+        let event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 48).unwrap(),
+            record: TelemetryRecord::Function(Value::String(
+                r#"{"level":"WARN","message":"warning with tags","ddtags":"env:staging,team:backend"}"#
+                    .to_string(),
+            )),
+        };
+        let lambda_message = processor.get_message(event).await.unwrap();
+        let intake_log = processor.get_intake_log(lambda_message).unwrap();
+
+        // Level should be extracted
+        assert_eq!(intake_log.message.status, "warn");
+        // Tags should be extracted and appended
+        assert!(intake_log.tags.contains("env:staging"));
+        assert!(intake_log.tags.contains("team:backend"));
+        // ddtags should be removed from the message
+        assert!(!intake_log.message.message.contains("ddtags"));
+    }
+
+    #[tokio::test]
+    async fn test_get_intake_log_status_field_fallback() {
+        let config = Arc::new(config::Config {
+            service: Some("test-service".to_string()),
+            tags: HashMap::from([("test".to_string(), "tags".to_string())]),
+            ..config::Config::default()
+        });
+
+        let tags_provider = Arc::new(provider::Provider::new(
+            Arc::clone(&config),
+            LAMBDA_RUNTIME_SLUG.to_string(),
+            &HashMap::from([("function_arn".to_string(), "test-arn".to_string())]),
+        ));
+
+        let (tx, _rx) = tokio::sync::mpsc::channel(2);
+        let mut processor =
+            LambdaProcessor::new(tags_provider, Arc::clone(&config), tx.clone(), false);
+
+        // Set request_id
+        let start_event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 47).unwrap(),
+            record: TelemetryRecord::PlatformStart {
+                request_id: "test-request-id".to_string(),
+                version: Some("test".to_string()),
+            },
+        };
+        let start_msg = processor.get_message(start_event).await.unwrap();
+        processor.get_intake_log(start_msg).unwrap();
+
+        // JSON log with "status" field instead of "level" (Datadog convention)
+        let event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 48).unwrap(),
+            record: TelemetryRecord::Function(Value::String(
+                r#"{"status":"error","message":"something failed"}"#.to_string(),
+            )),
+        };
+        let lambda_message = processor.get_message(event).await.unwrap();
+        let intake_log = processor.get_intake_log(lambda_message).unwrap();
+        assert_eq!(intake_log.message.status, "error");
+
+        // "level" takes priority over "status" when both are present
+        let event = TelemetryEvent {
+            time: Utc.with_ymd_and_hms(2023, 1, 7, 3, 23, 49).unwrap(),
+            record: TelemetryRecord::Function(Value::String(
+                r#"{"level":"WARN","status":"error","message":"both fields"}"#.to_string(),
+            )),
+        };
+        let lambda_message = processor.get_message(event).await.unwrap();
+        let intake_log = processor.get_intake_log(lambda_message).unwrap();
+        assert_eq!(intake_log.message.status, "warn");
+    }
 }