refactor(tests_integration): separate sequencer setup from test setup

crates/papyrus_network/src/network_manager/test_utils.rs

@@ -188,7 +188,7 @@ where
     const BUFFER_SIZE: usize = 1000;
 
     let mut channels_configs = create_connected_network_configs(n_configs + 1);
-    let broadcast_channels = channels_configs.pop().unwrap();
+    let broadcast_channels = channels_configs.remove(0);
 
     let mut channels_network_manager = NetworkManager::new(broadcast_channels, None);
     let broadcast_channels =

crates/starknet_integration_tests/src/flow_test_setup.rs

@@ -1,51 +1,100 @@
 use std::net::SocketAddr;
 
-use mempool_test_utils::starknet_api_test_utils::MultiAccountTransactionGenerator;
+use blockifier::context::ChainInfo;
+use mempool_test_utils::starknet_api_test_utils::{Contract, MultiAccountTransactionGenerator};
 use papyrus_network::network_manager::BroadcastTopicChannels;
 use papyrus_protobuf::consensus::ProposalPart;
 use starknet_api::rpc_transaction::RpcTransaction;
 use starknet_api::transaction::TransactionHash;
+use starknet_consensus_manager::config::ConsensusManagerConfig;
 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;
 use starknet_task_executor::tokio_executor::TokioExecutor;
 use tempfile::TempDir;
 use tokio::runtime::Handle;
 use tokio::task::JoinHandle;
+use tracing::{debug, instrument};
 
 use crate::state_reader::{spawn_test_rpc_state_reader, StorageTestSetup};
-use crate::utils::{create_chain_info, create_config};
+use crate::utils::{
+    create_chain_info,
+    create_config,
+    create_consensus_manager_configs_and_channels,
+};
+
+const PROPOSER_ID: usize = 0;
 
 pub struct FlowTestSetup {
     pub task_executor: TokioExecutor,
-
-    // Client for adding transactions to the sequencer node.
-    pub add_tx_http_client: HttpTestClient,
-
-    // Handlers for the storage files, maintained so the files are not deleted.
-    pub batcher_storage_file_handle: TempDir,
-    pub rpc_storage_file_handle: TempDir,
-
-    // Handle of the sequencer node.
-    pub sequencer_node_handle: JoinHandle<Result<(), anyhow::Error>>,
+    pub proposer: SequencerTestSetup,
 
     // Channels for consensus proposals, used for asserting the right transactions are proposed.
     pub consensus_proposals_channels: BroadcastTopicChannels<ProposalPart>,
 }
 
 impl FlowTestSetup {
+    #[instrument(skip(tx_generator), level = "debug")]
     pub async fn new_from_tx_generator(tx_generator: &MultiAccountTransactionGenerator) -> Self {
         let handle = Handle::current();
         let task_executor = TokioExecutor::new(handle);
         let chain_info = create_chain_info();
 
-        // Configure and start tracing.
-        configure_tracing();
-
         let accounts = tx_generator.accounts();
+        let (mut consensus_manager_configs, consensus_proposals_channels) =
+            create_consensus_manager_configs_and_channels(1);
+
+        let proposer_consensus_manager_config = consensus_manager_configs.remove(0);
+        let proposer = SequencerTestSetup::new(
+            accounts.clone(),
+            PROPOSER_ID,
+            chain_info.clone(),
+            &task_executor,
+            proposer_consensus_manager_config,
+        )
+        .await;
+
+        Self { task_executor, proposer, consensus_proposals_channels }
+    }
+
+    pub async fn assert_add_tx_success(&self, tx: RpcTransaction) -> TransactionHash {
+        self.proposer.add_tx_http_client.assert_add_tx_success(tx).await
+    }
+
+    pub async fn assert_add_tx_error(&self, tx: RpcTransaction) -> GatewaySpecError {
+        self.proposer.add_tx_http_client.assert_add_tx_error(tx).await
+    }
+}
+
+pub struct SequencerTestSetup {
+    /// Used to differentiate between different sequencer nodes.
+    pub sequencer_id: usize,
+
+    // Client for adding transactions to the sequencer node.
+    pub add_tx_http_client: HttpTestClient,
+
+    // Handlers for the storage files, maintained so the files are not deleted.
+    pub batcher_storage_file_handle: TempDir,
+    pub rpc_storage_file_handle: TempDir,
+
+    // Handle of the sequencer node.
+    pub sequencer_node_handle: JoinHandle<Result<(), anyhow::Error>>,
+}
+
+impl SequencerTestSetup {
+    #[instrument(
+        skip(accounts, chain_info, task_executor, consensus_manager_config),
+        level = "debug"
+    )]
+    pub async fn new(
+        accounts: Vec<Contract>,
+        sequencer_id: usize,
+        chain_info: ChainInfo,
+        task_executor: &TokioExecutor,
+        consensus_manager_config: ConsensusManagerConfig,
+    ) -> Self {
         let storage_for_test = StorageTestSetup::new(accounts, chain_info.chain_id.clone());
 
         // Spawn a papyrus rpc server for a papyrus storage reader.
@@ -55,10 +104,19 @@ impl FlowTestSetup {
         )
         .await;
 
+        debug!("Rpc server spawned at: {}", rpc_server_addr);
+
         // Derive the configuration for the sequencer node.
-        let (config, _required_params, consensus_proposals_channels) =
-            create_config(chain_info, rpc_server_addr, storage_for_test.batcher_storage_config)
-                .await;
+        let (config, _required_params) = create_config(
+            sequencer_id,
+            chain_info,
+            rpc_server_addr,
+            storage_for_test.batcher_storage_config,
+            consensus_manager_config,
+        )
+        .await;
+
+        debug!("Sequencer config: {:#?}", config);
 
         let (_clients, servers) = create_node_modules(&config);
 
@@ -75,20 +133,11 @@ impl FlowTestSetup {
         tokio::time::sleep(std::time::Duration::from_millis(100)).await;
 
         Self {
-            task_executor,
+            sequencer_id,
             add_tx_http_client,
             batcher_storage_file_handle: storage_for_test.batcher_storage_handle,
             rpc_storage_file_handle: storage_for_test.rpc_storage_handle,
             sequencer_node_handle,
-            consensus_proposals_channels,
         }
     }
-
-    pub async fn assert_add_tx_success(&self, tx: RpcTransaction) -> TransactionHash {
-        self.add_tx_http_client.assert_add_tx_success(tx).await
-    }
-
-    pub async fn assert_add_tx_error(&self, tx: RpcTransaction) -> GatewaySpecError {
-        self.add_tx_http_client.assert_add_tx_error(tx).await
-    }
 }

