use crate::{
debug_print, error::*, DatabaseConnection, DbBackend, ExecResult, MockDatabase, QueryResult,
Statement, Transaction,
};
use futures::Stream;
use std::{
fmt::Debug,
pin::Pin,
sync::{
atomic::{AtomicUsize, Ordering},
Arc, Mutex,
},
};
use tracing::instrument;
#[derive(Debug)]
pub struct MockDatabaseConnector;
#[derive(Debug)]
pub struct MockDatabaseConnection {
execute_counter: AtomicUsize,
query_counter: AtomicUsize,
mocker: Mutex<Box<dyn MockDatabaseTrait>>,
}
pub trait MockDatabaseTrait: Send + Debug {
fn execute(&mut self, counter: usize, stmt: Statement) -> Result<ExecResult, DbErr>;
fn query(&mut self, counter: usize, stmt: Statement) -> Result<Vec<QueryResult>, DbErr>;
fn begin(&mut self);
fn commit(&mut self);
fn rollback(&mut self);
fn drain_transaction_log(&mut self) -> Vec<Transaction>;
fn get_database_backend(&self) -> DbBackend;
fn ping(&self) -> Result<(), DbErr>;
}
impl MockDatabaseConnector {
#[allow(unused_variables)]
pub fn accepts(string: &str) -> bool {
#[cfg(feature = "sqlx-mysql")]
if DbBackend::MySql.is_prefix_of(string) {
return true;
}
#[cfg(feature = "sqlx-postgres")]
if DbBackend::Postgres.is_prefix_of(string) {
return true;
}
#[cfg(feature = "sqlx-sqlite")]
if DbBackend::Sqlite.is_prefix_of(string) {
return true;
}
false
}
#[allow(unused_variables)]
#[instrument(level = "trace")]
pub async fn connect(string: &str) -> Result<DatabaseConnection, DbErr> {
macro_rules! connect_mock_db {
( $syntax: expr ) => {
Ok(DatabaseConnection::MockDatabaseConnection(Arc::new(
MockDatabaseConnection::new(MockDatabase::new($syntax)),
)))
};
}
#[cfg(feature = "sqlx-mysql")]
if crate::SqlxMySqlConnector::accepts(string) {
return connect_mock_db!(DbBackend::MySql);
}
#[cfg(feature = "sqlx-postgres")]
if crate::SqlxPostgresConnector::accepts(string) {
return connect_mock_db!(DbBackend::Postgres);
}
#[cfg(feature = "sqlx-sqlite")]
if crate::SqlxSqliteConnector::accepts(string) {
return connect_mock_db!(DbBackend::Sqlite);
}
connect_mock_db!(DbBackend::Postgres)
}
}
impl MockDatabaseConnection {
pub fn new<M: 'static>(m: M) -> Self
where
M: MockDatabaseTrait,
{
Self {
execute_counter: AtomicUsize::new(0),
query_counter: AtomicUsize::new(0),
mocker: Mutex::new(Box::new(m)),
}
}
pub(crate) fn get_mocker_mutex(&self) -> &Mutex<Box<dyn MockDatabaseTrait>> {
&self.mocker
}
pub fn get_database_backend(&self) -> DbBackend {
self.mocker
.lock()
.expect("Fail to acquire mocker")
.get_database_backend()
}
#[instrument(level = "trace")]
pub fn execute(&self, statement: Statement) -> Result<ExecResult, DbErr> {
debug_print!("{}", statement);
let counter = self.execute_counter.fetch_add(1, Ordering::SeqCst);
self.mocker
.lock()
.map_err(exec_err)?
.execute(counter, statement)
}
#[instrument(level = "trace")]
pub fn query_one(&self, statement: Statement) -> Result<Option<QueryResult>, DbErr> {
debug_print!("{}", statement);
let counter = self.query_counter.fetch_add(1, Ordering::SeqCst);
let result = self
.mocker
.lock()
.map_err(query_err)?
.query(counter, statement)?;
Ok(result.into_iter().next())
}
#[instrument(level = "trace")]
pub fn query_all(&self, statement: Statement) -> Result<Vec<QueryResult>, DbErr> {
debug_print!("{}", statement);
let counter = self.query_counter.fetch_add(1, Ordering::SeqCst);
self.mocker
.lock()
.map_err(query_err)?
.query(counter, statement)
}
#[instrument(level = "trace")]
pub fn fetch(
&self,
statement: &Statement,
) -> Pin<Box<dyn Stream<Item = Result<QueryResult, DbErr>> + Send>> {
match self.query_all(statement.clone()) {
Ok(v) => Box::pin(futures::stream::iter(v.into_iter().map(Ok))),
Err(e) => Box::pin(futures::stream::iter(Some(Err(e)).into_iter())),
}
}
#[instrument(level = "trace")]
pub fn begin(&self) {
self.mocker
.lock()
.expect("Failed to acquire mocker")
.begin()
}
#[instrument(level = "trace")]
pub fn commit(&self) {
self.mocker
.lock()
.expect("Failed to acquire mocker")
.commit()
}
#[instrument(level = "trace")]
pub fn rollback(&self) {
self.mocker
.lock()
.expect("Failed to acquire mocker")
.rollback()
}
pub fn ping(&self) -> Result<(), DbErr> {
self.mocker.lock().map_err(query_err)?.ping()
}
}
impl
From<(
Arc<crate::MockDatabaseConnection>,
Statement,
Option<crate::metric::Callback>,
)> for crate::QueryStream
{
fn from(
(conn, stmt, metric_callback): (
Arc<crate::MockDatabaseConnection>,
Statement,
Option<crate::metric::Callback>,
),
) -> Self {
crate::QueryStream::build(stmt, crate::InnerConnection::Mock(conn), metric_callback)
}
}
impl crate::DatabaseTransaction {
pub(crate) async fn new_mock(
inner: Arc<crate::MockDatabaseConnection>,
metric_callback: Option<crate::metric::Callback>,
) -> Result<crate::DatabaseTransaction, DbErr> {
use futures::lock::Mutex;
let backend = inner.get_database_backend();
Self::begin(
Arc::new(Mutex::new(crate::InnerConnection::Mock(inner))),
backend,
metric_callback,
None,
None,
)
.await
}
}