Add a universal method for synchronizing in tests

Some bors commands do not produce comments, so `expect_comments`/`get_comment` cannot be used for them.

Author: Kobzol

src/bors/handlers/review.rs

@@ -547,11 +547,12 @@ approve = ["+approved"]
     async fn set_priority(pool: sqlx::PgPool) {
         run_test(pool, |mut tester| async {
             tester.post_comment("@bors p=5").await?;
-            // Wait for db update.
-            tester.expect_comments(1).await;
-            let pr = tester.get_default_pr().await?;
-
-            assert_eq!(pr.priority, Some(5));
+            tester
+                .wait_for(|| async {
+                    let pr = tester.get_default_pr().await?;
+                    Ok(pr.priority == Some(5))
+                })
+                .await?;
             Ok(tester)
         })
         .await;
@@ -561,10 +562,12 @@ approve = ["+approved"]
     async fn priority_preserved_after_approve(pool: sqlx::PgPool) {
         run_test(pool, |mut tester| async {
             tester.post_comment("@bors p=5").await?;
-            tester.expect_comments(1).await;
-
-            let pr = tester.get_default_pr().await?;
-            assert_eq!(pr.priority, Some(5));
+            tester
+                .wait_for(|| async {
+                    let pr = tester.get_default_pr().await?;
+                    Ok(pr.priority == Some(5))
+                })
+                .await?;
 
             tester.post_comment("@bors r+").await?;
             tester.expect_comments(1).await;
@@ -582,10 +585,12 @@ approve = ["+approved"]
     async fn priority_overridden_on_approve_with_priority(pool: sqlx::PgPool) {
         run_test(pool, |mut tester| async {
             tester.post_comment("@bors p=5").await?;
-            tester.expect_comments(1).await;
-
-            let pr = tester.get_default_pr().await?;
-            assert_eq!(pr.priority, Some(5));
+            tester
+                .wait_for(|| async {
+                    let pr = tester.get_default_pr().await?;
+                    Ok(pr.priority == Some(5))
+                })
+                .await?;
 
             tester.post_comment("@bors r+ p=10").await?;
             tester.expect_comments(1).await;

src/tests/mocks/bors.rs

@@ -1,13 +1,13 @@
 use anyhow::Context;
-use std::collections::HashMap;
-use std::future::Future;
-use std::sync::Arc;
-
 use axum::Router;
 use parking_lot::lock_api::MappedMutexGuard;
 use parking_lot::{Mutex, MutexGuard, RawMutex};
 use serde::Serialize;
 use sqlx::PgPool;
+use std::collections::HashMap;
+use std::future::Future;
+use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::mpsc::Sender;
 use tokio::task::JoinHandle;
 use tower::Service;
@@ -351,6 +351,41 @@ impl BorsTester {
         Ok(())
     }
 
+    /// Wait until the given condition is true.
+    /// Checks the condition every 500ms.
+    /// Times out if it takes too long (more than 5s).
+    ///
+    /// This method is useful if you execute a command that produces no comment as an output
+    /// and you need to wait until it has been processed by bors.
+    /// Prefer using [BorsTester::expect_comments] or [BorsTester::get_comment] to synchronize
+    /// if you are waiting for a comment to be posted to a PR.
+    pub async fn wait_for<F, Fut>(&self, condition: F) -> anyhow::Result<()>
+    where
+        F: Fn() -> Fut,
+        Fut: Future<Output = anyhow::Result<bool>>,
+    {
+        let wait_fut = async move {
+            loop {
+                let fut = condition();
+                match fut.await {
+                    Ok(res) => {
+                        if res {
+                            return Ok(());
+                        } else {
+                            tokio::time::sleep(Duration::from_millis(500)).await;
+                        }
+                    }
+                    Err(error) => {
+                        return Err(error);
+                    }
+                }
+            }
+        };
+        tokio::time::timeout(Duration::from_secs(5), wait_fut)
+            .await
+            .unwrap_or_else(|_| Err(anyhow::anyhow!("Timed out waiting for condition")))
+    }
+
     pub async fn finish(self, bors: JoinHandle<()>) -> World {
         // Make sure that the event channel senders are closed
         drop(self.app);