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}
42
43impl Default for CsvConfig {
44 fn default() -> Self {
45 Self {
46 delimiter: ',',
47 quote: '"',
48 quote_mode: QuoteMode::Minimal,
49 double_quote: true,
50 escape: None,
51 record_separator: RecordSeparator::Crlf,
52 comment_marker: None,
53 null_string: None,
54 headers: None,
55 has_headers: true,
56 skip_header_record: false,
57 capture_header_record: false,
58 ignore_surrounding_spaces: true,
59 trim: false,
60 use_maps: true,
61 }
62 }
63}
64
65impl CsvConfig {
66 pub fn excel() -> Self {
67 Self::default()
68 }
69
70 pub fn tdf() -> Self {
71 Self {
72 delimiter: '\t',
73 record_separator: RecordSeparator::Lf,
74 ..Self::default()
75 }
76 }
77
78 pub fn mysql() -> Self {
79 Self {
80 delimiter: '\t',
81 quote: '\0',
82 record_separator: RecordSeparator::Lf,
83 escape: Some('\\'),
84 ..Self::default()
85 }
86 }
87
88 pub fn delimiter(mut self, c: char) -> Self {
89 self.delimiter = c;
90 self
91 }
92 pub fn quote(mut self, c: char) -> Self {
93 self.quote = c;
94 self
95 }
96 pub fn quote_mode(mut self, m: QuoteMode) -> Self {
97 self.quote_mode = m;
98 self
99 }
100 pub fn double_quote(mut self, b: bool) -> Self {
101 self.double_quote = b;
102 self
103 }
104 pub fn escape(mut self, c: Option<char>) -> Self {
105 self.escape = c;
106 self
107 }
108 pub fn record_separator(mut self, s: RecordSeparator) -> Self {
109 self.record_separator = s;
110 self
111 }
112 pub fn comment_marker(mut self, c: Option<char>) -> Self {
113 self.comment_marker = c;
114 self
115 }
116 pub fn null_string(mut self, s: Option<String>) -> Self {
117 self.null_string = s;
118 self
119 }
120 pub fn headers(mut self, h: Vec<String>) -> Self {
121 self.headers = Some(h);
122 self
123 }
124 pub fn has_headers(mut self, b: bool) -> Self {
125 self.has_headers = b;
126 self
127 }
128 pub fn skip_header_record(mut self, b: bool) -> Self {
129 self.skip_header_record = b;
130 self
131 }
132 pub fn capture_header_record(mut self, b: bool) -> Self {
133 self.capture_header_record = b;
134 self
135 }
136 pub fn ignore_surrounding_spaces(mut self, b: bool) -> Self {
137 self.ignore_surrounding_spaces = b;
138 self
139 }
140 pub fn trim(mut self, b: bool) -> Self {
141 self.trim = b;
142 self
143 }
144 pub fn use_maps(mut self, b: bool) -> Self {
145 self.use_maps = b;
146 self
147 }
148}
149
150#[derive(Clone, Default)]
151pub struct CsvDataFormat {
152 config: CsvConfig,
153}
154
155impl CsvDataFormat {
156 pub fn new(config: CsvConfig) -> Self {
157 Self { config }
158 }
159
160 pub fn config(&self) -> &CsvConfig {
161 &self.config
162 }
163
164 fn field_to_string(value: &serde_json::Value) -> String {
166 match value {
167 serde_json::Value::String(s) => s.clone(),
168 serde_json::Value::Number(n) => n.to_string(),
169 serde_json::Value::Bool(b) => b.to_string(),
170 serde_json::Value::Null => String::new(),
171 serde_json::Value::Array(arr) => arr
172 .iter()
173 .map(|v| v.to_string())
174 .collect::<Vec<_>>()
175 .join(","),
176 serde_json::Value::Object(_) => value.to_string(),
177 }
178 }
179
180 fn write_err(e: csv::Error) -> CamelError {
181 CamelError::TypeConversionFailed(format!("CSV write error: {e}"))
182 }
183
184 fn unmarshal_internal(&self, body: Body) -> Result<(Body, Option<Vec<String>>), CamelError> {
185 let input_text: String =
186 match body {
187 Body::Text(s) => s,
188 Body::Bytes(b) => String::from_utf8(b.to_vec()).map_err(|e| {
189 CamelError::TypeConversionFailed(format!("CSV bytes not UTF-8: {e}"))
190 })?,
191 Body::Json(v) => return Ok((Body::Json(v), None)),
192 Body::Stream(_) => return Err(CamelError::TypeConversionFailed(
193 "cannot unmarshal Body::Stream directly — add 'stream_cache' before this step"
194 .into(),
195 )),
196 Body::Empty => {
197 return Err(CamelError::TypeConversionFailed(
198 "CsvDataFormat::unmarshal expects Body::Text or Body::Bytes".into(),
199 ));
200 }
201 Body::Xml(_) => {
202 return Err(CamelError::TypeConversionFailed(
203 "CsvDataFormat::unmarshal does not accept Body::Xml".into(),
204 ));
205 }
206 };
207
208 let mut rdr_builder = csv::ReaderBuilder::new();
209 let effective_has_headers = if self.config.headers.is_some() {
210 false
211 } else {
212 self.config.has_headers
213 };
214 rdr_builder
215 .delimiter(self.config.delimiter as u8)
216 .has_headers(effective_has_headers)
217 .trim(if self.config.trim {
218 csv::Trim::All
219 } else if self.config.ignore_surrounding_spaces {
220 csv::Trim::Fields
221 } else {
222 csv::Trim::None
223 });
224 if let Some(c) = self.config.comment_marker {
225 rdr_builder.comment(Some(c as u8));
226 }
227 let mut rdr = rdr_builder.from_reader(input_text.as_bytes());
228
229 let header_keys: Option<Vec<String>> = if let Some(h) = &self.config.headers {
230 Some(h.clone())
231 } else if self.config.has_headers {
232 Some(
233 rdr.headers()
234 .map_err(|e| CamelError::TypeConversionFailed(format!("CSV header read: {e}")))?
235 .iter()
236 .map(|s| s.to_string())
237 .collect(),
238 )
239 } else {
240 None
241 };
242
243 let mut records: Vec<serde_json::Value> = Vec::new();
244 if self.config.use_maps {
245 let keys = match &header_keys {
246 Some(k) => k.clone(),
247 None => {
248 return Err(CamelError::TypeConversionFailed(
249 "CsvDataFormat::unmarshal with use_maps=true requires has_headers=true or configured headers".into(),
250 ));
251 }
252 };
253 let mut iter = rdr.records();
254 if self.config.skip_header_record && self.config.headers.is_some() {
255 iter.next();
256 }
257 for result in iter {
258 let record = result
259 .map_err(|e| CamelError::TypeConversionFailed(format!("CSV parse: {e}")))?;
260 let mut obj = serde_json::Map::new();
261 for (i, key) in keys.iter().enumerate() {
262 let val = record.get(i).unwrap_or("");
263 let parsed = if let Some(ns) = &self.config.null_string {
264 if val == ns {
265 serde_json::Value::Null
266 } else {
267 serde_json::Value::String(val.to_string())
268 }
269 } else {
270 serde_json::Value::String(val.to_string())
271 };
272 obj.insert(key.clone(), parsed);
273 }
274 records.push(serde_json::Value::Object(obj));
275 }
276 } else {
277 for result in rdr.records() {
278 let record = result
279 .map_err(|e| CamelError::TypeConversionFailed(format!("CSV parse: {e}")))?;
280 let arr: Vec<serde_json::Value> = record
281 .iter()
282 .map(|s| serde_json::Value::String(s.to_string()))
283 .collect();
284 records.push(serde_json::Value::Array(arr));
285 }
286 }
287
288 Ok((Body::Json(serde_json::Value::Array(records)), header_keys))
289 }
290}
291
292impl DataFormat for CsvDataFormat {
293 fn name(&self) -> &str {
294 "csv"
295 }
296
297 fn marshal(&self, body: Body) -> Result<Body, CamelError> {
298 let json = match body {
299 Body::Json(v) => v,
300 Body::Text(_) => {
301 return Err(CamelError::TypeConversionFailed(
302 "CsvDataFormat::marshal expects Body::Json; use convert_body_to first".into(),
303 ));
304 }
305 Body::Bytes(_) => {
306 return Err(CamelError::TypeConversionFailed(
307 "CsvDataFormat::marshal expects Body::Json; use convert_body_to first".into(),
308 ));
309 }
310 Body::Stream(_) => {
311 return Err(CamelError::TypeConversionFailed(
312 "cannot marshal Body::Stream — add 'stream_cache' before this step".into(),
313 ));
314 }
315 Body::Empty => {
316 return Err(CamelError::TypeConversionFailed(
317 "CsvDataFormat::marshal expects Body::Json".into(),
318 ));
319 }
320 Body::Xml(_) => return Err(CamelError::TypeConversionFailed(
321 "CsvDataFormat::marshal does not accept Body::Xml — use unmarshal(\"xml\") first"
322 .into(),
323 )),
324 };
325
326 let mut wtr_builder = csv::WriterBuilder::new();
327 wtr_builder
328 .delimiter(self.config.delimiter as u8)
329 .quote(self.config.quote as u8)
330 .double_quote(self.config.double_quote);
331 if let Some(esc) = self.config.escape {
332 wtr_builder.escape(esc as u8);
333 }
334 let quote_style = match self.config.quote_mode {
335 QuoteMode::All => csv::QuoteStyle::Always,
336 QuoteMode::Minimal => csv::QuoteStyle::Necessary,
337 QuoteMode::NonNumeric => csv::QuoteStyle::NonNumeric,
338 QuoteMode::None => csv::QuoteStyle::Never,
339 };
340 wtr_builder.quote_style(quote_style);
341 let terminator = match self.config.record_separator {
342 RecordSeparator::Crlf => csv::Terminator::CRLF,
343 RecordSeparator::Lf => csv::Terminator::Any(b'\n'),
344 };
345 wtr_builder.terminator(terminator);
346
347 let mut output = Vec::new();
348 {
349 let mut wtr = wtr_builder.from_writer(&mut output);
350 match &json {
351 serde_json::Value::Array(arr) => {
352 if arr.is_empty() {
353 if let (true, Some(h)) = (self.config.has_headers, &self.config.headers) {
354 wtr.write_record(h.iter().map(String::as_str))
355 .map_err(Self::write_err)?;
356 }
357 } else {
358 match &arr[0] {
359 serde_json::Value::Object(first_obj) => {
360 let header_keys: Vec<String> = self
361 .config
362 .headers
363 .clone()
364 .unwrap_or_else(|| first_obj.keys().cloned().collect());
365 if self.config.has_headers {
366 wtr.write_record(header_keys.iter().map(String::as_str))
367 .map_err(Self::write_err)?;
368 }
369 for row in arr {
370 if let Some(obj) = row.as_object() {
371 let record: Vec<String> = header_keys
372 .iter()
373 .map(|k| {
374 Self::field_to_string(
375 obj.get(k).unwrap_or(&serde_json::Value::Null),
376 )
377 })
378 .collect();
379 wtr.write_record(record.iter().map(String::as_str))
380 .map_err(Self::write_err)?;
381 } else {
382 return Err(CamelError::TypeConversionFailed(format!(
383 "CsvDataFormat::marshal expected all array elements to be objects, got {:?}",
384 row
385 )));
386 }
387 }
388 }
389 _ => {
390 for row in arr {
391 if let Some(items) = row.as_array() {
392 let record: Vec<String> =
393 items.iter().map(Self::field_to_string).collect();
394 wtr.write_record(record.iter().map(String::as_str))
395 .map_err(Self::write_err)?;
396 } else {
397 return Err(CamelError::TypeConversionFailed(format!(
398 "CsvDataFormat::marshal expected all array elements to be arrays, got {:?}",
399 row
400 )));
401 }
402 }
403 }
404 }
405 }
406 }
407 serde_json::Value::Object(obj) => {
408 let header_keys: Vec<String> = self
409 .config
410 .headers
411 .clone()
412 .unwrap_or_else(|| obj.keys().cloned().collect());
413 if self.config.has_headers {
414 wtr.write_record(header_keys.iter().map(String::as_str))
415 .map_err(Self::write_err)?;
416 }
417 let record: Vec<String> = header_keys
418 .iter()
419 .map(|k| {
420 Self::field_to_string(obj.get(k).unwrap_or(&serde_json::Value::Null))
421 })
422 .collect();
423 wtr.write_record(record.iter().map(String::as_str))
424 .map_err(Self::write_err)?;
425 }
426 _ => {
427 return Err(CamelError::TypeConversionFailed(format!(
428 "CsvDataFormat::marshal only supports Json Array or Object, got {json:?}"
429 )));
430 }
431 }
432 wtr.flush()
433 .map_err(|e| CamelError::TypeConversionFailed(format!("CSV flush: {e}")))?;
434 }
435
436 let text = String::from_utf8(output)
437 .map_err(|e| CamelError::TypeConversionFailed(format!("CSV output not UTF-8: {e}")))?;
438 Ok(Body::Text(text))
439 }
440
441 fn unmarshal(&self, body: Body) -> Result<Body, CamelError> {
442 self.unmarshal_internal(body).map(|(b, _)| b)
443 }
444
445 fn unmarshal_in_exchange(
446 &self,
447 exchange: &mut Exchange,
448 body: Body,
449 ) -> Result<Body, CamelError> {
450 let (result, header_keys) = self.unmarshal_internal(body)?;
451 if let Some(keys) = header_keys
452 .filter(|_| self.config.capture_header_record)
453 .filter(|k| !k.is_empty())
454 {
455 exchange.input.headers.insert(
456 CAMEL_CSV_HEADER_RECORD.to_string(),
457 serde_json::Value::Array(keys.into_iter().map(serde_json::Value::String).collect()),
458 );
459 }
460 Ok(result)
461 }
462}
463
464#[cfg(test)]
465mod tests {
466 use super::*;
467 use serde_json::json;
468
469 #[test]
470 fn test_default_config_values() {
471 let cfg = CsvConfig::default();
472 assert_eq!(cfg.delimiter, ',');
473 assert_eq!(cfg.quote, '"');
474 assert!(matches!(cfg.quote_mode, QuoteMode::Minimal));
475 assert!(cfg.double_quote);
476 assert!(cfg.escape.is_none());
477 assert!(matches!(cfg.record_separator, RecordSeparator::Crlf));
478 assert!(cfg.comment_marker.is_none());
479 assert!(cfg.null_string.is_none());
480 assert!(cfg.headers.is_none());
481 assert!(cfg.has_headers);
482 assert!(!cfg.skip_header_record);
483 assert!(!cfg.capture_header_record);
484 assert!(cfg.ignore_surrounding_spaces);
485 assert!(!cfg.trim);
486 assert!(cfg.use_maps);
487 }
488
489 #[test]
490 fn test_tdf_preset_uses_tab() {
491 let cfg = CsvConfig::tdf();
492 assert_eq!(cfg.delimiter, '\t');
493 assert!(matches!(cfg.record_separator, RecordSeparator::Lf));
494 }
495
496 #[test]
497 fn test_mysql_preset_uses_tab_and_escape() {
498 let cfg = CsvConfig::mysql();
499 assert_eq!(cfg.delimiter, '\t');
500 assert_eq!(cfg.escape, Some('\\'));
501 }
502
503 #[test]
504 fn test_builder_methods() {
505 let cfg = CsvConfig::default()
506 .delimiter('|')
507 .headers(vec!["a".into(), "b".into()])
508 .use_maps(false);
509 assert_eq!(cfg.delimiter, '|');
510 assert_eq!(cfg.headers, Some(vec!["a".into(), "b".into()]));
511 assert!(!cfg.use_maps);
512 }
513
514 #[test]
515 fn test_marshal_array_of_objects_with_header() {
516 let df = CsvDataFormat::default();
517 let body = Body::Json(json!([
518 { "name": "Alice", "age": 30 },
519 { "name": "Bob", "age": 25 }
520 ]));
521 let result = df.marshal(body).unwrap();
522 match result {
523 Body::Text(s) => {
524 let lines: Vec<&str> = s.lines().collect();
525 let header_fields: Vec<&str> = lines[0].split(',').collect();
528 assert_eq!(header_fields.len(), 2);
529 assert!(header_fields.contains(&"name"));
530 assert!(header_fields.contains(&"age"));
531 assert!(lines[1].contains("Alice"));
532 assert!(lines[2].contains("Bob"));
533 }
534 _ => panic!("expected Body::Text"),
535 }
536 }
537
538 #[test]
539 fn test_marshal_array_of_arrays_no_header() {
540 let df = CsvDataFormat::default();
541 let body = Body::Json(json!([["a", "b", "c"], [1, 2, 3]]));
542 let result = df.marshal(body).unwrap();
543 match result {
544 Body::Text(s) => {
545 let lines: Vec<&str> = s.lines().collect();
546 assert_eq!(lines.len(), 2);
547 assert!(lines[0].contains("a"));
548 }
549 _ => panic!("expected Body::Text"),
550 }
551 }
552
553 #[test]
554 fn test_marshal_single_object() {
555 let df = CsvDataFormat::default();
556 let body = Body::Json(json!({ "x": 1, "y": 2 }));
557 let result = df.marshal(body).unwrap();
558 match result {
559 Body::Text(s) => {
560 let lines: Vec<&str> = s.lines().collect();
561 assert_eq!(lines.len(), 2);
562 }
563 _ => panic!("expected Body::Text"),
564 }
565 }
566
567 #[test]
568 fn test_marshal_empty_array_with_headers() {
569 let cfg = CsvConfig::default().headers(vec!["h1".into(), "h2".into()]);
570 let df = CsvDataFormat::new(cfg);
571 let body = Body::Json(json!([]));
572 let result = df.marshal(body).unwrap();
573 match result {
574 Body::Text(s) => {
575 assert_eq!(s.trim(), "h1,h2");
576 }
577 _ => panic!("expected Body::Text"),
578 }
579 }
580
581 #[test]
582 fn test_marshal_empty_array_no_headers() {
583 let cfg = CsvConfig::default().has_headers(false);
584 let df = CsvDataFormat::new(cfg);
585 let body = Body::Json(json!([]));
586 let result = df.marshal(body).unwrap();
587 match result {
588 Body::Text(s) => assert!(s.is_empty()),
589 _ => panic!("expected Body::Text"),
590 }
591 }
592
593 #[test]
594 fn test_marshal_quoted_fields_with_commas() {
595 let df = CsvDataFormat::default();
596 let body = Body::Json(json!([
597 { "name": "Doe, John", "age": 30 }
598 ]));
599 let result = df.marshal(body).unwrap();
600 match result {
601 Body::Text(s) => assert!(s.contains("\"Doe, John\"")),
602 _ => panic!("expected Body::Text"),
603 }
604 }
605
606 #[test]
607 fn test_marshal_text_returns_error() {
608 let df = CsvDataFormat::default();
609 let result = df.marshal(Body::Text("already text".into()));
610 assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
611 }
612
613 #[test]
614 fn test_marshal_stream_returns_error() {
615 use bytes::Bytes;
616 use camel_api::body::{StreamBody, StreamMetadata};
617 let df = CsvDataFormat::default();
618 let empty_stream = futures::stream::empty::<Result<Bytes, CamelError>>();
619 let stream_body = StreamBody {
620 stream: std::sync::Arc::new(tokio::sync::Mutex::new(Some(Box::pin(empty_stream)))),
621 metadata: StreamMetadata::default(),
622 };
623 let result = df.marshal(Body::Stream(stream_body));
624 assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
625 }
626
627 #[test]
628 fn test_marshal_quote_mode_all() {
629 let cfg = CsvConfig::default().quote_mode(QuoteMode::All);
630 let df = CsvDataFormat::new(cfg);
631 let body = Body::Json(json!([{ "x": 1 }]));
632 let result = df.marshal(body).unwrap();
633 match result {
634 Body::Text(s) => {
635 assert!(s.contains("\"x\""));
636 }
637 _ => panic!("expected Body::Text"),
638 }
639 }
640
641 #[test]
642 fn test_marshal_mysql_preset_quote_char_null() {
643 let cfg = CsvConfig::mysql();
644 let df = CsvDataFormat::new(cfg);
645 let body = Body::Json(json!([["a", "b"]]));
646 let result = df.marshal(body);
647 assert!(result.is_ok(), "mysql preset should marshal: {result:?}");
648 }
649
650 #[test]
651 fn test_unmarshal_text_to_json_maps() {
652 let df = CsvDataFormat::default();
653 let body = Body::Text("name,age\nAlice,30\nBob,25".into());
654 let result = df.unmarshal(body).unwrap();
655 match result {
656 Body::Json(v) => {
657 let arr = v.as_array().expect("expected array");
658 assert_eq!(arr.len(), 2);
659 assert_eq!(arr[0]["name"], json!("Alice"));
660 assert_eq!(arr[0]["age"], json!("30"));
661 }
662 _ => panic!("expected Body::Json"),
663 }
664 }
665
666 #[test]
667 fn test_unmarshal_text_to_json_lists() {
668 let cfg = CsvConfig::default().use_maps(false);
669 let df = CsvDataFormat::new(cfg);
670 let body = Body::Text("a,b,c\n1,2,3".into());
671 let result = df.unmarshal(body).unwrap();
672 match result {
673 Body::Json(v) => {
674 let arr = v.as_array().expect("expected array");
675 assert_eq!(arr.len(), 1);
676 assert_eq!(arr[0][0], json!("1"));
677 }
678 _ => panic!("expected Body::Json"),
679 }
680 }
681
682 #[test]
683 fn test_unmarshal_bytes_to_json() {
684 let df = CsvDataFormat::default();
685 let body = Body::Bytes(bytes::Bytes::from_static(b"x,y\n1,2"));
686 let result = df.unmarshal(body).unwrap();
687 match result {
688 Body::Json(v) => {
689 let arr = v.as_array().unwrap();
690 assert_eq!(arr[0]["x"], json!("1"));
691 }
692 _ => panic!("expected Body::Json"),
693 }
694 }
695
696 #[test]
697 fn test_unmarshal_configured_headers() {
698 let cfg = CsvConfig::default().headers(vec!["col1".into(), "col2".into()]);
699 let df = CsvDataFormat::new(cfg);
700 let body = Body::Text("val1,val2".into());
701 let result = df.unmarshal(body).unwrap();
702 match result {
703 Body::Json(v) => {
704 let arr = v.as_array().unwrap();
705 assert_eq!(arr[0]["col1"], json!("val1"));
706 }
707 _ => panic!("expected Body::Json"),
708 }
709 }
710
711 #[test]
712 fn test_unmarshal_skip_header_record_with_configured_headers() {
713 let cfg = CsvConfig::default()
714 .headers(vec!["col1".into(), "col2".into()])
715 .skip_header_record(true);
716 let df = CsvDataFormat::new(cfg);
717 let body = Body::Text("ignored1,ignored2\nval1,val2".into());
718 let result = df.unmarshal(body).unwrap();
719 match result {
720 Body::Json(v) => {
721 let arr = v.as_array().unwrap();
722 assert_eq!(arr.len(), 1);
723 assert_eq!(arr[0]["col1"], json!("val1"));
724 }
725 _ => panic!("expected Body::Json"),
726 }
727 }
728
729 #[test]
730 fn test_unmarshal_quoted_fields() {
731 let df = CsvDataFormat::default();
732 let body = Body::Text("name,note\n\"Doe, John\",hi".into());
733 let result = df.unmarshal(body).unwrap();
734 match result {
735 Body::Json(v) => {
736 let arr = v.as_array().unwrap();
737 assert_eq!(arr[0]["name"], json!("Doe, John"));
738 }
739 _ => panic!("expected Body::Json"),
740 }
741 }
742
743 #[test]
744 fn test_unmarshal_delimiter_tdf() {
745 let cfg = CsvConfig::tdf();
746 let df = CsvDataFormat::new(cfg);
747 let body = Body::Text("a\tb\n1\t2".into());
748 let result = df.unmarshal(body).unwrap();
749 match result {
750 Body::Json(v) => {
751 let arr = v.as_array().unwrap();
752 assert_eq!(arr[0]["a"], json!("1"));
753 }
754 _ => panic!("expected Body::Json"),
755 }
756 }
757
758 #[test]
759 fn test_unmarshal_comment_marker() {
760 let cfg = CsvConfig::default().comment_marker(Some('#'));
761 let df = CsvDataFormat::new(cfg);
762 let body = Body::Text("# this is a comment\na,b\n1,2".into());
763 let result = df.unmarshal(body).unwrap();
764 match result {
765 Body::Json(v) => {
766 let arr = v.as_array().unwrap();
767 assert_eq!(arr.len(), 1);
768 }
769 _ => panic!("expected Body::Json"),
770 }
771 }
772
773 #[test]
774 fn test_unmarshal_empty_returns_error() {
775 let df = CsvDataFormat::default();
776 let result = df.unmarshal(Body::Empty);
777 assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
778 }
779
780 #[test]
781 fn test_unmarshal_stream_returns_error() {
782 use bytes::Bytes;
783 use camel_api::body::{StreamBody, StreamMetadata};
784 let df = CsvDataFormat::default();
785 let empty_stream = futures::stream::empty::<Result<Bytes, CamelError>>();
786 let stream_body = StreamBody {
787 stream: std::sync::Arc::new(tokio::sync::Mutex::new(Some(Box::pin(empty_stream)))),
788 metadata: StreamMetadata::default(),
789 };
790 let result = df.unmarshal(Body::Stream(stream_body));
791 let msg = match result {
792 Err(CamelError::TypeConversionFailed(m)) => m,
793 _ => panic!("expected TypeConversionFailed"),
794 };
795 assert!(msg.contains("stream_cache"));
796 }
797
798 #[test]
799 fn test_unmarshal_no_header_map_mode_error() {
800 let cfg = CsvConfig::default().has_headers(false);
801 let df = CsvDataFormat::new(cfg);
802 let body = Body::Text("1,2,3".into());
803 let result = df.unmarshal(body);
804 let msg = match result {
805 Err(CamelError::TypeConversionFailed(m)) => m,
806 _ => panic!("expected TypeConversionFailed"),
807 };
808 assert!(msg.contains("has_headers") || msg.contains("headers"));
809 }
810
811 #[test]
812 fn test_unmarshal_json_identity() {
813 let df = CsvDataFormat::default();
814 let input = json!([{"x": 1}]);
815 let body = Body::Json(input.clone());
816 let result = df.unmarshal(body).unwrap();
817 assert!(matches!(result, Body::Json(_)));
818 }
819
820 #[test]
821 fn test_unmarshal_null_string() {
822 let cfg = CsvConfig::default().null_string(Some("NULL".into()));
823 let df = CsvDataFormat::new(cfg);
824 let body = Body::Text("a,b\n1,NULL".into());
825 let result = df.unmarshal(body).unwrap();
826 match result {
827 Body::Json(v) => {
828 let arr = v.as_array().unwrap();
829 assert_eq!(arr[0]["b"], serde_json::Value::Null);
830 }
831 _ => panic!("expected Body::Json"),
832 }
833 }
834
835 #[test]
836 fn test_marshal_roundtrip() {
837 let df = CsvDataFormat::default();
838 let original = "name,age\nAlice,30\nBob,25";
839 let json = df.unmarshal(Body::Text(original.into())).unwrap();
840 let back = df.marshal(json).unwrap();
841 match back {
842 Body::Text(s) => {
843 assert!(s.contains("Alice"));
844 assert!(s.contains("name"));
845 }
846 _ => panic!("expected Body::Text"),
847 }
848 }
849
850 #[test]
851 fn test_unmarshal_capture_header_maps_mode() {
852 let cfg = CsvConfig::default().capture_header_record(true);
853 let df = CsvDataFormat::new(cfg);
854 let body = Body::Text("name,age\nAlice,30".into());
855
856 let mut ex = Exchange::default();
857 let result = df.unmarshal_in_exchange(&mut ex, body).unwrap();
858 assert!(matches!(result, Body::Json(_)));
859
860 let captured = ex.input.headers.get(CAMEL_CSV_HEADER_RECORD);
861 assert_eq!(captured, Some(&serde_json::json!(["name", "age"])));
862 }
863
864 #[test]
865 fn test_unmarshal_capture_header_lists_mode() {
866 let cfg = CsvConfig::default()
867 .capture_header_record(true)
868 .use_maps(false);
869 let df = CsvDataFormat::new(cfg);
870 let body = Body::Text("a,b\n1,2".into());
871
872 let mut ex = Exchange::default();
873 let result = df.unmarshal_in_exchange(&mut ex, body).unwrap();
874 assert!(matches!(result, Body::Json(_)));
875
876 let captured = ex.input.headers.get(CAMEL_CSV_HEADER_RECORD);
877 assert_eq!(captured, Some(&serde_json::json!(["a", "b"])));
878 }
879
880 #[test]
881 fn test_unmarshal_capture_header_configured_headers() {
882 let cfg = CsvConfig::default()
883 .capture_header_record(true)
884 .headers(vec!["col1".into(), "col2".into()])
885 .skip_header_record(true);
886 let df = CsvDataFormat::new(cfg);
887 let body = Body::Text("ignored1,ignored2\nval1,val2".into());
888
889 let mut ex = Exchange::default();
890 let _ = df.unmarshal_in_exchange(&mut ex, body).unwrap();
891
892 let captured = ex.input.headers.get(CAMEL_CSV_HEADER_RECORD);
893 assert_eq!(captured, Some(&serde_json::json!(["col1", "col2"])));
894 }
895
896 #[test]
897 fn test_unmarshal_no_capture_default() {
898 let df = CsvDataFormat::default();
899 let body = Body::Text("a,b\n1,2".into());
900
901 let mut ex = Exchange::default();
902 let _ = df.unmarshal_in_exchange(&mut ex, body).unwrap();
903
904 assert!(ex.input.headers.get(CAMEL_CSV_HEADER_RECORD).is_none());
905 }
906
907 #[test]
908 fn test_marshal_mixed_array_objects_returns_error() {
909 let df = CsvDataFormat::default();
910 let body = Body::Json(json!([
911 { "name": "Alice" },
912 "scalar string",
913 { "name": "Bob" }
914 ]));
915 let result = df.marshal(body);
916 assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
917 }
918
919 #[test]
920 fn test_marshal_mixed_array_lists_returns_error() {
921 let cfg = CsvConfig::default().use_maps(false);
922 let df = CsvDataFormat::new(cfg);
923 let body = Body::Json(json!([
924 ["a", "b"],
925 { "not": "array" },
926 ["c", "d"]
927 ]));
928 let result = df.marshal(body);
929 assert!(matches!(result, Err(CamelError::TypeConversionFailed(_))));
930 }
931}