test(mempool): tx added to the mempool are forwarded to the p2p propa…

…gator client

crates/starknet_mempool/Cargo.toml

@@ -19,15 +19,21 @@ starknet_api.workspace = true
 starknet_sequencer_infra.workspace = true
 starknet_mempool_p2p_types.workspace = true
 starknet_mempool_types.workspace = true
+tokio.workspace = true
 tracing.workspace = true
 
 [dev-dependencies]
 assert_matches.workspace = true
 itertools.workspace = true
+mockall.workspace = true
+papyrus_network = { workspace = true, features = ["testing"] }
+papyrus_network_types = { workspace = true, features = ["testing"] }
+papyrus_test_utils.workspace = true
 rstest.workspace = true
 starknet_api = { workspace = true, features = ["testing"] }
 # Enable test utils feature for integration tests.
 starknet_mempool = { workspace = true, features = ["testing"] }
+starknet_mempool_p2p.workspace = true
 
 [features]
 testing = ["mempool_test_utils", "pretty_assertions", "starknet-types-core"]

crates/starknet_mempool/src/communication.rs

@@ -78,7 +78,10 @@ impl MempoolCommunicationWrapper {
         }
     }
 
-    async fn add_tx(&mut self, args_wrapper: AddTransactionArgsWrapper) -> MempoolResult<()> {
+    pub(crate) async fn add_tx(
+        &mut self,
+        args_wrapper: AddTransactionArgsWrapper,
+    ) -> MempoolResult<()> {
         self.mempool.add_tx(args_wrapper.args.clone())?;
         // TODO: Verify that only transactions that were added to the mempool are sent.
         // TODO: handle declare correctly and remove this match.

crates/starknet_mempool/src/mempool_test.rs

@@ -1,11 +1,24 @@
+use std::sync::Arc;
+
+use mockall::predicate;
+use papyrus_network_types::network_types::BroadcastedMessageMetadata;
+use papyrus_test_utils::{get_rng, GetTestInstance};
 use pretty_assertions::assert_eq;
 use rstest::{fixture, rstest};
 use starknet_api::block::GasPrice;
 use starknet_api::executable_transaction::AccountTransaction;
+use starknet_api::rpc_transaction::{
+    RpcDeployAccountTransaction,
+    RpcInvokeTransaction,
+    RpcTransaction,
+};
 use starknet_api::{contract_address, nonce};
+use starknet_mempool_p2p_types::communication::MockMempoolP2pPropagatorClient;
+use starknet_mempool_types::communication::AddTransactionArgsWrapper;
 use starknet_mempool_types::errors::MempoolError;
 use starknet_mempool_types::mempool_types::AddTransactionArgs;
 
+use crate::communication::MempoolCommunicationWrapper;
 use crate::mempool::{Mempool, MempoolConfig, TransactionReference};
 use crate::test_utils::{add_tx, add_tx_expect_error, commit_block, get_txs_and_assert_expected};
 use crate::transaction_pool::TransactionPool;
@@ -687,3 +700,60 @@ fn test_update_gas_price_threshold_decreases_threshold() {
         .build();
     expected_mempool_content.assert_eq(&mempool);
 }
+
+#[rstest]
+#[tokio::test]
+async fn test_new_tx_sent_to_p2p(mempool: Mempool) {
+    // add_tx_input! creates an Invoke Transaction
+    let tx_1_args = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 2, account_nonce: 2);
+    let new_tx_args =
+        AddTransactionArgsWrapper { args: tx_1_args.clone(), p2p_message_metadata: None };
+    let new_rpc_tx = match tx_1_args.tx {
+        AccountTransaction::Declare(_declare_tx) => {
+            panic!("No implementation for converting DeclareTransaction to an RpcTransaction")
+        }
+        AccountTransaction::DeployAccount(deploy_account_transaction) => {
+            RpcTransaction::DeployAccount(RpcDeployAccountTransaction::V3(
+                deploy_account_transaction.clone().into(),
+            ))
+        }
+        AccountTransaction::Invoke(invoke_transaction) => {
+            RpcTransaction::Invoke(RpcInvokeTransaction::V3(invoke_transaction.clone().into()))
+        }
+    };
+
+    let mut mock_mempool_p2p_propagator_client = MockMempoolP2pPropagatorClient::new();
+    mock_mempool_p2p_propagator_client
+        .expect_add_transaction()
+        .times(1)
+        .with(predicate::eq(new_rpc_tx))
+        .returning(|_| Ok(()));
+    let mut mempool_wrapper =
+        MempoolCommunicationWrapper::new(mempool, Arc::new(mock_mempool_p2p_propagator_client));
+
+    mempool_wrapper.add_tx(new_tx_args).await.unwrap();
+}
+
+#[rstest]
+#[tokio::test]
+async fn test_propagated_tx_sent_to_p2p(mempool: Mempool) {
+    // add_tx_input! creates an Invoke Transaction
+    let tx_2_args = add_tx_input!(tx_hash: 2, address: "0x0", tx_nonce: 3, account_nonce: 2);
+    let expected_message_metadata = BroadcastedMessageMetadata::get_test_instance(&mut get_rng());
+    let propagated_tx_args = AddTransactionArgsWrapper {
+        args: tx_2_args.clone(),
+        p2p_message_metadata: Some(expected_message_metadata.clone()),
+    };
+
+    let mut mock_mempool_p2p_propagator_client = MockMempoolP2pPropagatorClient::new();
+    mock_mempool_p2p_propagator_client
+        .expect_continue_propagation()
+        .times(1)
+        .with(predicate::eq(expected_message_metadata.clone()))
+        .returning(|_| Ok(()));
+
+    let mut mempool_wrapper =
+        MempoolCommunicationWrapper::new(mempool, Arc::new(mock_mempool_p2p_propagator_client));
+
+    mempool_wrapper.add_tx(propagated_tx_args).await.unwrap();
+}

crates/starknet_mempool_p2p_types/Cargo.toml

@@ -10,6 +10,7 @@ workspace = true
 
 [dependencies]
 async-trait.workspace = true
+mockall.workspace = true
 papyrus_network_types.workspace = true
 papyrus_proc_macros.workspace = true
 serde = { workspace = true, features = ["derive"] }

crates/starknet_mempool_p2p_types/src/communication.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use mockall::automock;
 use papyrus_network_types::network_types::BroadcastedMessageMetadata;
 use papyrus_proc_macros::handle_response_variants;
 use serde::{Deserialize, Serialize};
@@ -19,6 +20,7 @@ use thiserror::Error;
 use crate::errors::MempoolP2pPropagatorError;
 use crate::mempool_p2p_types::MempoolP2pPropagatorResult;
 
+#[automock]
 #[async_trait]
 pub trait MempoolP2pPropagatorClient: Send + Sync {
     /// Adds a transaction to be propagated to other peers. This should only be called on a new