rws_server/test_client/
db.rs1use std::future::{Future, IntoFuture};
54use std::pin::Pin;
55
56use crate::model::{Backend, DbError, DbPool};
57
58pub struct TestDb {
60 pool: DbPool,
61 transactional: bool,
63}
64
65enum Target {
66 #[cfg(feature = "model-sqlite")]
67 Sqlite,
68 #[cfg(feature = "model-postgres")]
69 Postgres(String),
70 #[cfg(feature = "model-mysql")]
71 MySql(String),
72}
73
74#[must_use = "a TestDbBuilder does nothing until it is awaited"]
77pub struct TestDbBuilder {
78 target: Target,
79 migrations: Vec<String>,
80}
81
82impl TestDb {
83 #[cfg(feature = "model-sqlite")]
85 pub fn sqlite() -> TestDbBuilder {
86 TestDbBuilder { target: Target::Sqlite, migrations: Vec::new() }
87 }
88
89 #[cfg(feature = "model-postgres")]
92 pub fn postgres(url: &str) -> TestDbBuilder {
93 TestDbBuilder { target: Target::Postgres(url.to_string()), migrations: Vec::new() }
94 }
95
96 #[cfg(feature = "model-mysql")]
99 pub fn mysql(url: &str) -> TestDbBuilder {
100 TestDbBuilder { target: Target::MySql(url.to_string()), migrations: Vec::new() }
101 }
102
103 #[cfg(feature = "model-postgres")]
120 pub fn postgres_from_env() -> Option<TestDbBuilder> {
121 url_from_env("RWS_TEST_POSTGRES_URL", "postgres://user:password@localhost/app_test").map(|url| TestDb::postgres(&url))
122 }
123
124 #[cfg(feature = "model-mysql")]
127 pub fn mysql_from_env() -> Option<TestDbBuilder> {
128 url_from_env("RWS_TEST_MYSQL_URL", "mysql://user:password@localhost/app_test").map(|url| TestDb::mysql(&url))
129 }
130
131 pub fn pool(&self) -> DbPool {
134 self.pool.clone()
135 }
136
137 pub fn backend(&self) -> Backend {
139 self.pool.backend()
140 }
141
142 pub fn is_transactional(&self) -> bool {
145 self.transactional
146 }
147
148 pub async fn rollback(mut self) -> Result<(), DbError> {
152 let transactional = std::mem::replace(&mut self.transactional, false);
153 if !transactional {
154 return Ok(());
155 }
156 match &self.pool {
157 #[cfg(feature = "model-sqlite")]
158 DbPool::Sqlite(_) => Ok(()),
159 #[cfg(feature = "model-postgres")]
160 DbPool::Postgres(pool) => rollback_and_close(pool).await,
161 #[cfg(feature = "model-mysql")]
162 DbPool::MySql(pool) => rollback_and_close(pool).await,
163 }
164 }
165}
166
167impl std::ops::Deref for TestDb {
168 type Target = DbPool;
169 fn deref(&self) -> &DbPool {
170 &self.pool
171 }
172}
173
174impl Drop for TestDb {
175 fn drop(&mut self) {
176 if !self.transactional {
177 return;
178 }
179 match &self.pool {
180 #[cfg(feature = "model-sqlite")]
181 DbPool::Sqlite(_) => {}
182 #[cfg(feature = "model-postgres")]
183 DbPool::Postgres(pool) => close_soon(pool.clone()),
184 #[cfg(feature = "model-mysql")]
185 DbPool::MySql(pool) => close_soon(pool.clone()),
186 }
187 }
188}
189
190impl TestDbBuilder {
191 pub fn migrations(mut self, dir: &str) -> Self {
195 self.migrations.push(dir.to_string());
196 self
197 }
198
199 pub async fn build(self) -> Result<TestDb, DbError> {
201 match self.target {
202 #[cfg(feature = "model-sqlite")]
203 Target::Sqlite => {
204 let pool = sqlx::pool::PoolOptions::<sqlx::Sqlite>::new()
207 .max_connections(1)
208 .idle_timeout(None)
209 .max_lifetime(None)
210 .connect("sqlite::memory:")
211 .await
212 .map_err(|e| DbError::new(e.to_string()))?;
213 let pool = DbPool::Sqlite(pool);
214 for dir in &self.migrations {
215 pool.migrate(dir).await?;
216 }
217 Ok(TestDb { pool, transactional: false })
218 }
219 #[cfg(feature = "model-postgres")]
220 Target::Postgres(url) => {
221 let _ = crate::crypto_provider().install_default();
222 let options: sqlx::postgres::PgConnectOptions =
223 url.parse().map_err(|e: sqlx::Error| DbError::new(format!("RWS test database URL: {e}")))?;
224 migrate_once(&url, &self.migrations, || async {
225 let pool = sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
226 .max_connections(1)
227 .connect_with(options.clone())
228 .await
229 .map_err(|e| DbError::new(e.to_string()))?;
230 Ok(DbPool::Postgres(pool))
231 })
232 .await?;
233 let pool = transactional_pool::<sqlx::Postgres>()
234 .connect_with(options)
235 .await
236 .map_err(|e| DbError::new(e.to_string()))?;
237 Ok(TestDb { pool: DbPool::Postgres(pool), transactional: true })
238 }
239 #[cfg(feature = "model-mysql")]
240 Target::MySql(url) => {
241 let _ = crate::crypto_provider().install_default();
242 let options: sqlx::mysql::MySqlConnectOptions =
243 url.parse().map_err(|e: sqlx::Error| DbError::new(format!("RWS test database URL: {e}")))?;
244 migrate_once(&url, &self.migrations, || async {
245 let pool = sqlx::pool::PoolOptions::<sqlx::MySql>::new()
246 .max_connections(1)
247 .connect_with(options.clone())
248 .await
249 .map_err(|e| DbError::new(e.to_string()))?;
250 Ok(DbPool::MySql(pool))
251 })
252 .await?;
253 let pool = transactional_pool::<sqlx::MySql>()
254 .connect_with(options)
255 .await
256 .map_err(|e| DbError::new(e.to_string()))?;
257 Ok(TestDb { pool: DbPool::MySql(pool), transactional: true })
258 }
259 }
260 }
261}
262
263impl IntoFuture for TestDbBuilder {
264 type Output = Result<TestDb, DbError>;
265 type IntoFuture = Pin<Box<dyn Future<Output = Result<TestDb, DbError>> + Send>>;
266
267 fn into_future(self) -> Self::IntoFuture {
268 Box::pin(self.build())
269 }
270}
271
272#[cfg(any(feature = "model-postgres", feature = "model-mysql"))]
273fn url_from_env(name: &str, example: &str) -> Option<String> {
274 match std::env::var(name).map(|v| v.trim().to_string()) {
275 Ok(url) if !url.is_empty() => Some(url),
276 _ => {
277 eprintln!("skipped: {name} is not set (the URL of a database for tests, e.g. {example})");
278 None
279 }
280 }
281}
282
283#[cfg(any(feature = "model-postgres", feature = "model-mysql"))]
288fn transactional_pool<DB: sqlx::Database>() -> sqlx::pool::PoolOptions<DB> {
289 sqlx::pool::PoolOptions::<DB>::new()
290 .max_connections(1)
291 .min_connections(0)
292 .idle_timeout(None)
293 .max_lifetime(None)
294 .after_connect(|conn, _meta| <DB::TransactionManager as sqlx::TransactionManager>::begin(conn, None))
295}
296
297#[cfg(any(feature = "model-postgres", feature = "model-mysql"))]
298async fn rollback_and_close<DB: sqlx::Database>(pool: &sqlx::Pool<DB>) -> Result<(), DbError> {
299 let result = async {
300 let mut conn = pool.acquire().await.map_err(|e| DbError::new(e.to_string()))?;
301 <DB::TransactionManager as sqlx::TransactionManager>::rollback(&mut conn).await.map_err(|e| DbError::new(e.to_string()))
302 }
303 .await;
304 pool.close().await;
305 result
306}
307
308#[cfg(any(feature = "model-postgres", feature = "model-mysql"))]
314fn close_soon<DB: sqlx::Database>(pool: sqlx::Pool<DB>) {
315 drop(pool.close());
316 if let Ok(runtime) = tokio::runtime::Handle::try_current() {
317 runtime.spawn(async move { pool.close().await });
318 }
319}
320
321#[cfg(any(feature = "model-postgres", feature = "model-mysql"))]
325async fn migrate_once<F, Fut>(url: &str, dirs: &[String], connect: F) -> Result<(), DbError>
326where
327 F: FnOnce() -> Fut,
328 Fut: Future<Output = Result<DbPool, DbError>>,
329{
330 use std::collections::HashSet;
331 use std::sync::OnceLock;
332 static DONE: OnceLock<tokio::sync::Mutex<HashSet<String>>> = OnceLock::new();
333 if dirs.is_empty() {
334 return Ok(());
335 }
336 let mut done = DONE.get_or_init(Default::default).lock().await;
337 let keys: Vec<String> = dirs.iter().map(|dir| format!("{url}\n{dir}")).collect();
338 if keys.iter().all(|key| done.contains(key)) {
339 return Ok(());
340 }
341 let pool = connect().await?;
342 let mut result = Ok(());
343 for (dir, key) in dirs.iter().zip(keys) {
344 if done.contains(&key) {
345 continue;
346 }
347 if let Err(e) = pool.migrate(dir).await {
348 result = Err(e);
349 break;
350 }
351 done.insert(key);
352 }
353 match &pool {
354 #[cfg(feature = "model-sqlite")]
355 DbPool::Sqlite(p) => p.close().await,
356 #[cfg(feature = "model-postgres")]
357 DbPool::Postgres(p) => p.close().await,
358 #[cfg(feature = "model-mysql")]
359 DbPool::MySql(p) => p.close().await,
360 }
361 result
362}