crates/starknet_integration_tests/src/integration_test_setup.rs

@@ -11,7 +11,11 @@ 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};
+use crate::utils::{
+    create_chain_info,
+    create_config,
+    create_consensus_manager_configs_and_channels,
+};
 
 pub struct IntegrationTestSetup {
     // Client for adding transactions to the sequencer node.
@@ -47,10 +51,18 @@ impl IntegrationTestSetup {
         )
         .await;
 
+        let (mut consensus_manager_configs, _consensus_proposals_channels) =
+            create_consensus_manager_configs_and_channels(1);
+
         // Derive the configuration for the sequencer node.
-        let (config, required_params, _) =
-            create_config(chain_info, rpc_server_addr, storage_for_test.batcher_storage_config)
-                .await;
+        let (config, required_params) = create_config(
+            0,
+            chain_info,
+            rpc_server_addr,
+            storage_for_test.batcher_storage_config,
+            consensus_manager_configs.pop().unwrap(),
+        )
+        .await;
 
         let node_config_dir_handle = tempdir().unwrap();
         let node_config_path = dump_config_file_changes(

crates/starknet_integration_tests/src/utils.rs

@@ -7,13 +7,14 @@ use blockifier::test_utils::contracts::FeatureContract;
 use blockifier::test_utils::CairoVersion;
 use mempool_test_utils::starknet_api_test_utils::{AccountId, MultiAccountTransactionGenerator};
 use papyrus_consensus::config::ConsensusConfig;
+use papyrus_consensus::types::ValidatorId;
 use papyrus_network::network_manager::test_utils::create_network_configs_connected_to_broadcast_channels;
 use papyrus_network::network_manager::BroadcastTopicChannels;
 use papyrus_protobuf::consensus::ProposalPart;
 use papyrus_storage::StorageConfig;
 use starknet_api::block::BlockNumber;
 use starknet_api::contract_address;
-use starknet_api::core::ContractAddress;
+use starknet_api::core::{ChainId, ContractAddress};
 use starknet_api::rpc_transaction::RpcTransaction;
 use starknet_api::transaction::TransactionHash;
 use starknet_batcher::block_builder::BlockBuilderConfig;
@@ -26,9 +27,16 @@ use starknet_gateway::config::{
     StatelessTransactionValidatorConfig,
 };
 use starknet_http_server::config::HttpServerConfig;
+use starknet_mempool_p2p::config::MempoolP2pConfig;
+use starknet_monitoring_endpoint::config::MonitoringEndpointConfig;
 use starknet_sequencer_infra::test_utils::get_available_socket;
 use starknet_sequencer_node::config::node_config::SequencerNodeConfig;
 use starknet_sequencer_node::config::test_utils::RequiredParams;
+use starknet_types_core::felt::Felt;
+
+// Currently the orchestrator expects the sequencer addresses in the form of [0..n_sequencers).
+// TODO(yair): Change this offset to be a non-zero value once the orchestrator is updated.
+pub const SEQUENCER_ADDRESS_OFFSET: usize = 0;
 
 pub fn create_chain_info() -> ChainInfo {
     let mut chain_info = ChainInfo::create_for_testing();
@@ -38,39 +46,45 @@ pub fn create_chain_info() -> ChainInfo {
     chain_info
 }
 
+// TODO(yair, itay): Create config presets for tests.
 pub async fn create_config(
+    sequencer_id: usize,
     chain_info: ChainInfo,
     rpc_server_addr: SocketAddr,
     batcher_storage_config: StorageConfig,
-) -> (SequencerNodeConfig, RequiredParams, BroadcastTopicChannels<ProposalPart>) {
+    mut consensus_manager_config: ConsensusManagerConfig,
+) -> (SequencerNodeConfig, RequiredParams) {
+    set_validator_id(&mut consensus_manager_config, sequencer_id);
     let fee_token_addresses = chain_info.fee_token_addresses.clone();
     let batcher_config = create_batcher_config(batcher_storage_config, chain_info.clone());
     let gateway_config = create_gateway_config(chain_info.clone()).await;
     let http_server_config = create_http_server_config().await;
     let rpc_state_reader_config = test_rpc_state_reader_config(rpc_server_addr);
-    let (mut consensus_manager_configs, consensus_proposals_channels) =
-        create_consensus_manager_configs_and_channels(1);
-    let consensus_manager_config = consensus_manager_configs.pop().unwrap();
+    let mempool_p2p_config = create_mempool_p2p_config(sequencer_id, chain_info.chain_id.clone());
+    let monitoring_endpoint_config = create_monitoring_endpoint_config(sequencer_id);
+    let sequencer_address = create_sequencer_address(sequencer_id);
+
     (
         SequencerNodeConfig {
             batcher_config,
             consensus_manager_config,
             gateway_config,
             http_server_config,
             rpc_state_reader_config,
-            ..SequencerNodeConfig::default()
+            mempool_p2p_config,
+            monitoring_endpoint_config,
+            ..Default::default()
         },
         RequiredParams {
             chain_id: chain_info.chain_id,
             eth_fee_token_address: fee_token_addresses.eth_fee_token_address,
             strk_fee_token_address: fee_token_addresses.strk_fee_token_address,
-            sequencer_address: ContractAddress::from(1312_u128), // Arbitrary non-zero value.
+            sequencer_address,
         },
-        consensus_proposals_channels,
     )
 }
 
-fn create_consensus_manager_configs_and_channels(
+pub fn create_consensus_manager_configs_and_channels(
     n_managers: usize,
 ) -> (Vec<ConsensusManagerConfig>, BroadcastTopicChannels<ProposalPart>) {
     let (network_configs, broadcast_channels) =
@@ -88,6 +102,7 @@ fn create_consensus_manager_configs_and_channels(
                 start_height: BlockNumber(1),
                 consensus_delay: Duration::from_secs(1),
                 network_config,
+                num_validators: u64::try_from(n_managers).unwrap(),
                 ..Default::default()
             },
         })
@@ -234,3 +249,29 @@ pub fn create_batcher_config(
         ..Default::default()
     }
 }
+
+fn set_validator_id(consensus_manager_config: &mut ConsensusManagerConfig, sequencer_id: usize) {
+    consensus_manager_config.consensus_config.validator_id = ValidatorId::try_from(
+        Felt::from(consensus_manager_config.consensus_config.validator_id)
+            + Felt::from(sequencer_id),
+    )
+    .unwrap();
+}
+
+fn create_sequencer_address(sequencer_id: usize) -> ContractAddress {
+    ContractAddress::from(u128::try_from(SEQUENCER_ADDRESS_OFFSET + sequencer_id).unwrap())
+}
+
+fn create_mempool_p2p_config(sequencer_id: usize, chain_id: ChainId) -> MempoolP2pConfig {
+    let mut config = MempoolP2pConfig::default();
+    config.network_config.tcp_port += u16::try_from(sequencer_id).unwrap();
+    config.network_config.quic_port += u16::try_from(sequencer_id).unwrap();
+    config.network_config.chain_id = chain_id;
+    config
+}
+
+fn create_monitoring_endpoint_config(sequencer_id: usize) -> MonitoringEndpointConfig {
+    let mut config = MonitoringEndpointConfig::default();
+    config.port += u16::try_from(sequencer_id).unwrap();
+    config
+}

crates/starknet_integration_tests/tests/end_to_end_flow_test.rs

@@ -29,6 +29,8 @@ fn tx_generator() -> MultiAccountTransactionGenerator {
 #[rstest]
 #[tokio::test]
 async fn end_to_end(mut tx_generator: MultiAccountTransactionGenerator) {
+    starknet_sequencer_infra::trace_util::configure_tracing();
+
     const LISTEN_TO_BROADCAST_MESSAGES_TIMEOUT: std::time::Duration =
         std::time::Duration::from_secs(5);
     // Setup.