authored_durability_policy_tests.rs (1735B)
1 use super::*; 2 use radroots_storage::event::SourceGeneration; 3 use tempfile::TempDir; 4 5 #[tokio::test] 6 async fn authored_durability_policy_holds_for_every_owned_pool_connection() { 7 let temp = TempDir::new().unwrap(); 8 let store = SqliteStorage::open( 9 OpenOptions::new( 10 Paths::from_directory(temp.path()).unwrap(), 11 OpenMode::Create, 12 ) 13 .with_source_generation(SourceGeneration::new([93; 32]).unwrap(), 9) 14 .unwrap(), 15 ) 16 .await 17 .unwrap(); 18 for (pool, database) in [ 19 (store.pool(), RUNTIME_DATABASE_NAME), 20 (store.private_pool(), PRIVATE_DATABASE_NAME), 21 ] { 22 let mut connections = Vec::new(); 23 for _ in 0..MAX_CONNECTIONS_PER_DATABASE { 24 connections.push(pool.acquire().await.unwrap()); 25 } 26 assert_eq!(connections.len(), 4); 27 for connection in &mut connections { 28 verify_connection(connection, database, Duration::from_millis(5_000)) 29 .await 30 .unwrap(); 31 sqlx::query("PRAGMA fullfsync = OFF") 32 .execute(&mut **connection) 33 .await 34 .unwrap(); 35 assert!(matches!( 36 verify_connection(connection, database, Duration::from_millis(5_000)).await, 37 Err(Error::ConnectionPolicyMismatch { database: actual }) if actual == database 38 )); 39 sqlx::query("PRAGMA fullfsync = ON") 40 .execute(&mut **connection) 41 .await 42 .unwrap(); 43 verify_connection(connection, database, Duration::from_millis(5_000)) 44 .await 45 .unwrap(); 46 } 47 } 48 store.close().await.unwrap(); 49 }