camel_component_log/
lib.rs

1//! Log component for rust-camel — routes exchange body and headers to the
2//! `tracing` subsystem at a configurable log level, primarily for debugging.
3//!
4//! Main types: `LogComponent`, `LogEndpoint`, `LogProducer`, `LogLevel`.
5
6use std::future::Future;
7use std::pin::Pin;
8use std::str::FromStr;
9use std::sync::Arc;
10use std::sync::atomic::{AtomicUsize, Ordering};
11use std::task::{Context, Poll};
12
13use tower::Service;
14use tracing::{debug, error, info, trace, warn};
15
16use camel_api::component_metadata::{
17    ComponentCapabilities, ComponentMetadata, OptionKind, UriOption,
18};
19use camel_component_api::UriConfig;
20use camel_component_api::parse_uri;
21use camel_component_api::{BoxProcessor, CamelError, Exchange};
22use camel_component_api::{Component, Consumer, Endpoint, ProducerContext, RuntimeObservability};
23
24// ---------------------------------------------------------------------------
25// LogLevel
26// ---------------------------------------------------------------------------
27
28/// Log level for the log component.
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
30pub enum LogLevel {
31    Trace,
32    Debug,
33    #[default]
34    Info,
35    Warn,
36    Error,
37}
38
39impl FromStr for LogLevel {
40    type Err = String;
41
42    fn from_str(s: &str) -> Result<Self, Self::Err> {
43        parse_log_level(s).map_err(|e| e.to_string())
44    }
45}
46
47fn parse_log_level(s: &str) -> Result<LogLevel, CamelError> {
48    match s.to_uppercase().as_str() {
49        "TRACE" => Ok(LogLevel::Trace),
50        "DEBUG" => Ok(LogLevel::Debug),
51        "INFO" => Ok(LogLevel::Info),
52        "WARN" | "WARNING" => Ok(LogLevel::Warn),
53        "ERROR" => Ok(LogLevel::Error),
54        _ => Err(CamelError::Config(format!(
55            "unknown log level: '{}'. Valid: TRACE, DEBUG, INFO, WARN, ERROR",
56            s
57        ))),
58    }
59}
60
61// ---------------------------------------------------------------------------
62// LogConfig
63// ---------------------------------------------------------------------------
64
65/// Configuration parsed from a log URI.
66///
67/// Format: `log:category?level=info&showHeaders=true&showBody=true&logMask=false&showStreamInfo=false&groupSize=10`
68#[derive(Debug, Clone)]
69pub struct LogConfig {
70    /// Log category (the path portion of the URI).
71    pub category: String,
72    /// Log level. Default: Info.
73    pub level: LogLevel,
74    /// Whether to include headers in the log output.
75    pub show_headers: bool,
76    /// Whether to include the body in the log output.
77    pub show_body: bool,
78    /// Maximum number of characters for the body in log output.
79    /// Bodies longer than this are truncated. `None` means no limit.
80    pub max_chars: Option<usize>,
81    /// When true, redact sensitive headers and body in log output.
82    /// Headers matching `/(?i)(password|secret|token|key|auth|credential)/` → `[REDACTED]`.
83    /// Body → `[Body redacted by logMask]`.
84    pub log_mask: bool,
85    /// When true, show stream origin metadata. When false (default), show `[Stream]`.
86    pub show_stream_info: bool,
87    /// When set, only emit a log every `n` exchanges (group logging).
88    /// The log message includes the exchange count.
89    pub group_size: Option<usize>,
90}
91
92impl UriConfig for LogConfig {
93    fn scheme() -> &'static str {
94        "log"
95    }
96
97    fn from_uri(uri: &str) -> Result<Self, CamelError> {
98        let parts = parse_uri(uri)?;
99        Self::from_components(parts)
100    }
101
102    fn from_components(parts: camel_component_api::UriComponents) -> Result<Self, CamelError> {
103        if parts.scheme != Self::scheme() {
104            return Err(CamelError::InvalidUri(format!(
105                "expected scheme '{}' but got '{}'",
106                Self::scheme(),
107                parts.scheme
108            )));
109        }
110
111        let level = match parts.params.get("level") {
112            Some(raw) => parse_log_level(raw)?,
113            None => LogLevel::Info,
114        };
115
116        let show_headers = match parts.params.get("showHeaders") {
117            Some(raw) => raw.parse::<bool>().map_err(|_| {
118                CamelError::InvalidUri(format!("invalid boolean value for showHeaders: {raw}"))
119            })?,
120            None => false,
121        };
122
123        let show_body = match parts.params.get("showBody") {
124            Some(raw) => raw.parse::<bool>().map_err(|_| {
125                CamelError::InvalidUri(format!("invalid boolean value for showBody: {raw}"))
126            })?,
127            None => true,
128        };
129
130        let max_chars = match parts.params.get("maxChars") {
131            Some(raw) => Some(raw.parse::<usize>().map_err(|_| {
132                CamelError::InvalidUri(format!("invalid integer value for maxChars: {raw}"))
133            })?),
134            None => None,
135        };
136
137        let log_mask = match parts.params.get("logMask") {
138            Some(raw) => raw.parse::<bool>().map_err(|_| {
139                CamelError::InvalidUri(format!("invalid boolean value for logMask: {raw}"))
140            })?,
141            None => false,
142        };
143
144        let show_stream_info = match parts.params.get("showStreamInfo") {
145            Some(raw) => raw.parse::<bool>().map_err(|_| {
146                CamelError::InvalidUri(format!("invalid boolean value for showStreamInfo: {raw}"))
147            })?,
148            None => false,
149        };
150
151        let group_size = match parts.params.get("groupSize") {
152            Some(raw) => Some(raw.parse::<usize>().map_err(|_| {
153                CamelError::InvalidUri(format!("invalid integer value for groupSize: {raw}"))
154            })?),
155            None => None,
156        };
157
158        Ok(Self {
159            category: parts.path,
160            level,
161            show_headers,
162            show_body,
163            max_chars,
164            log_mask,
165            show_stream_info,
166            group_size,
167        })
168    }
169}
170
171// ---------------------------------------------------------------------------
172// LogComponent
173// ---------------------------------------------------------------------------
174
175/// The Log component logs exchange information using `tracing`.
176pub struct LogComponent;
177
178impl LogComponent {
179    pub fn new() -> Self {
180        Self
181    }
182}
183
184impl Default for LogComponent {
185    fn default() -> Self {
186        Self::new()
187    }
188}
189
190impl Component for LogComponent {
191    fn scheme(&self) -> &str {
192        "log"
193    }
194
195    fn metadata(&self) -> ComponentMetadata {
196        ComponentMetadata {
197            scheme: "log".to_string(),
198            version: env!("CARGO_PKG_VERSION").to_string(),
199            description: "Logs exchanges at a configurable level".to_string(),
200            uri_syntax: "log:category?level=info&showHeaders=false".to_string(),
201            capabilities: ComponentCapabilities {
202                supports_producer: true,
203                ..Default::default()
204            },
205            uri_options: vec![
206                UriOption::new(
207                    "level",
208                    "Log level",
209                    OptionKind::Enum(vec![
210                        "trace".to_string(),
211                        "debug".to_string(),
212                        "info".to_string(),
213                        "warn".to_string(),
214                        "error".to_string(),
215                    ]),
216                )
217                .with_default("info"),
218                UriOption::new(
219                    "showHeaders",
220                    "Include headers in log output",
221                    OptionKind::Bool,
222                )
223                .with_default("false"),
224                UriOption::new("showBody", "Include body in log output", OptionKind::Bool)
225                    .with_default("true"),
226                UriOption::new(
227                    "maxChars",
228                    "Truncate body to N characters. Omit for no limit",
229                    OptionKind::Int,
230                ),
231                UriOption::new(
232                    "logMask",
233                    "Redact sensitive headers and body",
234                    OptionKind::Bool,
235                )
236                .with_default("false"),
237                UriOption::new(
238                    "showStreamInfo",
239                    "Show stream origin info",
240                    OptionKind::Bool,
241                )
242                .with_default("false"),
243                UriOption::new(
244                    "groupSize",
245                    "Emit log every N exchanges for group logging",
246                    OptionKind::Int,
247                ),
248            ],
249            ..ComponentMetadata::minimal("log")
250        }
251    }
252
253    fn create_endpoint(
254        &self,
255        uri: &str,
256        _ctx: &dyn camel_component_api::ComponentContext,
257    ) -> Result<Box<dyn Endpoint>, CamelError> {
258        let config = LogConfig::from_uri(uri)?;
259        Ok(Box::new(LogEndpoint {
260            uri: uri.to_string(),
261            config,
262        }))
263    }
264}
265
266// ---------------------------------------------------------------------------
267// LogEndpoint
268// ---------------------------------------------------------------------------
269
270struct LogEndpoint {
271    uri: String,
272    config: LogConfig,
273}
274
275impl Endpoint for LogEndpoint {
276    fn uri(&self) -> &str {
277        &self.uri
278    }
279
280    fn create_consumer(
281        &self,
282        _rt: Arc<dyn RuntimeObservability>,
283    ) -> Result<Box<dyn Consumer>, CamelError> {
284        Err(CamelError::EndpointCreationFailed(
285            "log endpoint does not support consumers".to_string(),
286        ))
287    }
288
289    fn create_producer(
290        &self,
291        _rt: Arc<dyn RuntimeObservability>,
292        _ctx: &ProducerContext,
293    ) -> Result<BoxProcessor, CamelError> {
294        Ok(BoxProcessor::new(LogProducer::new(self.config.clone())))
295    }
296}
297
298// ---------------------------------------------------------------------------
299// LogProducer
300// ---------------------------------------------------------------------------
301
302#[derive(Clone)]
303struct LogProducer {
304    config: LogConfig,
305    exchange_count: Arc<AtomicUsize>,
306}
307
308impl LogProducer {
309    fn new(config: LogConfig) -> Self {
310        Self {
311            config,
312            exchange_count: Arc::new(AtomicUsize::new(0)),
313        }
314    }
315
316    /// Returns true if the header key matches sensitive patterns.
317    fn is_sensitive_header(key: &str) -> bool {
318        let lower = key.to_lowercase();
319        let sensitive_keywords = [
320            "password",
321            "passwd",
322            "secret",
323            "token",
324            "apikey",
325            "api-key",
326            "api_key",
327            "authorization",
328            "auth",
329            "credential",
330            "private",
331            "signature",
332        ];
333        sensitive_keywords.iter().any(|kw| {
334            lower == *kw
335                || lower.ends_with(&format!("-{kw}"))
336                || lower.ends_with(&format!("_{kw}"))
337                || lower.starts_with(&format!("{kw}-"))
338                || lower.starts_with(&format!("{kw}_"))
339        })
340    }
341
342    fn format_exchange(&self, exchange: &Exchange, count: usize) -> String {
343        let mut parts = Vec::new();
344
345        if self.config.show_body {
346            let body_str = if self.config.log_mask {
347                "[Body redacted by logMask]".to_string()
348            } else {
349                match &exchange.input.body {
350                    camel_component_api::Body::Empty => "[empty]".to_string(),
351                    camel_component_api::Body::Text(s) => s.clone(),
352                    camel_component_api::Body::Json(v) => v.to_string(),
353                    camel_component_api::Body::Xml(s) => s.clone(),
354                    camel_component_api::Body::Bytes(b) => format!("[{} bytes]", b.len()),
355                    camel_component_api::Body::Stream(s) => {
356                        if self.config.show_stream_info {
357                            format!("[Stream: origin={:?}]", s.metadata.origin)
358                        } else {
359                            "[Stream]".to_string()
360                        }
361                    }
362                }
363            };
364
365            let mut body_str = body_str;
366            if let Some(limit) = self.config.max_chars
367                && body_str.len() > limit
368            {
369                body_str.truncate(limit);
370            }
371
372            parts.push(format!("Body: {body_str}"));
373        }
374
375        if self.config.show_headers && !exchange.input.headers.is_empty() {
376            let headers: Vec<String> = exchange
377                .input
378                .headers
379                .iter()
380                .map(|(k, v)| {
381                    if self.config.log_mask && Self::is_sensitive_header(k) {
382                        format!("{k}=[REDACTED]")
383                    } else {
384                        format!("{k}={v}")
385                    }
386                })
387                .collect();
388            parts.push(format!("Headers: {{{}}}", headers.join(", ")));
389        }
390
391        if parts.is_empty() {
392            format!("[{}] Exchange received", self.config.category)
393        } else if self.config.group_size.is_some() {
394            format!(
395                "[{}] Group of {count}: {}",
396                self.config.category,
397                parts.join(" | ")
398            )
399        } else {
400            format!("[{}] {}", self.config.category, parts.join(" | "))
401        }
402    }
403}
404
405impl Service<Exchange> for LogProducer {
406    type Response = Exchange;
407    type Error = CamelError;
408    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
409
410    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
411        Poll::Ready(Ok(()))
412    }
413
414    fn call(&mut self, exchange: Exchange) -> Self::Future {
415        let count = self.exchange_count.fetch_add(1, Ordering::Relaxed) + 1;
416
417        // Group logging: only emit every group_size exchanges
418        if let Some(group_size) = self.config.group_size
419            && !count.is_multiple_of(group_size)
420        {
421            return Box::pin(async move { Ok(exchange) });
422        }
423
424        let msg = self.format_exchange(&exchange, count);
425        let level = self.config.level;
426
427        Box::pin(async move {
428            match level {
429                LogLevel::Trace => trace!("{msg}"),
430                LogLevel::Debug => debug!("{msg}"),
431                LogLevel::Info => info!("{msg}"),
432                LogLevel::Warn => warn!("{msg}"),
433                LogLevel::Error => error!("{msg}"),
434            }
435
436            Ok(exchange)
437        })
438    }
439}
440
441// ---------------------------------------------------------------------------
442// Tests
443// ---------------------------------------------------------------------------
444
445#[cfg(test)]
446mod tests {
447    use camel_component_api::test_support::PanicRuntimeObservability;
448    fn rt() -> std::sync::Arc<dyn camel_component_api::RuntimeObservability> {
449        std::sync::Arc::new(PanicRuntimeObservability)
450    }
451    use super::*;
452    use camel_component_api::Body;
453    use camel_component_api::Message;
454    use camel_component_api::NoOpComponentContext;
455    use tower::ServiceExt;
456
457    fn test_producer_ctx() -> ProducerContext {
458        ProducerContext::new()
459    }
460
461    #[test]
462    fn test_log_config_defaults() {
463        let config = LogConfig::from_uri("log:myCategory").unwrap();
464        assert_eq!(config.category, "myCategory");
465        assert_eq!(config.level, LogLevel::Info);
466        assert!(!config.show_headers);
467        assert!(config.show_body);
468    }
469
470    #[test]
471    fn test_log_config_with_params() {
472        let config =
473            LogConfig::from_uri("log:app?level=debug&showHeaders=true&showBody=false").unwrap();
474        assert_eq!(config.category, "app");
475        assert_eq!(config.level, LogLevel::Debug);
476        assert!(config.show_headers);
477        assert!(!config.show_body);
478    }
479
480    #[test]
481    fn test_log_config_wrong_scheme() {
482        let result = LogConfig::from_uri("timer:tick");
483        assert!(result.is_err());
484    }
485
486    #[test]
487    fn test_log_component_scheme() {
488        let component = LogComponent::new();
489        assert_eq!(component.scheme(), "log");
490    }
491
492    #[test]
493    fn test_log_component_default() {
494        let component = LogComponent;
495        assert_eq!(component.scheme(), "log");
496    }
497
498    #[test]
499    fn test_log_level_from_str_variants() {
500        assert_eq!("trace".parse::<LogLevel>().unwrap(), LogLevel::Trace);
501        assert_eq!("DEBUG".parse::<LogLevel>().unwrap(), LogLevel::Debug);
502        assert_eq!("Info".parse::<LogLevel>().unwrap(), LogLevel::Info);
503        assert_eq!("warning".parse::<LogLevel>().unwrap(), LogLevel::Warn);
504        assert_eq!("error".parse::<LogLevel>().unwrap(), LogLevel::Error);
505    }
506
507    #[test]
508    fn test_log_level_from_str_invalid() {
509        let err = "nope".parse::<LogLevel>().unwrap_err();
510        assert_eq!(
511            err,
512            "Configuration error: unknown log level: 'nope'. Valid: TRACE, DEBUG, INFO, WARN, ERROR"
513        );
514    }
515
516    #[test]
517    fn test_log_config_invalid_level_rejected() {
518        let err = LogConfig::from_uri("log:test?level=invalid").unwrap_err();
519        assert!(
520            err.to_string()
521                .contains("unknown log level: 'invalid'. Valid: TRACE, DEBUG, INFO, WARN, ERROR")
522        );
523    }
524
525    #[test]
526    fn test_valid_log_levels_accepted() {
527        assert!(parse_log_level("DEBUG").is_ok());
528        assert!(parse_log_level("info").is_ok());
529        assert!(parse_log_level("WARN").is_ok());
530        assert!(parse_log_level("WARNING").is_ok());
531    }
532
533    #[test]
534    fn test_invalid_log_level_rejected() {
535        assert!(parse_log_level("VERBOSE").is_err());
536        assert!(parse_log_level("").is_err());
537        assert!(parse_log_level("log").is_err());
538    }
539
540    #[test]
541    fn test_log_endpoint_uri() {
542        let component = LogComponent::new();
543        let endpoint = component
544            .create_endpoint("log:uri-check", &NoOpComponentContext)
545            .unwrap();
546        assert_eq!(endpoint.uri(), "log:uri-check");
547    }
548
549    #[test]
550    fn test_log_endpoint_no_consumer() {
551        let component = LogComponent::new();
552        let endpoint = component
553            .create_endpoint("log:info", &NoOpComponentContext)
554            .unwrap();
555        assert!(endpoint.create_consumer(rt()).is_err());
556    }
557
558    #[test]
559    fn test_log_endpoint_creates_producer() {
560        let ctx = test_producer_ctx();
561        let component = LogComponent::new();
562        let endpoint = component
563            .create_endpoint("log:info", &NoOpComponentContext)
564            .unwrap();
565        assert!(endpoint.create_producer(rt(), &ctx).is_ok());
566    }
567
568    #[tokio::test]
569    async fn test_log_producer_processes_exchange() {
570        let ctx = test_producer_ctx();
571        let component = LogComponent::new();
572        let endpoint = component
573            .create_endpoint("log:test?showHeaders=true", &NoOpComponentContext)
574            .unwrap();
575        let producer = endpoint.create_producer(rt(), &ctx).unwrap();
576
577        let mut exchange = Exchange::new(Message::new("hello world"));
578        exchange
579            .input
580            .set_header("source", serde_json::Value::String("test".into()));
581
582        let result = producer.oneshot(exchange).await.unwrap();
583        // Log producer passes exchange through unchanged
584        assert_eq!(result.input.body.as_text(), Some("hello world"));
585    }
586
587    #[test]
588    fn test_format_exchange_without_body_or_headers() {
589        let producer = LogProducer::new(LogConfig {
590            category: "cat".to_string(),
591            level: LogLevel::Info,
592            show_headers: false,
593            show_body: false,
594            max_chars: None,
595            log_mask: false,
596            show_stream_info: false,
597            group_size: None,
598        });
599        let exchange = Exchange::new(Message::new("ignored"));
600        let formatted = producer.format_exchange(&exchange, 1);
601        assert_eq!(formatted, "[cat] Exchange received");
602    }
603
604    #[test]
605    fn test_format_exchange_body_variants() {
606        let base = LogProducer::new(LogConfig {
607            category: "cat".to_string(),
608            level: LogLevel::Info,
609            show_headers: false,
610            show_body: true,
611            max_chars: None,
612            log_mask: false,
613            show_stream_info: true,
614            group_size: None,
615        });
616
617        let empty = Exchange::new(Message::default());
618        assert!(base.format_exchange(&empty, 1).contains("Body: [empty]"));
619
620        let mut json_msg = Message::new("");
621        json_msg.body = Body::Json(serde_json::json!({"k":"v"}));
622        let json_ex = Exchange::new(json_msg);
623        assert!(
624            base.format_exchange(&json_ex, 2)
625                .contains("Body: {\"k\":\"v\"}")
626        );
627
628        let mut xml_msg = Message::new("");
629        xml_msg.body = Body::Xml("<a/>".to_string());
630        let xml_ex = Exchange::new(xml_msg);
631        assert!(base.format_exchange(&xml_ex, 3).contains("Body: <a/>"));
632
633        let mut bytes_msg = Message::new("");
634        bytes_msg.body = Body::Bytes(b"abc".to_vec().into());
635        let bytes_ex = Exchange::new(bytes_msg);
636        assert!(
637            base.format_exchange(&bytes_ex, 4)
638                .contains("Body: [3 bytes]")
639        );
640    }
641
642    #[test]
643    fn test_log_truncates_large_body() {
644        let producer = LogProducer::new(LogConfig {
645            category: "trunc".to_string(),
646            level: LogLevel::Info,
647            show_headers: false,
648            show_body: true,
649            max_chars: Some(10),
650            log_mask: false,
651            show_stream_info: false,
652            group_size: None,
653        });
654
655        let long_body = "a".repeat(100);
656        let exchange = Exchange::new(Message::new(long_body));
657        let formatted = producer.format_exchange(&exchange, 1);
658
659        // Extract the body part from "[trunc] Body: ..."
660        let body_part = formatted.split_once("Body: ").unwrap().1;
661        assert!(
662            body_part.len() <= 10,
663            "expected body <= 10 chars, got {} chars: {body_part:?}",
664            body_part.len()
665        );
666    }
667
668    #[test]
669    fn test_log_no_truncation_when_max_chars_unset() {
670        let producer = LogProducer::new(LogConfig {
671            category: "notrunc".to_string(),
672            level: LogLevel::Info,
673            show_headers: false,
674            show_body: true,
675            max_chars: None,
676            log_mask: false,
677            show_stream_info: false,
678            group_size: None,
679        });
680
681        let long_body = "b".repeat(200);
682        let exchange = Exchange::new(Message::new(long_body));
683        let formatted = producer.format_exchange(&exchange, 1);
684
685        let body_part = formatted.split_once("Body: ").unwrap().1;
686        assert_eq!(body_part.len(), 200);
687    }
688
689    #[test]
690    fn test_log_config_max_chars_param() {
691        let config = LogConfig::from_uri("log:test?maxChars=50").unwrap();
692        assert_eq!(config.max_chars, Some(50));
693    }
694
695    #[test]
696    fn test_log_config_max_chars_default_unset() {
697        let config = LogConfig::from_uri("log:test").unwrap();
698        assert_eq!(config.max_chars, None);
699    }
700
701    // --- LOG-007: logMask tests ---
702
703    #[test]
704    fn test_log_config_log_mask_param() {
705        let config = LogConfig::from_uri("log:test?logMask=true").unwrap();
706        assert!(config.log_mask);
707    }
708
709    #[test]
710    fn test_log_config_log_mask_default_false() {
711        let config = LogConfig::from_uri("log:test").unwrap();
712        assert!(!config.log_mask);
713    }
714
715    #[test]
716    fn test_log_mask_redacts_sensitive_headers() {
717        let producer = LogProducer::new(LogConfig {
718            category: "cat".to_string(),
719            level: LogLevel::Info,
720            show_headers: true,
721            show_body: false,
722            max_chars: None,
723            log_mask: true,
724            show_stream_info: false,
725            group_size: None,
726        });
727
728        let mut exchange = Exchange::new(Message::new("body"));
729        exchange.input.set_header(
730            "X-Auth-Token",
731            serde_json::Value::String("secret123".into()),
732        );
733        exchange
734            .input
735            .set_header("password", serde_json::Value::String("hunter2".into()));
736        exchange
737            .input
738            .set_header("ApiKey", serde_json::Value::String("abc".into()));
739        exchange
740            .input
741            .set_header("normal-header", serde_json::Value::String("visible".into()));
742        exchange.input.set_header(
743            "user-credential",
744            serde_json::Value::String("sensitive".into()),
745        );
746        exchange
747            .input
748            .set_header("secret-value", serde_json::Value::String("hidden".into()));
749
750        let formatted = producer.format_exchange(&exchange, 1);
751        assert!(
752            formatted.contains("X-Auth-Token=[REDACTED]"),
753            "auth header must be redacted: {formatted}"
754        );
755        assert!(
756            formatted.contains("password=[REDACTED]"),
757            "password header must be redacted: {formatted}"
758        );
759        assert!(
760            formatted.contains("ApiKey=[REDACTED]"),
761            "key header must be redacted: {formatted}"
762        );
763        assert!(
764            formatted.contains("normal-header=\"visible\""),
765            "normal header must be visible: {formatted}"
766        );
767        assert!(
768            formatted.contains("user-credential=[REDACTED]"),
769            "credential header must be redacted: {formatted}"
770        );
771        assert!(
772            formatted.contains("secret-value=[REDACTED]"),
773            "secret header must be redacted: {formatted}"
774        );
775    }
776
777    #[test]
778    fn test_log_mask_redacts_body() {
779        let producer = LogProducer::new(LogConfig {
780            category: "cat".to_string(),
781            level: LogLevel::Info,
782            show_headers: false,
783            show_body: true,
784            max_chars: None,
785            log_mask: true,
786            show_stream_info: false,
787            group_size: None,
788        });
789
790        let exchange = Exchange::new(Message::new("sensitive body content"));
791        let formatted = producer.format_exchange(&exchange, 1);
792        assert!(
793            formatted.contains("[Body redacted by logMask]"),
794            "body must be redacted: {formatted}"
795        );
796        assert!(
797            !formatted.contains("sensitive body content"),
798            "body content must not appear: {formatted}"
799        );
800    }
801
802    #[test]
803    fn test_log_mask_off_shows_data() {
804        let producer = LogProducer::new(LogConfig {
805            category: "cat".to_string(),
806            level: LogLevel::Info,
807            show_headers: true,
808            show_body: true,
809            max_chars: None,
810            log_mask: false,
811            show_stream_info: false,
812            group_size: None,
813        });
814
815        let mut exchange = Exchange::new(Message::new("visible body"));
816        exchange
817            .input
818            .set_header("password", serde_json::Value::String("hunter2".into()));
819
820        let formatted = producer.format_exchange(&exchange, 1);
821        assert!(
822            formatted.contains("visible body"),
823            "body must be visible when mask off: {formatted}"
824        );
825        assert!(
826            formatted.contains("hunter2"),
827            "header value must be visible when mask off: {formatted}"
828        );
829    }
830
831    // --- LOG-002: showStreamInfo tests ---
832
833    #[test]
834    fn test_log_stream_show_info() {
835        // show_stream_info = false → just [Stream]
836        let producer_no_info = LogProducer::new(LogConfig {
837            category: "cat".to_string(),
838            level: LogLevel::Info,
839            show_headers: false,
840            show_body: true,
841            max_chars: None,
842            log_mask: false,
843            show_stream_info: false,
844            group_size: None,
845        });
846
847        let mut msg = Message::new("");
848        msg.body = Body::Stream(camel_component_api::StreamBody {
849            stream: std::sync::Arc::new(tokio::sync::Mutex::new(None)),
850            metadata: camel_component_api::StreamMetadata {
851                origin: Some("file:///data/test.txt".to_string()),
852                ..Default::default()
853            },
854        });
855        let exchange = Exchange::new(msg);
856        let formatted = producer_no_info.format_exchange(&exchange, 1);
857        assert!(
858            formatted.contains("Body: [Stream]"),
859            "must show [Stream] when show_stream_info=false: {formatted}"
860        );
861
862        // show_stream_info = true → [Stream: origin=...]
863        let producer_with_info = LogProducer::new(LogConfig {
864            category: "cat".to_string(),
865            level: LogLevel::Info,
866            show_headers: false,
867            show_body: true,
868            max_chars: None,
869            log_mask: false,
870            show_stream_info: true,
871            group_size: None,
872        });
873
874        let formatted = producer_with_info.format_exchange(&exchange, 1);
875        assert!(
876            formatted.contains("Body: [Stream: origin=Some(\"file:///data/test.txt\")]"),
877            "must show origin when show_stream_info=true: {formatted}"
878        );
879    }
880
881    // --- LOG-008: groupSize tests ---
882
883    #[test]
884    fn test_log_group_size() {
885        let producer = LogProducer::new(LogConfig {
886            category: "cat".to_string(),
887            level: LogLevel::Info,
888            show_headers: false,
889            show_body: true,
890            max_chars: None,
891            log_mask: false,
892            show_stream_info: false,
893            group_size: Some(3),
894        });
895
896        // Count 1 and 2 should not trigger logging, count 3 should
897        let ex1 = Exchange::new(Message::new("first"));
898        let formatted1 = producer.format_exchange(&ex1, 3);
899        assert!(
900            formatted1.contains("Group of 3:"),
901            "group_size=3 must include count: {formatted1}"
902        );
903        assert!(
904            formatted1.contains("Body: first"),
905            "group log must include body: {formatted1}"
906        );
907    }
908
909    #[test]
910    fn test_log_config_group_size_param() {
911        let config = LogConfig::from_uri("log:test?groupSize=10").unwrap();
912        assert_eq!(config.group_size, Some(10));
913    }
914
915    #[test]
916    fn test_log_config_group_size_default_unset() {
917        let config = LogConfig::from_uri("log:test").unwrap();
918        assert_eq!(config.group_size, None);
919    }
920
921    #[test]
922    fn test_log_config_show_stream_info_param() {
923        let config = LogConfig::from_uri("log:test?showStreamInfo=true").unwrap();
924        assert!(config.show_stream_info);
925    }
926
927    #[test]
928    fn test_log_config_show_stream_info_default_false() {
929        let config = LogConfig::from_uri("log:test").unwrap();
930        assert!(!config.show_stream_info);
931    }
932}
camel_component_log/lib.rs

camel_component_log/
lib.rs
