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

camel_processor/data_format/
csv.rs
