test: replace sqlx::test with postgres testcontainers

Signed-off-by: Joseph Livesey <[email protected]>

Author: suchapalaver

Cargo.lock

@@ -88,6 +88,7 @@ tap_core = { version = "4.1.4", default-features = false }
 tap_graph = { version = "0.3.4", features = ["v2"] }
 tempfile = "3.8.0"
 test-log = { version = "0.2.12", default-features = false }
+testcontainers-modules = { version = "0.12.1", features = ["postgres"] }
 test-with = "0.14.6"
 thegraph-core = { version = "0.15.1", features = [
     "attestation",

Cargo.toml

@@ -39,6 +39,8 @@ serde_json.workspace = true
 [dev-dependencies]
 rand.workspace = true
 indexer-watcher = { path = "../watcher" }
+testcontainers-modules = { workspace = true, features = ["postgres"] }
+test-assets = { path = "../test-assets" }
 
 [build-dependencies]
 tonic-build = { workspace = true, optional = true }

crates/dips/Cargo.toml

@@ -135,7 +135,6 @@ pub(crate) mod test {
     use std::sync::Arc;
 
     use build_info::chrono::Duration;
-    use sqlx::PgPool;
     use thegraph_core::alloy::{
         primitives::{ruint::aliases::U256, Address},
         sol_types::SolValue,
@@ -145,9 +144,10 @@ pub(crate) mod test {
     use super::*;
     use crate::{CancellationRequest, IndexingAgreementVoucher};
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_store_agreement(pool: PgPool) {
-        let store = Arc::new(PsqlAgreementStore { pool });
+    #[tokio::test]
+    async fn test_store_agreement() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let store = Arc::new(PsqlAgreementStore { pool: test_db.pool });
         let id = Uuid::now_v7();
 
         // Create metadata first
@@ -196,9 +196,10 @@ pub(crate) mod test {
         assert_eq!(row.subgraph_deployment_id, "Qm123");
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_get_agreement_by_id(pool: PgPool) {
-        let store = Arc::new(PsqlAgreementStore { pool });
+    #[tokio::test]
+    async fn test_get_agreement_by_id() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let store = Arc::new(PsqlAgreementStore { pool: test_db.pool });
         let id = Uuid::parse_str("a1a2a3a4b1b2c1c2d1d2d3d4d5d6d7d9").unwrap();
 
         // Create metadata first
@@ -277,9 +278,10 @@ pub(crate) mod test {
         assert!(!stored_agreement.cancelled);
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_cancel_agreement(pool: PgPool) {
-        let store = Arc::new(PsqlAgreementStore { pool });
+    #[tokio::test]
+    async fn test_cancel_agreement() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let store = Arc::new(PsqlAgreementStore { pool: test_db.pool });
         let id = Uuid::parse_str("a1a2a3a4b1b2c1c2d1d2d3d4d5d6d7e9").unwrap();
 
         // Create metadata first

crates/dips/src/database.rs

@@ -251,8 +251,10 @@ pub(crate) mod test {
         ]
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn success_cost_models(pool: PgPool) {
+    #[tokio::test]
+    async fn success_cost_models() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pool = test_db.pool;
         let test_models = test_data();
         let test_deployments = test_models
             .iter()
@@ -313,8 +315,10 @@ pub(crate) mod test {
         }
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn global_fallback_cost_models(pool: PgPool) {
+    #[tokio::test]
+    async fn global_fallback_cost_models() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pool = test_db.pool;
         let test_models = test_data();
         let test_deployments = test_models
             .iter()
@@ -403,8 +407,10 @@ pub(crate) mod test {
         assert_eq!(missing_model.model, global_model.model);
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn success_cost_model(pool: PgPool) {
+    #[tokio::test]
+    async fn success_cost_model() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pool = test_db.pool;
         add_cost_models(&pool, to_db_models(test_data())).await;
 
         let deployment_id_from_bytes = DeploymentId::from_str(
@@ -425,8 +431,10 @@ pub(crate) mod test {
         assert_eq!(model.model, Some("default => 0.00025;".to_string()));
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn global_fallback_cost_model(pool: PgPool) {
+    #[tokio::test]
+    async fn global_fallback_cost_model() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pool = test_db.pool;
         let test_models = test_data();
         let global_model = global_cost_model();
 

crates/service/src/database/cost_model.rs

@@ -69,9 +69,10 @@ mod tests {
         service
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_composition_header_valid(pgpool: PgPool) {
-        let mut service = service(pgpool.clone()).await;
+    #[tokio::test]
+    async fn test_composition_header_valid() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let mut service = service(test_db.pool.clone()).await;
         // should allow queries that contains the free token
         // if the token does not match, return payment required
         let mut req = Request::new(Default::default());
@@ -83,9 +84,10 @@ mod tests {
         assert_eq!(res.status(), StatusCode::OK);
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_composition_header_invalid(pgpool: PgPool) {
-        let mut service = service(pgpool.clone()).await;
+    #[tokio::test]
+    async fn test_composition_header_invalid() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let mut service = service(test_db.pool.clone()).await;
 
         // if the token exists but is wrong, try the receipt
         let mut req = Request::new(Default::default());
@@ -96,9 +98,10 @@ mod tests {
         assert_eq!(res.status(), StatusCode::PAYMENT_REQUIRED);
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_composition_with_receipt(pgpool: PgPool) {
-        let mut service = service(pgpool.clone()).await;
+    #[tokio::test]
+    async fn test_composition_with_receipt() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let mut service = service(test_db.pool.clone()).await;
 
         let receipt = create_signed_receipt(SignedReceiptRequest::builder().build()).await;
 
@@ -111,16 +114,17 @@ mod tests {
         // verify receipts
         assert_while_retry!({
             let result = sqlx::query!("SELECT * FROM scalar_tap_receipts")
-                .fetch_all(&pgpool)
+                .fetch_all(&test_db.pool)
                 .await
                 .unwrap();
             result.is_empty()
         });
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_composition_without_header_or_receipt(pgpool: PgPool) {
-        let mut service = service(pgpool.clone()).await;
+    #[tokio::test]
+    async fn test_composition_without_header_or_receipt() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let mut service = service(test_db.pool.clone()).await;
         // if it has neither, should return payment required
         let req = Request::new(Default::default());
         let res = service.call(req).await.unwrap();

crates/service/src/middleware/auth.rs

@@ -164,12 +164,10 @@ mod tests {
     }
 
     #[rstest]
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_tap_valid_receipt(
-        metric: &'static prometheus::CounterVec,
-        #[ignore] pgpool: PgPool,
-    ) {
-        let mut service = service(metric, pgpool.clone()).await;
+    #[tokio::test]
+    async fn test_tap_valid_receipt(metric: &'static prometheus::CounterVec) {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let mut service = service(metric, test_db.pool.clone()).await;
 
         let receipt = create_signed_receipt(SignedReceiptRequest::builder().build()).await;
 
@@ -182,20 +180,18 @@ mod tests {
         // verify receipts
         assert_while_retry!({
             sqlx::query!("SELECT * FROM scalar_tap_receipts")
-                .fetch_all(&pgpool)
+                .fetch_all(&test_db.pool)
                 .await
                 .unwrap()
                 .is_empty()
         })
     }
 
     #[rstest]
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_invalid_receipt_with_failed_metric(
-        metric: &'static prometheus::CounterVec,
-        #[ignore] pgpool: PgPool,
-    ) {
-        let mut service = service(metric, pgpool.clone()).await;
+    #[tokio::test]
+    async fn test_invalid_receipt_with_failed_metric(metric: &'static prometheus::CounterVec) {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let mut service = service(metric, test_db.pool.clone()).await;
         // if it fails tap receipt, should return failed to process payment + tap message
 
         assert_eq!(metric.collect().first().unwrap().get_metric().len(), 0);
@@ -224,12 +220,10 @@ mod tests {
     }
 
     #[rstest]
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_tap_missing_signed_receipt(
-        metric: &'static prometheus::CounterVec,
-        #[ignore] pgpool: PgPool,
-    ) {
-        let mut service = service(metric, pgpool.clone()).await;
+    #[tokio::test]
+    async fn test_tap_missing_signed_receipt(metric: &'static prometheus::CounterVec) {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let mut service = service(metric, test_db.pool.clone()).await;
         // if it doesnt contain the signed receipt
         // should return payment required
         let req = Request::new(Body::default());

crates/service/src/middleware/auth/tap.rs

@@ -262,8 +262,10 @@ mod tests {
         DenyListCheck::new(pgpool).await
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_sender_denylist(pgpool: PgPool) {
+    #[tokio::test]
+    async fn test_sender_denylist() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         // Add the sender to the denylist
         sqlx::query!(
             r#"
@@ -292,8 +294,10 @@ mod tests {
             .is_err());
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn test_sender_denylist_updates(pgpool: PgPool) {
+    #[tokio::test]
+    async fn test_sender_denylist_updates() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         let signed_receipt = create_signed_receipt(SignedReceiptRequest::builder().build()).await;
 
         let deny_list_check = new_deny_list_check(pgpool.clone()).await;

crates/service/src/tap/checks/deny_list_check.rs

@@ -368,7 +368,6 @@ enum CostModelNotification {
 mod tests {
     use std::time::Duration;
 
-    use sqlx::PgPool;
     use tap_core::receipt::{checks::Check, Context};
     use test_assets::{create_signed_receipt, flush_messages, SignedReceiptRequest};
     use tokio::time::sleep;
@@ -379,14 +378,18 @@ mod tests {
         tap::{CheckingReceipt, TapReceipt},
     };
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn initialize_check(pgpool: PgPool) {
+    #[tokio::test]
+    async fn initialize_check() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         let check = MinimumValue::new(pgpool, Duration::from_secs(0)).await;
         assert_eq!(check.cost_model_map.read().unwrap().len(), 0);
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn should_initialize_check_with_models(pgpool: PgPool) {
+    #[tokio::test]
+    async fn should_initialize_check_with_models() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         // insert 2 cost models for different deployment_id
         let test_models = test::test_data();
 
@@ -399,8 +402,10 @@ mod tests {
         assert!(check.global_model.read().unwrap().is_none());
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn should_watch_model_insert(pgpool: PgPool) {
+    #[tokio::test]
+    async fn should_watch_model_insert() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         let mut check = MinimumValue::new(pgpool.clone(), Duration::from_secs(0)).await;
         assert_eq!(check.cost_model_map.read().unwrap().len(), 0);
 
@@ -416,8 +421,10 @@ mod tests {
         );
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn should_watch_model_remove(pgpool: PgPool) {
+    #[tokio::test]
+    async fn should_watch_model_remove() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         // insert 2 cost models for different deployment_id
         let test_models = test::test_data();
         add_cost_models(&pgpool, to_db_models(test_models.clone())).await;
@@ -436,17 +443,21 @@ mod tests {
         assert_eq!(check.cost_model_map.read().unwrap().len(), 0);
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn should_start_global_model(pgpool: PgPool) {
+    #[tokio::test]
+    async fn should_start_global_model() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         let global_model = global_cost_model();
         add_cost_models(&pgpool, vec![global_model.clone()]).await;
 
         let check = MinimumValue::new(pgpool.clone(), Duration::from_secs(0)).await;
         assert!(check.global_model.read().unwrap().is_some());
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn should_watch_global_model(pgpool: PgPool) {
+    #[tokio::test]
+    async fn should_watch_global_model() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         let mut check = MinimumValue::new(pgpool.clone(), Duration::from_secs(0)).await;
 
         let global_model = global_cost_model();
@@ -457,8 +468,10 @@ mod tests {
         assert!(check.global_model.read().unwrap().is_some());
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn should_remove_global_model(pgpool: PgPool) {
+    #[tokio::test]
+    async fn should_remove_global_model() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         let global_model = global_cost_model();
         add_cost_models(&pgpool, vec![global_model.clone()]).await;
 
@@ -475,8 +488,10 @@ mod tests {
         assert_eq!(check.cost_model_map.read().unwrap().len(), 0);
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn should_check_minimal_value(pgpool: PgPool) {
+    #[tokio::test]
+    async fn should_check_minimal_value() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         // insert cost models for different deployment_id
         let test_models = test::test_data();
 
@@ -565,8 +580,10 @@ mod tests {
             .expect("should accept more than minimal");
     }
 
-    #[sqlx::test(migrations = "../../migrations")]
-    async fn should_check_using_global(pgpool: PgPool) {
+    #[tokio::test]
+    async fn should_check_using_global() {
+        let test_db = test_assets::setup_shared_test_db().await;
+        let pgpool = test_db.pool;
         // insert cost models for different deployment_id
         let test_models = test::test_data();
         let global_model = global_cost_model();