camel_processor/data_format/
csv.rs

1use camel_api::Exchange;
2use camel_api::body::Body;
3use camel_api::data_format::DataFormat;
4use camel_api::error::CamelError;
5use serde::Deserialize;
6
7pub const CAMEL_CSV_HEADER_RECORD: &str = "CamelCsvHeaderRecord";
8
9#[derive(Clone, Copy, Debug, Default, Deserialize)]
10#[serde(rename_all = "snake_case")]
11pub enum RecordSeparator {
12    #[default]
13    Crlf,
14    Lf,
15}
16
17#[derive(Clone, Copy, Debug, Default, Deserialize)]
18#[serde(rename_all = "snake_case")]
19pub enum QuoteMode {
20    All,
21    #[default]
22    Minimal,
23    NonNumeric,
24    None,
25}
26
27/// R4-L9: policy for header/row width mismatch during CSV unmarshal (map mode).
28#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq, Eq)]
29#[serde(rename_all = "snake_case")]
30pub enum WidthMismatchPolicy {
31    /// No warning; silently pad/drop.
32    Lenient,
33    /// One aggregate warning per unmarshal (default — reduces per-row flood).
34    #[default]
35    Warn,
36    /// Error on first mismatch.
37    Error,
38}
39
40#[derive(Clone, Debug, Deserialize)]
41#[serde(default, deny_unknown_fields)]
42pub struct CsvConfig {
43    pub delimiter: char,
44    pub quote: char,
45    pub quote_mode: QuoteMode,
46    pub double_quote: bool,
47    pub escape: Option<char>,
48    pub record_separator: RecordSeparator,
49    pub comment_marker: Option<char>,
50    pub null_string: Option<String>,
51    pub headers: Option<Vec<String>>,
52    pub has_headers: bool,
53    pub skip_header_record: bool,
54    pub capture_header_record: bool,
55    pub ignore_surrounding_spaces: bool,
56    pub trim: bool,
57    pub use_maps: bool,
58    /// Maximum number of data records accepted by `unmarshal` (DoS cap, R3-M1).
59    /// Default 100_000. `None` disables the cap (not recommended for untrusted input).
60    pub max_records: Option<usize>,
61    /// Maximum byte length of a single CSV field accepted by `unmarshal` (DoS cap, R3-M1).
62    /// Default 1_048_576 (1 MiB). `None` disables the cap.
63    pub max_field_size: Option<usize>,
64    /// R4-L9: policy for header/row width mismatch during CSV unmarshal (map mode).
65    pub width_mismatch: WidthMismatchPolicy,
66}
67
68impl Default for CsvConfig {
69    fn default() -> Self {
70        Self {
71            delimiter: ',',
72            quote: '"',
73            quote_mode: QuoteMode::Minimal,
74            double_quote: true,
75            escape: None,
76            record_separator: RecordSeparator::Crlf,
77            comment_marker: None,
78            null_string: None,
79            headers: None,
80            has_headers: true,
81            skip_header_record: false,
82            capture_header_record: false,
83            ignore_surrounding_spaces: true,
84            trim: false,
85            use_maps: true,
86            max_records: Some(100_000),
87            max_field_size: Some(1_048_576),
88            width_mismatch: WidthMismatchPolicy::Warn,
89        }
90    }
91}
92
93impl CsvConfig {
94    pub fn excel() -> Self {
95        Self::default()
96    }
97
98    pub fn tdf() -> Self {
99        Self {
100            delimiter: '\t',
101            record_separator: RecordSeparator::Lf,
102            ..Self::default()
103        }
104    }
105
106    pub fn mysql() -> Self {
107        Self {
108            delimiter: '\t',
109            quote: '\0',
110            record_separator: RecordSeparator::Lf,
111            escape: Some('\\'),
112            ..Self::default()
113        }
114    }
115
116    pub fn delimiter(mut self, c: char) -> Self {
117        self.delimiter = c;
118        self
119    }
120    pub fn quote(mut self, c: char) -> Self {
121        self.quote = c;
122        self
123    }
124    pub fn quote_mode(mut self, m: QuoteMode) -> Self {
125        self.quote_mode = m;
126        self
127    }
128    pub fn double_quote(mut self, b: bool) -> Self {
129        self.double_quote = b;
130        self
131    }
132    pub fn escape(mut self, c: Option<char>) -> Self {
133        self.escape = c;
134        self
135    }
136    pub fn record_separator(mut self, s: RecordSeparator) -> Self {
137        self.record_separator = s;
138        self
139    }
140    pub fn comment_marker(mut self, c: Option<char>) -> Self {
141        self.comment_marker = c;
142        self
143    }
144    pub fn null_string(mut self, s: Option<String>) -> Self {
145        self.null_string = s;
146        self
147    }
148    pub fn headers(mut self, h: Vec<String>) -> Self {
149        self.headers = Some(h);
150        self
151    }
152    pub fn has_headers(mut self, b: bool) -> Self {
153        self.has_headers = b;
154        self
155    }
156    pub fn skip_header_record(mut self, b: bool) -> Self {
157        self.skip_header_record = b;
158        self
159    }
160    pub fn capture_header_record(mut self, b: bool) -> Self {
161        self.capture_header_record = b;
162        self
163    }
164    pub fn ignore_surrounding_spaces(mut self, b: bool) -> Self {
165        self.ignore_surrounding_spaces = b;
166        self
167    }
168    pub fn trim(mut self, b: bool) -> Self {
169        self.trim = b;
170        self
171    }
172    pub fn use_maps(mut self, b: bool) -> Self {
173        self.use_maps = b;
174        self
175    }
176    pub fn max_records(mut self, max: usize) -> Self {
177        self.max_records = Some(max);
178        self
179    }
180    pub fn max_field_size(mut self, max: usize) -> Self {
181        self.max_field_size = Some(max);
182        self
183    }
184    pub fn width_mismatch(mut self, policy: WidthMismatchPolicy) -> Self {
185        self.width_mismatch = policy;
186        self
187    }
188}
189
190#[derive(Clone, Default)]
191pub struct CsvDataFormat {
192    config: CsvConfig,
193}
194
195impl CsvDataFormat {
196    pub fn new(config: CsvConfig) -> Self {
197        Self { config }
198    }
199
200    pub fn config(&self) -> &CsvConfig {
201        &self.config
202    }
203
204    /// Render a single JSON value as a CSV field string.
205    fn field_to_string(value: &serde_json::Value) -> String {
206        match value {
207            serde_json::Value::String(s) => s.clone(),
208            serde_json::Value::Number(n) => n.to_string(),
209            serde_json::Value::Bool(b) => b.to_string(),
210            serde_json::Value::Null => String::new(),
211            serde_json::Value::Array(arr) => arr
212                .iter()
213                .map(|v| v.to_string())
214                .collect::<Vec<_>>()
215                .join(","),
216            serde_json::Value::Object(_) => value.to_string(),
217        }
218    }
219
220    /// Prefix cells starting with CSV formula-injection characters
221    /// (`=`, `+`, `-`, `@`, `\t`, `\r`) with a single quote to
222    /// neutralize them (OWASP CSV Injection prevention).
223    fn neutralize_formula_injection(s: &str) -> String {
224        if s.starts_with('=')
225            || s.starts_with('+')
226            || s.starts_with('-')
227            || s.starts_with('@')
228            || s.starts_with('\t')
229            || s.starts_with('\r')
230        {
231            let mut out = String::with_capacity(s.len() + 1);
232            out.push('\'');
233            out.push_str(s);
234            out
235        } else {
236            s.to_string()
237        }
238    }
239
240    fn write_err(e: csv::Error) -> CamelError {
241        CamelError::TypeConversionFailed(format!("CSV write error: {e}"))
242    }
243
244    /// Enforce the DoS caps (max_records, max_field_size) against a parsed record.
245    /// Returns Err on exceed; `count` is 1-based.
246    fn check_record_caps(
247        record: &csv::StringRecord,
248        count: usize,
249        max_records: Option<usize>,
250        max_field_size: Option<usize>,
251    ) -> Result<(), CamelError> {
252        if let Some(max) = max_records
253            && count > max
254        {
255            return Err(CamelError::TypeConversionFailed(format!(
256                "CSV unmarshal exceeded max_records {max}"
257            )));
258        }
259        if let Some(max_field) = max_field_size {
260            for s in record.iter() {
261                if s.len() > max_field {
262                    return Err(CamelError::TypeConversionFailed(format!(
263                        "CSV unmarshal field length {} exceeds max_field_size {max_field}",
264                        s.len()
265                    )));
266                }
267            }
268        }
269        Ok(())
270    }
271
272    fn unmarshal_internal(&self, body: Body) -> Result<(Body, Option<Vec<String>>), CamelError> {
273        let input_text: String =
274            match body {
275                Body::Text(s) => s,
276                Body::Bytes(b) => String::from_utf8(b.to_vec()).map_err(|e| {
277                    CamelError::TypeConversionFailed(format!("CSV bytes not UTF-8: {e}"))
278                })?,
279                Body::Json(v) => return Ok((Body::Json(v), None)),
280                Body::Stream(_) => return Err(CamelError::TypeConversionFailed(
281                    "cannot unmarshal Body::Stream directly — add 'stream_cache' before this step"
282                        .into(),
283                )),
284                Body::Empty => {
285                    return Err(CamelError::TypeConversionFailed(
286                        "CsvDataFormat::unmarshal expects Body::Text or Body::Bytes".into(),
287                    ));
288                }
289                Body::Xml(_) => {
290                    return Err(CamelError::TypeConversionFailed(
291                        "CsvDataFormat::unmarshal does not accept Body::Xml".into(),
292                    ));
293                }
294                _ => {
295                    return Err(CamelError::TypeConversionFailed(
296                        "CsvDataFormat::unmarshal expects Body::Text or Body::Bytes".into(),
297                    ));
298                }
299            };
300
301        let mut rdr_builder = csv::ReaderBuilder::new();
302        let effective_has_headers = if self.config.headers.is_some() {
303            false
304        } else {
305            self.config.has_headers
306        };
307        rdr_builder
308            .delimiter(self.config.delimiter as u8)
309            .has_headers(effective_has_headers)
310            .flexible(true)
311            .trim(if self.config.trim {
312                csv::Trim::All
313            } else if self.config.ignore_surrounding_spaces {
314                csv::Trim::Fields
315            } else {
316                csv::Trim::None
317            });
318        if let Some(c) = self.config.comment_marker {
319            rdr_builder.comment(Some(c as u8));
320        }
321        let mut rdr = rdr_builder.from_reader(input_text.as_bytes());
322
323        let header_keys: Option<Vec<String>> = if let Some(h) = &self.config.headers {
324            Some(h.clone())
325        } else if self.config.has_headers {
326            Some(
327                rdr.headers()
328                    .map_err(|e| CamelError::TypeConversionFailed(format!("CSV header read: {e}")))?
329                    .iter()
330                    .map(|s| s.to_string())
331                    .collect(),
332            )
333        } else {
334            None
335        };
336
337        let mut records: Vec<serde_json::Value> = Vec::new();
338        if self.config.use_maps {
339            let keys = match &header_keys {
340                Some(k) => k.clone(),
341                None => {
342                    return Err(CamelError::TypeConversionFailed(
343                        "CsvDataFormat::unmarshal with use_maps=true requires has_headers=true or configured headers".into(),
344                    ));
345                }
346            };
347            let mut iter = rdr.records();
348            if self.config.skip_header_record && self.config.headers.is_some() {
349                iter.next();
350            }
351            let mut count: usize = 0;
352            let mut mismatch_count: usize = 0;
353            let mut first_mismatch: Option<(usize, usize, usize)> = None;
354            for result in iter {
355                let record = result
356                    .map_err(|e| CamelError::TypeConversionFailed(format!("CSV parse: {e}")))?;
357                count += 1;
358                Self::check_record_caps(
359                    &record,
360                    count,
361                    self.config.max_records,
362                    self.config.max_field_size,
363                )?;
364                // R4-L9: configurable policy for header/row width mismatch.
365                if record.len() != keys.len() {
366                    match self.config.width_mismatch {
367                        WidthMismatchPolicy::Error => {
368                            return Err(CamelError::TypeConversionFailed(format!(
369                                "CSV record {} width {} differs from header width {}",
370                                count,
371                                record.len(),
372                                keys.len()
373                            )));
374                        }
375                        WidthMismatchPolicy::Warn => {
376                            mismatch_count += 1;
377                            if first_mismatch.is_none() {
378                                first_mismatch = Some((count, keys.len(), record.len()));
379                            }
380                        }
381                        WidthMismatchPolicy::Lenient => {}
382                    }
383                }
384                let mut obj = serde_json::Map::new();
385                for (i, key) in keys.iter().enumerate() {
386                    let val = record.get(i).unwrap_or("");
387                    let parsed = if let Some(ns) = &self.config.null_string {
388                        if val == ns {
389                            serde_json::Value::Null
390                        } else {
391                            serde_json::Value::String(val.to_string())
392                        }
393                    } else {
394                        serde_json::Value::String(val.to_string())
395                    };
396                    obj.insert(key.clone(), parsed);
397                }
398                records.push(serde_json::Value::Object(obj));
399            }
400            // R4-L9: emit one aggregate warn for all width mismatches.
401            if let (
402                WidthMismatchPolicy::Warn,
403                Some((first_record, header_width, first_row_width)),
404            ) = (self.config.width_mismatch, first_mismatch)
405            {
406                tracing::warn!(
407                    first_record,
408                    header_width,
409                    first_row_width,
410                    total_mismatches = mismatch_count,
411                    "CSV records had width mismatch with header; fields were padded or truncated"
412                );
413            }
414        } else {
415            let mut count: usize = 0;
416            for result in rdr.records() {
417                let record = result
418                    .map_err(|e| CamelError::TypeConversionFailed(format!("CSV parse: {e}")))?;
419                count += 1;
420                Self::check_record_caps(
421                    &record,
422                    count,
423                    self.config.max_records,
424                    self.config.max_field_size,
425                )?;
426                let arr: Vec<serde_json::Value> = record
427                    .iter()
428                    .map(|s| serde_json::Value::String(s.to_string()))
429                    .collect();
430                records.push(serde_json::Value::Array(arr));
431            }
432        }
433
434        Ok((Body::Json(serde_json::Value::Array(records)), header_keys))
435    }
436}
437
438impl DataFormat for CsvDataFormat {
439    fn name(&self) -> &str {
440        "csv"
441    }
442
443    fn marshal(&self, body: Body) -> Result<Body, CamelError> {
444        let json = match body {
445            Body::Json(v) => v,
446            Body::Text(_) => {
447                return Err(CamelError::TypeConversionFailed(
448                    "CsvDataFormat::marshal expects Body::Json; use convert_body_to first".into(),
449                ));
450            }
451            Body::Bytes(_) => {
452                return Err(CamelError::TypeConversionFailed(
453                    "CsvDataFormat::marshal expects Body::Json; use convert_body_to first".into(),
454                ));
455            }
456            Body::Stream(_) => {
457                return Err(CamelError::TypeConversionFailed(
458                    "cannot marshal Body::Stream — add 'stream_cache' before this step".into(),
459                ));
460            }
461            Body::Empty => {
462                return Err(CamelError::TypeConversionFailed(
463                    "CsvDataFormat::marshal expects Body::Json".into(),
464                ));
465            }
466            Body::Xml(_) => return Err(CamelError::TypeConversionFailed(
467                "CsvDataFormat::marshal does not accept Body::Xml — use unmarshal(\"xml\") first"
468                    .into(),
469            )),
470            _ => {
471                return Err(CamelError::TypeConversionFailed(
472                    "CsvDataFormat::marshal expects Body::Json".into(),
473                ));
474            }
475        };
476
477        let mut wtr_builder = csv::WriterBuilder::new();
478        wtr_builder
479            .delimiter(self.config.delimiter as u8)
480            .quote(self.config.quote as u8)
481            .double_quote(self.config.double_quote);
482        if let Some(esc) = self.config.escape {
483            wtr_builder.escape(esc as u8);
484        }
485        let quote_style = match self.config.quote_mode {
486            QuoteMode::All => csv::QuoteStyle::Always,
487            QuoteMode::Minimal => csv::QuoteStyle::Necessary,
488            QuoteMode::NonNumeric => csv::QuoteStyle::NonNumeric,
489            QuoteMode::None => csv::QuoteStyle::Never,
490        };
491        wtr_builder.quote_style(quote_style);
492        let terminator = match self.config.record_separator {
493            RecordSeparator::Crlf => csv::Terminator::CRLF,
494            RecordSeparator::Lf => csv::Terminator::Any(b'\n'),
495        };
496        wtr_builder.terminator(terminator);
497
498        let mut output = Vec::new();
499        {
500            let mut wtr = wtr_builder.from_writer(&mut output);
501            match &json {
502                serde_json::Value::Array(arr) => {
503                    if arr.is_empty() {
504                        if let (true, Some(h)) = (self.config.has_headers, &self.config.headers) {
505                            let neutralized: Vec<String> = h
506                                .iter()
507                                .map(|s| Self::neutralize_formula_injection(s))
508                                .collect();
509                            wtr.write_record(neutralized.iter().map(String::as_str))
510                                .map_err(Self::write_err)?;
511                        }
512                    } else {
513                        match &arr[0] {
514                            serde_json::Value::Object(first_obj) => {
515                                let header_keys: Vec<String> = self
516                                    .config
517                                    .headers
518                                    .clone()
519                                    .unwrap_or_else(|| first_obj.keys().cloned().collect());
520                                if self.config.has_headers {
521                                    let neutralized: Vec<String> = header_keys
522                                        .iter()
523                                        .map(|s| Self::neutralize_formula_injection(s))
524                                        .collect();
525                                    wtr.write_record(neutralized.iter().map(String::as_str))
526                                        .map_err(Self::write_err)?;
527                                }
528                                for row in arr {
529                                    if let Some(obj) = row.as_object() {
530                                        let record: Vec<String> = header_keys
531                                            .iter()
532                                            .map(|k| {
533                                                Self::neutralize_formula_injection(
534                                                    &Self::field_to_string(
535                                                        obj.get(k)
536                                                            .unwrap_or(&serde_json::Value::Null),
537                                                    ),
538                                                )
539                                            })
540                                            .collect();
541                                        wtr.write_record(record.iter().map(String::as_str))
542                                            .map_err(Self::write_err)?;
543                                    } else {
544                                        return Err(CamelError::TypeConversionFailed(format!(
545                                            "CsvDataFormat::marshal expected all array elements to be objects, got {:?}",
546                                            row
547                                        )));
548                                    }
549                                }
550                            }
551                            _ => {
552                                for row in arr {
553                                    if let Some(items) = row.as_array() {
554                                        let record: Vec<String> = items
555                                            .iter()
556                                            .map(|v| {
557                                                Self::neutralize_formula_injection(
558                                                    &Self::field_to_string(v),
559                                                )
560                                            })
561                                            .collect();
562                                        wtr.write_record(record.iter().map(String::as_str))
563                                            .map_err(Self::write_err)?;
564                                    } else {
565                                        return Err(CamelError::TypeConversionFailed(format!(
566                                            "CsvDataFormat::marshal expected all array elements to be arrays, got {:?}",
567                                            row
568                                        )));
569                                    }
570                                }
571                            }
572                        }
573                    }
574                }
575                serde_json::Value::Object(obj) => {
576                    let header_keys: Vec<String> = self
577                        .config
578                        .headers
579                        .clone()
580                        .unwrap_or_else(|| obj.keys().cloned().collect());
581                    if self.config.has_headers {
582                        let neutralized: Vec<String> = header_keys
583                            .iter()
584                            .map(|s| Self::neutralize_formula_injection(s))
585                            .collect();
586                        wtr.write_record(neutralized.iter().map(String::as_str))
587                            .map_err(Self::write_err)?;
588                    }
589                    let record: Vec<String> = header_keys
590                        .iter()
591                        .map(|k| {
592                            Self::neutralize_formula_injection(&Self::field_to_string(
593                                obj.get(k).unwrap_or(&serde_json::Value::Null),
594                            ))
595                        })
596                        .collect();
597                    wtr.write_record(record.iter().map(String::as_str))
598                        .map_err(Self::write_err)?;
599                }
600                _ => {
601                    return Err(CamelError::TypeConversionFailed(format!(
602                        "CsvDataFormat::marshal only supports Json Array or Object, got {json:?}"
603                    )));
604                }
605            }
606            wtr.flush()
607                .map_err(|e| CamelError::TypeConversionFailed(format!("CSV flush: {e}")))?;
608        }
609
610        let text = String::from_utf8(output)
611            .map_err(|e| CamelError::TypeConversionFailed(format!("CSV output not UTF-8: {e}")))?;
612        Ok(Body::Text(text))
613    }
614
615    fn unmarshal(&self, body: Body) -> Result<Body, CamelError> {
616        self.unmarshal_internal(body).map(|(b, _)| b)
617    }
618
619    fn unmarshal_in_exchange(
620        &self,
621        exchange: &mut Exchange,
622        body: Body,
623    ) -> Result<Body, CamelError> {
624        let (result, header_keys) = self.unmarshal_internal(body)?;
625        if let Some(keys) = header_keys
626            .filter(|_| self.config.capture_header_record)
627            .filter(|k| !k.is_empty())
628        {
629            exchange.input.headers.insert(
630                CAMEL_CSV_HEADER_RECORD.to_string(),
631                serde_json::Value::Array(keys.into_iter().map(serde_json::Value::String).collect()),
632            );
633        }
634        Ok(result)
635    }
636}
637
638#[cfg(test)]
639mod tests {
640    use super::*;
641    use serde_json::json;
642
643    #[test]
644    fn test_default_config_values() {
645        let cfg = CsvConfig::default();
646        assert_eq!(cfg.delimiter, ',');
647        assert_eq!(cfg.quote, '"');
648        assert!(matches!(cfg.quote_mode, QuoteMode::Minimal));
649        assert!(cfg.double_quote);
650        assert!(cfg.escape.is_none());
651        assert!(matches!(cfg.record_separator, RecordSeparator::Crlf));
652        assert!(cfg.comment_marker.is_none());
653        assert!(cfg.null_string.is_none());
654        assert!(cfg.headers.is_none());
655        assert!(cfg.has_headers);
656        assert!(!cfg.skip_header_record);
657        assert!(!cfg.capture_header_record);
658        assert!(cfg.ignore_surrounding_spaces);
659        assert!(!cfg.trim);
660        assert!(cfg.use_maps);
661    }
662
663    #[test]
664    fn test_tdf_preset_uses_tab() {
665        let cfg = CsvConfig::tdf();
666        assert_eq!(cfg.delimiter, '\t');
667        assert!(matches!(cfg.record_separator, RecordSeparator::Lf));
668    }
669
670    #[test]
671    fn test_mysql_preset_uses_tab_and_escape() {
672        let cfg = CsvConfig::mysql();
673        assert_eq!(cfg.delimiter, '\t');
674        assert_eq!(cfg.escape, Some('\\'));
675    }
676
677    #[test]
678    fn test_builder_methods() {
679        let cfg = CsvConfig::default()
680            .delimiter('|')
681            .headers(vec!["a".into(), "b".into()])
682            .use_maps(false);
683        assert_eq!(cfg.delimiter, '|');
684        assert_eq!(cfg.headers, Some(vec!["a".into(), "b".into()]));
685        assert!(!cfg.use_maps);
686    }
687
688    #[test]
689    fn test_marshal_array_of_objects_with_header() {
690        let df = CsvDataFormat::default();
691        let body = Body::Json(json!([
692            { "name": "Alice", "age": 30 },
693            { "name": "Bob", "age": 25 }
694        ]));
695        let result = df.marshal(body).unwrap();
696        match result {
697            Body::Text(s) => {
698                let lines: Vec<&str> = s.lines().collect();
699                // serde_json key order is not guaranteed (depends on preserve_order
700                // feature); assert the header carries both keys in any order.
701                let header_fields: Vec<&str> = lines[0].split(',').collect();
702                assert_eq!(header_fields.len(), 2);
703                assert!(header_fields.contains(&"name"));
704                assert!(header_fields.contains(&"age"));
705                assert!(lines[1].contains("Alice"));
706                assert!(lines[2].contains("Bob"));
707            }
708            _ => panic!("expected Body::Text"),
709        }
710    }
711
712    #[test]
713    fn test_marshal_array_of_arrays_no_header() {
714        let df = CsvDataFormat::default();
715        let body = Body::Json(json!([["a", "b", "c"], [1, 2, 3]]));
716        let result = df.marshal(body).unwrap();
717        match result {
718            Body::Text(s) => {
719                let lines: Vec<&str> = s.lines().collect();
720                assert_eq!(lines.len(), 2);
721                assert!(lines[0].contains("a"));
722            }
723            _ => panic!("expected Body::Text"),
724        }
725    }
726
727    #[test]
728    fn test_marshal_single_object() {
729        let df = CsvDataFormat::default();
730        let body = Body::Json(json!({ "x": 1, "y": 2 }));
731        let result = df.marshal(body).unwrap();
732        match result {
733            Body::Text(s) => {
734                let lines: Vec<&str> = s.lines().collect();
735                assert_eq!(lines.len(), 2);
736            }
737            _ => panic!("expected Body::Text"),
738        }
739    }
740
741    #[test]
742    fn test_marshal_empty_array_with_headers() {
743        let cfg = CsvConfig::default().headers(vec!["h1".into(), "h2".into()]);
744        let df = CsvDataFormat::new(cfg);
745        let body = Body::Json(json!([]));
746        let result = df.marshal(body).unwrap();
747        match result {
748            Body::Text(s) => {
749                assert_eq!(s.trim(), "h1,h2");
750            }
751            _ => panic!("expected Body::Text"),
752        }
753    }
754
755    #[test]
756    fn test_marshal_empty_array_no_headers() {
757        let cfg = CsvConfig::default().has_headers(false);
758        let df = CsvDataFormat::new(cfg);
759        let body = Body::Json(json!([]));
760        let result = df.marshal(body).unwrap();
761        match result {
762            Body::Text(s) => assert!(s.is_empty()),
763            _ => panic!("expected Body::Text"),
764        }
765    }
766
767    #[test]
768    fn test_marshal_quoted_fields_with_commas() {
769        let df = CsvDataFormat::default();
770        let body = Body::Json(json!([
771            { "name": "Doe, John", "age": 30 }
772        ]));
773        let result = df.marshal(body).unwrap();
774        match result {
775            Body::Text(s) => assert!(s.contains("\"Doe, John\"")),
776            _ => panic!("expected Body::Text"),
777        }
778    }
779
780    #[test]
781    fn test_marshal_text_returns_error() {
782        let df = CsvDataFormat::default();
783        let result = df.marshal(Body::Text("already text".into()));
784        assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
785    }
786
787    #[test]
788    fn test_marshal_stream_returns_error() {
789        use bytes::Bytes;
790        use camel_api::body::{StreamBody, StreamMetadata};
791        let df = CsvDataFormat::default();
792        let empty_stream = futures::stream::empty::<Result<Bytes, CamelError>>();
793        let stream_body = StreamBody {
794            stream: std::sync::Arc::new(tokio::sync::Mutex::new(Some(Box::pin(empty_stream)))),
795            metadata: StreamMetadata::default(),
796        };
797        let result = df.marshal(Body::Stream(stream_body));
798        assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
799    }
800
801    #[test]
802    fn test_marshal_quote_mode_all() {
803        let cfg = CsvConfig::default().quote_mode(QuoteMode::All);
804        let df = CsvDataFormat::new(cfg);
805        let body = Body::Json(json!([{ "x": 1 }]));
806        let result = df.marshal(body).unwrap();
807        match result {
808            Body::Text(s) => {
809                assert!(s.contains("\"x\""));
810            }
811            _ => panic!("expected Body::Text"),
812        }
813    }
814
815    #[test]
816    fn test_marshal_mysql_preset_quote_char_null() {
817        let cfg = CsvConfig::mysql();
818        let df = CsvDataFormat::new(cfg);
819        let body = Body::Json(json!([["a", "b"]]));
820        let result = df.marshal(body);
821        assert!(result.is_ok(), "mysql preset should marshal: {result:?}");
822    }
823
824    #[test]
825    fn test_unmarshal_text_to_json_maps() {
826        let df = CsvDataFormat::default();
827        let body = Body::Text("name,age\nAlice,30\nBob,25".into());
828        let result = df.unmarshal(body).unwrap();
829        match result {
830            Body::Json(v) => {
831                let arr = v.as_array().expect("expected array");
832                assert_eq!(arr.len(), 2);
833                assert_eq!(arr[0]["name"], json!("Alice"));
834                assert_eq!(arr[0]["age"], json!("30"));
835            }
836            _ => panic!("expected Body::Json"),
837        }
838    }
839
840    #[test]
841    fn test_unmarshal_text_to_json_lists() {
842        let cfg = CsvConfig::default().use_maps(false);
843        let df = CsvDataFormat::new(cfg);
844        let body = Body::Text("a,b,c\n1,2,3".into());
845        let result = df.unmarshal(body).unwrap();
846        match result {
847            Body::Json(v) => {
848                let arr = v.as_array().expect("expected array");
849                assert_eq!(arr.len(), 1);
850                assert_eq!(arr[0][0], json!("1"));
851            }
852            _ => panic!("expected Body::Json"),
853        }
854    }
855
856    #[test]
857    fn test_unmarshal_bytes_to_json() {
858        let df = CsvDataFormat::default();
859        let body = Body::Bytes(bytes::Bytes::from_static(b"x,y\n1,2"));
860        let result = df.unmarshal(body).unwrap();
861        match result {
862            Body::Json(v) => {
863                let arr = v.as_array().unwrap();
864                assert_eq!(arr[0]["x"], json!("1"));
865            }
866            _ => panic!("expected Body::Json"),
867        }
868    }
869
870    #[test]
871    fn test_unmarshal_configured_headers() {
872        let cfg = CsvConfig::default().headers(vec!["col1".into(), "col2".into()]);
873        let df = CsvDataFormat::new(cfg);
874        let body = Body::Text("val1,val2".into());
875        let result = df.unmarshal(body).unwrap();
876        match result {
877            Body::Json(v) => {
878                let arr = v.as_array().unwrap();
879                assert_eq!(arr[0]["col1"], json!("val1"));
880            }
881            _ => panic!("expected Body::Json"),
882        }
883    }
884
885    #[test]
886    fn test_unmarshal_skip_header_record_with_configured_headers() {
887        let cfg = CsvConfig::default()
888            .headers(vec!["col1".into(), "col2".into()])
889            .skip_header_record(true);
890        let df = CsvDataFormat::new(cfg);
891        let body = Body::Text("ignored1,ignored2\nval1,val2".into());
892        let result = df.unmarshal(body).unwrap();
893        match result {
894            Body::Json(v) => {
895                let arr = v.as_array().unwrap();
896                assert_eq!(arr.len(), 1);
897                assert_eq!(arr[0]["col1"], json!("val1"));
898            }
899            _ => panic!("expected Body::Json"),
900        }
901    }
902
903    #[test]
904    fn test_unmarshal_quoted_fields() {
905        let df = CsvDataFormat::default();
906        let body = Body::Text("name,note\n\"Doe, John\",hi".into());
907        let result = df.unmarshal(body).unwrap();
908        match result {
909            Body::Json(v) => {
910                let arr = v.as_array().unwrap();
911                assert_eq!(arr[0]["name"], json!("Doe, John"));
912            }
913            _ => panic!("expected Body::Json"),
914        }
915    }
916
917    #[test]
918    fn test_unmarshal_delimiter_tdf() {
919        let cfg = CsvConfig::tdf();
920        let df = CsvDataFormat::new(cfg);
921        let body = Body::Text("a\tb\n1\t2".into());
922        let result = df.unmarshal(body).unwrap();
923        match result {
924            Body::Json(v) => {
925                let arr = v.as_array().unwrap();
926                assert_eq!(arr[0]["a"], json!("1"));
927            }
928            _ => panic!("expected Body::Json"),
929        }
930    }
931
932    #[test]
933    fn test_unmarshal_comment_marker() {
934        let cfg = CsvConfig::default().comment_marker(Some('#'));
935        let df = CsvDataFormat::new(cfg);
936        let body = Body::Text("# this is a comment\na,b\n1,2".into());
937        let result = df.unmarshal(body).unwrap();
938        match result {
939            Body::Json(v) => {
940                let arr = v.as_array().unwrap();
941                assert_eq!(arr.len(), 1);
942            }
943            _ => panic!("expected Body::Json"),
944        }
945    }
946
947    #[test]
948    fn test_unmarshal_empty_returns_error() {
949        let df = CsvDataFormat::default();
950        let result = df.unmarshal(Body::Empty);
951        assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
952    }
953
954    #[test]
955    fn test_unmarshal_stream_returns_error() {
956        use bytes::Bytes;
957        use camel_api::body::{StreamBody, StreamMetadata};
958        let df = CsvDataFormat::default();
959        let empty_stream = futures::stream::empty::<Result<Bytes, CamelError>>();
960        let stream_body = StreamBody {
961            stream: std::sync::Arc::new(tokio::sync::Mutex::new(Some(Box::pin(empty_stream)))),
962            metadata: StreamMetadata::default(),
963        };
964        let result = df.unmarshal(Body::Stream(stream_body));
965        let msg = match result {
966            Err(CamelError::TypeConversionFailed(m)) => m,
967            _ => panic!("expected TypeConversionFailed"),
968        };
969        assert!(msg.contains("stream_cache"));
970    }
971
972    #[test]
973    fn test_unmarshal_no_header_map_mode_error() {
974        let cfg = CsvConfig::default().has_headers(false);
975        let df = CsvDataFormat::new(cfg);
976        let body = Body::Text("1,2,3".into());
977        let result = df.unmarshal(body);
978        let msg = match result {
979            Err(CamelError::TypeConversionFailed(m)) => m,
980            _ => panic!("expected TypeConversionFailed"),
981        };
982        assert!(msg.contains("has_headers") || msg.contains("headers"));
983    }
984
985    #[test]
986    fn test_unmarshal_json_identity() {
987        let df = CsvDataFormat::default();
988        let input = json!([{"x": 1}]);
989        let body = Body::Json(input.clone());
990        let result = df.unmarshal(body).unwrap();
991        assert!(matches!(result, Body::Json(_)));
992    }
993
994    #[test]
995    fn test_unmarshal_null_string() {
996        let cfg = CsvConfig::default().null_string(Some("NULL".into()));
997        let df = CsvDataFormat::new(cfg);
998        let body = Body::Text("a,b\n1,NULL".into());
999        let result = df.unmarshal(body).unwrap();
1000        match result {
1001            Body::Json(v) => {
1002                let arr = v.as_array().unwrap();
1003                assert_eq!(arr[0]["b"], serde_json::Value::Null);
1004            }
1005            _ => panic!("expected Body::Json"),
1006        }
1007    }
1008
1009    #[test]
1010    fn test_marshal_roundtrip() {
1011        let df = CsvDataFormat::default();
1012        let original = "name,age\nAlice,30\nBob,25";
1013        let json = df.unmarshal(Body::Text(original.into())).unwrap();
1014        let back = df.marshal(json).unwrap();
1015        match back {
1016            Body::Text(s) => {
1017                assert!(s.contains("Alice"));
1018                assert!(s.contains("name"));
1019            }
1020            _ => panic!("expected Body::Text"),
1021        }
1022    }
1023
1024    #[test]
1025    fn test_unmarshal_capture_header_maps_mode() {
1026        let cfg = CsvConfig::default().capture_header_record(true);
1027        let df = CsvDataFormat::new(cfg);
1028        let body = Body::Text("name,age\nAlice,30".into());
1029
1030        let mut ex = Exchange::default();
1031        let result = df.unmarshal_in_exchange(&mut ex, body).unwrap();
1032        assert!(matches!(result, Body::Json(_)));
1033
1034        let captured = ex.input.headers.get(CAMEL_CSV_HEADER_RECORD);
1035        assert_eq!(captured, Some(&serde_json::json!(["name", "age"])));
1036    }
1037
1038    #[test]
1039    fn test_unmarshal_capture_header_lists_mode() {
1040        let cfg = CsvConfig::default()
1041            .capture_header_record(true)
1042            .use_maps(false);
1043        let df = CsvDataFormat::new(cfg);
1044        let body = Body::Text("a,b\n1,2".into());
1045
1046        let mut ex = Exchange::default();
1047        let result = df.unmarshal_in_exchange(&mut ex, body).unwrap();
1048        assert!(matches!(result, Body::Json(_)));
1049
1050        let captured = ex.input.headers.get(CAMEL_CSV_HEADER_RECORD);
1051        assert_eq!(captured, Some(&serde_json::json!(["a", "b"])));
1052    }
1053
1054    #[test]
1055    fn test_unmarshal_capture_header_configured_headers() {
1056        let cfg = CsvConfig::default()
1057            .capture_header_record(true)
1058            .headers(vec!["col1".into(), "col2".into()])
1059            .skip_header_record(true);
1060        let df = CsvDataFormat::new(cfg);
1061        let body = Body::Text("ignored1,ignored2\nval1,val2".into());
1062
1063        let mut ex = Exchange::default();
1064        let _ = df.unmarshal_in_exchange(&mut ex, body).unwrap();
1065
1066        let captured = ex.input.headers.get(CAMEL_CSV_HEADER_RECORD);
1067        assert_eq!(captured, Some(&serde_json::json!(["col1", "col2"])));
1068    }
1069
1070    #[test]
1071    fn test_unmarshal_no_capture_default() {
1072        let df = CsvDataFormat::default();
1073        let body = Body::Text("a,b\n1,2".into());
1074
1075        let mut ex = Exchange::default();
1076        let _ = df.unmarshal_in_exchange(&mut ex, body).unwrap();
1077
1078        assert!(!ex.input.headers.contains_key(CAMEL_CSV_HEADER_RECORD));
1079    }
1080
1081    #[test]
1082    fn test_marshal_mixed_array_objects_returns_error() {
1083        let df = CsvDataFormat::default();
1084        let body = Body::Json(json!([
1085            { "name": "Alice" },
1086            "scalar string",
1087            { "name": "Bob" }
1088        ]));
1089        let result = df.marshal(body);
1090        assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
1091    }
1092
1093    #[test]
1094    fn test_marshal_mixed_array_lists_returns_error() {
1095        let cfg = CsvConfig::default().use_maps(false);
1096        let df = CsvDataFormat::new(cfg);
1097        let body = Body::Json(json!([
1098            ["a", "b"],
1099            { "not": "array" },
1100            ["c", "d"]
1101        ]));
1102        let result = df.marshal(body);
1103        assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
1104    }
1105
1106    #[test]
1107    fn test_unmarshal_max_records_exceeded() {
1108        let cfg = CsvConfig::default().max_records(3);
1109        let df = CsvDataFormat::new(cfg);
1110        let body = Body::Text("a,b\n1,2\n3,4\n5,6\n7,8".into());
1111        let result = df.unmarshal(body);
1112        let msg = match result {
1113            Err(CamelError::TypeConversionFailed(m)) => m,
1114            other => panic!("expected TypeConversionFailed, got {other:?}"),
1115        };
1116        assert!(
1117            msg.contains("max_records"),
1118            "msg should mention max_records: {msg}"
1119        );
1120    }
1121
1122    #[test]
1123    fn test_unmarshal_max_records_exceeded_lists_mode() {
1124        let cfg = CsvConfig::default().use_maps(false).max_records(2);
1125        let df = CsvDataFormat::new(cfg);
1126        // 3 data rows with lists-mode → cap at 2 => 3rd row triggers max_records
1127        let body = Body::Text("a,b,c\n1,2,3\n4,5,6\n7,8,9".into());
1128        let result = df.unmarshal(body);
1129        let msg = match result {
1130            Err(CamelError::TypeConversionFailed(m)) => m,
1131            other => panic!("expected TypeConversionFailed, got {other:?}"),
1132        };
1133        assert!(
1134            msg.contains("max_records"),
1135            "msg should mention max_records: {msg}"
1136        );
1137    }
1138
1139    #[test]
1140    fn test_unmarshal_max_field_size_exceeded() {
1141        let cfg = CsvConfig::default().max_field_size(4);
1142        let df = CsvDataFormat::new(cfg);
1143        let body = Body::Text("a,b\n12345,1".into());
1144        let result = df.unmarshal(body);
1145        let msg = match result {
1146            Err(CamelError::TypeConversionFailed(m)) => m,
1147            other => panic!("expected TypeConversionFailed, got {other:?}"),
1148        };
1149        assert!(
1150            msg.contains("max_field_size"),
1151            "msg should mention max_field_size: {msg}"
1152        );
1153    }
1154
1155    #[test]
1156    fn test_unmarshal_default_caps_accept_normal_input() {
1157        // Default caps must not reject ordinary CSV.
1158        let df = CsvDataFormat::default();
1159        let body = Body::Text("a,b\n1,2\n3,4".into());
1160        let result = df.unmarshal(body).unwrap();
1161        assert!(matches!(result, Body::Json(_)));
1162    }
1163
1164    #[test]
1165    fn marshal_neutralizes_formula_injection_cells() {
1166        let df = CsvDataFormat::default();
1167        let body = Body::Json(json!([
1168            { "name": "=SUM(1,1)", "note": "+danger", "val": "-1", "at": "@foo", "tab": "\t=cmd", "cr": "\r=cmd" }
1169        ]));
1170        let result = df.marshal(body).unwrap();
1171        match result {
1172            Body::Text(s) => {
1173                // Headers should also be neutralized
1174                assert!(
1175                    s.contains("'=SUM(1,1)"),
1176                    "cell starting with = should be neutralized: {s:?}"
1177                );
1178                assert!(
1179                    s.contains("'+danger"),
1180                    "cell starting with + should be neutralized: {s:?}"
1181                );
1182                assert!(
1183                    s.contains("'-1"),
1184                    "cell starting with - should be neutralized: {s:?}"
1185                );
1186                assert!(
1187                    s.contains("'@foo"),
1188                    "cell starting with @ should be neutralized: {s:?}"
1189                );
1190                assert!(
1191                    s.contains("'\t=cmd"),
1192                    "cell starting with tab should be neutralized: {s:?}"
1193                );
1194                assert!(
1195                    s.contains("'\r=cmd"),
1196                    "cell starting with CR should be neutralized: {s:?}"
1197                );
1198            }
1199            _ => panic!("expected Body::Text"),
1200        }
1201    }
1202
1203    #[test]
1204    fn test_unmarshal_width_mismatch_does_not_error() {
1205        // Row has fewer fields than the header — must not panic, must pad.
1206        let df = CsvDataFormat::default();
1207        let body = Body::Text("a,b,c\n1,2".into());
1208        let result = df.unmarshal(body).unwrap();
1209        match result {
1210            Body::Json(serde_json::Value::Array(rows)) => {
1211                assert_eq!(rows.len(), 1);
1212                assert_eq!(rows[0]["a"], json!("1"));
1213                assert_eq!(rows[0]["b"], json!("2"));
1214                assert_eq!(rows[0]["c"], json!(""));
1215            }
1216            other => panic!("expected JSON array, got {other:?}"),
1217        }
1218    }
1219
1220    #[test]
1221    fn test_csv_config_deserialize_from_json() {
1222        let json = serde_json::json!({
1223            "delimiter": "|",
1224            "quote_mode": "non_numeric",
1225            "record_separator": "lf",
1226            "max_records": 5000000
1227        });
1228        let cfg: CsvConfig = serde_json::from_value(json).unwrap();
1229        assert_eq!(cfg.delimiter, '|');
1230        assert!(matches!(cfg.quote_mode, QuoteMode::NonNumeric));
1231        assert!(matches!(cfg.record_separator, RecordSeparator::Lf));
1232        assert_eq!(cfg.max_records, Some(5000000));
1233    }
1234
1235    #[test]
1236    fn test_csv_config_deny_unknown_fields() {
1237        let json = serde_json::json!({"unknown_key": 42});
1238        let result: Result<CsvConfig, _> = serde_json::from_value(json);
1239        assert!(result.is_err(), "unknown field should fail closed");
1240    }
1241
1242    #[test]
1243    fn csv_width_default_is_warn() {
1244        assert_eq!(
1245            CsvConfig::default().width_mismatch,
1246            WidthMismatchPolicy::Warn
1247        );
1248    }
1249
1250    #[test]
1251    fn csv_width_warn_default_aggregate_single_log() {
1252        // Default policy (Warn): mismatched rows must not error; result is Ok.
1253        // Per-row warns are replaced by a single aggregate warn (verified by code path).
1254        let df = CsvDataFormat::default();
1255        let body = Body::Text("a,b\n1,2,3\n4".into());
1256        let result = df.unmarshal(body).unwrap();
1257        match result {
1258            Body::Json(serde_json::Value::Array(rows)) => {
1259                assert_eq!(rows.len(), 2);
1260                // Row 1: "1,2,3" → 3 cols, header has 2 → padded/truncated to 2
1261                assert_eq!(rows[0]["a"], json!("1"));
1262                assert_eq!(rows[0]["b"], json!("2"));
1263                // Row 2: "4" → 1 col, header has 2 → padded with ""
1264                assert_eq!(rows[1]["a"], json!("4"));
1265                assert_eq!(rows[1]["b"], json!(""));
1266            }
1267            other => panic!("expected JSON array, got {other:?}"),
1268        }
1269    }
1270
1271    #[test]
1272    fn csv_width_error_returns_err() {
1273        let cfg = CsvConfig::default().width_mismatch(WidthMismatchPolicy::Error);
1274        let df = CsvDataFormat::new(cfg);
1275        let body = Body::Text("a,b\n1,2,3".into());
1276        let result = df.unmarshal(body);
1277        let msg = match result {
1278            Err(CamelError::TypeConversionFailed(m)) => m,
1279            other => panic!("expected TypeConversionFailed, got {other:?}"),
1280        };
1281        assert!(
1282            msg.contains("width") && msg.contains("differs"),
1283            "error should mention width mismatch: {msg}"
1284        );
1285    }
1286
1287    #[test]
1288    fn csv_width_lenient_no_warn() {
1289        let cfg = CsvConfig::default().width_mismatch(WidthMismatchPolicy::Lenient);
1290        let df = CsvDataFormat::new(cfg);
1291        let body = Body::Text("a,b\n1,2,3\n4".into());
1292        let result = df.unmarshal(body).unwrap();
1293        match result {
1294            Body::Json(serde_json::Value::Array(rows)) => {
1295                assert_eq!(rows.len(), 2);
1296                assert_eq!(rows[0]["a"], json!("1"));
1297                assert_eq!(rows[1]["a"], json!("4"));
1298            }
1299            other => panic!("expected JSON array, got {other:?}"),
1300        }
1301    }
1302
1303    #[test]
1304    fn csv_width_unknown_enum_rejected() {
1305        let json = serde_json::json!({"width_mismatch": "invalid"});
1306        let result: Result<CsvConfig, _> = serde_json::from_value(json);
1307        assert!(
1308            result.is_err(),
1309            "invalid enum variant should fail deserialization"
1310        );
1311    }
1312}
camel_processor/data_format/csv.rs

camel_processor/data_format/
csv.rs
