pgwire_replication/config.rs
1//! Configuration types for PostgreSQL replication connections.
2//!
3//! This module provides configuration structures for establishing replication
4//! connections to PostgreSQL, including TLS settings and replication parameters.
5
6use std::path::PathBuf;
7use std::time::Duration;
8
9use crate::lsn::Lsn;
10
11/// SSL/TLS connection mode.
12///
13/// These modes match PostgreSQL's `sslmode` connection parameter.
14/// See [PostgreSQL SSL Support](https://www.postgresql.org/docs/current/libpq-ssl.html)
15/// for detailed documentation.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
17pub enum SslMode {
18 /// Never use TLS. Connection will fail if server requires TLS.
19 #[default]
20 Disable,
21
22 /// Try TLS first, fall back to unencrypted if server doesn't support it.
23 ///
24 /// **Warning**: Vulnerable to downgrade attacks. Not recommended for production.
25 Prefer,
26
27 /// Require TLS but don't verify the server certificate.
28 ///
29 /// Protects against passive eavesdropping but not active MITM attacks.
30 Require,
31
32 /// Require TLS and verify the server certificate chain against trusted CAs.
33 ///
34 /// Does NOT verify that the certificate hostname matches the connection target.
35 VerifyCa,
36
37 /// Require TLS, verify certificate chain, AND verify hostname matches.
38 ///
39 /// **Recommended for production**. Provides full protection against MITM attacks.
40 VerifyFull,
41}
42
43impl SslMode {
44 /// Returns `true` if this mode requires TLS (won't fall back to plain).
45 #[inline]
46 pub fn requires_tls(&self) -> bool {
47 !matches!(self, SslMode::Disable | SslMode::Prefer)
48 }
49
50 /// Returns `true` if this mode verifies the certificate chain.
51 #[inline]
52 pub fn verifies_certificate(&self) -> bool {
53 matches!(self, SslMode::VerifyCa | SslMode::VerifyFull)
54 }
55
56 /// Returns `true` if this mode verifies the server hostname.
57 #[inline]
58 pub fn verifies_hostname(&self) -> bool {
59 matches!(self, SslMode::VerifyFull)
60 }
61}
62
63/// TLS/SSL configuration for PostgreSQL connections.
64#[derive(Debug, Clone, Default, PartialEq, Eq)]
65pub struct TlsConfig {
66 /// SSL mode controlling connection security level.
67 pub mode: SslMode,
68
69 /// Path to PEM file containing trusted CA certificates.
70 ///
71 /// If `None` and verification is enabled (`VerifyCa`/`VerifyFull`),
72 /// the Mozilla root certificates (webpki-roots) are used.
73 pub ca_pem_path: Option<PathBuf>,
74
75 /// Override SNI hostname sent during TLS handshake.
76 ///
77 /// Useful when:
78 /// - Connecting via IP address but certificate has a DNS name
79 /// - Using a load balancer with different internal/external names
80 ///
81 /// If `None`, the connection `host` is used for SNI.
82 pub sni_hostname: Option<String>,
83
84 /// Path to PEM file containing client certificate chain.
85 ///
86 /// Required for mutual TLS (mTLS) authentication.
87 /// Must be paired with `client_key_pem_path`.
88 pub client_cert_pem_path: Option<PathBuf>,
89
90 /// Path to PEM file containing client private key.
91 ///
92 /// Required for mutual TLS (mTLS) authentication.
93 /// Must be paired with `client_cert_pem_path`.
94 /// Supports PKCS#8, PKCS#1 (RSA), and SEC1 (EC) formats.
95 pub client_key_pem_path: Option<PathBuf>,
96}
97
98impl TlsConfig {
99 /// Create a configuration with TLS disabled.
100 ///
101 /// # Example
102 /// ```
103 /// use pgwire_replication::config::TlsConfig;
104 ///
105 /// let tls = TlsConfig::disabled();
106 /// assert!(!tls.mode.requires_tls());
107 /// ```
108 pub fn disabled() -> Self {
109 Self::default()
110 }
111
112 /// Create a configuration requiring TLS without certificate verification.
113 ///
114 /// **Warning**: This mode is vulnerable to MITM attacks.
115 /// Use `verify_ca()` or `verify_full()` for production.
116 ///
117 /// # Example
118 /// ```
119 /// use pgwire_replication::config::TlsConfig;
120 ///
121 /// let tls = TlsConfig::require();
122 /// assert!(tls.mode.requires_tls());
123 /// assert!(!tls.mode.verifies_certificate());
124 /// ```
125 pub fn require() -> Self {
126 Self {
127 mode: SslMode::Require,
128 ..Default::default()
129 }
130 }
131
132 /// Create a configuration with certificate chain verification.
133 ///
134 /// # Arguments
135 /// * `ca_path` - Path to CA certificate PEM file, or `None` for system roots
136 ///
137 /// # Example
138 /// ```
139 /// use pgwire_replication::config::TlsConfig;
140 ///
141 /// // Using system/Mozilla roots
142 /// let tls = TlsConfig::verify_ca(None);
143 ///
144 /// // Using custom CA
145 /// let tls = TlsConfig::verify_ca(Some("/path/to/ca.pem".into()));
146 /// ```
147 pub fn verify_ca(ca_path: Option<PathBuf>) -> Self {
148 Self {
149 mode: SslMode::VerifyCa,
150 ca_pem_path: ca_path,
151 ..Default::default()
152 }
153 }
154
155 /// Create a configuration with full verification (chain + hostname).
156 ///
157 /// **Recommended for production**.
158 ///
159 /// # Arguments
160 /// * `ca_path` - Path to CA certificate PEM file, or `None` for system roots
161 ///
162 /// # Example
163 /// ```
164 /// use pgwire_replication::config::TlsConfig;
165 ///
166 /// let tls = TlsConfig::verify_full(Some("/etc/ssl/certs/ca.pem".into()));
167 /// assert!(tls.mode.verifies_hostname());
168 /// ```
169 pub fn verify_full(ca_path: Option<PathBuf>) -> Self {
170 Self {
171 mode: SslMode::VerifyFull,
172 ca_pem_path: ca_path,
173 ..Default::default()
174 }
175 }
176
177 /// Set SNI hostname override.
178 ///
179 /// # Example
180 /// ```
181 /// use pgwire_replication::config::TlsConfig;
182 ///
183 /// let tls = TlsConfig::verify_full(None)
184 /// .with_sni_hostname("db.example.com");
185 /// ```
186 pub fn with_sni_hostname(mut self, hostname: impl Into<String>) -> Self {
187 self.sni_hostname = Some(hostname.into());
188 self
189 }
190
191 /// Configure client certificate for mutual TLS.
192 ///
193 /// # Example
194 /// ```
195 /// use pgwire_replication::config::TlsConfig;
196 ///
197 /// let tls = TlsConfig::verify_full(Some("/ca.pem".into()))
198 /// .with_client_cert("/client.pem", "/client.key");
199 /// ```
200 pub fn with_client_cert(
201 mut self,
202 cert_path: impl Into<PathBuf>,
203 key_path: impl Into<PathBuf>,
204 ) -> Self {
205 self.client_cert_pem_path = Some(cert_path.into());
206 self.client_key_pem_path = Some(key_path.into());
207 self
208 }
209
210 /// Returns `true` if mutual TLS (client certificate) is configured.
211 #[inline]
212 pub fn is_mtls(&self) -> bool {
213 self.client_cert_pem_path.is_some() && self.client_key_pem_path.is_some()
214 }
215}
216
217/// One or more publication names a replication slot subscribes to.
218///
219/// A single slot can stream changes from several publications at once.
220/// Construct from a single name or from a collection:
221///
222/// ```
223/// use pgwire_replication::config::Publication;
224///
225/// let one: Publication = "orders".into();
226/// let many: Publication = ["orders", "customers"].into();
227/// let dynamic: Publication = vec![String::from("a"), String::from("b")].into();
228/// ```
229///
230/// `Publication` is an immutable value fixed for the connection's lifetime: the
231/// `publication_names` set is bound once at `START_REPLICATION` and cannot be
232/// changed on a live stream. To replicate a different set, reconnect with a new
233/// [`ReplicationConfig`].
234///
235/// # Limitations
236///
237/// Names are joined into PostgreSQL's comma-separated `publication_names` list,
238/// so a name containing a comma or whitespace cannot be represented (a
239/// pre-existing PostgreSQL constraint). Single quotes in names are escaped. An
240/// empty `Publication` is invalid and the server will reject the stream.
241#[derive(Debug, Clone, PartialEq, Eq)]
242pub struct Publication(Vec<String>);
243
244impl Publication {
245 /// The publication names, in order.
246 #[inline]
247 pub fn names(&self) -> &[String] {
248 &self.0
249 }
250
251 /// Render the value for the `publication_names` `START_REPLICATION` option:
252 /// single quotes escaped, names comma-separated.
253 pub(crate) fn to_option_value(&self) -> String {
254 self.0
255 .iter()
256 .map(|name| name.replace('\'', "''"))
257 .collect::<Vec<_>>()
258 .join(",")
259 }
260}
261
262impl From<&str> for Publication {
263 fn from(value: &str) -> Self {
264 Self(vec![value.to_string()])
265 }
266}
267
268impl From<String> for Publication {
269 fn from(value: String) -> Self {
270 Self(vec![value])
271 }
272}
273
274impl<A: Into<String>, const N: usize> From<[A; N]> for Publication {
275 fn from(value: [A; N]) -> Self {
276 Self(value.into_iter().map(Into::into).collect())
277 }
278}
279
280impl<A: Into<String>> From<Vec<A>> for Publication {
281 fn from(value: Vec<A>) -> Self {
282 Self(value.into_iter().map(Into::into).collect())
283 }
284}
285
286impl<A: Into<String>> FromIterator<A> for Publication {
287 fn from_iter<T: IntoIterator<Item = A>>(iter: T) -> Self {
288 Self(iter.into_iter().map(Into::into).collect())
289 }
290}
291
292impl std::fmt::Display for Publication {
293 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
294 write!(f, "{}", self.0.join(","))
295 }
296}
297
298/// Configuration for PostgreSQL logical replication connections.
299///
300/// # Example
301///
302/// ```
303/// use pgwire_replication::config::{ReplicationConfig, TlsConfig};
304///
305/// let config = ReplicationConfig::new(
306/// "db.example.com",
307/// "replicator",
308/// "secret",
309/// "mydb",
310/// "my_slot",
311/// "my_publication",
312/// )
313/// .with_tls(TlsConfig::verify_full(Some("/path/to/ca.pem".into())));
314/// ```
315///
316/// This struct is `#[non_exhaustive]`: construct it with [`ReplicationConfig::new`]
317/// (or [`unix`](ReplicationConfig::unix)) plus the `with_*` builder methods
318/// rather than a struct literal, so new fields can be added without breaking you.
319#[derive(Debug, Clone, PartialEq, Eq)]
320#[non_exhaustive]
321pub struct ReplicationConfig {
322 /// PostgreSQL server hostname or IP address.
323 pub host: String,
324
325 /// PostgreSQL server port (default: 5432).
326 pub port: u16,
327
328 /// PostgreSQL username with replication privileges.
329 ///
330 /// The user must have the `REPLICATION` attribute or be a superuser.
331 pub user: String,
332
333 /// Password for authentication.
334 pub password: String,
335
336 /// Database name to connect to.
337 pub database: String,
338
339 /// TLS/SSL configuration.
340 pub tls: TlsConfig,
341
342 /// Name of the replication slot to use.
343 ///
344 /// The slot must already exist and be a logical replication slot
345 /// using the `pgoutput` plugin.
346 pub slot: String,
347
348 /// Publication(s) to subscribe to.
349 ///
350 /// Each publication must exist and include the tables you want to replicate.
351 /// Accepts a single name or a collection — see [`Publication`].
352 pub publication: Publication,
353
354 /// LSN position to start replication from.
355 ///
356 /// - `Lsn(0)`: Start from slot's `confirmed_flush_lsn`
357 /// - Specific LSN: Resume from that position (must be >= slot's restart_lsn)
358 pub start_lsn: Lsn,
359
360 /// Optional LSN to stop replication at.
361 ///
362 /// When set, replication will stop once a commit with `end_lsn >= stop_at_lsn`
363 /// is received. Useful for:
364 /// - Bounded replay (e.g., point-in-time recovery)
365 /// - Testing with known data ranges
366 ///
367 /// If `None`, replication continues indefinitely (normal CDC mode).
368 pub stop_at_lsn: Option<Lsn>,
369
370 /// Interval for sending standby status updates to the server.
371 ///
372 /// Status updates inform PostgreSQL of the client's replay position,
373 /// allowing the server to release WAL segments. Too infrequent updates
374 /// may cause WAL accumulation; too frequent updates add overhead.
375 ///
376 /// Default: 1 second (matches pg_recvlogical)
377 pub status_interval: Duration,
378
379 /// Maximum time to wait for server messages before waking up.
380 ///
381 /// Silence is normal during logical replication. When this interval elapses
382 /// with no incoming messages, the client will send a standby status update
383 /// (feedback) and continue waiting.
384 ///
385 /// This effectively bounds how long the worker can stay blocked in a read
386 /// while idle.
387 ///
388 /// Default: 10 seconds
389 pub idle_wakeup_interval: Duration,
390
391 /// Size of the bounded event buffer between replication worker and consumer.
392 ///
393 /// Larger buffers can smooth out processing latency spikes but use more memory.
394 /// Each event is typically 100-1000 bytes depending on row size.
395 ///
396 /// Default: 8192 events
397 pub buffer_events: usize,
398
399 /// Request column values in PostgreSQL binary wire format.
400 ///
401 /// When `true`, the `binary 'true'` option is added to `START_REPLICATION`
402 /// and `pgoutput` encodes column values using each type's binary send
403 /// function instead of text output.
404 ///
405 /// This does not change anything the library parses — the worker decodes
406 /// only Begin/Commit/Message boundaries and forwards `XLogData` tuple bytes
407 /// raw — so it is the **consumer's** decoder that must handle binary values.
408 ///
409 /// Leave as `false` (the default) unless you control both ends and have
410 /// measured a real win: any column whose type lacks a binary send function
411 /// makes the walsender error and close the replication stream.
412 ///
413 /// Requires PostgreSQL 14 or newer.
414 ///
415 /// Default: `false`
416 pub binary: bool,
417
418 /// Value for the startup `options` parameter, passed through to the
419 /// server the same way `libpq`'s `options` connection parameter (or the
420 /// `PGOPTIONS` environment variable) would on an ordinary connection.
421 ///
422 /// PostgreSQL honors `options` on replication connections too, so this
423 /// is typically used to pin session GUCs (e.g.
424 /// `"-c datestyle=ISO,MDY -c intervalstyle=iso_8601"`) so that values
425 /// decoded from the replication stream are rendered in text the same
426 /// way as values read through a regular connection whose GUCs are also
427 /// pinned.
428 ///
429 /// See [`PGOPTIONS`](https://www.postgresql.org/docs/current/libpq-envars.html)
430 /// for the `-c name=value` syntax.
431 ///
432 /// Default: `None` (server defaults apply).
433 pub options: Option<String>,
434}
435
436impl Default for ReplicationConfig {
437 fn default() -> Self {
438 Self {
439 host: "127.0.0.1".into(),
440 port: 5432,
441 user: "postgres".into(),
442 password: "postgres".into(),
443 database: "postgres".into(),
444 tls: TlsConfig::default(),
445 slot: "slot".into(),
446 publication: "pub".into(),
447 start_lsn: Lsn(0),
448 stop_at_lsn: None,
449 status_interval: Duration::from_secs(10),
450 idle_wakeup_interval: Duration::from_secs(10),
451 buffer_events: 8192,
452 binary: false,
453 options: None,
454 }
455 }
456}
457
458impl ReplicationConfig {
459 /// Create a new configuration with required fields.
460 ///
461 /// Other fields use defaults and can be customized with builder methods.
462 ///
463 /// # Example
464 /// ```
465 /// use pgwire_replication::config::ReplicationConfig;
466 ///
467 /// let config = ReplicationConfig::new(
468 /// "db.example.com",
469 /// "replicator",
470 /// "secret",
471 /// "mydb",
472 /// "my_slot",
473 /// "my_pub",
474 /// );
475 /// ```
476 pub fn new(
477 host: impl Into<String>,
478 user: impl Into<String>,
479 password: impl Into<String>,
480 database: impl Into<String>,
481 slot: impl Into<String>,
482 publication: impl Into<Publication>,
483 ) -> Self {
484 Self {
485 host: host.into(),
486 user: user.into(),
487 password: password.into(),
488 database: database.into(),
489 slot: slot.into(),
490 publication: publication.into(),
491 ..Default::default()
492 }
493 }
494
495 /// Returns `true` if `host` refers to a Unix domain socket directory.
496 ///
497 /// Following libpq convention, a host starting with `/` is treated as
498 /// the directory containing the PostgreSQL Unix socket file.
499 #[inline]
500 pub fn is_unix_socket(&self) -> bool {
501 self.host.starts_with('/')
502 }
503
504 /// Returns the full Unix socket path: `{host}/.s.PGSQL.{port}`.
505 ///
506 /// # Panics
507 ///
508 /// Panics if `host` does not start with `/` (i.e. `is_unix_socket()` is false).
509 pub fn unix_socket_path(&self) -> std::path::PathBuf {
510 assert!(
511 self.is_unix_socket(),
512 "unix_socket_path() called but host is not a socket directory: {:?}",
513 self.host
514 );
515 std::path::Path::new(&self.host).join(format!(".s.PGSQL.{}", self.port))
516 }
517
518 /// Create a configuration for connecting via Unix domain socket.
519 ///
520 /// `socket_dir` is the directory containing the PostgreSQL socket file
521 /// (e.g. `/var/run/postgresql`). The actual socket path will be
522 /// `{socket_dir}/.s.PGSQL.{port}`.
523 ///
524 /// TLS is automatically disabled for Unix socket connections.
525 ///
526 /// # Example
527 /// ```
528 /// use pgwire_replication::config::ReplicationConfig;
529 ///
530 /// let config = ReplicationConfig::unix(
531 /// "/var/run/postgresql",
532 /// 5432,
533 /// "replicator",
534 /// "secret",
535 /// "mydb",
536 /// "my_slot",
537 /// "my_pub",
538 /// );
539 /// assert!(config.is_unix_socket());
540 /// ```
541 pub fn unix(
542 socket_dir: impl Into<String>,
543 port: u16,
544 user: impl Into<String>,
545 password: impl Into<String>,
546 database: impl Into<String>,
547 slot: impl Into<String>,
548 publication: impl Into<Publication>,
549 ) -> Self {
550 Self {
551 host: socket_dir.into(),
552 port,
553 user: user.into(),
554 password: password.into(),
555 database: database.into(),
556 tls: TlsConfig::disabled(),
557 slot: slot.into(),
558 publication: publication.into(),
559 ..Default::default()
560 }
561 }
562
563 /// Set the server port.
564 pub fn with_port(mut self, port: u16) -> Self {
565 self.port = port;
566 self
567 }
568
569 /// Set TLS configuration.
570 pub fn with_tls(mut self, tls: TlsConfig) -> Self {
571 self.tls = tls;
572 self
573 }
574
575 /// Set the starting LSN.
576 pub fn with_start_lsn(mut self, lsn: Lsn) -> Self {
577 self.start_lsn = lsn;
578 self
579 }
580
581 /// Set an optional stop LSN for bounded replay.
582 pub fn with_stop_lsn(mut self, lsn: Lsn) -> Self {
583 self.stop_at_lsn = Some(lsn);
584 self
585 }
586
587 /// Set the status update interval.
588 pub fn with_status_interval(mut self, interval: Duration) -> Self {
589 self.status_interval = interval;
590 self
591 }
592
593 /// Set the idle wakeup interval.
594 pub fn with_wakeup_interval(mut self, timeout: Duration) -> Self {
595 self.idle_wakeup_interval = timeout;
596 self
597 }
598
599 /// Set the event buffer size.
600 pub fn with_buffer_size(mut self, size: usize) -> Self {
601 self.buffer_events = size;
602 self
603 }
604
605 /// Request binary-format column values from `pgoutput` (requires PG 14+).
606 ///
607 /// See [`binary`](Self::binary) for caveats — off by default.
608 pub fn with_binary(mut self, binary: bool) -> Self {
609 self.binary = binary;
610 self
611 }
612
613 /// Set the startup `options` parameter (equivalent to `libpq`'s `options`
614 /// connection parameter / `PGOPTIONS`).
615 ///
616 /// # Example
617 /// ```
618 /// use pgwire_replication::config::ReplicationConfig;
619 ///
620 /// let config = ReplicationConfig::new("h", "u", "p", "db", "slot", "pub")
621 /// .with_options("-c datestyle=ISO,MDY -c intervalstyle=iso_8601");
622 /// assert_eq!(
623 /// config.options.as_deref(),
624 /// Some("-c datestyle=ISO,MDY -c intervalstyle=iso_8601")
625 /// );
626 /// ```
627 pub fn with_options(mut self, options: impl Into<String>) -> Self {
628 self.options = Some(options.into());
629 self
630 }
631
632 /// Returns the connection string for display (password masked).
633 ///
634 /// Useful for logging without exposing credentials.
635 pub fn display_connection(&self) -> String {
636 if self.is_unix_socket() {
637 format!(
638 "postgresql://{}:***@[{}]:{}/{}",
639 self.user,
640 self.unix_socket_path().display(),
641 self.port,
642 self.database
643 )
644 } else {
645 format!(
646 "postgresql://{}:***@{}:{}/{}",
647 self.user, self.host, self.port, self.database
648 )
649 }
650 }
651}
652
653#[cfg(test)]
654mod tests {
655 use super::*;
656
657 #[test]
658 fn publication_from_single_str() {
659 let p: Publication = "orders".into();
660 assert_eq!(p.names(), ["orders"]);
661 assert_eq!(p.to_option_value(), "orders");
662 }
663
664 #[test]
665 fn publication_from_array_is_comma_joined() {
666 let p: Publication = ["orders", "customers"].into();
667 assert_eq!(p.names(), ["orders", "customers"]);
668 assert_eq!(p.to_option_value(), "orders,customers");
669 }
670
671 #[test]
672 fn publication_from_vec_and_from_iter() {
673 let from_vec: Publication = vec![String::from("a"), String::from("b")].into();
674 assert_eq!(from_vec.to_option_value(), "a,b");
675
676 let collected: Publication = ["x", "y", "z"].into_iter().collect();
677 assert_eq!(collected.to_option_value(), "x,y,z");
678 }
679
680 #[test]
681 fn publication_escapes_single_quotes_per_name() {
682 let p: Publication = ["a'b", "c"].into();
683 // Each single quote is doubled; names remain comma-separated.
684 assert_eq!(p.to_option_value(), "a''b,c");
685 }
686
687 #[test]
688 fn publication_display_is_plain_join_without_escaping() {
689 let p: Publication = ["a'b", "c"].into();
690 assert_eq!(p.to_string(), "a'b,c");
691 }
692
693 #[test]
694 fn sslmode_verification_predicates() {
695 assert!(SslMode::VerifyCa.verifies_certificate());
696 assert!(SslMode::VerifyFull.verifies_certificate());
697 assert!(!SslMode::Require.verifies_certificate());
698
699 assert!(SslMode::VerifyFull.verifies_hostname());
700 assert!(!SslMode::VerifyCa.verifies_hostname());
701 assert!(!SslMode::Require.verifies_hostname());
702 }
703
704 #[test]
705 fn verify_ca_sets_mode_and_ca_path() {
706 let tls = TlsConfig::verify_ca(Some("/ca.pem".into()));
707 assert_eq!(tls.mode, SslMode::VerifyCa);
708 assert_eq!(tls.ca_pem_path, Some("/ca.pem".into()));
709 }
710
711 #[test]
712 fn is_mtls_requires_both_cert_and_key() {
713 let both = TlsConfig::verify_full(None).with_client_cert("/c.pem", "/k.pem");
714 assert!(both.is_mtls());
715
716 let neither = TlsConfig::verify_full(None);
717 assert!(!neither.is_mtls());
718
719 let cert_only = TlsConfig {
720 client_key_pem_path: None,
721 ..TlsConfig::verify_full(None).with_client_cert("/c.pem", "/k.pem")
722 };
723 assert!(!cert_only.is_mtls());
724 }
725
726 #[test]
727 fn is_unix_socket_and_path() {
728 let tcp = ReplicationConfig::new("127.0.0.1", "u", "p", "db", "s", "pub");
729 assert!(!tcp.is_unix_socket());
730
731 let unix = ReplicationConfig::unix("/var/run/postgresql", 5432, "u", "p", "db", "s", "pub");
732 assert!(unix.is_unix_socket());
733 assert_eq!(
734 unix.unix_socket_path(),
735 std::path::PathBuf::from("/var/run/postgresql/.s.PGSQL.5432")
736 );
737 }
738
739 #[test]
740 fn display_connection_masks_password_and_shows_target() {
741 let cfg = ReplicationConfig::new("db.example.com", "alice", "secret", "mydb", "s", "pub")
742 .with_port(6432);
743 let shown = cfg.display_connection();
744 assert!(shown.contains("alice"));
745 assert!(shown.contains("db.example.com"));
746 assert!(shown.contains("6432"));
747 assert!(shown.contains("mydb"));
748 assert!(shown.contains("***"));
749 assert!(!shown.contains("secret"));
750 }
751
752 #[test]
753 fn binary_defaults_off_and_builder_sets_it() {
754 assert!(!ReplicationConfig::default().binary);
755 let cfg = ReplicationConfig::new("h", "u", "p", "db", "slot", "pub").with_binary(true);
756 assert!(cfg.binary);
757 }
758
759 #[test]
760 fn options_defaults_none_and_builder_sets_it() {
761 assert_eq!(ReplicationConfig::default().options, None);
762 let cfg = ReplicationConfig::new("h", "u", "p", "db", "slot", "pub")
763 .with_options("-c datestyle=ISO,MDY");
764 assert_eq!(cfg.options.as_deref(), Some("-c datestyle=ISO,MDY"));
765 }
766
767 #[test]
768 fn new_accepts_single_and_multiple_publications() {
769 let single = ReplicationConfig::new("h", "u", "p", "db", "slot", "solo");
770 assert_eq!(single.publication.to_option_value(), "solo");
771
772 let multi = ReplicationConfig::new("h", "u", "p", "db", "slot", ["one", "two"]);
773 assert_eq!(multi.publication.to_option_value(), "one,two");
774 }
775}