From 51f02c89eebe9d7d9c2915330eeb9fdb9f0e1853 Mon Sep 17 00:00:00 2001 From: Steven Fackler Date: Fri, 26 Jul 2019 20:11:34 -0700 Subject: [PATCH] Fully working prepare --- tokio-postgres/src/client.rs | 95 +-- tokio-postgres/src/prepare.rs | 265 ++++++-- tokio-postgres/src/query.rs | 55 +- tokio-postgres/src/types/mod.rs | 16 +- tokio-postgres/src/types/serde_json_1.rs | 2 +- tokio-postgres/tests/test/main.rs | 307 ++++----- tokio-postgres/tests/test/runtime.rs | 10 +- tokio-postgres/tests/test/types/bit_vec_07.rs | 14 +- tokio-postgres/tests/test/types/chrono_04.rs | 62 +- tokio-postgres/tests/test/types/eui48_04.rs | 7 +- tokio-postgres/tests/test/types/geo_010.rs | 27 +- tokio-postgres/tests/test/types/mod.rs | 622 +++++++++--------- .../tests/test/types/serde_json_1.rs | 14 +- tokio-postgres/tests/test/types/uuid_07.rs | 7 +- 14 files changed, 826 insertions(+), 677 deletions(-) diff --git a/tokio-postgres/src/client.rs b/tokio-postgres/src/client.rs index 0f731ae4..e49c68da 100644 --- a/tokio-postgres/src/client.rs +++ b/tokio-postgres/src/client.rs @@ -1,7 +1,7 @@ use crate::codec::BackendMessages; use crate::connection::{Request, RequestMessages}; use crate::prepare::prepare; -use crate::query::{execute, query, Query}; +use crate::query::{self, Query}; use crate::types::{Oid, ToSql, Type}; use crate::{Error, Statement}; use fallible_iterator::FallibleIterator; @@ -42,9 +42,9 @@ impl Responses { } struct State { - has_typeinfo: bool, - has_typeinfo_composite: bool, - has_typeinfo_enum: bool, + typeinfo: Option, + typeinfo_composite: Option, + typeinfo_enum: Option, types: HashMap, } @@ -67,36 +67,36 @@ impl InnerClient { }) } - pub fn has_typeinfo(&self) -> bool { - self.state.lock().has_typeinfo + pub fn typeinfo(&self) -> Option { + self.state.lock().typeinfo.clone() } - pub fn set_has_typeinfo(&self) { - self.state.lock().has_typeinfo = true; + pub fn set_typeinfo(&self, statement: &Statement) { + self.state.lock().typeinfo = Some(statement.clone()); } - pub fn has_typeinfo_composite(&self) -> bool { - self.state.lock().has_typeinfo_composite + pub fn typeinfo_composite(&self) -> Option { + self.state.lock().typeinfo_composite.clone() } - pub fn set_has_typeinfo_composite(&self) { - self.state.lock().has_typeinfo_composite = true; + pub fn set_typeinfo_composite(&self, statement: &Statement) { + self.state.lock().typeinfo_composite = Some(statement.clone()); } - pub fn has_typeinfo_enum(&self) -> bool { - self.state.lock().has_typeinfo_enum + pub fn typeinfo_enum(&self) -> Option { + self.state.lock().typeinfo_enum.clone() } - pub fn set_has_typeinfo_enum(&self) { - self.state.lock().has_typeinfo_enum = true; + pub fn set_typeinfo_enum(&self, statement: &Statement) { + self.state.lock().typeinfo_enum = Some(statement.clone()); } pub fn type_(&self, oid: Oid) -> Option { self.state.lock().types.get(&oid).cloned() } - pub fn set_type(&self, oid: Oid, type_: Type) { - self.state.lock().types.insert(oid, type_); + pub fn set_type(&self, oid: Oid, type_: &Type) { + self.state.lock().types.insert(oid, type_.clone()); } } @@ -116,9 +116,9 @@ impl Client { inner: Arc::new(InnerClient { sender, state: Mutex::new(State { - has_typeinfo: false, - has_typeinfo_composite: false, - has_typeinfo_enum: false, + typeinfo: None, + typeinfo_composite: None, + typeinfo_enum: None, types: HashMap::new(), }), }), @@ -131,58 +131,59 @@ impl Client { self.inner.clone() } - pub fn prepare<'a>( - &mut self, - query: &'a str, - ) -> impl Future> + 'a { + pub fn prepare(&mut self, query: &str) -> impl Future> { self.prepare_typed(query, &[]) } - pub fn prepare_typed<'a>( + pub fn prepare_typed( &mut self, - query: &'a str, - parameter_types: &'a [Type], - ) -> impl Future> + 'a { + query: &str, + parameter_types: &[Type], + ) -> impl Future> { prepare(self.inner(), query, parameter_types) } - pub fn query<'a>( + pub fn query( &mut self, - statement: &'a Statement, - params: &'a [&dyn ToSql], - ) -> impl Future> + 'a { - self.query_iter(statement, params.iter().cloned()) + statement: &Statement, + params: &[&dyn ToSql], + ) -> impl Future> { + let buf = query::encode(statement, params.iter().cloned()); + query::query(self.inner(), statement.clone(), buf) } pub fn query_iter<'a, I>( &mut self, - statement: &'a Statement, + statement: &Statement, params: I, - ) -> impl Future> + 'a + ) -> impl Future> where - I: IntoIterator + 'a, + I: IntoIterator, I::IntoIter: ExactSizeIterator, { - query(self.inner(), statement, params) + let buf = query::encode(statement, params); + query::query(self.inner(), statement.clone(), buf) } - pub fn execute<'a>( + pub fn execute( &mut self, - statement: &'a Statement, - params: &'a [&dyn ToSql], - ) -> impl Future> + 'a { - self.execute_iter(statement, params.iter().cloned()) + statement: &Statement, + params: &[&dyn ToSql], + ) -> impl Future> { + let buf = query::encode(statement, params.iter().cloned()); + query::execute(self.inner(), buf) } pub fn execute_iter<'a, I>( &mut self, - statement: &'a Statement, + statement: &Statement, params: I, - ) -> impl Future> + 'a + ) -> impl Future> where - I: IntoIterator + 'a, + I: IntoIterator, I::IntoIter: ExactSizeIterator, { - execute(self.inner(), statement, params) + let buf = query::encode(statement, params); + query::execute(self.inner(), buf) } } diff --git a/tokio-postgres/src/prepare.rs b/tokio-postgres/src/prepare.rs index 056ec3dc..8cc88091 100644 --- a/tokio-postgres/src/prepare.rs +++ b/tokio-postgres/src/prepare.rs @@ -1,64 +1,119 @@ use crate::client::InnerClient; use crate::codec::FrontendMessage; use crate::connection::RequestMessages; -use crate::types::{Oid, Type}; +use crate::error::SqlState; +use crate::query; +use crate::types::{Field, Kind, Oid, ToSql, Type}; use crate::{Column, Error, Statement}; use fallible_iterator::FallibleIterator; +use futures::{future, StreamExt, TryStreamExt}; use postgres_protocol::message::backend::Message; use postgres_protocol::message::frontend; +use std::future::Future; +use std::pin::Pin; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +const TYPEINFO_STMT: &str = "typeinfo"; + +const TYPEINFO_QUERY: &str = "\ +SELECT t.typname, t.typtype, t.typelem, r.rngsubtype, t.typbasetype, n.nspname, t.typrelid +FROM pg_catalog.pg_type t +LEFT OUTER JOIN pg_catalog.pg_range r ON r.rngtypid = t.oid +INNER JOIN pg_catalog.pg_namespace n ON t.typnamespace = n.oid +WHERE t.oid = $1 +"; + +// Range types weren't added until Postgres 9.2, so pg_range may not exist +const TYPEINFO_FALLBACK_QUERY: &str = "\ +SELECT t.typname, t.typtype, t.typelem, NULL::OID, t.typbasetype, n.nspname, t.typrelid +FROM pg_catalog.pg_type t +INNER JOIN pg_catalog.pg_namespace n ON t.typnamespace = n.oid +WHERE t.oid = $1 +"; + +const TYPEINFO_ENUM_QUERY: &str = "\ +SELECT enumlabel +FROM pg_catalog.pg_enum +WHERE enumtypid = $1 +ORDER BY enumsortorder +"; + +// Postgres 9.0 didn't have enumsortorder +const TYPEINFO_ENUM_FALLBACK_QUERY: &str = "\ +SELECT enumlabel +FROM pg_catalog.pg_enum +WHERE enumtypid = $1 +ORDER BY oid +"; + +const TYPEINFO_COMPOSITE_QUERY: &str = "\ +SELECT attname, atttypid +FROM pg_catalog.pg_attribute +WHERE attrelid = $1 +AND NOT attisdropped +AND attnum > 0 +ORDER BY attnum +"; + static NEXT_ID: AtomicUsize = AtomicUsize::new(0); -pub async fn prepare( +pub fn prepare( client: Arc, query: &str, types: &[Type], -) -> Result { +) -> impl Future> { let name = format!("s{}", NEXT_ID.fetch_add(1, Ordering::SeqCst)); + let buf = encode(&name, query, types); + async move { + let buf = buf?; + let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?; + + match responses.next().await? { + Message::ParseComplete => {} + _ => return Err(Error::unexpected_message()), + } + + let parameter_description = match responses.next().await? { + Message::ParameterDescription(body) => body, + _ => return Err(Error::unexpected_message()), + }; + + let row_description = match responses.next().await? { + Message::RowDescription(body) => Some(body), + Message::NoData => None, + _ => return Err(Error::unexpected_message()), + }; + + let mut parameters = vec![]; + let mut it = parameter_description.parameters(); + while let Some(oid) = it.next().map_err(Error::parse)? { + let type_ = get_type(&client, oid).await?; + parameters.push(type_); + } + + let mut columns = vec![]; + if let Some(row_description) = row_description { + let mut it = row_description.fields(); + while let Some(field) = it.next().map_err(Error::parse)? { + let type_ = get_type(&client, field.type_oid()).await?; + let column = Column::new(field.name().to_string(), type_); + columns.push(column); + } + } + + Ok(Statement::new(&client, name, parameters, columns)) + } +} + +fn encode(name: &str, query: &str, types: &[Type]) -> Result, Error> { let mut buf = vec![]; - frontend::parse(&name, query, types.iter().map(Type::oid), &mut buf).map_err(Error::encode)?; + frontend::parse(name, query, types.iter().map(Type::oid), &mut buf).map_err(Error::encode)?; frontend::describe(b'S', &name, &mut buf).map_err(Error::encode)?; frontend::sync(&mut buf); - let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?; - - match responses.next().await? { - Message::ParseComplete => {} - _ => return Err(Error::unexpected_message()), - } - - let parameter_description = match responses.next().await? { - Message::ParameterDescription(body) => body, - _ => return Err(Error::unexpected_message()), - }; - - let row_description = match responses.next().await? { - Message::RowDescription(body) => Some(body), - Message::NoData => None, - _ => return Err(Error::unexpected_message()), - }; - - let mut parameters = vec![]; - let mut it = parameter_description.parameters(); - while let Some(oid) = it.next().map_err(Error::parse)? { - let type_ = get_type(&client, oid).await?; - parameters.push(type_); - } - - let mut columns = vec![]; - if let Some(row_description) = row_description { - let mut it = row_description.fields(); - while let Some(field) = it.next().map_err(Error::parse)? { - let type_ = get_type(&client, field.type_oid()).await?; - let column = Column::new(field.name().to_string(), type_); - columns.push(column); - } - } - - Ok(Statement::new(&client, name, parameters, columns)) + Ok(buf) } async fn get_type(client: &Arc, oid: Oid) -> Result { @@ -70,5 +125,133 @@ async fn get_type(client: &Arc, oid: Oid) -> Result { return Ok(type_); } - unimplemented!() + let stmt = typeinfo_statement(client).await?; + + let params: &[&dyn ToSql] = &[&oid]; + let buf = query::encode(&stmt, params.iter().cloned()); + let mut rows = query::query(client.clone(), stmt, buf).await?; + + let row = match rows.try_next().await? { + Some(row) => row, + None => return Err(Error::unexpected_message()), + }; + + let name: String = row.try_get(0)?; + let type_: i8 = row.try_get(1)?; + let elem_oid: Oid = row.try_get(2)?; + let rngsubtype: Option = row.try_get(3)?; + let basetype: Oid = row.try_get(4)?; + let schema: String = row.try_get(5)?; + let relid: Oid = row.try_get(6)?; + + let kind = if type_ == b'e' as i8 { + let variants = get_enum_variants(client, oid).await?; + Kind::Enum(variants) + } else if type_ == b'p' as i8 { + Kind::Pseudo + } else if basetype != 0 { + let type_ = get_type_rec(client, basetype).await?; + Kind::Domain(type_) + } else if elem_oid != 0 { + let type_ = get_type_rec(client, elem_oid).await?; + Kind::Array(type_) + } else if relid != 0 { + let fields = get_composite_fields(client, relid).await?; + Kind::Composite(fields) + } else if let Some(rngsubtype) = rngsubtype { + let type_ = get_type_rec(client, rngsubtype).await?; + Kind::Range(type_) + } else { + Kind::Simple + }; + + let type_ = Type::new(name, oid, kind, schema); + client.set_type(oid, &type_); + + Ok(type_) +} + +fn get_type_rec<'a>( + client: &'a Arc, + oid: Oid, +) -> Pin> + 'a>> { + Box::pin(get_type(client, oid)) +} + +async fn typeinfo_statement(client: &Arc) -> Result { + if let Some(stmt) = client.typeinfo() { + return Ok(stmt); + } + + let stmt = match Box::pin(prepare(client.clone(), TYPEINFO_QUERY, &[])).await { + Ok(stmt) => stmt, + Err(ref e) if e.code() == Some(&SqlState::UNDEFINED_TABLE) => { + Box::pin(prepare(client.clone(), TYPEINFO_FALLBACK_QUERY, &[])).await? + } + Err(e) => return Err(e), + }; + + client.set_typeinfo(&stmt); + Ok(stmt) +} + +async fn get_enum_variants(client: &Arc, oid: Oid) -> Result, Error> { + let stmt = typeinfo_enum_statement(client).await?; + + let params: &[&dyn ToSql] = &[&oid]; + let buf = query::encode(&stmt, params.iter().cloned()); + query::query(client.clone(), stmt, buf) + .await? + .and_then(|row| future::ready(row.try_get(0))) + .try_collect() + .await +} + +async fn typeinfo_enum_statement(client: &Arc) -> Result { + if let Some(stmt) = client.typeinfo_enum() { + return Ok(stmt); + } + + let stmt = match Box::pin(prepare(client.clone(), TYPEINFO_ENUM_QUERY, &[])).await { + Ok(stmt) => stmt, + Err(ref e) if e.code() == Some(&SqlState::UNDEFINED_COLUMN) => { + Box::pin(prepare(client.clone(), TYPEINFO_ENUM_FALLBACK_QUERY, &[])).await? + } + Err(e) => return Err(e), + }; + + client.set_typeinfo_enum(&stmt); + Ok(stmt) +} + +async fn get_composite_fields(client: &Arc, oid: Oid) -> Result, Error> { + let stmt = typeinfo_composite_statement(client).await?; + + let params: &[&dyn ToSql] = &[&oid]; + let buf = query::encode(&stmt, params.iter().cloned()); + let rows = query::query(client.clone(), stmt, buf) + .await? + .try_collect::>() + .await?; + + let mut fields = vec![]; + for row in rows { + let name = row.try_get(0)?; + let oid = row.try_get(1)?; + let type_ = Box::pin(get_type(client, oid)).await?; + fields.push(Field::new(name, type_)); + } + + Ok(fields) +} + +async fn typeinfo_composite_statement(client: &Arc) -> Result { + if let Some(stmt) = client.typeinfo_composite() { + return Ok(stmt); + } + + let stmt = Box::pin(prepare(client.clone(), TYPEINFO_COMPOSITE_QUERY, &[])).await?; + + client.set_typeinfo_composite(&stmt); + Ok(stmt) } diff --git a/tokio-postgres/src/query.rs b/tokio-postgres/src/query.rs index 4da69197..e9702f91 100644 --- a/tokio-postgres/src/query.rs +++ b/tokio-postgres/src/query.rs @@ -11,33 +11,21 @@ use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; -pub async fn query<'a, I>( +pub async fn query( client: Arc, - statement: &Statement, - params: I, -) -> Result -where - I: IntoIterator, - I::IntoIter: ExactSizeIterator, -{ - let responses = start(&client, &statement, params).await?; + statement: Statement, + buf: Result, Error>, +) -> Result { + let responses = start(client, buf).await?; Ok(Query { - statement: statement.clone(), + statement, responses, }) } -pub async fn execute<'a, I>( - client: Arc, - statement: &Statement, - params: I, -) -> Result -where - I: IntoIterator, - I::IntoIter: ExactSizeIterator, -{ - let mut responses = start(&client, &statement, params).await?; +pub async fn execute(client: Arc, buf: Result, Error>) -> Result { + let mut responses = start(client, buf).await?; loop { match responses.next().await? { @@ -59,11 +47,19 @@ where } } -async fn start<'a, I>( - client: &Arc, - statement: &Statement, - params: I, -) -> Result +async fn start(client: Arc, buf: Result, Error>) -> Result { + let buf = buf?; + let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?; + + match responses.next().await? { + Message::BindComplete => {} + _ => return Err(Error::unexpected_message()), + } + + Ok(responses) +} + +pub fn encode<'a, I>(statement: &Statement, params: I) -> Result, Error> where I: IntoIterator, I::IntoIter: ExactSizeIterator, @@ -105,14 +101,7 @@ where frontend::execute("", 0, &mut buf).map_err(Error::encode)?; frontend::sync(&mut buf); - let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?; - - match responses.next().await? { - Message::BindComplete => {} - _ => return Err(Error::unexpected_message()), - } - - Ok(responses) + Ok(buf) } pub struct Query { diff --git a/tokio-postgres/src/types/mod.rs b/tokio-postgres/src/types/mod.rs index 9cf5a1cd..baa98a8a 100644 --- a/tokio-postgres/src/types/mod.rs +++ b/tokio-postgres/src/types/mod.rs @@ -107,7 +107,7 @@ impl fmt::Display for Type { } impl Type { - pub(crate) fn _new(name: String, oid: Oid, kind: Kind, schema: String) -> Type { + pub(crate) fn new(name: String, oid: Oid, kind: Kind, schema: String) -> Type { Type(Inner::Other(Arc::new(Other { name, oid, @@ -175,6 +175,10 @@ pub struct Field { } impl Field { + pub(crate) fn new(name: String, type_: Type) -> Field { + Field { name, type_ } + } + /// Returns the name of the field. pub fn name(&self) -> &str { &self.name @@ -186,12 +190,6 @@ impl Field { } } -impl Field { - pub(crate) fn new(name: String, type_: Type) -> Field { - Field { name, type_ } - } -} - /// An error indicating that a `NULL` Postgres value was passed to a `FromSql` /// implementation that does not support `NULL` values. #[derive(Debug, Clone, Copy)] @@ -543,7 +541,7 @@ pub enum IsNull { /// /// `ToSql` is implemented for `Vec` and `&[T]` where `T` implements `ToSql`, /// and corresponds to one-dimensional Postgres arrays with an index offset of 1. -pub trait ToSql: fmt::Debug + Sync + Send { +pub trait ToSql: fmt::Debug { /// Converts the value of `self` into the binary format of the specified /// Postgres `Type`, appending it to `out`. /// @@ -744,7 +742,7 @@ simple_to!(f64, float8_to_sql, FLOAT8); impl ToSql for HashMap, H> where - H: BuildHasher + Sync + Send, + H: BuildHasher, { fn to_sql(&self, _: &Type, w: &mut Vec) -> Result> { types::hstore_to_sql( diff --git a/tokio-postgres/src/types/serde_json_1.rs b/tokio-postgres/src/types/serde_json_1.rs index 5ee098be..b3b2e3c9 100644 --- a/tokio-postgres/src/types/serde_json_1.rs +++ b/tokio-postgres/src/types/serde_json_1.rs @@ -33,7 +33,7 @@ where impl ToSql for Json where - T: Serialize + Debug + Sync + Send, + T: Serialize + Debug, { fn to_sql(&self, ty: &Type, out: &mut Vec) -> Result> { if *ty == Type::JSONB { diff --git a/tokio-postgres/tests/test/main.rs b/tokio-postgres/tests/test/main.rs index 5d27a43d..7bc39876 100644 --- a/tokio-postgres/tests/test/main.rs +++ b/tokio-postgres/tests/test/main.rs @@ -1,19 +1,17 @@ #![warn(rust_2018_idioms)] #![feature(async_await)] -use futures::{try_join, FutureExt, StreamExt, TryFutureExt, TryStreamExt}; +use futures::{try_join, FutureExt, TryFutureExt, TryStreamExt}; use tokio::net::TcpStream; use tokio_postgres::error::SqlState; use tokio_postgres::tls::{NoTls, NoTlsStream}; -use tokio_postgres::types::Type; +use tokio_postgres::types::{Kind, Type}; use tokio_postgres::{Client, Config, Connection, Error}; mod parse; #[cfg(feature = "runtime")] mod runtime; -/* mod types; -*/ async fn connect_raw(s: &str) -> Result<(Client, Connection), Error> { let socket = TcpStream::connect(&"127.0.0.1:5433".parse().unwrap()) @@ -100,13 +98,13 @@ async fn scram_password_ok() { async fn pipelined_prepare() { let mut client = connect("user=postgres").await; - let prepare1 = client.prepare("SELECT $1::TEXT"); + let prepare1 = client.prepare("SELECT $1::HSTORE[]"); let prepare2 = client.prepare("SELECT $1::BIGINT"); let (statement1, statement2) = try_join!(prepare1, prepare2).unwrap(); - assert_eq!(statement1.params()[0], Type::TEXT); - assert_eq!(statement1.columns()[0].type_(), &Type::TEXT); + assert_eq!(statement1.params()[0].name(), "_hstore"); + assert_eq!(statement1.columns()[0].type_().name(), "_hstore"); assert_eq!(statement2.params()[0], Type::INT8); assert_eq!(statement2.columns()[0].type_(), &Type::INT8); @@ -140,6 +138,125 @@ async fn insert_select() { assert_eq!(rows[1].get::<_, &str>(1), "bob"); } +#[tokio::test] +async fn custom_enum() { + let mut client = connect("user=postgres").await; + + let create = client + .prepare( + "CREATE TYPE pg_temp.mood AS ENUM ( + 'sad', + 'ok', + 'happy' + )", + ) + .await + .unwrap(); + client.execute(&create, &[]).await.unwrap(); + + let select = client.prepare("SELECT $1::mood").await.unwrap(); + + let ty = &select.params()[0]; + assert_eq!("mood", ty.name()); + assert_eq!( + &Kind::Enum(vec![ + "sad".to_string(), + "ok".to_string(), + "happy".to_string(), + ]), + ty.kind(), + ); +} + +#[tokio::test] +async fn custom_domain() { + let mut client = connect("user=postgres").await; + + let create = client + .prepare("CREATE DOMAIN pg_temp.session_id AS bytea CHECK(octet_length(VALUE) = 16)") + .await + .unwrap(); + client.execute(&create, &[]).await.unwrap(); + + let select = client.prepare("SELECT $1::session_id").await.unwrap(); + + let ty = &select.params()[0]; + assert_eq!("session_id", ty.name()); + assert_eq!(&Kind::Domain(Type::BYTEA), ty.kind()); +} + +#[tokio::test] +async fn custom_array() { + let mut client = connect("user=postgres").await; + + let select = client.prepare("SELECT $1::HSTORE[]").await.unwrap(); + + let ty = &select.params()[0]; + assert_eq!("_hstore", ty.name()); + match ty.kind() { + Kind::Array(ty) => { + assert_eq!("hstore", ty.name()); + assert_eq!(&Kind::Simple, ty.kind()); + } + _ => panic!("unexpected kind"), + } +} + +#[tokio::test] +async fn custom_composite() { + let mut client = connect("user=postgres").await; + + let create = client + .prepare( + "CREATE TYPE pg_temp.inventory_item AS ( + name TEXT, + supplier INTEGER, + price NUMERIC + )", + ) + .await + .unwrap(); + client.execute(&create, &[]).await.unwrap(); + + let select = client.prepare("SELECT $1::inventory_item").await.unwrap(); + + let ty = &select.params()[0]; + assert_eq!(ty.name(), "inventory_item"); + match ty.kind() { + Kind::Composite(fields) => { + assert_eq!(fields[0].name(), "name"); + assert_eq!(fields[0].type_(), &Type::TEXT); + assert_eq!(fields[1].name(), "supplier"); + assert_eq!(fields[1].type_(), &Type::INT4); + assert_eq!(fields[2].name(), "price"); + assert_eq!(fields[2].type_(), &Type::NUMERIC); + } + _ => panic!("unexpected kind"), + } +} + +#[tokio::test] +async fn custom_range() { + let mut client = connect("user=postgres").await; + + let create = client + .prepare( + "CREATE TYPE pg_temp.floatrange AS RANGE ( + subtype = float8, + subtype_diff = float8mi + )", + ) + .await + .unwrap(); + client.execute(&create, &[]).await.unwrap(); + + let select = client.prepare("SELECT $1::floatrange").await.unwrap(); + + let ty = &select.params()[0]; + assert_eq!("floatrange", ty.name()); + assert_eq!(&Kind::Range(Type::FLOAT8), ty.kind()); +} + /* #[test] fn query_portal() { @@ -219,182 +336,6 @@ fn cancel_query_raw() { let ((), ()) = runtime.block_on(sleep.join(cancel)).unwrap(); } -#[test] -fn custom_enum() { - let _ = env_logger::try_init(); - let mut runtime = Runtime::new().unwrap(); - - let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.handle().spawn(connection).unwrap(); - - runtime - .block_on( - client - .simple_query( - "CREATE TYPE pg_temp.mood AS ENUM ( - 'sad', - 'ok', - 'happy' - )", - ) - .for_each(|_| Ok(())), - ) - .unwrap(); - - let select = client.prepare("SELECT $1::mood"); - let select = runtime.block_on(select).unwrap(); - - let ty = &select.params()[0]; - assert_eq!("mood", ty.name()); - assert_eq!( - &Kind::Enum(vec![ - "sad".to_string(), - "ok".to_string(), - "happy".to_string(), - ]), - ty.kind() - ); -} - -#[test] -fn custom_domain() { - let _ = env_logger::try_init(); - let mut runtime = Runtime::new().unwrap(); - - let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.handle().spawn(connection).unwrap(); - - runtime - .block_on( - client - .simple_query( - "CREATE DOMAIN pg_temp.session_id AS bytea CHECK(octet_length(VALUE) = 16)", - ) - .for_each(|_| Ok(())), - ) - .unwrap(); - - let select = client.prepare("SELECT $1::session_id"); - let select = runtime.block_on(select).unwrap(); - - let ty = &select.params()[0]; - assert_eq!("session_id", ty.name()); - assert_eq!(&Kind::Domain(Type::BYTEA), ty.kind()); -} - -#[test] -fn custom_array() { - let _ = env_logger::try_init(); - let mut runtime = Runtime::new().unwrap(); - - let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.handle().spawn(connection).unwrap(); - - let select = client.prepare("SELECT $1::HSTORE[]"); - let select = runtime.block_on(select).unwrap(); - - let ty = &select.params()[0]; - assert_eq!("_hstore", ty.name()); - match *ty.kind() { - Kind::Array(ref ty) => { - assert_eq!("hstore", ty.name()); - assert_eq!(&Kind::Simple, ty.kind()); - } - _ => panic!("unexpected kind"), - } -} - -#[test] -fn custom_composite() { - let _ = env_logger::try_init(); - let mut runtime = Runtime::new().unwrap(); - - let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.handle().spawn(connection).unwrap(); - - runtime - .block_on( - client - .simple_query( - "CREATE TYPE pg_temp.inventory_item AS ( - name TEXT, - supplier INTEGER, - price NUMERIC - )", - ) - .for_each(|_| Ok(())), - ) - .unwrap(); - - let select = client.prepare("SELECT $1::inventory_item"); - let select = runtime.block_on(select).unwrap(); - - let ty = &select.params()[0]; - assert_eq!(ty.name(), "inventory_item"); - match *ty.kind() { - Kind::Composite(ref fields) => { - assert_eq!(fields[0].name(), "name"); - assert_eq!(fields[0].type_(), &Type::TEXT); - assert_eq!(fields[1].name(), "supplier"); - assert_eq!(fields[1].type_(), &Type::INT4); - assert_eq!(fields[2].name(), "price"); - assert_eq!(fields[2].type_(), &Type::NUMERIC); - } - ref t => panic!("bad type {:?}", t), - } -} - -#[test] -fn custom_range() { - let _ = env_logger::try_init(); - let mut runtime = Runtime::new().unwrap(); - - let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.handle().spawn(connection).unwrap(); - - runtime - .block_on( - client - .simple_query( - "CREATE TYPE pg_temp.floatrange AS RANGE ( - subtype = float8, - subtype_diff = float8mi - )", - ) - .for_each(|_| Ok(())), - ) - .unwrap(); - - let select = client.prepare("SELECT $1::floatrange"); - let select = runtime.block_on(select).unwrap(); - - let ty = &select.params()[0]; - assert_eq!("floatrange", ty.name()); - assert_eq!(&Kind::Range(Type::FLOAT8), ty.kind()); -} - -#[test] -fn custom_simple() { - let _ = env_logger::try_init(); - let mut runtime = Runtime::new().unwrap(); - - let (mut client, connection) = runtime.block_on(connect("user=postgres")).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.handle().spawn(connection).unwrap(); - - let select = client.prepare("SELECT $1::HSTORE"); - let select = runtime.block_on(select).unwrap(); - - let ty = &select.params()[0]; - assert_eq!("hstore", ty.name()); - assert_eq!(&Kind::Simple, ty.kind()); -} - #[test] fn notifications() { let _ = env_logger::try_init(); diff --git a/tokio-postgres/tests/test/runtime.rs b/tokio-postgres/tests/test/runtime.rs index b9939e51..b66f81fc 100644 --- a/tokio-postgres/tests/test/runtime.rs +++ b/tokio-postgres/tests/test/runtime.rs @@ -1,6 +1,5 @@ -use futures::{Future, FutureExt, Stream}; +use futures::{FutureExt, TryStreamExt}; use std::time::{Duration, Instant}; -use tokio::runtime::current_thread::Runtime; use tokio::timer::Delay; use tokio_postgres::error::SqlState; use tokio_postgres::NoTls; @@ -10,10 +9,9 @@ async fn smoke_test(s: &str) { let connection = connection.map(|e| e.unwrap()); tokio::spawn(connection); - /* - let execute = client.simple_query("SELECT 1").for_each(|_| Ok(())); - runtime.block_on(execute).unwrap(); - */ + let stmt = client.prepare("SELECT $1::INT").await.unwrap(); + let rows = client.query(&stmt, &[&1i32]).await.unwrap().try_collect::>().await.unwrap(); + assert_eq!(rows[0].get::<_, i32>(0), 1i32); } #[tokio::test] diff --git a/tokio-postgres/tests/test/types/bit_vec_07.rs b/tokio-postgres/tests/test/types/bit_vec_07.rs index 3ab29ee7..4c0e426f 100644 --- a/tokio-postgres/tests/test/types/bit_vec_07.rs +++ b/tokio-postgres/tests/test/types/bit_vec_07.rs @@ -2,28 +2,30 @@ use bit_vec_07::BitVec; use crate::types::test_type; -#[test] -fn test_bit_params() { +#[tokio::test] +async fn test_bit_params() { let mut bv = BitVec::from_bytes(&[0b0110_1001, 0b0000_0111]); bv.pop(); bv.pop(); test_type( "BIT(14)", - &[(Some(bv), "B'01101001000001'"), (None, "NULL")], + vec![(Some(bv), "B'01101001000001'"), (None, "NULL")], ) + .await } -#[test] -fn test_varbit_params() { +#[tokio::test] +async fn test_varbit_params() { let mut bv = BitVec::from_bytes(&[0b0110_1001, 0b0000_0111]); bv.pop(); bv.pop(); test_type( "VARBIT", - &[ + vec![ (Some(bv), "B'01101001000001'"), (Some(BitVec::from_bytes(&[])), "B''"), (None, "NULL"), ], ) + .await } diff --git a/tokio-postgres/tests/test/types/chrono_04.rs b/tokio-postgres/tests/test/types/chrono_04.rs index eb001806..e36aeafa 100644 --- a/tokio-postgres/tests/test/types/chrono_04.rs +++ b/tokio-postgres/tests/test/types/chrono_04.rs @@ -3,8 +3,8 @@ use tokio_postgres::types::{Date, Timestamp}; use crate::types::test_type; -#[test] -fn test_naive_date_time_params() { +#[tokio::test] +async fn test_naive_date_time_params() { fn make_check(time: &str) -> (Option, &str) { ( Some(NaiveDateTime::parse_from_str(time, "'%Y-%m-%d %H:%M:%S.%f'").unwrap()), @@ -13,17 +13,17 @@ fn test_naive_date_time_params() { } test_type( "TIMESTAMP", - &[ + vec![ make_check("'1970-01-01 00:00:00.010000000'"), make_check("'1965-09-25 11:19:33.100314000'"), make_check("'2010-02-09 23:11:45.120200000'"), (None, "NULL"), ], - ); + ).await; } -#[test] -fn test_with_special_naive_date_time_params() { +#[tokio::test] +async fn test_with_special_naive_date_time_params() { fn make_check(time: &str) -> (Timestamp, &str) { ( Timestamp::Value( @@ -34,18 +34,19 @@ fn test_with_special_naive_date_time_params() { } test_type( "TIMESTAMP", - &[ + vec![ make_check("'1970-01-01 00:00:00.010000000'"), make_check("'1965-09-25 11:19:33.100314000'"), make_check("'2010-02-09 23:11:45.120200000'"), (Timestamp::PosInfinity, "'infinity'"), (Timestamp::NegInfinity, "'-infinity'"), ], - ); + ) + .await; } -#[test] -fn test_date_time_params() { +#[tokio::test] +async fn test_date_time_params() { fn make_check(time: &str) -> (Option>, &str) { ( Some( @@ -57,17 +58,18 @@ fn test_date_time_params() { } test_type( "TIMESTAMP WITH TIME ZONE", - &[ + vec![ make_check("'1970-01-01 00:00:00.010000000'"), make_check("'1965-09-25 11:19:33.100314000'"), make_check("'2010-02-09 23:11:45.120200000'"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_with_special_date_time_params() { +#[tokio::test] +async fn test_with_special_date_time_params() { fn make_check(time: &str) -> (Timestamp>, &str) { ( Timestamp::Value( @@ -79,18 +81,19 @@ fn test_with_special_date_time_params() { } test_type( "TIMESTAMP WITH TIME ZONE", - &[ + vec![ make_check("'1970-01-01 00:00:00.010000000'"), make_check("'1965-09-25 11:19:33.100314000'"), make_check("'2010-02-09 23:11:45.120200000'"), (Timestamp::PosInfinity, "'infinity'"), (Timestamp::NegInfinity, "'-infinity'"), ], - ); + ) + .await; } -#[test] -fn test_date_params() { +#[tokio::test] +async fn test_date_params() { fn make_check(time: &str) -> (Option, &str) { ( Some(NaiveDate::parse_from_str(time, "'%Y-%m-%d'").unwrap()), @@ -99,17 +102,18 @@ fn test_date_params() { } test_type( "DATE", - &[ + vec![ make_check("'1970-01-01'"), make_check("'1965-09-25'"), make_check("'2010-02-09'"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_with_special_date_params() { +#[tokio::test] +async fn test_with_special_date_params() { fn make_check(date: &str) -> (Date, &str) { ( Date::Value(NaiveDate::parse_from_str(date, "'%Y-%m-%d'").unwrap()), @@ -118,18 +122,19 @@ fn test_with_special_date_params() { } test_type( "DATE", - &[ + vec![ make_check("'1970-01-01'"), make_check("'1965-09-25'"), make_check("'2010-02-09'"), (Date::PosInfinity, "'infinity'"), (Date::NegInfinity, "'-infinity'"), ], - ); + ) + .await; } -#[test] -fn test_time_params() { +#[tokio::test] +async fn test_time_params() { fn make_check(time: &str) -> (Option, &str) { ( Some(NaiveTime::parse_from_str(time, "'%H:%M:%S.%f'").unwrap()), @@ -138,11 +143,12 @@ fn test_time_params() { } test_type( "TIME", - &[ + vec![ make_check("'00:00:00.010000000'"), make_check("'11:19:33.100314000'"), make_check("'23:11:45.120200000'"), (None, "NULL"), ], - ); + ) + .await; } diff --git a/tokio-postgres/tests/test/types/eui48_04.rs b/tokio-postgres/tests/test/types/eui48_04.rs index a881e24f..0cfb7cb5 100644 --- a/tokio-postgres/tests/test/types/eui48_04.rs +++ b/tokio-postgres/tests/test/types/eui48_04.rs @@ -2,11 +2,11 @@ use eui48_04::MacAddress; use crate::types::test_type; -#[test] -fn test_eui48_params() { +#[tokio::test] +async fn test_eui48_params() { test_type( "MACADDR", - &[ + vec![ ( Some(MacAddress::parse_str("12-34-56-AB-CD-EF").unwrap()), "'12-34-56-ab-cd-ef'", @@ -14,4 +14,5 @@ fn test_eui48_params() { (None, "NULL"), ], ) + .await } diff --git a/tokio-postgres/tests/test/types/geo_010.rs b/tokio-postgres/tests/test/types/geo_010.rs index a782a736..ecb1cbfc 100644 --- a/tokio-postgres/tests/test/types/geo_010.rs +++ b/tokio-postgres/tests/test/types/geo_010.rs @@ -2,23 +2,24 @@ use geo_010::{Coordinate, LineString, Point, Rect}; use crate::types::test_type; -#[test] -fn test_point_params() { +#[tokio::test] +async fn test_point_params() { test_type( "POINT", - &[ + vec![ (Some(Point::new(0.0, 0.0)), "POINT(0, 0)"), (Some(Point::new(-3.14, 1.618)), "POINT(-3.14, 1.618)"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_box_params() { +#[tokio::test] +async fn test_box_params() { test_type( "BOX", - &[ + vec![ ( Some(Rect { min: Coordinate { x: -3.14, y: 1.618 }, @@ -31,11 +32,12 @@ fn test_box_params() { ), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_path_params() { +#[tokio::test] +async fn test_path_params() { let points = vec![ Coordinate { x: 0., y: 0. }, Coordinate { x: -3.14, y: 1.618 }, @@ -46,12 +48,13 @@ fn test_path_params() { ]; test_type( "PATH", - &[ + vec![ ( Some(LineString(points)), "path '((0, 0), (-3.14, 1.618), (160.0, 69701.5615))'", ), (None, "NULL"), ], - ); + ) + .await; } diff --git a/tokio-postgres/tests/test/types/mod.rs b/tokio-postgres/tests/test/types/mod.rs index 18858d56..8332a959 100644 --- a/tokio-postgres/tests/test/types/mod.rs +++ b/tokio-postgres/tests/test/types/mod.rs @@ -1,4 +1,4 @@ -use futures::{Future, Stream}; +use futures::{TryStreamExt}; use std::collections::HashMap; use std::error::Error; use std::f32; @@ -7,7 +7,6 @@ use std::fmt; use std::net::IpAddr; use std::result; use std::time::{Duration, UNIX_EPOCH}; -use tokio::runtime::current_thread::Runtime; use tokio_postgres::to_sql_checked; use tokio_postgres::types::{FromSql, FromSqlOwned, IsNull, Kind, ToSql, Type, WrongType}; @@ -26,53 +25,63 @@ mod serde_json_1; #[cfg(feature = "with-uuid-0_7")] mod uuid_07; -fn test_type(sql_type: &str, checks: &[(T, S)]) +async fn test_type(sql_type: &str, checks: Vec<(T, S)>) where T: PartialEq + for<'a> FromSqlOwned + ToSql, S: fmt::Display, { - let mut runtime = Runtime::new().unwrap(); + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - for &(ref val, ref repr) in checks.iter() { - let prepare = client.prepare(&format!("SELECT {}::{}", repr, sql_type)); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[]).collect(); - let rows = runtime.block_on(query).unwrap(); + for (val, repr) in checks { + let stmt = client + .prepare(&format!("SELECT {}::{}", repr, sql_type)) + .await + .unwrap(); + let rows = client + .query(&stmt, &[]) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); let result = rows[0].get(0); - assert_eq!(val, &result); + assert_eq!(val, result); - let prepare = client.prepare(&format!("SELECT $1::{}", sql_type)); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[val]).collect(); - let rows = runtime.block_on(query).unwrap(); + let stmt = client + .prepare(&format!("SELECT $1::{}", sql_type)) + .await + .unwrap(); + let rows = client + .query(&stmt, &[&val]) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); let result = rows[0].get(0); - assert_eq!(val, &result); + assert_eq!(val, result); } } -#[test] -fn test_bool_params() { +#[tokio::test] +async fn test_bool_params() { test_type( "BOOL", - &[(Some(true), "'t'"), (Some(false), "'f'"), (None, "NULL")], - ); + vec![(Some(true), "'t'"), (Some(false), "'f'"), (None, "NULL")], + ) + .await; } -#[test] -fn test_i8_params() { - test_type("\"char\"", &[(Some('a' as i8), "'a'"), (None, "NULL")]); +#[tokio::test] +async fn test_i8_params() { + test_type("\"char\"", vec![(Some('a' as i8), "'a'"), (None, "NULL")]).await; } -#[test] -fn test_name_params() { +#[tokio::test] +async fn test_name_params() { test_type( "NAME", - &[ + vec![ (Some("hello world".to_owned()), "'hello world'"), ( Some("イロハニホヘト チリヌルヲ".to_owned()), @@ -80,88 +89,95 @@ fn test_name_params() { ), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_i16_params() { +#[tokio::test] +async fn test_i16_params() { test_type( "SMALLINT", - &[ + vec![ (Some(15001i16), "15001"), (Some(-15001i16), "-15001"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_i32_params() { +#[tokio::test] +async fn test_i32_params() { test_type( "INT", - &[ + vec![ (Some(2_147_483_548i32), "2147483548"), (Some(-2_147_483_548i32), "-2147483548"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_oid_params() { +#[tokio::test] +async fn test_oid_params() { test_type( "OID", - &[ + vec![ (Some(2_147_483_548u32), "2147483548"), (Some(4_000_000_000), "4000000000"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_i64_params() { +#[tokio::test] +async fn test_i64_params() { test_type( "BIGINT", - &[ + vec![ (Some(9_223_372_036_854_775_708i64), "9223372036854775708"), (Some(-9_223_372_036_854_775_708i64), "-9223372036854775708"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_f32_params() { +#[tokio::test] +async fn test_f32_params() { test_type( "REAL", - &[ + vec![ (Some(f32::INFINITY), "'infinity'"), (Some(f32::NEG_INFINITY), "'-infinity'"), (Some(1000.55), "1000.55"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_f64_params() { +#[tokio::test] +async fn test_f64_params() { test_type( "DOUBLE PRECISION", - &[ + vec![ (Some(f64::INFINITY), "'infinity'"), (Some(f64::NEG_INFINITY), "'-infinity'"), (Some(10000.55), "10000.55"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_varchar_params() { +#[tokio::test] +async fn test_varchar_params() { test_type( "VARCHAR", - &[ + vec![ (Some("hello world".to_owned()), "'hello world'"), ( Some("イロハニホヘト チリヌルヲ".to_owned()), @@ -169,14 +185,15 @@ fn test_varchar_params() { ), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_text_params() { +#[tokio::test] +async fn test_text_params() { test_type( "TEXT", - &[ + vec![ (Some("hello world".to_owned()), "'hello world'"), ( Some("イロハニホヘト チリヌルヲ".to_owned()), @@ -184,138 +201,149 @@ fn test_text_params() { ), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_borrowed_text() { - let mut runtime = Runtime::new().unwrap(); +#[tokio::test] +async fn test_borrowed_text() { + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let prepare = client.prepare("SELECT 'foo'"); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[]).collect(); - let rows = runtime.block_on(query).unwrap(); + let stmt = client.prepare("SELECT 'foo'").await.unwrap(); + let rows = client + .query(&stmt, &[]) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); let s: &str = rows[0].get(0); assert_eq!(s, "foo"); } -#[test] -fn test_bpchar_params() { - let mut runtime = Runtime::new().unwrap(); +#[tokio::test] +async fn test_bpchar_params() { + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let batch = client - .simple_query( + let stmt = client + .prepare( "CREATE TEMPORARY TABLE foo ( id SERIAL PRIMARY KEY, b CHAR(5) )", ) - .for_each(|_| Ok(())); - runtime.block_on(batch).unwrap(); + .await + .unwrap(); + client.execute(&stmt, &[]).await.unwrap(); - let prepare = client.prepare("INSERT INTO foo (b) VALUES ($1), ($2), ($3)"); - let stmt = runtime.block_on(prepare).unwrap(); - let execute = client.execute(&stmt, &[&"12345", &"123", &None::<&'static str>]); - runtime.block_on(execute).unwrap(); + let stmt = client + .prepare("INSERT INTO foo (b) VALUES ($1), ($2), ($3)") + .await + .unwrap(); + client + .execute(&stmt, &[&"12345", &"123", &None::<&'static str>]) + .await + .unwrap(); - let prepare = client.prepare("SELECT b FROM foo ORDER BY id"); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[]).collect(); - let res = runtime.block_on(query).unwrap(); + let stmt = client + .prepare("SELECT b FROM foo ORDER BY id") + .await + .unwrap(); + let rows = client + .query(&stmt, &[]) + .await + .unwrap() + .map_ok(|row| row.get(0)) + .try_collect::>>() + .await + .unwrap(); assert_eq!( vec![Some("12345".to_owned()), Some("123 ".to_owned()), None], - res.iter().map(|row| row.get(0)).collect::>() + rows, ); } -#[test] -fn test_citext_params() { - let mut runtime = Runtime::new().unwrap(); +#[tokio::test] +async fn test_citext_params() { + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let batch = client - .simple_query( + let stmt = client + .prepare( "CREATE TEMPORARY TABLE foo ( id SERIAL PRIMARY KEY, b CITEXT )", ) - .for_each(|_| Ok(())); - runtime.block_on(batch).unwrap(); + .await + .unwrap(); + client.execute(&stmt, &[]).await.unwrap(); - let prepare = client.prepare("INSERT INTO foo (b) VALUES ($1), ($2), ($3)"); - let stmt = runtime.block_on(prepare).unwrap(); - let execute = client.execute(&stmt, &[&"foobar", &"FooBar", &None::<&'static str>]); - runtime.block_on(execute).unwrap(); + let stmt = client + .prepare("INSERT INTO foo (b) VALUES ($1), ($2), ($3)") + .await + .unwrap(); + client + .execute(&stmt, &[&"foobar", &"FooBar", &None::<&'static str>]) + .await + .unwrap(); - let prepare = client.prepare("SELECT b FROM foo WHERE b = 'FOOBAR' ORDER BY id"); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[]).collect(); - let res = runtime.block_on(query).unwrap(); + let stmt = client + .prepare("SELECT b FROM foo WHERE b = 'FOOBAR' ORDER BY id") + .await + .unwrap(); + let rows = client + .query(&stmt, &[]) + .await + .unwrap() + .map_ok(|row| row.get(0)) + .try_collect::>() + .await + .unwrap(); - assert_eq!( - vec!["foobar".to_string(), "FooBar".to_string()], - res.iter() - .map(|row| row.get::<_, String>(0)) - .collect::>() - ); + assert_eq!(vec!["foobar".to_string(), "FooBar".to_string()], rows,); } -#[test] -fn test_bytea_params() { +#[tokio::test] +async fn test_bytea_params() { test_type( "BYTEA", - &[ + vec![ (Some(vec![0u8, 1, 2, 3, 254, 255]), "'\\x00010203feff'"), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_borrowed_bytea() { - let mut runtime = Runtime::new().unwrap(); - - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let prepare = client.prepare("SELECT 'foo'::BYTEA"); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[]).collect(); - let rows = runtime.block_on(query).unwrap(); +#[tokio::test] +async fn test_borrowed_bytea() { + let mut client = connect("user=postgres").await; + let stmt = client.prepare("SELECT 'foo'::BYTEA").await.unwrap(); + let rows = client + .query(&stmt, &[]) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); let s: &[u8] = rows[0].get(0); assert_eq!(s, b"foo"); } -#[test] -fn test_hstore_params() { - macro_rules! make_map { - ($($k:expr => $v:expr),+) => ({ - let mut map = HashMap::new(); - $(map.insert($k, $v);)+ - map - }) - } +macro_rules! make_map { + ($($k:expr => $v:expr),+) => ({ + let mut map = HashMap::new(); + $(map.insert($k, $v);)+ + map + }) +} + +#[tokio::test] +async fn test_hstore_params() { test_type( "hstore", - &[ + vec![ ( Some(make_map!("a".to_owned() => Some("1".to_owned()))), "'a=>1'", @@ -328,149 +356,149 @@ fn test_hstore_params() { ), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn test_array_params() { +#[tokio::test] +async fn test_array_params() { test_type( "integer[]", - &[ + vec![ (Some(vec![1i32, 2i32]), "ARRAY[1,2]"), (Some(vec![1i32]), "ARRAY[1]"), (Some(vec![]), "ARRAY[]"), (None, "NULL"), ], - ); + ) + .await; } #[allow(clippy::eq_op)] -fn test_nan_param(sql_type: &str) +async fn test_nan_param(sql_type: &str) where T: PartialEq + ToSql + FromSqlOwned, { - let mut runtime = Runtime::new().unwrap(); + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let prepare = client.prepare(&format!("SELECT 'NaN'::{}", sql_type)); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[]).collect(); - let rows = runtime.block_on(query).unwrap(); + let stmt = client + .prepare(&format!("SELECT 'NaN'::{}", sql_type)) + .await + .unwrap(); + let rows = client + .query(&stmt, &[]) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); let val: T = rows[0].get(0); assert!(val != val); } -#[test] -fn test_f32_nan_param() { - test_nan_param::("REAL"); +#[tokio::test] +async fn test_f32_nan_param() { + test_nan_param::("REAL").await; } -#[test] -fn test_f64_nan_param() { - test_nan_param::("DOUBLE PRECISION"); +#[tokio::test] +async fn test_f64_nan_param() { + test_nan_param::("DOUBLE PRECISION").await; } -#[test] -fn test_pg_database_datname() { - let mut runtime = Runtime::new().unwrap(); - - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let prepare = client.prepare("SELECT datname FROM pg_database"); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[]).collect(); - let rows = runtime.block_on(query).unwrap(); +#[tokio::test] +async fn test_pg_database_datname() { + let mut client = connect("user=postgres").await; + let stmt = client + .prepare("SELECT datname FROM pg_database") + .await + .unwrap(); + let rows = client + .query(&stmt, &[]) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); assert_eq!(rows[0].get::<_, &str>(0), "postgres"); } -#[test] -fn test_slice() { - let mut runtime = Runtime::new().unwrap(); +#[tokio::test] +async fn test_slice() { + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let batch = client - .simple_query( + let stmt = client + .prepare( "CREATE TEMPORARY TABLE foo ( - id SERIAL PRIMARY KEY, - f TEXT - ); - - INSERT INTO foo(f) VALUES ('a'), ('b'), ('c'), ('d');", + id SERIAL PRIMARY KEY, + f TEXT + )", ) - .for_each(|_| Ok(())); - runtime.block_on(batch).unwrap(); + .await + .unwrap(); + client.execute(&stmt, &[]).await.unwrap(); - let prepare = client.prepare("SELECT f FROM foo WHERE id = ANY($1)"); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client + let stmt = client + .prepare("INSERT INTO foo (f) VALUES ('a'), ('b'), ('c'), ('d')") + .await + .unwrap(); + client.execute(&stmt, &[]).await.unwrap(); + + let stmt = client + .prepare("SELECT f FROM foo WHERE id = ANY($1)") + .await + .unwrap(); + let rows = client .query(&stmt, &[&&[1i32, 3, 4][..]]) - .map(|r| r.get::<_, String>(0)) - .collect(); - let rows = runtime.block_on(query).unwrap(); + .await + .unwrap() + .map_ok(|r| r.get(0)) + .try_collect::>() + .await + .unwrap(); assert_eq!(vec!["a".to_owned(), "c".to_owned(), "d".to_owned()], rows); } -#[test] -fn test_slice_wrong_type() { - let mut runtime = Runtime::new().unwrap(); +#[tokio::test] +async fn test_slice_wrong_type() { + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let batch = client - .simple_query( + let stmt = client + .prepare( "CREATE TEMPORARY TABLE foo ( - id SERIAL PRIMARY KEY - )", + id SERIAL PRIMARY KEY + )", ) - .for_each(|_| Ok(())); - runtime.block_on(batch).unwrap(); + .await + .unwrap(); + client.execute(&stmt, &[]).await.unwrap(); - let prepare = client.prepare("SELECT * FROM foo WHERE id = ANY($1)"); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[&&[&"hi"][..]]).collect(); - let err = runtime.block_on(query).err().unwrap(); + let stmt = client + .prepare("SELECT * FROM foo WHERE id = ANY($1)") + .await + .unwrap(); + let err = client.query(&stmt, &[&&[&"hi"][..]]).await.err().unwrap(); match err.source() { Some(e) if e.is::() => {} _ => panic!("Unexpected error {:?}", err), }; } -#[test] -fn test_slice_range() { - let mut runtime = Runtime::new().unwrap(); +#[tokio::test] +async fn test_slice_range() { + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let prepare = client.prepare("SELECT $1::INT8RANGE"); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[&&[&1i64][..]]).collect(); - let err = runtime.block_on(query).err().unwrap(); + let stmt = client.prepare("SELECT $1::INT8RANGE").await.unwrap(); + let err = client.query(&stmt, &[&&[&1i64][..]]).await.err().unwrap(); match err.source() { Some(e) if e.is::() => {} _ => panic!("Unexpected error {:?}", err), }; } -#[test] -fn domain() { +#[tokio::test] +async fn domain() { #[derive(Debug, PartialEq)] struct SessionId(Vec); @@ -509,57 +537,56 @@ fn domain() { } } - let mut runtime = Runtime::new().unwrap(); + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); + let stmt = client + .prepare("CREATE DOMAIN pg_temp.session_id AS bytea CHECK(octet_length(VALUE) = 16)") + .await + .unwrap(); + client.execute(&stmt, &[]).await.unwrap(); - let batch = client - .simple_query( - "CREATE DOMAIN pg_temp.session_id AS bytea CHECK(octet_length(VALUE) = 16); - CREATE TABLE pg_temp.foo (id pg_temp.session_id);", - ) - .for_each(|_| Ok(())); - runtime.block_on(batch).unwrap(); + let stmt = client + .prepare("CREATE TABLE pg_temp.foo (id pg_temp.session_id)") + .await + .unwrap(); + client.execute(&stmt, &[]).await.unwrap(); let id = SessionId(b"0123456789abcdef".to_vec()); - let prepare = client.prepare("INSERT INTO pg_temp.foo (id) VALUES ($1)"); - let stmt = runtime.block_on(prepare).unwrap(); - let execute = client.execute(&stmt, &[&id]); - runtime.block_on(execute).unwrap(); + let stmt = client + .prepare("INSERT INTO pg_temp.foo (id) VALUES ($1)") + .await + .unwrap(); + client.execute(&stmt, &[&id]).await.unwrap(); - let prepare = client.prepare("SELECT id FROM pg_temp.foo"); - let stmt = runtime.block_on(prepare).unwrap(); - let query = client.query(&stmt, &[]).collect(); - let rows = runtime.block_on(query).unwrap(); + let stmt = client.prepare("SELECT id FROM pg_temp.foo").await.unwrap(); + let rows = client + .query(&stmt, &[]) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); assert_eq!(id, rows[0].get(0)); } -#[test] -fn composite() { - let mut runtime = Runtime::new().unwrap(); +#[tokio::test] +async fn composite() { + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); - - let batch = client - .simple_query( + let stmt = client + .prepare( "CREATE TYPE pg_temp.inventory_item AS ( - name TEXT, - supplier INTEGER, - price NUMERIC - )", + name TEXT, + supplier INTEGER, + price NUMERIC + )", ) - .for_each(|_| Ok(())); - runtime.block_on(batch).unwrap(); + .await + .unwrap(); + client.execute(&stmt, &[]).await.unwrap(); - let prepare = client.prepare("SELECT $1::inventory_item"); - let stmt = runtime.block_on(prepare).unwrap(); + let stmt = client.prepare("SELECT $1::inventory_item").await.unwrap(); let type_ = &stmt.params()[0]; assert_eq!(type_.name(), "inventory_item"); match *type_.kind() { @@ -575,22 +602,17 @@ fn composite() { } } -#[test] -fn enum_() { - let mut runtime = Runtime::new().unwrap(); +#[tokio::test] +async fn enum_() { + let mut client = connect("user=postgres").await; - let handshake = connect("user=postgres"); - let (mut client, connection) = runtime.block_on(handshake).unwrap(); - let connection = connection.map_err(|e| panic!("{}", e)); - runtime.spawn(connection); + let stmt = client + .prepare("CREATE TYPE pg_temp.mood AS ENUM ('sad', 'ok', 'happy')") + .await + .unwrap(); + client.execute(&stmt, &[]).await.unwrap(); - let batch = client - .simple_query("CREATE TYPE pg_temp.mood AS ENUM ('sad', 'ok', 'happy');") - .for_each(|_| Ok(())); - runtime.block_on(batch).unwrap(); - - let prepare = client.prepare("SELECT $1::mood"); - let stmt = runtime.block_on(prepare).unwrap(); + let stmt = client.prepare("SELECT $1::mood").await.unwrap(); let type_ = &stmt.params()[0]; assert_eq!(type_.name(), "mood"); match *type_.kind() { @@ -604,11 +626,11 @@ fn enum_() { } } -#[test] -fn system_time() { +#[tokio::test] +async fn system_time() { test_type( "TIMESTAMP", - &[ + vec![ ( Some(UNIX_EPOCH + Duration::from_millis(1_010)), "'1970-01-01 00:00:01.01'", @@ -623,14 +645,15 @@ fn system_time() { ), (None, "NULL"), ], - ); + ) + .await; } -#[test] -fn inet() { +#[tokio::test] +async fn inet() { test_type( "INET", - &[ + vec![ (Some("127.0.0.1".parse::().unwrap()), "'127.0.0.1'"), ( Some("127.0.0.1".parse::().unwrap()), @@ -653,5 +676,6 @@ fn inet() { "'2001:4f8:3:ba:2e0:81ff:fe22:d1f1/128'", ), ], - ); + ) + .await; } diff --git a/tokio-postgres/tests/test/types/serde_json_1.rs b/tokio-postgres/tests/test/types/serde_json_1.rs index 37d431f6..f48c188f 100644 --- a/tokio-postgres/tests/test/types/serde_json_1.rs +++ b/tokio-postgres/tests/test/types/serde_json_1.rs @@ -2,11 +2,11 @@ use serde_json_1::Value; use crate::types::test_type; -#[test] -fn test_json_params() { +#[tokio::test] +async fn test_json_params() { test_type( "JSON", - &[ + vec![ ( Some(serde_json_1::from_str::("[10, 11, 12]").unwrap()), "'[10, 11, 12]'", @@ -18,13 +18,14 @@ fn test_json_params() { (None, "NULL"), ], ) + .await } -#[test] -fn test_jsonb_params() { +#[tokio::test] +async fn test_jsonb_params() { test_type( "JSONB", - &[ + vec![ ( Some(serde_json_1::from_str::("[10, 11, 12]").unwrap()), "'[10, 11, 12]'", @@ -36,4 +37,5 @@ fn test_jsonb_params() { (None, "NULL"), ], ) + .await } diff --git a/tokio-postgres/tests/test/types/uuid_07.rs b/tokio-postgres/tests/test/types/uuid_07.rs index 1d51a8ad..8e60fe3a 100644 --- a/tokio-postgres/tests/test/types/uuid_07.rs +++ b/tokio-postgres/tests/test/types/uuid_07.rs @@ -2,11 +2,11 @@ use uuid_07::Uuid; use crate::types::test_type; -#[test] -fn test_uuid_params() { +#[tokio::test] +async fn test_uuid_params() { test_type( "UUID", - &[ + vec![ ( Some(Uuid::parse_str("a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11").unwrap()), "'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'", @@ -14,4 +14,5 @@ fn test_uuid_params() { (None, "NULL"), ], ) + .await }