toasty_driver_postgresql/
lib.rs

1#![warn(missing_docs)]
2
3//! Toasty driver for [PostgreSQL](https://www.postgresql.org/) using
4//! [`tokio-postgres`](https://docs.rs/tokio-postgres).
5//!
6//! # Examples
7//!
8//! ```no_run
9//! use toasty_driver_postgresql::PostgreSQL;
10//!
11//! let driver = PostgreSQL::new("postgresql://localhost/mydb").unwrap();
12//! ```
13
14mod oid_cache;
15mod statement_cache;
16#[cfg(feature = "tls")]
17mod tls;
18mod r#type;
19mod value;
20
21pub(crate) use value::Value;
22
23use async_trait::async_trait;
24use std::{borrow::Cow, sync::Arc};
25use toasty_core::{
26    Result, Schema,
27    driver::{
28        Capability, ConnectContext, ConnectionUrl, Driver, ExecResponse, Operation, QueryLogConfig,
29        log::QueryLog,
30        operation::{RawSqlRet, Transaction, TransactionMode, TypedValue},
31    },
32    schema::{
33        db::{self, Migration, Table},
34        diff,
35    },
36    stmt,
37    stmt::ValueRecord,
38};
39use toasty_sql::{self as sql};
40use tokio_postgres::{Client, Config, Socket, tls::MakeTlsConnect, types::ToSql};
41
42enum SqlReturn {
43    Count,
44    Infer,
45    Types(Vec<stmt::Type>),
46}
47
48use crate::{oid_cache::OidCache, statement_cache::StatementCache};
49
50/// Classifies a `tokio_postgres::Error` into a Toasty error.
51///
52/// Errors that carry a server-side `DbError` are mapped to typed
53/// variants where one exists (`SerializationFailure`,
54/// `ReadOnlyTransaction`); everything else with a `DbError` becomes
55/// `DriverOperationFailed`. Errors *without* a `DbError` are
56/// classified as `ConnectionLost`: per `tokio-postgres`, those
57/// originate from the underlying socket or protocol layer (closed
58/// socket, IO error, end-of-stream during handshake), which the
59/// pool treats as evictable.
60fn classify_pg_error(e: tokio_postgres::Error) -> toasty_core::Error {
61    if let Some(db_err) = e.as_db_error() {
62        match db_err.code().code() {
63            "40001" => toasty_core::Error::serialization_failure(db_err.message()),
64            "25006" => toasty_core::Error::read_only_transaction(db_err.message()),
65            _ => toasty_core::Error::driver_operation_failed(e),
66        }
67    } else {
68        toasty_core::Error::connection_lost(e)
69    }
70}
71
72/// A PostgreSQL [`Driver`] that connects via `tokio-postgres`.
73///
74/// # Examples
75///
76/// ```no_run
77/// use toasty_driver_postgresql::PostgreSQL;
78///
79/// let driver = PostgreSQL::new("postgresql://localhost/mydb").unwrap();
80/// ```
81pub struct PostgreSQL {
82    url: String,
83    config: Config,
84    #[cfg(feature = "tls")]
85    tls: Option<tokio_postgres_rustls::MakeRustlsConnect>,
86}
87
88impl std::fmt::Debug for PostgreSQL {
89    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
90        let Self {
91            url,
92            config,
93            #[cfg(feature = "tls")]
94            tls,
95        } = self;
96        let mut s = f.debug_struct("PostgreSQL");
97        s.field("url", url);
98        s.field("config", config);
99        #[cfg(feature = "tls")]
100        s.field("tls", &tls.as_ref().map(|_| "MakeRustlsConnect"));
101        s.finish()
102    }
103}
104
105impl PostgreSQL {
106    /// Create a new PostgreSQL driver from a connection URL.
107    ///
108    /// The URL takes the `postgresql://user:password@host:port/dbname`
109    /// form (`postgres://` also works). Query parameters supplement or
110    /// override the URL components, following libpq: `host`, `port`,
111    /// `user`, `password`, `dbname`, `application_name`, and `options`
112    /// (server startup options such as `-c search_path=my_schema`,
113    /// applied to every connection the pool creates), plus the TLS
114    /// parameters `sslmode`, `sslrootcert`, `sslcert`, `sslkey`,
115    /// `channel_binding`, and `sslnegotiation`. Unrecognized parameters
116    /// are ignored.
117    pub fn new(url: impl Into<String>) -> Result<Self> {
118        let url_str = url.into();
119        let url = ConnectionUrl::parse(&url_str)?;
120
121        if !(url.has_scheme("postgresql") || url.has_scheme("postgres")) {
122            return Err(toasty_core::Error::invalid_connection_url(format!(
123                "connection URL does not have a `postgresql` scheme; url={}",
124                url.as_str()
125            )));
126        }
127
128        if url.path().is_empty() {
129            return Err(toasty_core::Error::invalid_connection_url(format!(
130                "no database specified - missing path in connection URL; url={}",
131                url.as_str()
132            )));
133        }
134
135        let mut config = Config::new();
136
137        let dbname = url.decoded_path().map_err(|_| {
138            toasty_core::Error::invalid_connection_url("database name is not valid UTF-8")
139        })?;
140        config.dbname(dbname.trim_start_matches('/'));
141
142        if let Some(user) = url.username()?.filter(|user| !user.is_empty()) {
143            config.user(&*user);
144        }
145
146        if let Some(password) = url.password() {
147            config.password(&*password);
148        }
149
150        // libpq lets standard connection parameters appear in the query
151        // string; honor the ones a Toasty user can reasonably set so that
152        // `postgresql:///mydb?host=/tmp&user=alice` reaches the server.
153        // Single-valued setters (user, password, dbname, application_name,
154        // options) replace earlier calls, so we can apply them inline; host and
155        // port are list-valued — staged into Options below so a query
156        // parameter cleanly overrides the URL component instead of being
157        // appended as a fallback tokio-postgres would try first.
158        let mut host: Option<String> = None;
159        let mut port: Option<u16> = None;
160
161        for (key, value) in url.query_pairs() {
162            match key.as_ref() {
163                "host" => host = Some(value.into_owned()),
164                "port" => {
165                    port = Some(value.parse::<u16>().map_err(|_| {
166                        toasty_core::Error::invalid_connection_url(format!(
167                            "invalid port in connection URL query parameter: {value}"
168                        ))
169                    })?);
170                }
171                "user" => {
172                    config.user(&*value);
173                }
174                "password" => {
175                    config.password(value.as_bytes());
176                }
177                "dbname" => {
178                    config.dbname(&*value);
179                }
180                "application_name" => {
181                    config.application_name(&*value);
182                }
183                "options" => {
184                    config.options(&*value);
185                }
186                _ => {}
187            }
188        }
189
190        let host = host.or(url.host()?.map(String::from)).ok_or_else(|| {
191            toasty_core::Error::invalid_connection_url(format!(
192                "missing host in connection URL; url={}",
193                url.as_str()
194            ))
195        })?;
196        config.host(&host);
197
198        if let Some(port) = port.or(url.port()?) {
199            config.port(port);
200        }
201
202        #[cfg(feature = "tls")]
203        let tls = tls::configure_tls(&url, &mut config)?;
204
205        #[cfg(not(feature = "tls"))]
206        for (key, value) in url.query_pairs() {
207            if key == "sslmode" && value != "disable" {
208                return Err(toasty_core::Error::invalid_connection_url(
209                    "TLS not available: compile with the `tls` feature",
210                ));
211            }
212        }
213
214        Ok(Self {
215            url: url_str,
216            config,
217            #[cfg(feature = "tls")]
218            tls,
219        })
220    }
221
222    async fn connect_with_config(&self, config: Config) -> Result<Connection> {
223        #[cfg(feature = "tls")]
224        if let Some(ref tls) = self.tls {
225            return Connection::connect(config, tls.clone()).await;
226        }
227        Connection::connect(config, tokio_postgres::NoTls).await
228    }
229}
230
231#[async_trait]
232impl Driver for PostgreSQL {
233    fn url(&self) -> Cow<'_, str> {
234        Cow::Borrowed(&self.url)
235    }
236
237    fn capability(&self) -> &'static Capability {
238        &Capability::POSTGRESQL
239    }
240
241    async fn connect(
242        &self,
243        cx: &ConnectContext,
244    ) -> toasty_core::Result<Box<dyn toasty_core::driver::Connection>> {
245        let mut connection = self.connect_with_config(self.config.clone()).await?;
246        connection.query_log = cx.query_log;
247        Ok(Box::new(connection))
248    }
249
250    fn generate_migration(&self, schema_diff: &diff::Schema<'_>) -> Migration {
251        let statements = sql::MigrationStatement::from_diff(schema_diff, &Capability::POSTGRESQL);
252
253        let sql_strings: Vec<String> = statements
254            .iter()
255            .map(|stmt| sql::Serializer::postgresql(stmt.schema()).serialize(stmt.statement()))
256            .collect();
257
258        Migration::new_sql(sql_strings.join("\n"))
259    }
260
261    async fn reset_db(&self) -> toasty_core::Result<()> {
262        let dbname = self
263            .config
264            .get_dbname()
265            .ok_or_else(|| {
266                toasty_core::Error::invalid_connection_url("no database name configured")
267            })?
268            .to_string();
269
270        // We cannot drop a database we are currently connected to, so we need a temp database.
271        let temp_dbname = "__toasty_reset_temp";
272
273        let connect = |dbname: &str| {
274            let mut config = self.config.clone();
275            config.dbname(dbname);
276            self.connect_with_config(config)
277        };
278
279        // Step 1: Connect to the target DB and create a temp DB
280        let conn = connect(&dbname).await?;
281        conn.client
282            .execute(&format!("DROP DATABASE IF EXISTS \"{}\"", temp_dbname), &[])
283            .await
284            .map_err(classify_pg_error)?;
285        conn.client
286            .execute(&format!("CREATE DATABASE \"{}\"", temp_dbname), &[])
287            .await
288            .map_err(classify_pg_error)?;
289        drop(conn);
290
291        // Step 2: Connect to the temp DB, drop and recreate the target
292        let conn = connect(temp_dbname).await?;
293        conn.client
294            .execute(
295                "SELECT pg_terminate_backend(pid) \
296                 FROM pg_stat_activity \
297                 WHERE datname = $1 AND pid <> pg_backend_pid()",
298                &[&dbname],
299            )
300            .await
301            .map_err(classify_pg_error)?;
302        conn.client
303            .execute(&format!("DROP DATABASE IF EXISTS \"{}\"", dbname), &[])
304            .await
305            .map_err(classify_pg_error)?;
306        conn.client
307            .execute(&format!("CREATE DATABASE \"{}\"", dbname), &[])
308            .await
309            .map_err(classify_pg_error)?;
310        drop(conn);
311
312        // Step 3: Connect back to the target and clean up the temp DB
313        let conn = connect(&dbname).await?;
314        conn.client
315            .execute(&format!("DROP DATABASE IF EXISTS \"{}\"", temp_dbname), &[])
316            .await
317            .map_err(classify_pg_error)?;
318
319        Ok(())
320    }
321}
322
323/// An open connection to a PostgreSQL database.
324#[derive(Debug)]
325pub struct Connection {
326    client: Client,
327    statement_cache: StatementCache,
328    oid_cache: OidCache,
329    query_log: QueryLogConfig,
330}
331
332impl Connection {
333    /// Initialize a Toasty PostgreSQL connection using an initialized client.
334    pub fn new(client: Client) -> Self {
335        Self {
336            client,
337            statement_cache: StatementCache::new(100),
338            oid_cache: OidCache::new(),
339            query_log: QueryLogConfig::default(),
340        }
341    }
342
343    /// Connects to a PostgreSQL database using a [`postgres::Config`].
344    ///
345    /// See [`postgres::Client::configure`] for more information.
346    pub async fn connect<T>(config: Config, tls: T) -> Result<Self>
347    where
348        T: MakeTlsConnect<Socket> + 'static,
349        T::Stream: Send,
350    {
351        let (client, connection) = config.connect(tls).await.map_err(classify_pg_error)?;
352
353        tokio::spawn(async move {
354            if let Err(e) = connection.await {
355                eprintln!("connection error: {e}");
356            }
357        });
358
359        Ok(Self::new(client))
360    }
361
362    async fn exec_sql(
363        &mut self,
364        sql_as_str: &str,
365        typed_params: Vec<TypedValue>,
366        ret: SqlReturn,
367    ) -> Result<ExecResponse> {
368        let mut log = QueryLog::sql(
369            &self.query_log,
370            "postgresql",
371            sql_as_str,
372            typed_params.iter().map(|tv| &tv.value),
373        );
374        let result = self
375            .exec_sql_inner(sql_as_str, typed_params, ret, &mut log)
376            .await;
377        log.finish(&result);
378        result
379    }
380
381    async fn exec_sql_inner(
382        &mut self,
383        sql_as_str: &str,
384        typed_params: Vec<TypedValue>,
385        ret: SqlReturn,
386        log: &mut QueryLog<'_>,
387    ) -> Result<ExecResponse> {
388        self.oid_cache
389            .preload(&self.client, typed_params.iter().map(|tv| &tv.ty))
390            .await?;
391        let param_types: Vec<_> = typed_params
392            .iter()
393            .map(|tv| self.oid_cache.get(&tv.ty).clone())
394            .collect();
395
396        let values: Vec<_> = typed_params
397            .into_iter()
398            .map(|tv| Value::from(tv.value))
399            .collect();
400        let params = values
401            .iter()
402            .map(|param| param as &(dyn ToSql + Sync))
403            .collect::<Vec<_>>();
404
405        let statement = self
406            .statement_cache
407            .prepare_typed(&mut self.client, sql_as_str, &param_types)
408            .await
409            .map_err(classify_pg_error)?;
410
411        if matches!(ret, SqlReturn::Count) {
412            let count = self
413                .client
414                .execute(&statement, &params)
415                .await
416                .map_err(classify_pg_error)?;
417            return Ok(ExecResponse::count(count));
418        }
419
420        let rows = self
421            .client
422            .query(&statement, &params)
423            .await
424            .map_err(classify_pg_error)?;
425
426        log.rows(rows.len() as u64);
427
428        let results = rows.into_iter().map(move |row| {
429            let mut results = Vec::new();
430
431            match &ret {
432                SqlReturn::Count => unreachable!(),
433                SqlReturn::Infer => {
434                    for (i, column) in row.columns().iter().enumerate() {
435                        results.push(Value::from_sql_infer(i, &row, column).into_inner());
436                    }
437                }
438                SqlReturn::Types(ret_tys) => {
439                    for (i, column) in row.columns().iter().enumerate() {
440                        results.push(Value::from_sql(i, &row, column, &ret_tys[i]).into_inner());
441                    }
442                }
443            }
444
445            Ok(ValueRecord::from_vec(results))
446        });
447
448        Ok(ExecResponse::value_stream(stmt::ValueStream::from_iter(
449            results,
450        )))
451    }
452
453    /// Creates a table.
454    pub async fn create_table(&mut self, schema: &db::Schema, table: &Table) -> Result<()> {
455        let serializer = sql::Serializer::postgresql(schema);
456
457        let sql = serializer.serialize(&sql::Statement::create_table(
458            table,
459            &Capability::POSTGRESQL,
460        ));
461
462        self.client
463            .execute(&sql, &[])
464            .await
465            .map_err(classify_pg_error)?;
466
467        for index in &table.indices {
468            if index.primary_key {
469                continue;
470            }
471
472            let sql = serializer.serialize(&sql::Statement::create_index(index));
473
474            self.client
475                .execute(&sql, &[])
476                .await
477                .map_err(classify_pg_error)?;
478        }
479
480        Ok(())
481    }
482}
483
484impl From<Client> for Connection {
485    fn from(client: Client) -> Self {
486        Self::new(client)
487    }
488}
489
490#[async_trait]
491impl toasty_core::driver::Connection for Connection {
492    async fn exec(&mut self, schema: &Arc<Schema>, op: Operation) -> Result<ExecResponse> {
493        tracing::trace!(driver = "postgresql", op = %op.name(), "driver exec");
494
495        if let Operation::Transaction(ref t) = op {
496            // PostgreSQL has no `BEGIN IMMEDIATE` / `BEGIN EXCLUSIVE`
497            // analogue; reject non-Default modes loudly rather than
498            // silently dropping them at the serializer.
499            if let Transaction::Start {
500                mode: mode @ (TransactionMode::Immediate | TransactionMode::Exclusive),
501                ..
502            } = t
503            {
504                return Err(toasty_core::Error::unsupported_feature(format!(
505                    "PostgreSQL does not support TransactionMode::{mode:?}"
506                )));
507            }
508            let sql = sql::Serializer::postgresql(&schema.db).serialize_transaction(t);
509            self.client
510                .batch_execute(&sql)
511                .await
512                .map_err(classify_pg_error)?;
513            return Ok(ExecResponse::count(0));
514        }
515
516        let (sql, typed_params, ret_tys) = match op {
517            Operation::Insert(op) => (sql::Statement::from(op.stmt), op.params, op.ret),
518            Operation::QuerySql(query) => {
519                (sql::Statement::from(query.stmt), query.params, query.ret)
520            }
521            Operation::RawSql(op) => {
522                let ret = match op.ret {
523                    RawSqlRet::None => SqlReturn::Count,
524                    RawSqlRet::Infer => SqlReturn::Infer,
525                    RawSqlRet::Types(types) => SqlReturn::Types(types),
526                };
527                return self.exec_sql(&op.sql, op.params, ret).await;
528            }
529            op => todo!("op={:#?}", op),
530        };
531
532        let sql_as_str = sql::Serializer::postgresql(&schema.db).serialize(&sql);
533
534        let ret = if sql.returning_len().is_some() {
535            SqlReturn::Types(ret_tys.unwrap())
536        } else {
537            SqlReturn::Count
538        };
539
540        self.exec_sql(&sql_as_str, typed_params, ret).await
541    }
542
543    async fn push_schema(&mut self, schema: &Schema) -> Result<()> {
544        let serializer = sql::Serializer::postgresql(&schema.db);
545
546        // Create PostgreSQL enum types before creating tables.
547        // Collect unique enum types across all columns.
548        let mut created_enum_types = hashbrown::HashSet::new();
549        for table in &schema.db.tables {
550            for column in &table.columns {
551                if let Some(type_enum) = column.storage_ty.named_enum()
552                    && created_enum_types.insert(type_enum.name.clone())
553                {
554                    let sql = serializer.serialize(&sql::Statement::create_enum_type(type_enum));
555
556                    tracing::debug!(enum_type = ?type_enum.name, "creating enum type");
557                    self.client
558                        .execute(&sql, &[])
559                        .await
560                        .map_err(classify_pg_error)?;
561                }
562            }
563        }
564
565        for table in &schema.db.tables {
566            tracing::debug!(table = %table.name, "creating table");
567            self.create_table(&schema.db, table).await?;
568        }
569        Ok(())
570    }
571
572    async fn applied_migrations(
573        &mut self,
574    ) -> Result<Vec<toasty_core::schema::db::AppliedMigration>> {
575        // Ensure the migrations table exists
576        self.client
577            .execute(
578                "CREATE TABLE IF NOT EXISTS __toasty_migrations (
579                id BIGINT PRIMARY KEY,
580                name TEXT NOT NULL,
581                applied_at TIMESTAMP NOT NULL
582            )",
583                &[],
584            )
585            .await
586            .map_err(classify_pg_error)?;
587
588        // Query all applied migrations
589        let rows = self
590            .client
591            .query(
592                "SELECT id FROM __toasty_migrations ORDER BY applied_at",
593                &[],
594            )
595            .await
596            .map_err(classify_pg_error)?;
597
598        Ok(rows
599            .iter()
600            .map(|row| {
601                let id: i64 = row.get(0);
602                toasty_core::schema::db::AppliedMigration::new(id as u64)
603            })
604            .collect())
605    }
606
607    async fn apply_migration(
608        &mut self,
609        id: u64,
610        name: &str,
611        migration: &toasty_core::schema::db::Migration,
612    ) -> Result<()> {
613        tracing::info!(id = id, name = %name, "applying migration");
614        // Ensure the migrations table exists
615        self.client
616            .execute(
617                "CREATE TABLE IF NOT EXISTS __toasty_migrations (
618                id BIGINT PRIMARY KEY,
619                name TEXT NOT NULL,
620                applied_at TIMESTAMP NOT NULL
621            )",
622                &[],
623            )
624            .await
625            .map_err(classify_pg_error)?;
626
627        // Start transaction
628        let transaction = self.client.transaction().await.map_err(classify_pg_error)?;
629
630        // Execute each migration statement
631        for statement in migration.statements() {
632            if let Err(e) = transaction
633                .batch_execute(statement)
634                .await
635                .map_err(classify_pg_error)
636            {
637                transaction.rollback().await.map_err(classify_pg_error)?;
638                return Err(e);
639            }
640        }
641
642        // Record the migration
643        if let Err(e) = transaction
644            .execute(
645                "INSERT INTO __toasty_migrations (id, name, applied_at) VALUES ($1, $2, NOW())",
646                &[&(id as i64), &name],
647            )
648            .await
649            .map_err(classify_pg_error)
650        {
651            transaction.rollback().await.map_err(classify_pg_error)?;
652            return Err(e);
653        }
654
655        // Commit transaction
656        transaction.commit().await.map_err(classify_pg_error)?;
657        Ok(())
658    }
659
660    fn is_valid(&self) -> bool {
661        !self.client.is_closed()
662    }
663
664    async fn ping(&mut self) -> Result<()> {
665        // An empty `simple_query` is the lightest sync round-trip in
666        // the PG protocol — it skips parsing entirely. Any failure is
667        // surfaced as `connection_lost`: the only meaningful outcome
668        // of a ping is "the connection is alive" or "evict it."
669        self.client
670            .simple_query("")
671            .await
672            .map(|_| ())
673            .map_err(toasty_core::Error::connection_lost)
674    }
675}
676
677#[cfg(test)]
678mod tests {
679    use super::*;
680    use tokio_postgres::config::Host;
681
682    fn cfg(url: &str) -> Config {
683        PostgreSQL::new(url).expect("valid URL").config
684    }
685
686    #[test]
687    fn host_in_url_authority() {
688        let c = cfg("postgresql://example.com/mydb");
689        assert_eq!(c.get_hosts(), &[Host::Tcp("example.com".into())]);
690        assert_eq!(c.get_dbname(), Some("mydb"));
691    }
692
693    // `Host::Unix` only exists on Unix targets in tokio-postgres, so the
694    // tests that assert socket-path resolution are gated to those
695    // platforms. Windows builds still exercise the URL-parsing path via
696    // the non-Unix tests below.
697    #[cfg(unix)]
698    mod unix_socket {
699        use super::*;
700        use std::path::PathBuf;
701
702        #[test]
703        fn unix_socket_via_host_query_param() {
704            // Regression for #984: libpq lets a Unix-socket directory be
705            // supplied via `?host=/path`, since URL syntax cannot put a
706            // filesystem path in the authority.
707            let c = cfg("postgresql:///mydb?host=/tmp&user=myuser");
708            assert_eq!(c.get_hosts(), &[Host::Unix(PathBuf::from("/tmp"))]);
709            assert_eq!(c.get_user(), Some("myuser"));
710            assert_eq!(c.get_dbname(), Some("mydb"));
711        }
712
713        #[test]
714        fn query_param_host_overrides_url_authority() {
715            // libpq semantics: a `host=` query parameter replaces (not
716            // appends to) the URL authority host. tokio-postgres's
717            // `Config::host` is additive across calls, so an authority
718            // host would otherwise be tried first and the configured
719            // socket reached only as a fallback.
720            let c = cfg("postgresql://example.com/mydb?host=/var/run/postgresql");
721            assert_eq!(
722                c.get_hosts(),
723                &[Host::Unix(PathBuf::from("/var/run/postgresql"))]
724            );
725        }
726
727        #[test]
728        fn query_param_port_user_password_dbname() {
729            let c = cfg(
730                "postgresql:///placeholder?host=/tmp&port=5433&user=alice&password=s3cret&dbname=real",
731            );
732            assert_eq!(c.get_hosts(), &[Host::Unix(PathBuf::from("/tmp"))]);
733            assert_eq!(c.get_ports(), &[5433]);
734            assert_eq!(c.get_user(), Some("alice"));
735            assert_eq!(c.get_password(), Some(&b"s3cret"[..]));
736            assert_eq!(c.get_dbname(), Some("real"));
737        }
738    }
739
740    #[test]
741    fn query_param_port_overrides_url_port() {
742        // `Config::port` is additive too, so a `port=` query parameter
743        // must replace the URL authority port for the same reason.
744        let c = cfg("postgresql://example.com:5432/mydb?port=5433");
745        assert_eq!(c.get_ports(), &[5433]);
746    }
747
748    #[test]
749    fn application_name_query_param() {
750        let c = cfg("postgresql://localhost/mydb?application_name=my_app");
751        assert_eq!(c.get_application_name(), Some("my_app"));
752    }
753
754    #[test]
755    fn options_query_param() {
756        // libpq's `options` startup parameter passes through, e.g. to set
757        // a per-connection `search_path`. Form-decoding maps `%20` (and
758        // `+`) to a space and `%3D` to `=`.
759        let c = cfg("postgresql://localhost/mydb?options=-c%20search_path%3Dtenant_a%2Cpublic");
760        assert_eq!(c.get_options(), Some("-c search_path=tenant_a,public"));
761    }
762
763    #[test]
764    fn missing_host_rejected() {
765        let err = PostgreSQL::new("postgresql:///mydb").unwrap_err();
766        assert!(
767            err.to_string().contains("missing host"),
768            "expected missing-host error, got: {err}"
769        );
770    }
771
772    #[test]
773    fn invalid_port_query_param_rejected() {
774        let err = PostgreSQL::new("postgresql:///mydb?host=/tmp&port=not-a-number").unwrap_err();
775        assert!(
776            err.to_string().contains("invalid port"),
777            "expected invalid-port error, got: {err}"
778        );
779    }
780}
toasty_driver_postgresql/lib.rs

toasty_driver_postgresql/
lib.rs
