1use std::future::Future;
32use std::pin::Pin;
33use std::sync::Arc;
34use sz_orm_core::{
35 Connection, ConnectionFactory, DbError, Dialect, MySqlDialect, OracleDialect,
36 PostgreSqlDialect, QueryRows, QueryStreamItem, QueryValues, SqlServerDialect, SqliteDialect,
37 Value,
38};
39
40use sz_orm_core::DbType;
41
42use crate::any::{
43 MySqlPoolHandle, PgPoolHandle, SqlitePoolHandle, SqlxMySqlConnectionFactory,
44 SqlxPgConnectionFactory, SqlxSqliteConnectionFactory,
45};
46
47#[cfg(feature = "oracle")]
48use sz_orm_oracle::{OracleConnectionFactory, OraclePoolHandle};
49
50#[cfg(feature = "mssql")]
51use sz_orm_mssql::{MssqlConnectionFactory, MssqlPoolHandle};
52
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59#[non_exhaustive]
60pub enum AnyBackend {
61 MySql,
63 Postgres,
65 Sqlite,
67 Oracle,
69 Mssql,
71}
72
73impl AnyBackend {
74 pub fn from_dsn(dsn: &str) -> Result<Self, DbError> {
85 if dsn.starts_with("mysql://") || dsn.starts_with("mariadb://") {
86 Ok(AnyBackend::MySql)
87 } else if dsn.starts_with("postgres://") || dsn.starts_with("postgresql://") {
88 Ok(AnyBackend::Postgres)
89 } else if dsn.starts_with("szrsql://") {
90 Ok(AnyBackend::Postgres)
93 } else if dsn.starts_with("sqlite://") || dsn.starts_with("sqlite:") {
94 Ok(AnyBackend::Sqlite)
95 } else if dsn.starts_with("oracle://") {
96 Ok(AnyBackend::Oracle)
97 } else if dsn.starts_with("mssql://") || dsn.starts_with("sqlserver://") {
98 Ok(AnyBackend::Mssql)
99 } else {
100 Err(DbError::ConnectionRefused(format!(
101 "未知的 DSN scheme: {}(支持 mysql/postgres/sqlite/oracle/mssql)",
102 dsn
103 )))
104 }
105 }
106
107 pub fn from_db_type(db_type: DbType) -> Option<Self> {
121 match db_type {
122 DbType::MySQL | DbType::MariaDB | DbType::TiDB | DbType::OceanBase => {
123 Some(AnyBackend::MySql)
124 }
125 DbType::PostgreSQL
126 | DbType::Kingbase
127 | DbType::PolarDB
128 | DbType::GaussDB
129 | DbType::Szrsql => Some(AnyBackend::Postgres),
130 DbType::Sqlite => Some(AnyBackend::Sqlite),
131 DbType::Oracle | DbType::Dameng => Some(AnyBackend::Oracle),
132 DbType::SqlServer | DbType::Sybase | DbType::GBase => Some(AnyBackend::Mssql),
133 _ => None,
134 }
135 }
136
137 pub fn name(&self) -> &'static str {
139 match self {
140 AnyBackend::MySql => "mysql",
141 AnyBackend::Postgres => "postgres",
142 AnyBackend::Sqlite => "sqlite",
143 AnyBackend::Oracle => "oracle",
144 AnyBackend::Mssql => "mssql",
145 }
146 }
147
148 pub fn dialect(&self) -> Box<dyn Dialect> {
156 match self {
157 AnyBackend::MySql => Box::new(MySqlDialect),
158 AnyBackend::Postgres => Box::new(PostgreSqlDialect),
159 AnyBackend::Sqlite => Box::new(SqliteDialect),
160 AnyBackend::Oracle => Box::new(OracleDialect),
161 AnyBackend::Mssql => Box::new(SqlServerDialect),
162 }
163 }
164}
165
166pub struct AnyPool {
168 backend: AnyBackend,
169 factory: Arc<dyn ConnectionFactory>,
170}
171
172impl AnyPool {
173 pub async fn connect(dsn: &str) -> Result<Self, DbError> {
181 let backend = AnyBackend::from_dsn(dsn)?;
182 let factory: Arc<dyn ConnectionFactory> = match backend {
183 AnyBackend::MySql => {
184 let handle = Arc::new(MySqlPoolHandle::connect(dsn).await?);
185 Arc::new(SqlxMySqlConnectionFactory::new(handle))
186 }
187 AnyBackend::Postgres => {
188 let handle = Arc::new(PgPoolHandle::connect(dsn).await?);
189 Arc::new(SqlxPgConnectionFactory::new(handle))
190 }
191 AnyBackend::Sqlite => {
192 let handle = Arc::new(SqlitePoolHandle::connect(dsn).await?);
193 Arc::new(SqlxSqliteConnectionFactory::new(handle))
194 }
195 AnyBackend::Oracle => {
196 #[cfg(feature = "oracle")]
197 {
198 let (username, password, connect_string) = parse_oracle_dsn(dsn)?;
199 let handle = Arc::new(OraclePoolHandle::connect(
200 &username,
201 &password,
202 &connect_string,
203 )?);
204 Arc::new(OracleConnectionFactory::new(handle))
205 }
206 #[cfg(not(feature = "oracle"))]
207 {
208 return Err(DbError::ConnectionRefused(
209 "Oracle 后端未启用,请在 Cargo.toml 中添加 features = [\"oracle\"]"
210 .to_string(),
211 ));
212 }
213 }
214 AnyBackend::Mssql => {
215 #[cfg(feature = "mssql")]
216 {
217 let ado_string = parse_mssql_dsn(dsn)?;
218 let handle = Arc::new(MssqlPoolHandle::connect(&ado_string).await?);
219 Arc::new(MssqlConnectionFactory::new(handle))
220 }
221 #[cfg(not(feature = "mssql"))]
222 {
223 return Err(DbError::ConnectionRefused(
224 "MSSQL 后端未启用,请在 Cargo.toml 中添加 features = [\"mssql\"]"
225 .to_string(),
226 ));
227 }
228 }
229 };
230 Ok(Self { backend, factory })
231 }
232
233 pub fn from_factory(backend: AnyBackend, factory: Arc<dyn ConnectionFactory>) -> Self {
235 Self { backend, factory }
236 }
237
238 pub fn backend(&self) -> AnyBackend {
240 self.backend
241 }
242
243 pub fn dialect(&self) -> Box<dyn Dialect> {
247 self.backend.dialect()
248 }
249
250 pub async fn create(&self) -> Result<AnyConnection, DbError> {
252 let conn = self.factory.create().await?;
253 Ok(AnyConnection {
254 backend: self.backend,
255 inner: conn,
256 })
257 }
258}
259
260pub async fn create_connection(dsn: &str) -> Result<Box<dyn Connection>, DbError> {
291 let pool = AnyPool::connect(dsn).await?;
292 let conn = pool.create().await?;
293 Ok(Box::new(conn))
294}
295
296pub async fn create_connection_by_type(
315 db_type: DbType,
316 dsn: &str,
317) -> Result<Box<dyn Connection>, DbError> {
318 let backend = AnyBackend::from_db_type(db_type).ok_or_else(|| {
319 DbError::ConnectionRefused(format!(
320 "DbType {:?} 不在 sz-orm-sqlx 支持范围内(支持 MySQL/PostgreSQL/SQLite/Oracle/MSSQL 及兼容方言)",
321 db_type
322 ))
323 })?;
324 let normalized = normalize_dsn_scheme(backend, dsn);
325 create_connection(&normalized).await
326}
327
328fn normalize_dsn_scheme(backend: AnyBackend, dsn: &str) -> String {
333 let expected_scheme = match backend {
334 AnyBackend::MySql => "mysql://",
335 AnyBackend::Postgres => "postgres://",
336 AnyBackend::Sqlite => "sqlite:",
337 AnyBackend::Oracle => "oracle://",
338 AnyBackend::Mssql => "mssql://",
339 };
340 let known_schemes = [
341 "mysql://",
342 "mariadb://",
343 "postgres://",
344 "postgresql://",
345 "szrsql://",
346 "sqlite://",
347 "sqlite:",
348 "oracle://",
349 "mssql://",
350 "sqlserver://",
351 ];
352 for scheme in &known_schemes {
353 if let Some(rest) = dsn.strip_prefix(scheme) {
354 if backend == AnyBackend::from_dsn(dsn).unwrap_or(AnyBackend::MySql) {
355 return dsn.to_string();
356 }
357 return format!("{}{}", expected_scheme, rest);
358 }
359 }
360 dsn.to_string()
361}
362
363#[allow(dead_code)]
369pub(crate) fn parse_oracle_dsn(dsn: &str) -> Result<(String, String, String), DbError> {
370 let rest = dsn
371 .strip_prefix("oracle://")
372 .ok_or_else(|| DbError::ConnectionRefused(format!("无效的 Oracle DSN: {}", dsn)))?;
373 parse_user_pass_host(rest, "Oracle")
374}
375
376#[allow(dead_code)]
382pub(crate) fn parse_mssql_dsn(dsn: &str) -> Result<String, DbError> {
383 let rest = dsn
384 .strip_prefix("mssql://")
385 .or_else(|| dsn.strip_prefix("sqlserver://"))
386 .ok_or_else(|| DbError::ConnectionRefused(format!("无效的 MSSQL DSN: {}", dsn)))?;
387 let (username, password, host_port_db) = parse_user_pass_host(rest, "MSSQL")?;
388 let (host_port, database) = host_port_db
389 .split_once('/')
390 .ok_or_else(|| DbError::ConnectionRefused(format!("MSSQL DSN 缺少 database: {}", dsn)))?;
391 let (host, port) = host_port.split_once(':').unwrap_or((host_port, "1433"));
392 Ok(format!(
393 "Server={},{};Database={};User Id={};Password={};",
394 host, port, database, username, password
395 ))
396}
397
398#[allow(dead_code)]
400fn parse_user_pass_host(rest: &str, backend: &str) -> Result<(String, String, String), DbError> {
401 let (userinfo, hostinfo) = rest
402 .split_once('@')
403 .ok_or_else(|| DbError::ConnectionRefused(format!("{} DSN 缺少 @: {}", backend, rest)))?;
404 let (username, password) = userinfo.split_once(':').ok_or_else(|| {
405 DbError::ConnectionRefused(format!("{} DSN 缺少 password: {}", backend, rest))
406 })?;
407 Ok((
408 username.to_string(),
409 password.to_string(),
410 hostinfo.to_string(),
411 ))
412}
413
414pub struct AnyConnection {
416 backend: AnyBackend,
417 inner: Box<dyn Connection>,
418}
419
420impl AnyConnection {
421 pub fn backend(&self) -> AnyBackend {
423 self.backend
424 }
425}
426
427impl Connection for AnyConnection {
428 fn db_type(&self) -> Option<DbType> {
429 Some(match self.backend {
430 AnyBackend::MySql => DbType::MySQL,
431 AnyBackend::Postgres => DbType::PostgreSQL,
432 AnyBackend::Sqlite => DbType::Sqlite,
433 AnyBackend::Oracle => DbType::Oracle,
434 AnyBackend::Mssql => DbType::SqlServer,
435 })
436 }
437
438 fn execute<'a>(
439 &'a mut self,
440 sql: &'a str,
441 ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
442 self.inner.execute(sql)
443 }
444
445 fn query<'a>(
446 &'a mut self,
447 sql: &'a str,
448 ) -> Pin<Box<dyn Future<Output = Result<QueryRows, DbError>> + Send + 'a>> {
449 self.inner.query(sql)
450 }
451
452 fn begin_transaction<'a>(
453 &'a mut self,
454 ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
455 self.inner.begin_transaction()
456 }
457
458 fn commit<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
459 self.inner.commit()
460 }
461
462 fn rollback<'a>(
463 &'a mut self,
464 ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
465 self.inner.rollback()
466 }
467
468 fn in_transaction(&self) -> bool {
469 self.inner.in_transaction()
470 }
471
472 fn session_reset_sql(&self) -> Option<&'static str> {
473 self.inner.session_reset_sql()
474 }
475
476 fn is_connected(&self) -> bool {
477 self.inner.is_connected()
478 }
479
480 fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
481 self.inner.ping()
482 }
483
484 fn close<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
485 self.inner.close()
486 }
487
488 fn execute_with_params<'a>(
489 &'a mut self,
490 sql: &'a str,
491 params: &'a [Value],
492 ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
493 self.inner.execute_with_params(sql, params)
494 }
495
496 fn query_with_params<'a>(
497 &'a mut self,
498 sql: &'a str,
499 params: &'a [Value],
500 ) -> Pin<Box<dyn Future<Output = Result<QueryRows, DbError>> + Send + 'a>> {
501 self.inner.query_with_params(sql, params)
502 }
503
504 fn query_values<'a>(
505 &'a mut self,
506 sql: &'a str,
507 ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
508 self.inner.query_values(sql)
509 }
510
511 fn query_values_with_params<'a>(
512 &'a mut self,
513 sql: &'a str,
514 params: &'a [Value],
515 ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
516 self.inner.query_values_with_params(sql, params)
517 }
518
519 fn query_stream<'a>(
520 &'a mut self,
521 sql: &'a str,
522 ) -> Pin<Box<dyn futures::Stream<Item = QueryStreamItem> + Send + 'a>> {
523 self.inner.query_stream(sql)
524 }
525
526 fn query_stream_cursor<'a>(
527 &'a mut self,
528 sql: &'a str,
529 batch_size: usize,
530 ) -> Pin<Box<dyn futures::Stream<Item = QueryStreamItem> + Send + 'a>> {
531 self.inner.query_stream_cursor(sql, batch_size)
532 }
533
534 fn execute_batch<'a>(
535 &'a mut self,
536 sqls: &'a [String],
537 ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
538 self.inner.execute_batch(sqls)
539 }
540}
541
542#[cfg(test)]
543mod tests {
544 use super::*;
545
546 #[test]
547 fn test_any_backend_from_dsn_mysql() {
548 assert_eq!(
549 AnyBackend::from_dsn("mysql://root:pass@127.0.0.1/db").unwrap(),
550 AnyBackend::MySql
551 );
552 assert_eq!(
553 AnyBackend::from_dsn("mariadb://root:pass@127.0.0.1/db").unwrap(),
554 AnyBackend::MySql
555 );
556 }
557
558 #[test]
559 fn test_any_backend_from_dsn_postgres() {
560 assert_eq!(
561 AnyBackend::from_dsn("postgres://user:pass@127.0.0.1/db").unwrap(),
562 AnyBackend::Postgres
563 );
564 assert_eq!(
565 AnyBackend::from_dsn("postgresql://user:pass@127.0.0.1/db").unwrap(),
566 AnyBackend::Postgres
567 );
568 }
569
570 #[test]
571 fn test_any_backend_from_dsn_sqlite() {
572 assert_eq!(
573 AnyBackend::from_dsn("sqlite::memory:").unwrap(),
574 AnyBackend::Sqlite
575 );
576 assert_eq!(
577 AnyBackend::from_dsn("sqlite://./test.db").unwrap(),
578 AnyBackend::Sqlite
579 );
580 }
581
582 #[test]
583 fn test_any_backend_from_dsn_oracle() {
584 assert_eq!(
585 AnyBackend::from_dsn("oracle://sys:test123@127.0.0.1:1521/freepdb1").unwrap(),
586 AnyBackend::Oracle
587 );
588 }
589
590 #[test]
591 fn test_any_backend_from_dsn_mssql() {
592 assert_eq!(
593 AnyBackend::from_dsn("mssql://sa:test123@127.0.0.1:1433/testdb").unwrap(),
594 AnyBackend::Mssql
595 );
596 }
597
598 #[test]
599 fn test_any_backend_from_dsn_sqlserver() {
600 assert_eq!(
601 AnyBackend::from_dsn("sqlserver://sa:test123@127.0.0.1:1433/testdb").unwrap(),
602 AnyBackend::Mssql
603 );
604 }
605
606 #[test]
607 fn test_any_backend_from_dsn_unknown_v22() {
608 let result = AnyBackend::from_dsn("redis://127.0.0.1");
609 assert!(result.is_err());
610 if let Err(e) = result {
611 let msg = format!("{}", e);
612 assert!(msg.contains("mysql"));
613 assert!(msg.contains("postgres"));
614 assert!(msg.contains("sqlite"));
615 assert!(msg.contains("oracle"));
616 assert!(msg.contains("mssql"));
617 }
618 }
619
620 #[test]
621 fn test_any_backend_name_v22() {
622 assert_eq!(AnyBackend::MySql.name(), "mysql");
623 assert_eq!(AnyBackend::Postgres.name(), "postgres");
624 assert_eq!(AnyBackend::Sqlite.name(), "sqlite");
625 assert_eq!(AnyBackend::Oracle.name(), "oracle");
626 assert_eq!(AnyBackend::Mssql.name(), "mssql");
627 }
628
629 #[test]
630 fn test_any_backend_equality() {
631 assert_eq!(AnyBackend::MySql, AnyBackend::MySql);
632 assert_ne!(AnyBackend::MySql, AnyBackend::Postgres);
633 assert_ne!(AnyBackend::Postgres, AnyBackend::Sqlite);
634 assert_ne!(AnyBackend::Oracle, AnyBackend::Mssql);
635 assert_ne!(AnyBackend::Oracle, AnyBackend::MySql);
636 }
637
638 #[test]
639 fn test_parse_oracle_dsn() {
640 let (user, pass, cs) =
641 parse_oracle_dsn("oracle://sys:test123@127.0.0.1:1521/freepdb1").unwrap();
642 assert_eq!(user, "sys");
643 assert_eq!(pass, "test123");
644 assert_eq!(cs, "127.0.0.1:1521/freepdb1");
645 }
646
647 #[test]
648 fn test_parse_mssql_dsn() {
649 let ado = parse_mssql_dsn("mssql://sa:test123@127.0.0.1:1433/testdb").unwrap();
650 assert!(ado.contains("Server=127.0.0.1,1433"));
651 assert!(ado.contains("Database=testdb"));
652 assert!(ado.contains("User Id=sa"));
653 assert!(ado.contains("Password=test123"));
654 }
655
656 #[test]
657 fn test_parse_mssql_dsn_sqlserver_scheme() {
658 let ado = parse_mssql_dsn("sqlserver://sa:pass@localhost/testdb").unwrap();
659 assert!(ado.contains("Server=localhost,1433"));
660 assert!(ado.contains("Database=testdb"));
661 }
662
663 #[tokio::test]
666 async fn test_any_pool_sqlite_connect_and_query() {
667 let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
668 assert_eq!(pool.backend(), AnyBackend::Sqlite);
669
670 let mut conn = pool.create().await.unwrap();
671 assert_eq!(conn.backend(), AnyBackend::Sqlite);
672
673 conn.execute("CREATE TABLE test_any (id INTEGER PRIMARY KEY, name TEXT)")
675 .await
676 .unwrap();
677 conn.execute("INSERT INTO test_any (name) VALUES ('Alice')")
678 .await
679 .unwrap();
680 conn.execute("INSERT INTO test_any (name) VALUES ('Bob')")
681 .await
682 .unwrap();
683
684 let rows = conn
686 .query("SELECT * FROM test_any ORDER BY id")
687 .await
688 .unwrap();
689 assert_eq!(rows.len(), 2);
690 assert_eq!(rows[0].get("name").and_then(|v| v.as_str()), Some("Alice"));
691 assert_eq!(rows[1].get("name").and_then(|v| v.as_str()), Some("Bob"));
692 }
693
694 #[tokio::test]
695 async fn test_any_pool_sqlite_transaction_commit() {
696 let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
697 let mut conn = pool.create().await.unwrap();
698
699 conn.execute("CREATE TABLE tx_test (id INTEGER PRIMARY KEY, val INTEGER)")
700 .await
701 .unwrap();
702
703 conn.begin_transaction().await.unwrap();
705 conn.execute("INSERT INTO tx_test (val) VALUES (1)")
706 .await
707 .unwrap();
708 conn.execute("INSERT INTO tx_test (val) VALUES (2)")
709 .await
710 .unwrap();
711 conn.commit().await.unwrap();
712
713 let rows = conn.query("SELECT * FROM tx_test").await.unwrap();
714 assert_eq!(rows.len(), 2);
715 }
716
717 #[tokio::test]
718 async fn test_any_pool_sqlite_transaction_rollback() {
719 let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
720 let mut conn = pool.create().await.unwrap();
721
722 conn.execute("CREATE TABLE tx_rb (id INTEGER PRIMARY KEY, val INTEGER)")
723 .await
724 .unwrap();
725
726 conn.begin_transaction().await.unwrap();
727 conn.execute("INSERT INTO tx_rb (val) VALUES (1)")
728 .await
729 .unwrap();
730 conn.rollback().await.unwrap();
731
732 let rows = conn.query("SELECT * FROM tx_rb").await.unwrap();
733 assert_eq!(rows.len(), 0);
734 }
735
736 #[tokio::test]
737 async fn test_any_connection_in_transaction_state_tracking() {
738 let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
739 let mut conn = pool.create().await.unwrap();
740
741 assert!(!conn.in_transaction(), "新建连接不应处于事务中");
742
743 conn.begin_transaction().await.unwrap();
744 assert!(conn.in_transaction(), "begin 后应处于事务中");
745
746 conn.execute("CREATE TABLE tx_track (id INTEGER PRIMARY KEY)")
747 .await
748 .unwrap();
749 assert!(conn.in_transaction(), "事务内执行 SQL 后仍应处于事务中");
750
751 conn.commit().await.unwrap();
752 assert!(!conn.in_transaction(), "commit 后不应处于事务中");
753
754 conn.begin_transaction().await.unwrap();
755 assert!(conn.in_transaction());
756 conn.rollback().await.unwrap();
757 assert!(!conn.in_transaction(), "rollback 后不应处于事务中");
758 }
759
760 #[tokio::test]
761 async fn test_any_connection_in_transaction_leak_detection() {
762 let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
763 let mut conn = pool.create().await.unwrap();
764
765 conn.execute("CREATE TABLE tx_leak (id INTEGER PRIMARY KEY)")
766 .await
767 .unwrap();
768
769 conn.begin_transaction().await.unwrap();
770 conn.execute("INSERT INTO tx_leak (id) VALUES (1)")
771 .await
772 .unwrap();
773
774 assert!(
775 conn.in_transaction(),
776 "未提交事务应被检测到(泄漏防护前提)"
777 );
778
779 conn.rollback().await.unwrap();
780 assert!(!conn.in_transaction());
781 }
782
783 #[tokio::test]
784 async fn test_any_pool_sqlite_ping() {
785 let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
786 let mut conn = pool.create().await.unwrap();
787 let ok = conn.ping().await;
788 assert!(ok);
789 assert!(conn.is_connected());
790 }
791
792 #[tokio::test]
793 async fn test_any_pool_invalid_dsn() {
794 let result = AnyPool::connect("invalid://dsn").await;
795 assert!(result.is_err());
796 }
797
798 #[tokio::test]
799 async fn test_any_pool_sqlite_count_query() {
800 let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
801 let mut conn = pool.create().await.unwrap();
802
803 conn.execute("CREATE TABLE cnt (id INTEGER PRIMARY KEY)")
804 .await
805 .unwrap();
806 for i in 1..=5 {
807 conn.execute(&format!("INSERT INTO cnt (id) VALUES ({})", i))
808 .await
809 .unwrap();
810 }
811
812 let rows = conn.query("SELECT * FROM cnt").await.unwrap();
814 assert_eq!(rows.len(), 5);
815 }
816
817 #[test]
820 fn test_any_backend_dialect_mapping() {
821 use sz_orm_core::DbType;
822
823 let mysql_d = AnyBackend::MySql.dialect();
824 assert_eq!(mysql_d.db_type(), DbType::MySQL);
825
826 let pg_d = AnyBackend::Postgres.dialect();
827 assert_eq!(pg_d.db_type(), DbType::PostgreSQL);
828
829 let sqlite_d = AnyBackend::Sqlite.dialect();
830 assert_eq!(sqlite_d.db_type(), DbType::Sqlite);
831
832 let oracle_d = AnyBackend::Oracle.dialect();
833 assert_eq!(oracle_d.db_type(), DbType::Oracle);
834
835 let mssql_d = AnyBackend::Mssql.dialect();
836 assert_eq!(mssql_d.db_type(), DbType::SqlServer);
837 }
838
839 #[test]
840 fn test_oracle_dialect_pagination() {
841 let d = AnyBackend::Oracle.dialect();
842 let sql = d.build_pagination("SELECT * FROM users", 2, 10);
843 let upper = sql.to_uppercase();
844 assert!(
845 upper.contains("OFFSET") || upper.contains("FETCH") || upper.contains("ROWNUM"),
846 "Oracle 分页 SQL 应含 OFFSET/FETCH/ROWNUM,实际: {}",
847 sql
848 );
849 assert!(
850 !upper.contains("LIMIT"),
851 "Oracle 分页 SQL 不应含 LIMIT,实际: {}",
852 sql
853 );
854 }
855
856 #[test]
857 fn test_mssql_dialect_pagination() {
858 let d = AnyBackend::Mssql.dialect();
859 let sql = d.build_pagination("SELECT * FROM users", 2, 10);
860 let upper = sql.to_uppercase();
861 assert!(
862 upper.contains("OFFSET") || upper.contains("FETCH"),
863 "MSSQL 分页 SQL 应含 OFFSET/FETCH,实际: {}",
864 sql
865 );
866 assert!(
867 !upper.contains("LIMIT"),
868 "MSSQL 分页 SQL 不应含 LIMIT,实际: {}",
869 sql
870 );
871 }
872
873 #[test]
874 fn test_oracle_dialect_no_placeholder() {
875 let d = AnyBackend::Oracle.dialect();
876 let _ = d.db_type();
878 let _ = d.quote("col");
879 let _ = d.quote_checked("col").unwrap();
880 let _ = d.escape_string("val");
881 let _ = d.supports_returning();
882 let _ = d.build_pagination("SELECT 1", 1, 10);
883 let _ = d.json_type();
884 let _ = d.json_extract("col", "$.key");
885 let _ = d.full_text_search(&["col"], "kw");
886 let _ = d.bool_to_int("expr");
887 let _ = d.concat(&["a", "b"]);
888 }
889
890 #[test]
891 fn test_mssql_dialect_no_placeholder() {
892 let d = AnyBackend::Mssql.dialect();
893 let _ = d.db_type();
894 let _ = d.quote("col");
895 let _ = d.quote_checked("col").unwrap();
896 let _ = d.escape_string("val");
897 let _ = d.supports_returning();
898 let _ = d.build_pagination("SELECT 1", 1, 10);
899 let _ = d.json_type();
900 let _ = d.json_extract("col", "$.key");
901 let _ = d.full_text_search(&["col"], "kw");
902 let _ = d.bool_to_int("expr");
903 let _ = d.concat(&["a", "b"]);
904 }
905
906 #[tokio::test]
907 async fn test_any_pool_dialect_sqlite() {
908 let pool = AnyPool::connect("sqlite::memory:").await.unwrap();
909 let d = pool.dialect();
910 assert_eq!(d.db_type(), sz_orm_core::DbType::Sqlite);
911 }
912
913 #[test]
916 fn test_from_db_type_supported() {
917 use sz_orm_core::DbType;
918 assert_eq!(
919 AnyBackend::from_db_type(DbType::MySQL),
920 Some(AnyBackend::MySql)
921 );
922 assert_eq!(
923 AnyBackend::from_db_type(DbType::MariaDB),
924 Some(AnyBackend::MySql)
925 );
926 assert_eq!(
927 AnyBackend::from_db_type(DbType::TiDB),
928 Some(AnyBackend::MySql)
929 );
930 assert_eq!(
931 AnyBackend::from_db_type(DbType::OceanBase),
932 Some(AnyBackend::MySql)
933 );
934 assert_eq!(
935 AnyBackend::from_db_type(DbType::PostgreSQL),
936 Some(AnyBackend::Postgres)
937 );
938 assert_eq!(
939 AnyBackend::from_db_type(DbType::Kingbase),
940 Some(AnyBackend::Postgres)
941 );
942 assert_eq!(
943 AnyBackend::from_db_type(DbType::PolarDB),
944 Some(AnyBackend::Postgres)
945 );
946 assert_eq!(
947 AnyBackend::from_db_type(DbType::GaussDB),
948 Some(AnyBackend::Postgres)
949 );
950 assert_eq!(
951 AnyBackend::from_db_type(DbType::Sqlite),
952 Some(AnyBackend::Sqlite)
953 );
954 assert_eq!(
955 AnyBackend::from_db_type(DbType::Oracle),
956 Some(AnyBackend::Oracle)
957 );
958 assert_eq!(
959 AnyBackend::from_db_type(DbType::Dameng),
960 Some(AnyBackend::Oracle)
961 );
962 assert_eq!(
963 AnyBackend::from_db_type(DbType::SqlServer),
964 Some(AnyBackend::Mssql)
965 );
966 assert_eq!(
967 AnyBackend::from_db_type(DbType::Sybase),
968 Some(AnyBackend::Mssql)
969 );
970 assert_eq!(
971 AnyBackend::from_db_type(DbType::GBase),
972 Some(AnyBackend::Mssql)
973 );
974 }
975
976 #[test]
977 fn test_from_db_type_unsupported() {
978 use sz_orm_core::DbType;
979 assert_eq!(AnyBackend::from_db_type(DbType::Redis), None);
980 assert_eq!(AnyBackend::from_db_type(DbType::MongoDB), None);
981 assert_eq!(AnyBackend::from_db_type(DbType::ClickHouse), None);
982 assert_eq!(AnyBackend::from_db_type(DbType::VectorDb), None);
983 }
984
985 #[tokio::test]
986 async fn test_create_connection_sqlite() {
987 let mut conn = create_connection("sqlite::memory:").await.unwrap();
988 conn.execute("CREATE TABLE cc (id INTEGER PRIMARY KEY, v TEXT)")
989 .await
990 .unwrap();
991 conn.execute("INSERT INTO cc (v) VALUES ('ok')")
992 .await
993 .unwrap();
994 let rows = conn.query("SELECT * FROM cc").await.unwrap();
995 assert_eq!(rows.len(), 1);
996 }
997
998 #[tokio::test]
999 async fn test_create_connection_invalid_dsn() {
1000 let result = create_connection("invalid://dsn").await;
1001 assert!(result.is_err());
1002 }
1003
1004 #[tokio::test]
1005 async fn test_create_connection_by_type_sqlite() {
1006 use sz_orm_core::DbType;
1007 let mut conn = create_connection_by_type(DbType::Sqlite, "sqlite::memory:")
1008 .await
1009 .unwrap();
1010 conn.execute("SELECT 1").await.unwrap();
1011 }
1012
1013 #[tokio::test]
1014 async fn test_create_connection_by_type_unsupported() {
1015 use sz_orm_core::DbType;
1016 let result = create_connection_by_type(DbType::Redis, "redis://127.0.0.1").await;
1017 assert!(result.is_err());
1018 }
1019
1020 #[test]
1021 fn test_normalize_dsn_scheme_already_correct() {
1022 assert_eq!(
1023 normalize_dsn_scheme(AnyBackend::MySql, "mysql://root:pass@host/db"),
1024 "mysql://root:pass@host/db"
1025 );
1026 assert_eq!(
1027 normalize_dsn_scheme(AnyBackend::Postgres, "postgres://user@host/db"),
1028 "postgres://user@host/db"
1029 );
1030 assert_eq!(
1031 normalize_dsn_scheme(AnyBackend::Oracle, "oracle://sys:pass@host/svc"),
1032 "oracle://sys:pass@host/svc"
1033 );
1034 }
1035
1036 #[test]
1037 fn test_normalize_dsn_scheme_fix_mismatch() {
1038 let fixed = normalize_dsn_scheme(AnyBackend::MySql, "postgres://root:pass@host/db");
1039 assert!(fixed.starts_with("mysql://"));
1040 assert!(fixed.contains("root:pass@host/db"));
1041 }
1042}