chore(infra): add run_until utility fn

crates/infra_utils/Cargo.toml

@@ -10,7 +10,10 @@ description = "Infrastructure utility."
 workspace = true
 
 [dependencies]
-tokio = { workspace = true, features = ["process"] }
+tokio = { workspace = true, features = ["process", "time"] }
+tracing.workspace = true
 
 [dev-dependencies]
+pretty_assertions.workspace = true
 rstest.workspace = true
+tokio = { workspace = true, features = ["macros", "rt"] }

crates/infra_utils/src/lib.rs

@@ -1,2 +1,3 @@
 pub mod command;
 pub mod path;
+pub mod run_until;

crates/infra_utils/src/run_until.rs

@@ -0,0 +1,95 @@
+use tokio::time::{sleep, Duration};
+use tracing::{debug, error, info, trace, warn};
+
+#[cfg(test)]
+#[path = "run_until_test.rs"]
+mod run_until_test;
+
+/// Struct to hold trace configuration
+pub struct TraceConfig {
+    pub level: LogLevel,
+    pub message: String,
+}
+
+/// Enum for dynamically setting trace level
+#[derive(Clone, Copy)]
+pub enum LogLevel {
+    Trace,
+    Debug,
+    Info,
+    Warn,
+    Error,
+}
+
+/// Runs an asynchronous function until a condition is met or max attempts are reached.
+///
+/// # Arguments
+/// - `interval`: Time between each attempt (in milliseconds).
+/// - `max_attempts`: Maximum number of attempts.
+/// - `executable`: An asynchronous function to execute, which returns a value of type `T`.
+/// - `condition`: A closure that takes a value of type `T` and returns `true` if the condition is
+///   met.
+/// - `trace_config`: Optional trace configuration for logging.
+///
+/// # Returns
+/// - `Option<T>`: Returns `Some(value)` if the condition is met within the attempts, otherwise
+///   `None`.
+pub async fn run_until<T, F, C>(
+    interval: u64,
+    max_attempts: usize,
+    mut executable: F,
+    condition: C,
+    trace_config: Option<TraceConfig>,
+) -> Option<T>
+where
+    T: Clone + Send + std::fmt::Debug + 'static,
+    F: FnMut() -> T + Send,
+    C: Fn(&T) -> bool + Send + Sync,
+{
+    for attempt in 1..=max_attempts {
+        let result = executable();
+
+        // Log attempt message.
+        if let Some(config) = &trace_config {
+            let attempt_message = format!(
+                "{}: Attempt {}/{}, Value {:?}",
+                config.message, attempt, max_attempts, result
+            );
+            log_message(config.level, &attempt_message);
+        }
+
+        // Check if the condition is met.
+        if condition(&result) {
+            if let Some(config) = &trace_config {
+                let success_message = format!(
+                    "{}: Condition met on attempt {}/{}",
+                    config.message, attempt, max_attempts
+                );
+                log_message(config.level, &success_message);
+            }
+            return Some(result);
+        }
+
+        // Wait for the interval before the next attempt.
+        sleep(Duration::from_millis(interval)).await;
+    }
+
+    if let Some(config) = &trace_config {
+        let failure_message =
+            format!("{}: Condition not met after {} attempts.", config.message, max_attempts);
+        log_message(config.level, &failure_message);
+    }
+
+    None
+}
+
+/// Logs a message at the specified level
+fn log_message(level: LogLevel, message: &str) {
+    match level {
+        LogLevel::Trace => trace!("{}", message),
+        LogLevel::Debug => debug!("{}", message),
+        LogLevel::Info => info!("{}", message),
+        LogLevel::Warn => warn!("{}", message),
+        LogLevel::Error => error!("{}", message),
+    }
+}

crates/infra_utils/src/run_until_test.rs

@@ -0,0 +1,46 @@
+use pretty_assertions::assert_eq;
+use rstest::rstest;
+
+use crate::run_until::run_until;
+
+#[rstest]
+#[tokio::test]
+async fn test_run_until_condition_met() {
+    // Mock executable that increments a counter.
+    let mut counter = 0;
+    let mock_executable = || {
+        counter += 1;
+        counter
+    };
+
+    // Condition: stop when the counter reaches 3.
+    let condition = |&result: &i32| result >= 3;
+
+    // Run the function with a short interval and a maximum of 5 attempts.
+    let result = run_until(100, 5, mock_executable, condition, None).await;
+
+    // Assert that the condition was met and the returned value is correct.
+    assert_eq!(result, Some(3));
+    assert_eq!(counter, 3); // Counter should stop at 3 since the condition is met.
+}
+
+#[rstest]
+#[tokio::test]
+async fn test_run_until_condition_not_met() {
+    // Mock executable that increments a counter.
+    let mut counter = 0;
+    let mock_executable = || {
+        counter += 1;
+        counter
+    };
+
+    // Condition: stop when the counter reaches 3.
+    let condition = |&result: &i32| result >= 3;
+
+    // Test that it stops when the maximum attempts are exceeded without meeting the condition.
+    let failed_result = run_until(100, 2, mock_executable, condition, None).await;
+
+    // The condition is not met within 2 attempts, so the result should be None.
+    assert_eq!(failed_result, None);
+    assert_eq!(counter, 2); // Counter should stop at 2 because of max attempts.
+}

crates/starknet_http_server/Cargo.toml

@@ -5,13 +5,18 @@ edition.workspace = true
 license.workspace = true
 repository.workspace = true
 
+[features]
+testing = ["mempool_test_utils", "reqwest"]
+
 [lints]
 workspace = true
 
 [dependencies]
 axum.workspace = true
 hyper.workspace = true
+mempool_test_utils = { workspace = true, optional = true }
 papyrus_config.workspace = true
