1use std::str::FromStr;
2use std::time::Duration;
3
4use camel_api::component_metadata::{ComponentMetadata, UriOption};
5use camel_component_api::CamelError;
6use camel_component_api::NetworkRetryPolicy;
7use camel_component_api::{UriComponents, UriConfig, parse_uri};
8use tracing::warn;
9
10fn redacted_opt(opt: &Option<String>) -> Option<&'static str> {
12 if opt.is_some() { Some("***") } else { None }
13}
14
15pub fn redact_db_url(db_url: &str) -> String {
18 match url::Url::parse(db_url) {
19 Ok(mut parsed) => {
20 if parsed.username().is_empty() && parsed.password().is_none() {
21 return db_url.to_string();
22 }
23 let _ = parsed.set_username("***");
24 let _ = parsed.set_password(Some("***"));
25 parsed.to_string()
26 }
27 Err(_) => {
28 if let Some(scheme_end) = db_url.find("://") {
34 let after_scheme = &db_url[scheme_end + 3..];
35 if let Some(at) = after_scheme.find('@') {
36 let scheme = &db_url[..scheme_end + 3];
37 let host_part = &after_scheme[at + 1..];
38 return format!("{scheme}***@{host_part}");
39 }
40 }
41 db_url.to_string()
42 }
43 }
44}
45
46#[derive(Debug, Clone, PartialEq, Default)]
48pub enum SqlOutputType {
49 #[default]
51 SelectList,
52 SelectOne,
54 StreamList,
56}
57
58impl FromStr for SqlOutputType {
59 type Err = CamelError;
60
61 fn from_str(s: &str) -> Result<Self, Self::Err> {
62 match s {
63 "SelectList" => Ok(SqlOutputType::SelectList),
64 "SelectOne" => Ok(SqlOutputType::SelectOne),
65 "StreamList" => Ok(SqlOutputType::StreamList),
66 _ => Err(CamelError::InvalidUri(format!(
67 "Unknown output type: {}",
68 s
69 ))),
70 }
71 }
72}
73
74#[derive(Debug, Clone, PartialEq, Default)]
82pub enum TransactionMode {
83 #[default]
85 Auto,
86 Managed,
88}
89
90impl FromStr for TransactionMode {
91 type Err = CamelError;
92
93 fn from_str(s: &str) -> Result<Self, Self::Err> {
94 match s {
95 "Auto" => Ok(TransactionMode::Auto),
96 "Managed" => Ok(TransactionMode::Managed),
97 _ => Err(CamelError::InvalidUri(format!(
98 "Unknown transaction mode: {}. Expected 'Auto' or 'Managed'",
99 s
100 ))),
101 }
102 }
103}
104
105impl std::fmt::Display for TransactionMode {
106 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
107 match self {
108 TransactionMode::Auto => write!(f, "Auto"),
109 TransactionMode::Managed => write!(f, "Managed"),
110 }
111 }
112}
113
114#[derive(Debug, Clone, PartialEq, Default)]
116pub enum ProcessingStrategy {
117 #[default]
119 Direct,
120 Scheduled,
122}
123
124impl FromStr for ProcessingStrategy {
125 type Err = CamelError;
126
127 fn from_str(s: &str) -> Result<Self, Self::Err> {
128 match s {
129 "Direct" => Ok(ProcessingStrategy::Direct),
130 "Scheduled" => Ok(ProcessingStrategy::Scheduled),
131 _ => Err(CamelError::InvalidUri(format!(
132 "Unknown processing strategy: {}. Expected 'Direct' or 'Scheduled'",
133 s
134 ))),
135 }
136 }
137}
138
139impl std::fmt::Display for ProcessingStrategy {
140 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
141 match self {
142 ProcessingStrategy::Direct => write!(f, "Direct"),
143 ProcessingStrategy::Scheduled => write!(f, "Scheduled"),
144 }
145 }
146}
147
148#[derive(Debug, Clone, PartialEq, Default)]
150pub enum PollStrategy {
151 #[default]
153 Sequential,
154 Burst,
156}
157
158impl FromStr for PollStrategy {
159 type Err = CamelError;
160
161 fn from_str(s: &str) -> Result<Self, Self::Err> {
162 match s {
163 "Sequential" => Ok(PollStrategy::Sequential),
164 "Burst" => Ok(PollStrategy::Burst),
165 _ => Err(CamelError::InvalidUri(format!(
166 "Unknown poll strategy: {}. Expected 'Sequential' or 'Burst'",
167 s
168 ))),
169 }
170 }
171}
172
173impl std::fmt::Display for PollStrategy {
174 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
175 match self {
176 PollStrategy::Sequential => write!(f, "Sequential"),
177 PollStrategy::Burst => write!(f, "Burst"),
178 }
179 }
180}
181
182#[derive(Clone, PartialEq, serde::Deserialize)]
189#[serde(default)]
190pub struct SqlGlobalConfig {
191 pub max_connections: u32,
192 pub min_connections: u32,
193 pub idle_timeout_secs: u64,
194 pub max_lifetime_secs: u64,
195 pub ssl_mode: Option<String>,
197 pub ssl_root_cert: Option<String>,
198 pub ssl_cert: Option<String>,
199 pub ssl_key: Option<String>,
200 #[serde(default)]
202 pub retry: NetworkRetryPolicy,
203}
204
205impl std::fmt::Debug for SqlGlobalConfig {
206 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
207 f.debug_struct("SqlGlobalConfig")
208 .field("max_connections", &self.max_connections)
209 .field("min_connections", &self.min_connections)
210 .field("idle_timeout_secs", &self.idle_timeout_secs)
211 .field("max_lifetime_secs", &self.max_lifetime_secs)
212 .field("ssl_mode", &self.ssl_mode)
213 .field("ssl_root_cert", &self.ssl_root_cert)
214 .field("ssl_cert", &self.ssl_cert)
215 .field("ssl_key", &redacted_opt(&self.ssl_key))
216 .field("retry", &self.retry)
217 .finish()
218 }
219}
220
221impl Default for SqlGlobalConfig {
222 fn default() -> Self {
223 Self {
224 max_connections: 5,
225 min_connections: 1,
226 idle_timeout_secs: 300,
227 max_lifetime_secs: 1800,
228 ssl_mode: None,
229 ssl_root_cert: None,
230 ssl_cert: None,
231 ssl_key: None,
232 retry: NetworkRetryPolicy::default(),
233 }
234 }
235}
236
237impl SqlGlobalConfig {
238 pub fn new() -> Self {
239 Self::default()
240 }
241
242 pub fn with_max_connections(mut self, value: u32) -> Self {
243 self.max_connections = value;
244 self
245 }
246
247 pub fn with_min_connections(mut self, value: u32) -> Self {
248 self.min_connections = value;
249 self
250 }
251
252 pub fn with_idle_timeout_secs(mut self, value: u64) -> Self {
253 self.idle_timeout_secs = value;
254 self
255 }
256
257 pub fn with_max_lifetime_secs(mut self, value: u64) -> Self {
258 self.max_lifetime_secs = value;
259 self
260 }
261
262 pub fn with_ssl_mode(mut self, value: impl Into<String>) -> Self {
263 self.ssl_mode = Some(value.into());
264 self
265 }
266
267 pub fn with_ssl_root_cert(mut self, value: impl Into<String>) -> Self {
268 self.ssl_root_cert = Some(value.into());
269 self
270 }
271
272 pub fn with_ssl_cert(mut self, value: impl Into<String>) -> Self {
273 self.ssl_cert = Some(value.into());
274 self
275 }
276
277 pub fn with_ssl_key(mut self, value: impl Into<String>) -> Self {
278 self.ssl_key = Some(value.into());
279 self
280 }
281
282 pub fn with_retry(mut self, value: NetworkRetryPolicy) -> Self {
283 self.retry = value;
284 self
285 }
286}
287
288#[derive(Clone)]
305pub struct SqlEndpointConfig {
306 pub db_url: String,
309 pub datasource_name: Option<String>,
311 pub max_connections: Option<u32>,
313 pub min_connections: Option<u32>,
315 pub idle_timeout_secs: Option<u64>,
317 pub max_lifetime_secs: Option<u64>,
319
320 pub query: String,
323 pub source_path: Option<String>,
325 pub output_type: SqlOutputType,
327 pub placeholder: char,
329 pub use_placeholder: bool,
331 pub noop: bool,
333 pub in_separator: String,
335
336 pub always_populate_statement: bool,
340
341 pub allow_named_parameters: bool,
345
346 pub fetch_size: Option<u32>,
349
350 pub transaction_mode: TransactionMode,
353
354 pub delay_ms: u64,
357 pub initial_delay_ms: u64,
359 pub max_messages_per_poll: Option<i32>,
361 pub on_consume: Option<String>,
363 pub on_consume_failed: Option<String>,
365 pub on_consume_batch_complete: Option<String>,
367 pub route_empty_result_set: bool,
369 pub use_iterator: bool,
371 pub expected_update_count: Option<i64>,
373 pub break_batch_on_consume_fail: bool,
375 pub bridge_error_handler: bool,
377
378 pub repeat_count: Option<u32>,
384
385 pub break_on_empty: bool,
390
391 pub processing_strategy: ProcessingStrategy,
394
395 pub poll_strategy: PollStrategy,
398
399 pub batch: bool,
402 pub use_message_body_for_sql: bool,
404 pub allow_dynamic_query: bool,
407
408 pub ssl_mode: Option<String>,
411 pub ssl_root_cert: Option<String>,
413 pub ssl_cert: Option<String>,
415 pub ssl_key: Option<String>,
417
418 pub retry: NetworkRetryPolicy,
420
421 retry_set_from_uri: bool,
425}
426
427impl std::fmt::Debug for SqlEndpointConfig {
428 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
429 f.debug_struct("SqlEndpointConfig")
430 .field("db_url", &redact_db_url(&self.db_url))
431 .field("datasource_name", &self.datasource_name)
432 .field("max_connections", &self.max_connections)
433 .field("min_connections", &self.min_connections)
434 .field("idle_timeout_secs", &self.idle_timeout_secs)
435 .field("max_lifetime_secs", &self.max_lifetime_secs)
436 .field("query", &self.query)
437 .field("source_path", &self.source_path)
438 .field("output_type", &self.output_type)
439 .field("placeholder", &self.placeholder)
440 .field("use_placeholder", &self.use_placeholder)
441 .field("noop", &self.noop)
442 .field("in_separator", &self.in_separator)
443 .field("always_populate_statement", &self.always_populate_statement)
444 .field("allow_named_parameters", &self.allow_named_parameters)
445 .field("fetch_size", &self.fetch_size)
446 .field("transaction_mode", &self.transaction_mode)
447 .field("delay_ms", &self.delay_ms)
448 .field("initial_delay_ms", &self.initial_delay_ms)
449 .field("max_messages_per_poll", &self.max_messages_per_poll)
450 .field("on_consume", &self.on_consume)
451 .field("on_consume_failed", &self.on_consume_failed)
452 .field("on_consume_batch_complete", &self.on_consume_batch_complete)
453 .field("route_empty_result_set", &self.route_empty_result_set)
454 .field("use_iterator", &self.use_iterator)
455 .field("expected_update_count", &self.expected_update_count)
456 .field(
457 "break_batch_on_consume_fail",
458 &self.break_batch_on_consume_fail,
459 )
460 .field("bridge_error_handler", &self.bridge_error_handler)
461 .field("repeat_count", &self.repeat_count)
462 .field("break_on_empty", &self.break_on_empty)
463 .field("processing_strategy", &self.processing_strategy)
464 .field("poll_strategy", &self.poll_strategy)
465 .field("batch", &self.batch)
466 .field("use_message_body_for_sql", &self.use_message_body_for_sql)
467 .field("allow_dynamic_query", &self.allow_dynamic_query)
468 .field("ssl_mode", &self.ssl_mode)
469 .field("ssl_root_cert", &self.ssl_root_cert)
470 .field("ssl_cert", &self.ssl_cert)
471 .field("ssl_key", &redacted_opt(&self.ssl_key))
472 .field("retry", &self.retry)
473 .finish()
474 }
475}
476
477#[derive(Debug, Clone, UriConfig)]
483#[allow(dead_code)]
484#[uri_scheme = "sql"]
485#[uri_config(
486 skip_impl,
487 metadata(
488 scheme = "sql",
489 description = "Execute SQL against a configured datasource",
490 producer,
491 consumer
492 ),
493 crate = "camel_component_api"
494)]
495struct SqlUriConfig {
496 #[uri_param(name = "db_url", desc = "Database connection URL", secret)]
498 pub db_url: Option<String>,
499 #[uri_param(
500 name = "datasource",
501 desc = "Named datasource reference (from CamelConfig.datasources)"
502 )]
503 pub datasource: Option<String>,
504 #[uri_param(name = "maxConnections", desc = "Maximum connections in the pool")]
505 pub max_connections: Option<u64>,
506 #[uri_param(name = "minConnections", desc = "Minimum connections in the pool")]
507 pub min_connections: Option<u64>,
508 #[uri_param(name = "idleTimeoutSecs", desc = "Idle timeout in seconds")]
509 pub idle_timeout_secs: Option<u64>,
510 #[uri_param(
511 name = "maxLifetimeSecs",
512 desc = "Maximum connection lifetime in seconds"
513 )]
514 pub max_lifetime_secs: Option<u64>,
515
516 #[uri_param(
518 name = "outputType",
519 kind = "enum:SelectList,SelectOne,StreamList",
520 default = "SelectList",
521 desc = "Output type for query results"
522 )]
523 pub output_type: Option<String>,
524 #[uri_param(
525 name = "placeholder",
526 default = "#",
527 desc = "Placeholder character for parameters"
528 )]
529 pub placeholder: Option<String>,
530 #[uri_param(
531 name = "usePlaceholder",
532 default = "true",
533 desc = "Process parameter placeholders in queries"
534 )]
535 pub use_placeholder: Option<bool>,
536 #[uri_param(
537 name = "noop",
538 default = "false",
539 desc = "Dry-run mode (don't execute the query)"
540 )]
541 pub noop: Option<bool>,
542 #[uri_param(
543 name = "inSeparator",
544 default = ", ",
545 desc = "Separator for IN clause expansion"
546 )]
547 pub in_separator: Option<String>,
548 #[uri_param(
549 name = "alwaysPopulateStatement",
550 default = "false",
551 desc = "Always bind parameters even if the exchange body is null or empty"
552 )]
553 pub always_populate_statement: Option<bool>,
554 #[uri_param(
555 name = "allowNamedParameters",
556 default = "true",
557 desc = "Recognize :name style placeholders from headers or body"
558 )]
559 pub allow_named_parameters: Option<bool>,
560 #[uri_param(name = "fetchSize", desc = "Fetch size hint for query results")]
561 pub fetch_size: Option<u64>,
562 #[uri_param(
563 name = "transactionMode",
564 kind = "enum:Auto,Managed",
565 default = "Auto",
566 desc = "Transaction mode for SQL operations"
567 )]
568 pub transaction_mode: Option<String>,
569
570 #[uri_param(
572 name = "delay",
573 default = "500",
574 desc = "Delay between polls in milliseconds"
575 )]
576 pub delay: Option<u64>,
577 #[uri_param(
578 name = "initialDelay",
579 default = "1000",
580 desc = "Initial delay before first poll in milliseconds"
581 )]
582 pub initial_delay: Option<u64>,
583 #[uri_param(name = "maxMessagesPerPoll", desc = "Maximum messages per poll")]
584 pub max_messages_per_poll: Option<u64>,
585 #[uri_param(
586 name = "onConsume",
587 desc = "SQL to execute after consuming each message"
588 )]
589 pub on_consume: Option<String>,
590 #[uri_param(name = "onConsumeFailed", desc = "SQL to execute if consumption fails")]
591 pub on_consume_failed: Option<String>,
592 #[uri_param(
593 name = "onConsumeBatchComplete",
594 desc = "SQL to execute after consuming a batch"
595 )]
596 pub on_consume_batch_complete: Option<String>,
597 #[uri_param(
598 name = "routeEmptyResultSet",
599 default = "false",
600 desc = "Route empty result sets"
601 )]
602 pub route_empty_result_set: Option<bool>,
603 #[uri_param(
604 name = "useIterator",
605 default = "true",
606 desc = "Use iterator for results"
607 )]
608 pub use_iterator: Option<bool>,
609 #[uri_param(
610 name = "expectedUpdateCount",
611 desc = "Expected number of rows affected"
612 )]
613 pub expected_update_count: Option<u64>,
614 #[uri_param(
615 name = "breakBatchOnConsumeFail",
616 default = "false",
617 desc = "Break batch on consume failure"
618 )]
619 pub break_batch_on_consume_fail: Option<bool>,
620 #[uri_param(
621 name = "bridgeErrorHandler",
622 default = "false",
623 desc = "Bridge poll errors into route error handling"
624 )]
625 pub bridge_error_handler: Option<bool>,
626 #[uri_param(
627 name = "repeatCount",
628 desc = "Maximum number of polls before the consumer stops"
629 )]
630 pub repeat_count: Option<u64>,
631 #[uri_param(
632 name = "breakOnEmpty",
633 default = "false",
634 desc = "Stop the consumer after a poll that returns zero rows"
635 )]
636 pub break_on_empty: Option<bool>,
637 #[uri_param(
638 name = "processingStrategy",
639 kind = "enum:Direct,Scheduled",
640 default = "Direct",
641 desc = "Processing strategy for the consumer"
642 )]
643 pub processing_strategy: Option<String>,
644 #[uri_param(
645 name = "pollStrategy",
646 kind = "enum:Sequential,Burst",
647 default = "Sequential",
648 desc = "Poll strategy for the consumer"
649 )]
650 pub poll_strategy: Option<String>,
651
652 #[uri_param(name = "batch", default = "false", desc = "Enable batch mode")]
654 pub batch: Option<bool>,
655 #[uri_param(
656 name = "useMessageBodyForSql",
657 default = "false",
658 desc = "Use message body for SQL"
659 )]
660 pub use_message_body_for_sql: Option<bool>,
661 #[uri_param(
662 name = "allowDynamicQuery",
663 default = "false",
664 desc = "Allow queries to be sourced from exchange headers or body"
665 )]
666 pub allow_dynamic_query: Option<bool>,
667
668 #[uri_param(name = "sslMode", desc = "SSL mode for the connection")]
670 pub ssl_mode: Option<String>,
671 #[uri_param(name = "sslRootCert", desc = "Path to SSL root certificate")]
672 pub ssl_root_cert: Option<String>,
673 #[uri_param(name = "sslCert", desc = "Path to SSL client certificate")]
674 pub ssl_cert: Option<String>,
675 #[uri_param(name = "sslKey", desc = "Path to SSL client key", secret)]
676 pub ssl_key: Option<String>,
677}
678
679impl SqlEndpointConfig {
680 pub fn metadata() -> ComponentMetadata {
683 SqlUriConfig::metadata()
684 }
685
686 pub fn uri_options() -> Vec<UriOption> {
689 SqlUriConfig::uri_options()
690 }
691
692 pub fn apply_defaults(&mut self, defaults: &SqlGlobalConfig) {
694 if self.max_connections.is_none() {
695 self.max_connections = Some(defaults.max_connections);
696 }
697 if self.min_connections.is_none() {
698 self.min_connections = Some(defaults.min_connections);
699 }
700 if self.idle_timeout_secs.is_none() {
701 self.idle_timeout_secs = Some(defaults.idle_timeout_secs);
702 }
703 if self.max_lifetime_secs.is_none() {
704 self.max_lifetime_secs = Some(defaults.max_lifetime_secs);
705 }
706 if self.ssl_mode.is_none() {
707 self.ssl_mode = defaults.ssl_mode.clone();
708 }
709 if self.ssl_root_cert.is_none() {
710 self.ssl_root_cert = defaults.ssl_root_cert.clone();
711 }
712 if self.ssl_cert.is_none() {
713 self.ssl_cert = defaults.ssl_cert.clone();
714 }
715 if self.ssl_key.is_none() {
716 self.ssl_key = defaults.ssl_key.clone();
717 }
718 if !self.retry_set_from_uri {
720 self.retry = defaults.retry.clone();
721 }
722 }
723
724 pub fn resolve_defaults(&mut self) {
726 let defaults = SqlGlobalConfig::default();
727 self.apply_defaults(&defaults);
728 }
729
730 pub async fn resolve_file_query(&mut self) -> Result<(), CamelError> {
739 if let Some(file_path) = self.source_path.take() {
740 let contents = tokio::fs::read_to_string(&file_path).await.map_err(|e| {
741 CamelError::Config(format!("Failed to read SQL file '{}': {}", file_path, e))
742 })?;
743 self.query = contents.trim().to_string();
744 self.source_path = Some(file_path);
746 }
747 Ok(())
748 }
749}
750
751struct SslParamMapping {
752 pg_key: &'static str,
753 mysql_key: &'static str,
754}
755
756const SSL_MAPPINGS: &[(&str, SslParamMapping)] = &[
757 (
758 "sslMode",
759 SslParamMapping {
760 pg_key: "sslmode",
761 mysql_key: "ssl-mode",
762 },
763 ),
764 (
765 "sslRootCert",
766 SslParamMapping {
767 pg_key: "sslrootcert",
768 mysql_key: "ssl-ca",
769 },
770 ),
771 (
772 "sslCert",
773 SslParamMapping {
774 pg_key: "sslcert",
775 mysql_key: "ssl-cert",
776 },
777 ),
778 (
779 "sslKey",
780 SslParamMapping {
781 pg_key: "sslkey",
782 mysql_key: "ssl-key",
783 },
784 ),
785];
786
787pub fn enrich_db_url_with_ssl(
788 db_url: &str,
789 config: &SqlEndpointConfig,
790) -> Result<String, CamelError> {
791 enrich_db_url_with_ssl_params(
792 db_url,
793 config.ssl_mode.as_deref(),
794 config.ssl_root_cert.as_deref(),
795 config.ssl_cert.as_deref(),
796 config.ssl_key.as_deref(),
797 )
798}
799
800pub(crate) fn enrich_db_url_with_ssl_params(
801 db_url: &str,
802 ssl_mode: Option<&str>,
803 ssl_root_cert: Option<&str>,
804 ssl_cert: Option<&str>,
805 ssl_key: Option<&str>,
806) -> Result<String, CamelError> {
807 let mut parsed = url::Url::parse(db_url).map_err(|e| {
808 CamelError::InvalidUri(format!(
809 "Cannot parse database URL for SSL enrichment: {}",
810 e
811 ))
812 })?;
813
814 let scheme = parsed.scheme();
815 if scheme.starts_with("sqlite") {
816 if ssl_mode.is_some() || ssl_root_cert.is_some() || ssl_cert.is_some() || ssl_key.is_some()
817 {
818 warn!(
819 "SSL options configured for SQLite database URL, but SQLite does not support SSL/TLS; ignoring sslMode/sslRootCert/sslCert/sslKey"
820 );
821 }
822 return Ok(db_url.to_string());
823 }
824
825 if scheme != "postgres" && scheme != "postgresql" && scheme != "mysql" {
826 return Ok(db_url.to_string());
827 }
828 let is_mysql = scheme == "mysql";
829
830 let effective_ssl_mode: &str = ssl_mode.unwrap_or(if is_mysql { "prefer" } else { "require" });
832
833 let ssl_params: Vec<(&str, &str)> = [
834 Some(("sslMode", effective_ssl_mode)),
835 ssl_root_cert.map(|v| ("sslRootCert", v)),
836 ssl_cert.map(|v| ("sslCert", v)),
837 ssl_key.map(|v| ("sslKey", v)),
838 ]
839 .into_iter()
840 .flatten()
841 .collect();
842
843 let mut query_pairs = parsed.query_pairs().collect::<Vec<_>>();
844 for (camel_name, value) in &ssl_params {
845 if let Some((_, mapping)) = SSL_MAPPINGS.iter().find(|(name, _)| *name == *camel_name) {
846 let driver_key = if is_mysql {
847 mapping.mysql_key
848 } else {
849 mapping.pg_key
850 };
851
852 if let Some(pos) = query_pairs.iter().position(|(k, _)| k == driver_key) {
853 query_pairs[pos].1 = (*value).into();
854 } else {
855 query_pairs.push((driver_key.into(), (*value).into()));
856 }
857 }
858 }
859
860 if !query_pairs.iter().any(|(k, _)| k == "connect_timeout") {
863 query_pairs.push(("connect_timeout".into(), "10".into()));
864 }
865
866 {
867 let mut serializer = url::form_urlencoded::Serializer::new(String::new());
868 for (k, v) in &query_pairs {
869 serializer.append_pair(k, v);
870 }
871 parsed.set_query(Some(&serializer.finish()));
872 }
873
874 Ok(parsed.to_string())
875}
876
877impl UriConfig for SqlEndpointConfig {
878 fn scheme() -> &'static str {
879 "sql"
880 }
881
882 fn from_uri(uri: &str) -> Result<Self, CamelError> {
883 let parts = parse_uri(uri)?;
884 Self::from_components(parts)
885 }
886
887 fn from_components(parts: UriComponents) -> Result<Self, CamelError> {
888 if parts.scheme != Self::scheme() {
890 return Err(CamelError::InvalidUri(format!(
891 "expected scheme '{}' but got '{}'",
892 Self::scheme(),
893 parts.scheme
894 )));
895 }
896
897 let params = &parts.params;
898
899 let (query, source_path) = if parts.path.starts_with("file:") {
904 let file_path = parts.path.trim_start_matches("file:").to_string();
905 (String::new(), Some(file_path))
906 } else {
907 (parts.path.clone(), None)
908 };
909
910 let db_url = params.get("db_url").cloned().unwrap_or_default();
912
913 let datasource_name = params.get("datasource").cloned();
915
916 let max_connections = params.get("maxConnections").and_then(|v| v.parse().ok());
918 let min_connections = params.get("minConnections").and_then(|v| v.parse().ok());
919 let idle_timeout_secs = params.get("idleTimeoutSecs").and_then(|v| v.parse().ok());
920 let max_lifetime_secs = params.get("maxLifetimeSecs").and_then(|v| v.parse().ok());
921
922 let output_type = params
924 .get("outputType")
925 .map(|s| s.parse())
926 .transpose()?
927 .unwrap_or_default();
928 let placeholder = params
929 .get("placeholder")
930 .filter(|v| !v.is_empty())
931 .map(|v| {
932 if v.chars().count() != 1 {
933 return Err(CamelError::InvalidUri(format!(
934 "placeholder must be exactly one character, got '{}'",
935 v
936 )));
937 }
938 if !v.is_ascii() {
939 return Err(CamelError::InvalidUri(
940 "placeholder must be a single ASCII character".to_string(),
941 ));
942 }
943 Ok(v.chars().next().unwrap()) })
945 .transpose()?
946 .unwrap_or('#');
947 fn parse_bool_param(name: &str, value: &str) -> Result<bool, CamelError> {
952 if value.eq_ignore_ascii_case("true") {
953 Ok(true)
954 } else if value.eq_ignore_ascii_case("false") {
955 Ok(false)
956 } else {
957 Err(CamelError::InvalidUri(format!(
958 "{} must be 'true' or 'false', got '{}'",
959 name, value
960 )))
961 }
962 }
963
964 let use_placeholder = params
965 .get("usePlaceholder")
966 .map(|v| parse_bool_param("usePlaceholder", v))
967 .transpose()?
968 .unwrap_or(true);
969 let noop = params
970 .get("noop")
971 .map(|v| parse_bool_param("noop", v))
972 .transpose()?
973 .unwrap_or(false);
974 let in_separator = params
975 .get("inSeparator")
976 .map(|v| v.to_string())
977 .unwrap_or_else(|| ", ".to_string());
978 if in_separator.is_empty() {
979 return Err(CamelError::InvalidUri(
980 "inSeparator must not be empty".to_string(),
981 ));
982 }
983
984 let always_populate_statement = params
986 .get("alwaysPopulateStatement")
987 .map(|v| parse_bool_param("alwaysPopulateStatement", v))
988 .transpose()?
989 .unwrap_or(false);
990
991 let allow_named_parameters = params
993 .get("allowNamedParameters")
994 .map(|v| parse_bool_param("allowNamedParameters", v))
995 .transpose()?
996 .unwrap_or(true);
997
998 let fetch_size = params.get("fetchSize").and_then(|v| v.parse().ok());
1000
1001 let transaction_mode = params
1003 .get("transactionMode")
1004 .map(|s| s.parse())
1005 .transpose()?
1006 .unwrap_or_default();
1007
1008 let delay_ms = params
1010 .get("delay")
1011 .and_then(|v| v.parse().ok())
1012 .unwrap_or(500);
1013 let initial_delay_ms = params
1014 .get("initialDelay")
1015 .and_then(|v| v.parse().ok())
1016 .unwrap_or(1000);
1017 let max_messages_per_poll = params
1018 .get("maxMessagesPerPoll")
1019 .and_then(|v| v.parse().ok());
1020 let on_consume = params.get("onConsume").cloned();
1021 let on_consume_failed = params.get("onConsumeFailed").cloned();
1022 let on_consume_batch_complete = params.get("onConsumeBatchComplete").cloned();
1023 let route_empty_result_set = params
1024 .get("routeEmptyResultSet")
1025 .map(|v| parse_bool_param("routeEmptyResultSet", v))
1026 .transpose()?
1027 .unwrap_or(false);
1028 let use_iterator = params
1029 .get("useIterator")
1030 .map(|v| parse_bool_param("useIterator", v))
1031 .transpose()?
1032 .unwrap_or(true);
1033 let expected_update_count = params
1034 .get("expectedUpdateCount")
1035 .and_then(|v| v.parse().ok());
1036 let break_batch_on_consume_fail = params
1037 .get("breakBatchOnConsumeFail")
1038 .map(|v| parse_bool_param("breakBatchOnConsumeFail", v))
1039 .transpose()?
1040 .unwrap_or(false);
1041 let bridge_error_handler = params
1042 .get("bridgeErrorHandler")
1043 .map(|v| parse_bool_param("bridgeErrorHandler", v))
1044 .transpose()?
1045 .unwrap_or(false);
1046
1047 let repeat_count = params.get("repeatCount").and_then(|v| v.parse().ok());
1049
1050 let break_on_empty = params
1052 .get("breakOnEmpty")
1053 .map(|v| parse_bool_param("breakOnEmpty", v))
1054 .transpose()?
1055 .unwrap_or(false);
1056
1057 let processing_strategy = params
1059 .get("processingStrategy")
1060 .map(|s| s.parse())
1061 .transpose()?
1062 .unwrap_or_default();
1063
1064 let poll_strategy = params
1066 .get("pollStrategy")
1067 .map(|s| s.parse())
1068 .transpose()?
1069 .unwrap_or_default();
1070
1071 let batch = params
1073 .get("batch")
1074 .map(|v| parse_bool_param("batch", v))
1075 .transpose()?
1076 .unwrap_or(false);
1077 let use_message_body_for_sql = params
1078 .get("useMessageBodyForSql")
1079 .map(|v| parse_bool_param("useMessageBodyForSql", v))
1080 .transpose()?
1081 .unwrap_or(false);
1082 let allow_dynamic_query = params
1083 .get("allowDynamicQuery")
1084 .map(|v| parse_bool_param("allowDynamicQuery", v))
1085 .transpose()?
1086 .unwrap_or(false);
1087 let ssl_mode = params.get("sslMode").cloned();
1088 let ssl_root_cert = params.get("sslRootCert").cloned();
1089 let ssl_cert = params.get("sslCert").cloned();
1090 let ssl_key = params.get("sslKey").cloned();
1091
1092 let mut retry = NetworkRetryPolicy::default();
1094 let mut retry_set_from_uri = false;
1095 if let Some(raw) = params.get("retryEnabled") {
1096 retry.enabled = raw.parse::<bool>().map_err(|_| {
1097 CamelError::InvalidUri(format!("retryEnabled must be a boolean, got '{raw}'"))
1098 })?;
1099 retry_set_from_uri = true;
1100 }
1101 if let Some(raw) = params.get("retryMaxAttempts") {
1102 retry.max_attempts = raw.parse::<u32>().map_err(|_| {
1103 CamelError::InvalidUri(format!("retryMaxAttempts must be a u32, got '{raw}'"))
1104 })?;
1105 retry_set_from_uri = true;
1106 }
1107 if let Some(raw) = params.get("retryInitialDelayMs") {
1108 retry.initial_delay = Duration::from_millis(raw.parse::<u64>().map_err(|_| {
1109 CamelError::InvalidUri(format!("retryInitialDelayMs must be a u64, got '{raw}'"))
1110 })?);
1111 retry_set_from_uri = true;
1112 }
1113 if let Some(raw) = params.get("retryMultiplier") {
1114 retry.multiplier = raw.parse::<f64>().map_err(|_| {
1115 CamelError::InvalidUri(format!("retryMultiplier must be a f64, got '{raw}'"))
1116 })?;
1117 retry_set_from_uri = true;
1118 }
1119 if let Some(raw) = params.get("retryMaxDelayMs") {
1120 retry.max_delay = Duration::from_millis(raw.parse::<u64>().map_err(|_| {
1121 CamelError::InvalidUri(format!("retryMaxDelayMs must be a u64, got '{raw}'"))
1122 })?);
1123 retry_set_from_uri = true;
1124 }
1125 if let Some(raw) = params.get("retryJitter") {
1126 retry.jitter_factor = raw.parse::<f64>().map_err(|_| {
1127 CamelError::InvalidUri(format!("retryJitter must be a f64, got '{raw}'"))
1128 })?;
1129 retry_set_from_uri = true;
1130 }
1131
1132 if datasource_name.is_none() && db_url.is_empty() {
1133 return Err(CamelError::Config(
1134 "either 'datasource' or 'db_url' parameter is required".to_string(),
1135 ));
1136 }
1137
1138 if datasource_name.is_some() && !db_url.is_empty() {
1139 return Err(CamelError::InvalidUri(
1140 "'db_url' not allowed with named datasource — use 'datasource' alone".to_string(),
1141 ));
1142 }
1143
1144 if datasource_name.is_some() {
1145 let overrides: Vec<&str> = {
1146 let mut v = Vec::new();
1147 if max_connections.is_some() {
1148 v.push("maxConnections");
1149 }
1150 if min_connections.is_some() {
1151 v.push("minConnections");
1152 }
1153 if idle_timeout_secs.is_some() {
1154 v.push("idleTimeoutSecs");
1155 }
1156 if max_lifetime_secs.is_some() {
1157 v.push("maxLifetimeSecs");
1158 }
1159 if ssl_mode.is_some() {
1160 v.push("sslMode");
1161 }
1162 if ssl_root_cert.is_some() {
1163 v.push("sslRootCert");
1164 }
1165 if ssl_cert.is_some() {
1166 v.push("sslCert");
1167 }
1168 if ssl_key.is_some() {
1169 v.push("sslKey");
1170 }
1171 v
1172 };
1173 if !overrides.is_empty() {
1174 return Err(CamelError::InvalidUri(format!(
1175 "pool-affecting params not allowed with named datasource: {}",
1176 overrides.join(", ")
1177 )));
1178 }
1179 }
1180
1181 Ok(Self {
1182 db_url,
1183 datasource_name,
1184 max_connections,
1185 min_connections,
1186 idle_timeout_secs,
1187 max_lifetime_secs,
1188 query,
1189 source_path,
1190 output_type,
1191 placeholder,
1192 use_placeholder,
1193 noop,
1194 in_separator,
1195 always_populate_statement,
1196 allow_named_parameters,
1197 fetch_size,
1198 transaction_mode,
1199 delay_ms,
1200 initial_delay_ms,
1201 max_messages_per_poll,
1202 on_consume,
1203 on_consume_failed,
1204 on_consume_batch_complete,
1205 route_empty_result_set,
1206 use_iterator,
1207 expected_update_count,
1208 break_batch_on_consume_fail,
1209 bridge_error_handler,
1210 repeat_count,
1211 break_on_empty,
1212 processing_strategy,
1213 poll_strategy,
1214 batch,
1215 use_message_body_for_sql,
1216 allow_dynamic_query,
1217 ssl_mode,
1218 ssl_root_cert,
1219 ssl_cert,
1220 ssl_key,
1221 retry,
1222 retry_set_from_uri,
1223 })
1224 }
1225}
1226
1227#[cfg(test)]
1228mod tests {
1229 use super::*;
1230 use camel_component_api::NetworkRetryPolicy;
1231
1232 #[test]
1233 fn config_defaults() {
1234 let mut c =
1235 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1236 c.resolve_defaults();
1237 assert_eq!(c.query, "select 1");
1238 assert_eq!(c.db_url, "postgres://localhost/test");
1239 assert_eq!(c.max_connections, Some(5));
1240 assert_eq!(c.min_connections, Some(1));
1241 assert_eq!(c.idle_timeout_secs, Some(300));
1242 assert_eq!(c.max_lifetime_secs, Some(1800));
1243 assert_eq!(c.output_type, SqlOutputType::SelectList);
1244 assert_eq!(c.placeholder, '#');
1245 assert!(!c.noop);
1246 assert_eq!(c.in_separator, ", ");
1247 assert_eq!(c.delay_ms, 500);
1248 assert_eq!(c.initial_delay_ms, 1000);
1249 assert!(c.max_messages_per_poll.is_none());
1250 assert!(c.on_consume.is_none());
1251 assert!(c.on_consume_failed.is_none());
1252 assert!(c.on_consume_batch_complete.is_none());
1253 assert!(!c.route_empty_result_set);
1254 assert!(c.use_iterator);
1255 assert!(c.expected_update_count.is_none());
1256 assert!(!c.break_batch_on_consume_fail);
1257 assert!(!c.batch);
1258 assert!(!c.use_message_body_for_sql);
1259 assert!(!c.allow_dynamic_query);
1260 assert!(c.ssl_mode.is_none());
1261 assert!(c.ssl_root_cert.is_none());
1262 assert!(c.ssl_cert.is_none());
1263 assert!(c.ssl_key.is_none());
1264 assert!(!c.always_populate_statement);
1266 assert!(c.allow_named_parameters);
1267 assert!(c.fetch_size.is_none());
1268 assert_eq!(c.transaction_mode, TransactionMode::Auto);
1269 assert!(c.repeat_count.is_none());
1270 assert_eq!(c.processing_strategy, ProcessingStrategy::Direct);
1271 assert_eq!(c.poll_strategy, PollStrategy::Sequential);
1272 }
1273
1274 #[test]
1275 fn ssl_none_by_default() {
1276 let c =
1277 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1278 assert!(c.ssl_mode.is_none());
1279 assert!(c.ssl_root_cert.is_none());
1280 assert!(c.ssl_cert.is_none());
1281 assert!(c.ssl_key.is_none());
1282 }
1283
1284 #[test]
1285 fn ssl_mode_from_uri() {
1286 let c = SqlEndpointConfig::from_uri(
1287 "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1288 )
1289 .unwrap();
1290 assert_eq!(c.ssl_mode, Some("require".to_string()));
1291 assert!(c.ssl_root_cert.is_none());
1292 }
1293
1294 #[test]
1295 fn ssl_all_params_from_uri() {
1296 let c = SqlEndpointConfig::from_uri(
1297 "sql:select 1?db_url=postgres://localhost/test&sslMode=require&sslRootCert=/ca.pem&sslCert=/cert.pem&sslKey=/key.pem",
1298 )
1299 .unwrap();
1300 assert_eq!(c.ssl_mode, Some("require".to_string()));
1301 assert_eq!(c.ssl_root_cert, Some("/ca.pem".to_string()));
1302 assert_eq!(c.ssl_cert, Some("/cert.pem".to_string()));
1303 assert_eq!(c.ssl_key, Some("/key.pem".to_string()));
1304 }
1305
1306 #[test]
1307 fn ssl_global_applied_to_endpoint() {
1308 let mut c =
1309 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1310 let global = SqlGlobalConfig::default()
1311 .with_ssl_mode("require")
1312 .with_ssl_root_cert("/etc/ssl/ca.pem");
1313 c.apply_defaults(&global);
1314 assert_eq!(c.ssl_mode, Some("require".to_string()));
1315 assert_eq!(c.ssl_root_cert, Some("/etc/ssl/ca.pem".to_string()));
1316 assert!(c.ssl_cert.is_none());
1317 assert!(c.ssl_key.is_none());
1318 }
1319
1320 #[test]
1321 fn ssl_uri_overrides_global() {
1322 let mut c = SqlEndpointConfig::from_uri(
1323 "sql:select 1?db_url=postgres://localhost/test&sslMode=verify-full",
1324 )
1325 .unwrap();
1326 let global = SqlGlobalConfig::default().with_ssl_mode("require");
1327 c.apply_defaults(&global);
1328 assert_eq!(c.ssl_mode, Some("verify-full".to_string()));
1329 }
1330
1331 #[test]
1332 fn config_wrong_scheme() {
1333 assert!(SqlEndpointConfig::from_uri("redis://localhost:6379").is_err());
1334 }
1335
1336 #[test]
1337 fn config_missing_db_url() {
1338 assert!(SqlEndpointConfig::from_uri("sql:select 1").is_err());
1339 }
1340
1341 #[test]
1342 fn config_output_type_select_one() {
1343 let c = SqlEndpointConfig::from_uri(
1344 "sql:select 1?db_url=postgres://localhost/test&outputType=SelectOne",
1345 )
1346 .unwrap();
1347 assert_eq!(c.output_type, SqlOutputType::SelectOne);
1348 }
1349
1350 #[test]
1351 fn config_output_type_stream_list() {
1352 let c = SqlEndpointConfig::from_uri(
1353 "sql:select 1?db_url=postgres://localhost/test&outputType=StreamList",
1354 )
1355 .unwrap();
1356 assert_eq!(c.output_type, SqlOutputType::StreamList);
1357 }
1358
1359 #[test]
1360 fn in_separator_default() {
1361 let c =
1362 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1363 assert_eq!(c.in_separator, ", ");
1364 }
1365
1366 #[test]
1367 fn in_separator_from_uri() {
1368 let c = SqlEndpointConfig::from_uri(
1369 "sql:select 1?db_url=postgres://localhost/test&inSeparator=;",
1370 )
1371 .unwrap();
1372 assert_eq!(c.in_separator, ";");
1373 }
1374
1375 #[test]
1376 fn in_separator_empty_rejected() {
1377 let result = SqlEndpointConfig::from_uri(
1378 "sql:select 1?db_url=postgres://localhost/test&inSeparator=",
1379 );
1380 assert!(result.is_err());
1381 let msg = format!("{:?}", result.unwrap_err());
1382 assert!(msg.contains("inSeparator") || msg.contains("empty"));
1383 }
1384
1385 #[test]
1386 fn config_consumer_options() {
1387 let c = SqlEndpointConfig::from_uri(
1388 "sql:select * from t?db_url=postgres://localhost/test&delay=2000&initialDelay=500&maxMessagesPerPoll=10&onConsume=update t set done=true where id=:#id&onConsumeFailed=update t set failed=true where id=:#id&onConsumeBatchComplete=delete from t where done=true&routeEmptyResultSet=true&useIterator=false&expectedUpdateCount=1&breakBatchOnConsumeFail=true"
1389 ).unwrap();
1390 assert_eq!(c.delay_ms, 2000);
1391 assert_eq!(c.initial_delay_ms, 500);
1392 assert_eq!(c.max_messages_per_poll, Some(10));
1393 assert_eq!(
1394 c.on_consume,
1395 Some("update t set done=true where id=:#id".to_string())
1396 );
1397 assert_eq!(
1398 c.on_consume_failed,
1399 Some("update t set failed=true where id=:#id".to_string())
1400 );
1401 assert_eq!(
1402 c.on_consume_batch_complete,
1403 Some("delete from t where done=true".to_string())
1404 );
1405 assert!(c.route_empty_result_set);
1406 assert!(!c.use_iterator);
1407 assert_eq!(c.expected_update_count, Some(1));
1408 assert!(c.break_batch_on_consume_fail);
1409 assert!(!c.bridge_error_handler);
1410 }
1411
1412 #[test]
1413 fn config_producer_options() {
1414 let c = SqlEndpointConfig::from_uri(
1415 "sql:insert into t values (#)?db_url=postgres://localhost/test&batch=true&useMessageBodyForSql=true&noop=true"
1416 ).unwrap();
1417 assert!(c.batch);
1418 assert!(c.use_message_body_for_sql);
1419 assert!(c.noop);
1420 }
1421
1422 #[test]
1423 fn config_pool_options() {
1424 let c = SqlEndpointConfig::from_uri(
1425 "sql:select 1?db_url=postgres://localhost/test&maxConnections=20&minConnections=3&idleTimeoutSecs=600&maxLifetimeSecs=3600"
1426 ).unwrap();
1427 assert_eq!(c.max_connections, Some(20));
1428 assert_eq!(c.min_connections, Some(3));
1429 assert_eq!(c.idle_timeout_secs, Some(600));
1430 assert_eq!(c.max_lifetime_secs, Some(3600));
1431 }
1432
1433 #[test]
1434 fn config_query_with_special_chars() {
1435 let c = SqlEndpointConfig::from_uri(
1436 "sql:select * from users where name = :#name and age > #?db_url=postgres://localhost/test",
1437 )
1438 .unwrap();
1439 assert_eq!(
1440 c.query,
1441 "select * from users where name = :#name and age > #"
1442 );
1443 }
1444
1445 #[test]
1446 fn output_type_from_str() {
1447 assert_eq!(
1448 "SelectList".parse::<SqlOutputType>().unwrap(),
1449 SqlOutputType::SelectList
1450 );
1451 assert_eq!(
1452 "SelectOne".parse::<SqlOutputType>().unwrap(),
1453 SqlOutputType::SelectOne
1454 );
1455 assert_eq!(
1456 "StreamList".parse::<SqlOutputType>().unwrap(),
1457 SqlOutputType::StreamList
1458 );
1459 assert!("Invalid".parse::<SqlOutputType>().is_err());
1460 }
1461
1462 #[tokio::test]
1464 async fn config_file_not_found() {
1465 let mut config = SqlEndpointConfig::from_uri(
1466 "sql:file:/nonexistent/path/query.sql?db_url=postgres://localhost/test",
1467 )
1468 .expect("from_uri should defer file reading");
1469 assert_eq!(
1471 config.source_path,
1472 Some("/nonexistent/path/query.sql".to_string())
1473 );
1474 assert!(config.query.is_empty());
1475
1476 let result = config.resolve_file_query().await;
1478 assert!(result.is_err());
1479 let msg = format!("{:?}", result.unwrap_err());
1480 assert!(msg.contains("Failed to read SQL file") || msg.contains("nonexistent"));
1481 }
1482
1483 #[tokio::test]
1485 async fn config_file_query() {
1486 use std::io::Write;
1487 let unique_name = format!(
1488 "test_sql_query_{}.sql",
1489 std::time::SystemTime::now()
1490 .duration_since(std::time::UNIX_EPOCH)
1491 .unwrap_or_default()
1492 .as_nanos()
1493 );
1494 let mut tmp = std::env::temp_dir();
1495 tmp.push(unique_name);
1496 {
1497 let mut f = std::fs::File::create(&tmp).unwrap();
1498 writeln!(f, "SELECT * FROM users").unwrap();
1499 }
1500 let uri = format!(
1501 "sql:file:{}?db_url=postgres://localhost/test",
1502 tmp.display()
1503 );
1504 let mut c = SqlEndpointConfig::from_uri(&uri).unwrap();
1505 assert!(c.query.is_empty());
1507 assert_eq!(c.source_path, Some(tmp.to_string_lossy().into_owned()));
1508
1509 c.resolve_file_query()
1511 .await
1512 .expect("file query should resolve");
1513 assert_eq!(c.query, "SELECT * FROM users");
1514 std::fs::remove_file(&tmp).ok();
1515 }
1516
1517 #[test]
1519 fn pool_fields_none_when_not_set() {
1520 let c =
1521 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1522 assert_eq!(c.max_connections, None);
1523 assert_eq!(c.min_connections, None);
1524 assert_eq!(c.idle_timeout_secs, None);
1525 assert_eq!(c.max_lifetime_secs, None);
1526 }
1527
1528 #[test]
1529 fn apply_defaults_fills_none() {
1530 let mut c =
1531 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1532 let global = SqlGlobalConfig {
1533 max_connections: 10,
1534 min_connections: 2,
1535 idle_timeout_secs: 600,
1536 max_lifetime_secs: 3600,
1537 ssl_mode: None,
1538 ssl_root_cert: None,
1539 ssl_cert: None,
1540 ssl_key: None,
1541 retry: NetworkRetryPolicy::default(),
1542 };
1543 c.apply_defaults(&global);
1544 assert_eq!(c.max_connections, Some(10));
1545 assert_eq!(c.min_connections, Some(2));
1546 assert_eq!(c.idle_timeout_secs, Some(600));
1547 assert_eq!(c.max_lifetime_secs, Some(3600));
1548 assert!(c.ssl_mode.is_none());
1549 assert!(c.ssl_root_cert.is_none());
1550 assert!(c.ssl_cert.is_none());
1551 assert!(c.ssl_key.is_none());
1552 }
1553
1554 #[test]
1555 fn apply_defaults_does_not_override() {
1556 let mut c = SqlEndpointConfig::from_uri(
1557 "sql:select 1?db_url=postgres://localhost/test&maxConnections=99&minConnections=5",
1558 )
1559 .unwrap();
1560 let global = SqlGlobalConfig {
1561 max_connections: 10,
1562 min_connections: 2,
1563 idle_timeout_secs: 600,
1564 max_lifetime_secs: 3600,
1565 ssl_mode: None,
1566 ssl_root_cert: None,
1567 ssl_cert: None,
1568 ssl_key: None,
1569 retry: NetworkRetryPolicy::default(),
1570 };
1571 c.apply_defaults(&global);
1572 assert_eq!(c.max_connections, Some(99));
1574 assert_eq!(c.min_connections, Some(5));
1575 assert_eq!(c.idle_timeout_secs, Some(600));
1577 assert_eq!(c.max_lifetime_secs, Some(3600));
1578 }
1579
1580 #[test]
1581 fn resolve_defaults_fills_remaining() {
1582 let mut c = SqlEndpointConfig::from_uri(
1583 "sql:select 1?db_url=postgres://localhost/test&maxConnections=7",
1584 )
1585 .unwrap();
1586 c.resolve_defaults();
1587 assert_eq!(c.max_connections, Some(7)); assert_eq!(c.min_connections, Some(1)); assert_eq!(c.idle_timeout_secs, Some(300)); assert_eq!(c.max_lifetime_secs, Some(1800)); }
1592
1593 #[test]
1594 fn global_config_builder() {
1595 let c = SqlGlobalConfig::default()
1596 .with_max_connections(20)
1597 .with_min_connections(3)
1598 .with_idle_timeout_secs(600)
1599 .with_max_lifetime_secs(3600)
1600 .with_ssl_mode("require")
1601 .with_ssl_root_cert("/ca.pem")
1602 .with_ssl_cert("/cert.pem")
1603 .with_ssl_key("/key.pem");
1604 assert_eq!(c.max_connections, 20);
1605 assert_eq!(c.min_connections, 3);
1606 assert_eq!(c.idle_timeout_secs, 600);
1607 assert_eq!(c.max_lifetime_secs, 3600);
1608 assert_eq!(c.ssl_mode, Some("require".to_string()));
1609 assert_eq!(c.ssl_root_cert, Some("/ca.pem".to_string()));
1610 assert_eq!(c.ssl_cert, Some("/cert.pem".to_string()));
1611 assert_eq!(c.ssl_key, Some("/key.pem".to_string()));
1612 }
1613
1614 #[test]
1615 fn enrich_postgres_ssl_mode() {
1616 let mut c = SqlEndpointConfig::from_uri(
1617 "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1618 )
1619 .unwrap();
1620 c.resolve_defaults();
1621 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1622 assert!(url.contains("sslmode=require"), "got: {}", url);
1623 }
1624
1625 #[test]
1626 fn enrich_postgres_all_ssl() {
1627 let mut c = SqlEndpointConfig::from_uri(
1628 "sql:select 1?db_url=postgres://localhost/test&sslMode=require&sslRootCert=/ca.pem&sslCert=/cert.pem&sslKey=/key.pem",
1629 )
1630 .unwrap();
1631 c.resolve_defaults();
1632 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1633 assert!(url.contains("sslmode=require"), "got: {}", url);
1634 assert!(url.contains("sslrootcert="), "got: {}", url);
1635 assert!(url.contains("sslcert="), "got: {}", url);
1636 assert!(url.contains("sslkey="), "got: {}", url);
1637 }
1638
1639 #[test]
1640 fn enrich_mysql_ssl() {
1641 let mut c = SqlEndpointConfig::from_uri(
1642 "sql:select 1?db_url=mysql://localhost/test&sslMode=require",
1643 )
1644 .unwrap();
1645 c.resolve_defaults();
1646 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1647 assert!(url.contains("ssl-mode=require"), "got: {}", url);
1648 }
1649
1650 #[test]
1651 fn enrich_existing_query_params() {
1652 let mut c = SqlEndpointConfig::from_uri(
1653 "sql:select 1?db_url=postgres://localhost/test?existing=1&sslMode=require",
1654 )
1655 .unwrap();
1656 c.resolve_defaults();
1657 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1658 assert!(url.contains("existing=1"), "got: {}", url);
1659 assert!(url.contains("sslmode=require"), "got: {}", url);
1660 }
1661
1662 #[test]
1663 fn enrich_override_existing() {
1664 let mut c = SqlEndpointConfig::from_uri(
1665 "sql:select 1?db_url=postgres://localhost/test?sslmode=allow&sslMode=require",
1666 )
1667 .unwrap();
1668 c.resolve_defaults();
1669 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1670 assert!(url.contains("sslmode=require"), "got: {}", url);
1671 assert!(!url.contains("sslmode=allow"), "got: {}", url);
1672 }
1673
1674 #[test]
1675 fn enrich_applies_postgres_defaults() {
1676 let mut c =
1677 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1678 c.resolve_defaults();
1679 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1680 assert!(
1682 url.contains("sslmode=require"),
1683 "expected sslmode=require, got: {}",
1684 url
1685 );
1686 assert!(
1687 url.contains("connect_timeout=10"),
1688 "expected connect_timeout=10, got: {}",
1689 url
1690 );
1691 }
1692
1693 #[test]
1694 fn enrich_mysql_defaults_to_prefer() {
1695 let mut c =
1696 SqlEndpointConfig::from_uri("sql:select 1?db_url=mysql://localhost/test").unwrap();
1697 c.resolve_defaults();
1698 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1699 assert!(
1700 url.contains("ssl-mode=prefer"),
1701 "expected ssl-mode=prefer, got: {}",
1702 url
1703 );
1704 assert!(
1705 url.contains("connect_timeout=10"),
1706 "expected connect_timeout=10, got: {}",
1707 url
1708 );
1709 }
1710
1711 #[test]
1712 fn enrich_preserves_explicit_connect_timeout() {
1713 let mut c = SqlEndpointConfig::from_uri(
1714 "sql:select 1?db_url=postgres://localhost/test?connect_timeout=5",
1715 )
1716 .unwrap();
1717 c.resolve_defaults();
1718 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1719 assert!(
1720 url.contains("connect_timeout=5"),
1721 "expected explicit connect_timeout=5 preserved, got: {}",
1722 url
1723 );
1724 assert!(
1725 url.contains("sslmode=require"),
1726 "expected sslmode=require, got: {}",
1727 url
1728 );
1729 }
1730
1731 #[test]
1732 fn enrich_url_encodes_paths() {
1733 let mut c = SqlEndpointConfig::from_uri(
1734 "sql:select 1?db_url=postgres://localhost/test&sslRootCert=/path/to/my%20cert.pem",
1735 )
1736 .unwrap();
1737 c.resolve_defaults();
1738 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1739 assert!(url.contains("sslrootcert="), "got: {}", url);
1740 }
1741
1742 #[test]
1743 fn enrich_unsupported_scheme_returns_unchanged() {
1744 let mut c = SqlEndpointConfig::from_uri(
1745 "sql:select 1?db_url=sqlite://localhost/test.db&sslMode=require",
1746 )
1747 .unwrap();
1748 c.resolve_defaults();
1749 let url = enrich_db_url_with_ssl(&c.db_url, &c).unwrap();
1750 assert_eq!(url, "sqlite://localhost/test.db");
1751 }
1752
1753 #[test]
1754 fn enrich_invalid_url_returns_error() {
1755 let mut c = SqlEndpointConfig::from_uri(
1756 "sql:select 1?db_url=postgres://localhost/test&sslMode=require",
1757 )
1758 .unwrap();
1759 c.resolve_defaults();
1760 let result = enrich_db_url_with_ssl("://not-a-valid-url", &c);
1761 assert!(result.is_err());
1762 }
1763
1764 #[test]
1768 fn debug_redacts_db_url_with_password() {
1769 let c = SqlEndpointConfig::from_uri(
1770 "sql:select 1?db_url=postgres://user:secret123@localhost/test",
1771 )
1772 .unwrap();
1773 let debug_output = format!("{:?}", c);
1774 assert!(
1775 !debug_output.contains("secret123"),
1776 "Debug output must not contain password: {}",
1777 debug_output
1778 );
1779 assert!(
1780 debug_output.contains("***"),
1781 "Debug output must contain redacted marker: {}",
1782 debug_output
1783 );
1784 }
1785
1786 #[test]
1787 fn debug_redacts_ssl_key() {
1788 let c = SqlEndpointConfig::from_uri(
1789 "sql:select 1?db_url=postgres://localhost/test&sslKey=/secret/key.pem",
1790 )
1791 .unwrap();
1792 let debug_output = format!("{:?}", c);
1793 assert!(
1794 !debug_output.contains("/secret/key.pem"),
1795 "Debug output must not contain ssl_key path: {}",
1796 debug_output
1797 );
1798 }
1799
1800 #[test]
1801 fn debug_global_config_redacts_ssl_key() {
1802 let c = SqlGlobalConfig::default().with_ssl_key("/secret/key.pem");
1803 let debug_output = format!("{:?}", c);
1804 assert!(
1805 !debug_output.contains("/secret/key.pem"),
1806 "Debug output must not contain ssl_key path: {}",
1807 debug_output
1808 );
1809 assert!(
1810 debug_output.contains("***"),
1811 "Debug output must contain redacted marker: {}",
1812 debug_output
1813 );
1814 }
1815
1816 #[test]
1817 fn redact_db_url_with_credentials() {
1818 assert_eq!(
1819 redact_db_url("postgres://user:pass@host/db"),
1820 "postgres://***:***@host/db"
1821 );
1822 }
1823
1824 #[test]
1825 fn redact_db_url_without_credentials() {
1826 assert_eq!(redact_db_url("sqlite::memory:"), "sqlite::memory:");
1827 }
1828
1829 #[test]
1830 fn redact_db_url_invalid_returns_original() {
1831 assert_eq!(redact_db_url("not-a-url"), "not-a-url");
1832 }
1833
1834 #[test]
1835 fn redact_db_url_invalid_with_credentials_redacts() {
1836 let redacted = redact_db_url("postgres://user:secret@host:99999/db");
1840 assert!(
1841 !redacted.contains("secret"),
1842 "password must be redacted in unparseable URL: {redacted}"
1843 );
1844 assert!(
1845 !redacted.contains("user:secret"),
1846 "userinfo must be redacted: {redacted}"
1847 );
1848 assert!(
1849 !redacted.contains("user"),
1850 "username must be redacted: {redacted}"
1851 );
1852 }
1853
1854 #[test]
1855 fn redact_db_url_at_before_scheme_returns_original() {
1856 assert_eq!(redact_db_url("foo@bar://baz"), "foo@bar://baz");
1859 }
1860
1861 #[test]
1863 fn use_placeholder_defaults_to_true() {
1864 let c =
1865 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
1866 assert!(c.use_placeholder);
1867 }
1868
1869 #[test]
1870 fn use_placeholder_false_from_uri() {
1871 let c = SqlEndpointConfig::from_uri(
1872 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=false",
1873 )
1874 .unwrap();
1875 assert!(!c.use_placeholder);
1876 }
1877
1878 #[test]
1879 fn use_placeholder_true_from_uri() {
1880 let c = SqlEndpointConfig::from_uri(
1881 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=true",
1882 )
1883 .unwrap();
1884 assert!(c.use_placeholder);
1885 }
1886
1887 #[test]
1889 fn use_placeholder_rejects_invalid_value() {
1890 let result = SqlEndpointConfig::from_uri(
1891 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=1",
1892 );
1893 assert!(result.is_err());
1894 let msg = format!("{:?}", result.unwrap_err());
1895 assert!(msg.contains("usePlaceholder") && msg.contains("true") && msg.contains("false"));
1896 }
1897
1898 #[test]
1899 fn use_placeholder_rejects_typo_tru() {
1900 let result = SqlEndpointConfig::from_uri(
1901 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=tru",
1902 );
1903 assert!(result.is_err());
1904 }
1905
1906 #[test]
1907 fn use_placeholder_rejects_yes() {
1908 let result = SqlEndpointConfig::from_uri(
1909 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=yes",
1910 );
1911 assert!(result.is_err());
1912 }
1913
1914 #[test]
1915 fn noop_rejects_invalid_value() {
1916 let result =
1917 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test&noop=1");
1918 assert!(result.is_err());
1919 let msg = format!("{:?}", result.unwrap_err());
1920 assert!(msg.contains("noop"));
1921 }
1922
1923 #[test]
1924 fn batch_rejects_invalid_value() {
1925 let result =
1926 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test&batch=yes");
1927 assert!(result.is_err());
1928 let msg = format!("{:?}", result.unwrap_err());
1929 assert!(msg.contains("batch"));
1930 }
1931
1932 #[test]
1933 fn route_empty_result_set_rejects_invalid_value() {
1934 let result = SqlEndpointConfig::from_uri(
1935 "sql:select 1?db_url=postgres://localhost/test&routeEmptyResultSet=on",
1936 );
1937 assert!(result.is_err());
1938 }
1939
1940 #[test]
1941 fn use_iterator_rejects_invalid_value() {
1942 let result = SqlEndpointConfig::from_uri(
1943 "sql:select 1?db_url=postgres://localhost/test&useIterator=1",
1944 );
1945 assert!(result.is_err());
1946 }
1947
1948 #[test]
1949 fn break_batch_on_consume_fail_rejects_invalid_value() {
1950 let result = SqlEndpointConfig::from_uri(
1951 "sql:select 1?db_url=postgres://localhost/test&breakBatchOnConsumeFail=yes",
1952 );
1953 assert!(result.is_err());
1954 }
1955
1956 #[test]
1957 fn use_message_body_for_sql_rejects_invalid_value() {
1958 let result = SqlEndpointConfig::from_uri(
1959 "sql:select 1?db_url=postgres://localhost/test&useMessageBodyForSql=1",
1960 );
1961 assert!(result.is_err());
1962 }
1963
1964 #[test]
1966 fn boolean_params_case_insensitive() {
1967 let c = SqlEndpointConfig::from_uri(
1968 "sql:select 1?db_url=postgres://localhost/test&usePlaceholder=TRUE&noop=FALSE&batch=True&useIterator=False&bridgeErrorHandler=TRUE",
1969 )
1970 .unwrap();
1971 assert!(c.use_placeholder);
1972 assert!(!c.noop);
1973 assert!(c.batch);
1974 assert!(!c.use_iterator);
1975 assert!(c.bridge_error_handler);
1976 }
1977
1978 #[test]
1980 fn multi_char_placeholder_rejected() {
1981 let result = SqlEndpointConfig::from_uri(
1982 "sql:select 1?db_url=postgres://localhost/test&placeholder=##",
1983 );
1984 assert!(result.is_err());
1985 let msg = format!("{:?}", result.unwrap_err());
1986 assert!(msg.contains("placeholder") && msg.contains("one character"));
1987 }
1988
1989 #[test]
1990 fn non_ascii_placeholder_rejected() {
1991 let result = SqlEndpointConfig::from_uri(
1992 "sql:select 1?db_url=postgres://localhost/test&placeholder=%C2%A2",
1993 );
1994 assert!(result.is_err());
1995 }
1996
1997 #[test]
1998 fn single_char_placeholder_accepted() {
1999 let c = SqlEndpointConfig::from_uri(
2000 "sql:select 1?db_url=postgres://localhost/test&placeholder=$",
2001 )
2002 .unwrap();
2003 assert_eq!(c.placeholder, '$');
2004 }
2005
2006 #[test]
2007 fn empty_placeholder_falls_back_to_default() {
2008 let c = SqlEndpointConfig::from_uri(
2010 "sql:select 1?db_url=postgres://localhost/test&placeholder=",
2011 )
2012 .unwrap();
2013 assert_eq!(c.placeholder, '#');
2014 }
2015
2016 #[tokio::test]
2018 async fn file_query_cached_in_config() {
2019 use std::io::Write;
2020 let unique_name = format!(
2021 "test_sql_cached_{}.sql",
2022 std::time::SystemTime::now()
2023 .duration_since(std::time::UNIX_EPOCH)
2024 .unwrap_or_default()
2025 .as_nanos()
2026 );
2027 let mut tmp = std::env::temp_dir();
2028 tmp.push(unique_name);
2029 {
2030 let mut f = std::fs::File::create(&tmp).unwrap();
2031 writeln!(f, "SELECT * FROM cached_test").unwrap();
2032 }
2033 let uri = format!(
2034 "sql:file:{}?db_url=postgres://localhost/test",
2035 tmp.display()
2036 );
2037 let mut c = SqlEndpointConfig::from_uri(&uri).unwrap();
2038 assert!(c.query.is_empty());
2040 assert_eq!(c.source_path, Some(tmp.to_string_lossy().into_owned()));
2041
2042 c.resolve_file_query()
2044 .await
2045 .expect("resolve should succeed");
2046 assert_eq!(c.query, "SELECT * FROM cached_test");
2047
2048 std::fs::remove_file(&tmp).ok();
2050 assert_eq!(c.query, "SELECT * FROM cached_test");
2051 }
2052
2053 #[test]
2057 fn always_populate_statement_defaults_to_false() {
2058 let c =
2059 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2060 assert!(!c.always_populate_statement);
2061 }
2062
2063 #[test]
2064 fn always_populate_statement_from_uri() {
2065 let c = SqlEndpointConfig::from_uri(
2066 "sql:select 1?db_url=postgres://localhost/test&alwaysPopulateStatement=true",
2067 )
2068 .unwrap();
2069 assert!(c.always_populate_statement);
2070 }
2071
2072 #[test]
2074 fn allow_named_parameters_defaults_to_true() {
2075 let c =
2076 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2077 assert!(c.allow_named_parameters);
2078 }
2079
2080 #[test]
2081 fn allow_named_parameters_false_from_uri() {
2082 let c = SqlEndpointConfig::from_uri(
2083 "sql:select 1?db_url=postgres://localhost/test&allowNamedParameters=false",
2084 )
2085 .unwrap();
2086 assert!(!c.allow_named_parameters);
2087 }
2088
2089 #[test]
2091 fn fetch_size_defaults_to_none() {
2092 let c =
2093 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2094 assert!(c.fetch_size.is_none());
2095 }
2096
2097 #[test]
2098 fn fetch_size_from_uri() {
2099 let c = SqlEndpointConfig::from_uri(
2100 "sql:select 1?db_url=postgres://localhost/test&fetchSize=1000",
2101 )
2102 .unwrap();
2103 assert_eq!(c.fetch_size, Some(1000));
2104 }
2105
2106 #[test]
2108 fn transaction_mode_defaults_to_auto() {
2109 let c =
2110 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2111 assert_eq!(c.transaction_mode, TransactionMode::Auto);
2112 }
2113
2114 #[test]
2115 fn transaction_mode_managed_from_uri() {
2116 let c = SqlEndpointConfig::from_uri(
2117 "sql:select 1?db_url=postgres://localhost/test&transactionMode=Managed",
2118 )
2119 .unwrap();
2120 assert_eq!(c.transaction_mode, TransactionMode::Managed);
2121 }
2122
2123 #[test]
2124 fn transaction_mode_invalid_rejected() {
2125 let result = SqlEndpointConfig::from_uri(
2126 "sql:select 1?db_url=postgres://localhost/test&transactionMode=Invalid",
2127 );
2128 assert!(result.is_err());
2129 }
2130
2131 #[test]
2133 fn repeat_count_defaults_to_none() {
2134 let c =
2135 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2136 assert!(c.repeat_count.is_none());
2137 }
2138
2139 #[test]
2140 fn repeat_count_from_uri() {
2141 let c = SqlEndpointConfig::from_uri(
2142 "sql:select 1?db_url=postgres://localhost/test&repeatCount=10",
2143 )
2144 .unwrap();
2145 assert_eq!(c.repeat_count, Some(10));
2146 }
2147
2148 #[test]
2150 fn break_on_empty_defaults_to_false() {
2151 let c =
2152 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2153 assert!(!c.break_on_empty);
2154 }
2155
2156 #[test]
2157 fn break_on_empty_true_from_uri() {
2158 let c = SqlEndpointConfig::from_uri(
2159 "sql:select 1?db_url=postgres://localhost/test&breakOnEmpty=true",
2160 )
2161 .unwrap();
2162 assert!(c.break_on_empty);
2163 }
2164
2165 #[test]
2166 fn break_on_empty_explicit_false_from_uri() {
2167 let c = SqlEndpointConfig::from_uri(
2168 "sql:select 1?db_url=postgres://localhost/test&breakOnEmpty=false",
2169 )
2170 .unwrap();
2171 assert!(!c.break_on_empty);
2172 }
2173
2174 #[test]
2176 fn processing_strategy_defaults_to_direct() {
2177 let c =
2178 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2179 assert_eq!(c.processing_strategy, ProcessingStrategy::Direct);
2180 }
2181
2182 #[test]
2183 fn processing_strategy_scheduled_from_uri() {
2184 let c = SqlEndpointConfig::from_uri(
2185 "sql:select 1?db_url=postgres://localhost/test&processingStrategy=Scheduled",
2186 )
2187 .unwrap();
2188 assert_eq!(c.processing_strategy, ProcessingStrategy::Scheduled);
2189 }
2190
2191 #[test]
2192 fn processing_strategy_invalid_rejected() {
2193 let result = SqlEndpointConfig::from_uri(
2194 "sql:select 1?db_url=postgres://localhost/test&processingStrategy=Invalid",
2195 );
2196 assert!(result.is_err());
2197 }
2198
2199 #[test]
2201 fn poll_strategy_defaults_to_sequential() {
2202 let c =
2203 SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test").unwrap();
2204 assert_eq!(c.poll_strategy, PollStrategy::Sequential);
2205 }
2206
2207 #[test]
2208 fn poll_strategy_burst_from_uri() {
2209 let c = SqlEndpointConfig::from_uri(
2210 "sql:select 1?db_url=postgres://localhost/test&pollStrategy=Burst",
2211 )
2212 .unwrap();
2213 assert_eq!(c.poll_strategy, PollStrategy::Burst);
2214 }
2215
2216 #[test]
2217 fn poll_strategy_invalid_rejected() {
2218 let result = SqlEndpointConfig::from_uri(
2219 "sql:select 1?db_url=postgres://localhost/test&pollStrategy=Invalid",
2220 );
2221 assert!(result.is_err());
2222 }
2223
2224 #[test]
2227 fn sql_endpoint_config_has_retry_policy() {
2228 let cfg = SqlEndpointConfig::from_uri(
2229 "sql:select 1?db_url=sqlite::memory:&retryMaxAttempts=3&retryInitialDelayMs=500",
2230 )
2231 .expect("parse");
2232 assert_eq!(cfg.retry.max_attempts, 3);
2233 assert_eq!(
2234 cfg.retry.initial_delay,
2235 std::time::Duration::from_millis(500)
2236 );
2237 assert!(cfg.retry.enabled);
2238 }
2239
2240 #[test]
2241 fn sql_endpoint_config_retry_defaults_when_unspecified() {
2242 let cfg =
2243 SqlEndpointConfig::from_uri("sql:select 1?db_url=sqlite::memory:").expect("parse");
2244 assert!(cfg.retry.enabled);
2246 assert_eq!(cfg.retry.max_attempts, 10); }
2248
2249 #[test]
2250 fn sql_global_config_has_retry_default() {
2251 let cfg = SqlGlobalConfig::default();
2252 assert!(cfg.retry.enabled);
2253 }
2254
2255 #[test]
2256 fn retry_policy_parse_full_uri_params() {
2257 let cfg = SqlEndpointConfig::from_uri(
2258 "sql:select 1?db_url=sqlite::memory:&retryEnabled=false&retryMaxAttempts=7&retryInitialDelayMs=1000&retryMultiplier=3.0&retryMaxDelayMs=60000&retryJitter=0.5",
2259 )
2260 .expect("parse");
2261 assert!(!cfg.retry.enabled);
2262 assert_eq!(cfg.retry.max_attempts, 7);
2263 assert_eq!(
2264 cfg.retry.initial_delay,
2265 std::time::Duration::from_millis(1000)
2266 );
2267 assert!((cfg.retry.multiplier - 3.0).abs() < f64::EPSILON);
2268 assert_eq!(cfg.retry.max_delay, std::time::Duration::from_millis(60000));
2269 assert!((cfg.retry.jitter_factor - 0.5).abs() < f64::EPSILON);
2270 }
2271
2272 #[test]
2273 fn retry_policy_from_uri_survives_apply_defaults_with_global() {
2274 let mut ep = SqlEndpointConfig::from_uri(
2275 "sql:select 1?db_url=sqlite::memory:&retryMaxAttempts=10&retryInitialDelayMs=500",
2276 )
2277 .expect("parse");
2278 let global = SqlGlobalConfig::default(); ep.apply_defaults(&global);
2280 assert_eq!(ep.retry.max_attempts, 10);
2282 assert_eq!(
2283 ep.retry.initial_delay,
2284 std::time::Duration::from_millis(500)
2285 );
2286 }
2287
2288 #[test]
2289 fn retry_policy_falls_back_to_global_when_uri_has_no_retry_params() {
2290 let mut ep =
2291 SqlEndpointConfig::from_uri("sql:select 1?db_url=sqlite::memory:").expect("parse");
2292 let mut global = SqlGlobalConfig::default();
2293 global.retry.max_attempts = 7;
2294 ep.apply_defaults(&global);
2295 assert_eq!(ep.retry.max_attempts, 7);
2297 }
2298
2299 #[test]
2300 fn from_uri_with_datasource_name() {
2301 let cfg = SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders").unwrap();
2302 assert_eq!(cfg.datasource_name.as_deref(), Some("orders"));
2303 assert!(cfg.db_url.is_empty());
2304 }
2305
2306 #[test]
2307 fn from_uri_with_datasource_and_behavior_override() {
2308 let cfg =
2309 SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&outputType=SelectOne")
2310 .unwrap();
2311 assert_eq!(cfg.datasource_name.as_deref(), Some("orders"));
2312 }
2313
2314 #[test]
2315 fn from_uri_datasource_rejects_pool_override() {
2316 let result =
2317 SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&maxConnections=50");
2318 assert!(result.is_err());
2319 let msg = result.unwrap_err().to_string();
2320 assert!(msg.contains("pool-affecting"));
2321 }
2322
2323 #[test]
2324 fn from_uri_neither_datasource_nor_db_url_is_error() {
2325 let result = SqlEndpointConfig::from_uri("sql:SELECT 1");
2326 assert!(result.is_err());
2327 }
2328
2329 #[test]
2330 fn from_uri_db_url_inline_still_works() {
2331 let cfg =
2332 SqlEndpointConfig::from_uri("sql:SELECT 1?db_url=postgres://localhost/test").unwrap();
2333 assert!(cfg.datasource_name.is_none());
2334 assert_eq!(cfg.db_url, "postgres://localhost/test");
2335 }
2336
2337 #[test]
2338 fn from_uri_datasource_rejects_ssl_mode() {
2339 let result = SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&sslMode=require");
2340 assert!(result.is_err());
2341 let msg = result.unwrap_err().to_string();
2342 assert!(msg.contains("pool-affecting"));
2343 }
2344
2345 #[test]
2346 fn from_uri_datasource_rejects_ssl_root_cert() {
2347 let result =
2348 SqlEndpointConfig::from_uri("sql:SELECT 1?datasource=orders&sslRootCert=/ca.pem");
2349 assert!(result.is_err());
2350 }
2351
2352 #[test]
2353 fn from_uri_datasource_rejects_db_url() {
2354 let result = SqlEndpointConfig::from_uri(
2355 "sql:SELECT 1?datasource=orders&db_url=postgres://evil:5432/pwned",
2356 );
2357 assert!(result.is_err());
2358 let msg = result.unwrap_err().to_string();
2359 assert!(msg.contains("db_url") && msg.contains("datasource"));
2360 }
2361}