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

camel_processor/data_format/
csv.rs