+reqwest = { workspace = true, optional = true }
 serde.workspace = true
 starknet_api.workspace = true
 starknet_gateway_types.workspace = true

crates/starknet_http_server/src/lib.rs

@@ -2,3 +2,5 @@ pub mod communication;
 pub mod config;
 pub mod errors;
 pub mod http_server;
+#[cfg(feature = "testing")]
+pub mod test_utils;

crates/starknet_http_server/src/test_utils.rs

@@ -0,0 +1,55 @@
+use std::net::SocketAddr;
+
+use axum::body::Body;
+use mempool_test_utils::starknet_api_test_utils::rpc_tx_to_json;
+use reqwest::{Client, Response};
+use starknet_api::rpc_transaction::RpcTransaction;
+use starknet_api::transaction::TransactionHash;
+use starknet_gateway_types::errors::GatewaySpecError;
+use starknet_sequencer_infra::test_utils::get_available_socket;
+
+use crate::config::HttpServerConfig;
+
+/// A test utility client for interacting with an http server.
+pub struct HttpTestClient {
+    socket: SocketAddr,
+    client: Client,
+}
+
+impl HttpTestClient {
+    pub fn new(socket: SocketAddr) -> Self {
+        let client = Client::new();
+        Self { socket, client }
+    }
+
+    pub async fn assert_add_tx_success(&self, rpc_tx: RpcTransaction) -> TransactionHash {
+        let response = self.add_tx(rpc_tx).await;
+        assert!(response.status().is_success());
+
+        response.json().await.unwrap()
+    }
+
+    // TODO: implement when usage eventually arises.
+    pub async fn assert_add_tx_error(&self, _tx: RpcTransaction) -> GatewaySpecError {
+        todo!()
+    }
+
+    // Prefer using assert_add_tx_success or other higher level methods of this client, to ensure
+    // tests are boilerplate and implementation-detail free.
+    pub async fn add_tx(&self, rpc_tx: RpcTransaction) -> Response {
+        let tx_json = rpc_tx_to_json(&rpc_tx);
+        self.client
+            .post(format!("http://{}/add_tx", self.socket))
+            .header("content-type", "application/json")
+            .body(Body::from(tx_json))
+            .send()
+            .await
+            .unwrap()
+    }
+}
+
+pub async fn create_http_server_config() -> HttpServerConfig {
+    // TODO(Tsabary): use ser_generated_param.
+    let socket = get_available_socket().await;
+    HttpServerConfig { ip: socket.ip(), port: socket.port() }
+}

crates/starknet_integration_tests/Cargo.toml

@@ -11,7 +11,6 @@ workspace = true
 [dependencies]
 anyhow.workspace = true
 assert_matches.workspace = true
-axum.workspace = true
 blockifier.workspace = true
 cairo-lang-starknet-classes.workspace = true
 futures.workspace = true
@@ -24,7 +23,6 @@ papyrus_network = { workspace = true, features = ["testing"] }
 papyrus_protobuf.workspace = true
 papyrus_rpc.workspace = true
 papyrus_storage = { workspace = true, features = ["testing"] }
-reqwest.workspace = true
 serde_json.workspace = true
 starknet-types-core.workspace = true
 starknet_api.workspace = true
@@ -33,7 +31,7 @@ starknet_client.workspace = true
 starknet_consensus_manager.workspace = true
 starknet_gateway = { workspace = true, features = ["testing"] }
 starknet_gateway_types.workspace = true
-starknet_http_server.workspace = true
+starknet_http_server = { workspace = true, features = ["testing"] }
 starknet_mempool_p2p.workspace = true
 starknet_monitoring_endpoint = { workspace = true, features = ["testing"] }
 starknet_sequencer_infra = { workspace = true, features = ["testing"] }

crates/starknet_integration_tests/src/flow_test_setup.rs

@@ -7,6 +7,7 @@ use starknet_api::rpc_transaction::RpcTransaction;
 use starknet_api::transaction::TransactionHash;
 use starknet_gateway_types::errors::GatewaySpecError;
 use starknet_http_server::config::HttpServerConfig;
+use starknet_http_server::test_utils::HttpTestClient;
 use starknet_sequencer_infra::trace_util::configure_tracing;
 use starknet_sequencer_node::servers::run_component_servers;
 use starknet_sequencer_node::utils::create_node_modules;
@@ -16,7 +17,7 @@ use tokio::runtime::Handle;
 use tokio::task::JoinHandle;
 
 use crate::state_reader::{spawn_test_rpc_state_reader, StorageTestSetup};
-use crate::utils::{create_chain_info, create_config, HttpTestClient};
+use crate::utils::{create_chain_info, create_config};
 
 pub struct FlowTestSetup {
     pub task_executor: TokioExecutor,

crates/starknet_integration_tests/src/integration_test_setup.rs

@@ -4,13 +4,14 @@ use std::path::PathBuf;
 use mempool_test_utils::starknet_api_test_utils::MultiAccountTransactionGenerator;
 use papyrus_storage::StorageConfig;
 use starknet_http_server::config::HttpServerConfig;
+use starknet_http_server::test_utils::HttpTestClient;
 use starknet_monitoring_endpoint::config::MonitoringEndpointConfig;
 use starknet_monitoring_endpoint::test_utils::IsAliveClient;
 use tempfile::{tempdir, TempDir};
 
 use crate::config_utils::dump_config_file_changes;
 use crate::state_reader::{spawn_test_rpc_state_reader, StorageTestSetup};
-use crate::utils::{create_chain_info, create_config, HttpTestClient};
+use crate::utils::{create_chain_info, create_config};
 
 pub struct IntegrationTestSetup {
     // Client for adding transactions to the sequencer node.