Skip to content

apollo_integration_tests: move end-to-end flow logic into library flow modules #14879

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
avi-starkware wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

apollo_integration_tests: move end-to-end flow logic into library flow modules #14879

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 2 additions & 56 deletions ...sequencer_node_end_to_end_integration_tests/integration_test_central_and_p2p_sync_flow.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,60 +1,6 @@
use apollo_infra_utils::test_utils::TestIdentifier;
use apollo_integration_tests::integration_test_manager::IntegrationTestManager;
use apollo_integration_tests::integration_test_utils::integration_test_setup;
use apollo_integration_tests::utils::NodeDescriptor;
use apollo_node_config::component_execution_config::{
ActiveComponentExecutionMode,
ReactiveComponentExecutionMode,
};
use apollo_node_config::node_config::SequencerNodeConfig;
use apollo_state_sync_config::config::CentralSyncClientConfig;
use starknet_api::block::BlockNumber;
use tracing::info;
use apollo_integration_tests::flows;

#[tokio::main]
async fn main() {
integration_test_setup("sync").await;
const BLOCK_TO_WAIT_FOR: BlockNumber = BlockNumber(20);
// Node layout (index → type): 0 = consolidated, 1 = consolidated.
// Node 1 is configured as the central sync node.
const CENTRAL_SYNC_NODE: usize = 1;

let node_descriptors = vec![NodeDescriptor::consolidated(), NodeDescriptor::consolidated()];

let mut integration_test_manager = IntegrationTestManager::new(
node_descriptors,
None,
TestIdentifier::SyncFlowIntegrationTest,
)
.await;

let update_config_disable_everything_but_sync = |config: &mut SequencerNodeConfig| {
config.components.batcher.execution_mode = ReactiveComponentExecutionMode::Disabled;
config.components.gateway.execution_mode = ReactiveComponentExecutionMode::Disabled;
config.components.mempool.execution_mode = ReactiveComponentExecutionMode::Disabled;
config.components.mempool_p2p.execution_mode = ReactiveComponentExecutionMode::Disabled;
config.components.l1_events_provider.execution_mode =
ReactiveComponentExecutionMode::Disabled;
config.components.consensus_manager.execution_mode = ActiveComponentExecutionMode::Disabled;
config.components.http_server.execution_mode = ActiveComponentExecutionMode::Disabled;
config.components.l1_events_scraper.execution_mode = ActiveComponentExecutionMode::Disabled;
};

let update_config_use_central_sync = |config: &mut SequencerNodeConfig| {
config.state_sync_config.as_mut().unwrap().static_config.central_sync_client_config =
Some(CentralSyncClientConfig::default());
config.state_sync_config.as_mut().unwrap().static_config.p2p_sync_client_config = None;
};

let node_indices = integration_test_manager.get_node_indices();
integration_test_manager
.modify_config_idle_nodes(node_indices.clone(), update_config_disable_everything_but_sync);
integration_test_manager
.modify_config_idle_nodes([CENTRAL_SYNC_NODE].into(), update_config_use_central_sync);

integration_test_manager.run_nodes(node_indices.clone()).await;

integration_test_manager.await_sync_block_on_all_running_nodes(BLOCK_TO_WAIT_FOR).await;

info!("Sync flow integration test completed successfully!");
flows::sync::run().await;
}
55 changes: 2 additions & 53 deletions ...sts/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_positive_flow.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,57 +1,6 @@
use apollo_infra_utils::test_utils::TestIdentifier;
use apollo_integration_tests::integration_test_manager::IntegrationTestManager;
use apollo_integration_tests::integration_test_utils::integration_test_setup;
use apollo_integration_tests::utils::NodeDescriptor;
use starknet_api::block::BlockNumber;
use tracing::info;
use apollo_integration_tests::flows;

#[tokio::main]
async fn main() {
integration_test_setup("positive").await;
const BLOCK_TO_WAIT_FOR: BlockNumber = BlockNumber(15);
const N_INVOKE_TXS: usize = 50;
const N_L1_HANDLER_TXS: usize = 2;

let node_descriptors = vec![
NodeDescriptor::consolidated(),
NodeDescriptor::consolidated(),
NodeDescriptor::consolidated(),
NodeDescriptor::distributed(),
NodeDescriptor::hybrid(),
];

// Get the sequencer configurations.
let mut integration_test_manager = IntegrationTestManager::new(
node_descriptors,
None,
TestIdentifier::PositiveFlowIntegrationTest,
)
.await;

// TODO(Tsabary): consider decreasing
// "consensus_manager_config.consensus_manager_config.static_config.startup_delay" and
// "batcher_config.static_config.block_builder_config.proposer_idle_detection_delay_millis".

let node_indices = integration_test_manager.get_node_indices();
// Run the nodes.
integration_test_manager.run_nodes(node_indices.clone()).await;

// Run the first block scenario to deploy the accounts.
integration_test_manager.send_deploy_and_invoke_txs_and_verify().await;

integration_test_manager.send_declare_txs_and_verify().await;

// Run the test.
integration_test_manager
.send_txs_and_verify(N_INVOKE_TXS, N_L1_HANDLER_TXS, BLOCK_TO_WAIT_FOR)
.await;

integration_test_manager
.verify_block_hash_across_all_running_nodes(Some(BLOCK_TO_WAIT_FOR.unchecked_next()))
.await;

info!("Shutting down nodes.");
integration_test_manager.shutdown_nodes(node_indices).await;

info!("Positive flow integration test completed successfully!");
flows::positive::run().await;
}
52 changes: 2 additions & 50 deletions ..._tests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_proof_flow.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,54 +1,6 @@
//! Integration test for the proof submission flow.
//!
//! The test relies on pre-generated fixture files in
//! `crates/apollo_integration_tests/resources/proof_flow/`.

use apollo_infra_utils::test_utils::TestIdentifier;
use apollo_integration_tests::integration_test_manager::{
IntegrationTestManager,
DEFAULT_SENDER_ACCOUNT,
};
use apollo_integration_tests::integration_test_utils::integration_test_setup;
use apollo_integration_tests::utils::{NodeDescriptor, ProofFlowTxs};
use blockifier::abi::constants::STORED_BLOCK_HASH_BUFFER;
use starknet_api::block::BlockNumber;
use tracing::info;
use apollo_integration_tests::flows;

#[tokio::main]
async fn main() {
integration_test_setup("proof_flow").await;

// The fixture's proof references block 0, so the proof-bearing tx is only valid once the
// chain has progressed past `STORED_BLOCK_HASH_BUFFER`. We first advance the chain past that
// buffer with filler invokes, then submit the proof tx and wait one more block.
const BLOCK_PAST_HASH_BUFFER: BlockNumber = BlockNumber(STORED_BLOCK_HASH_BUFFER);
const BLOCK_TO_WAIT_FOR: BlockNumber = BlockNumber(STORED_BLOCK_HASH_BUFFER + 1);
let node_descriptors = vec![
NodeDescriptor::consolidated(),
NodeDescriptor::distributed(),
NodeDescriptor::hybrid(),
];

let mut integration_test_manager = IntegrationTestManager::new(
node_descriptors,
None,
TestIdentifier::ProofFlowIntegrationTest,
)
.await;

let node_indices = integration_test_manager.get_node_indices();
integration_test_manager.run_nodes(node_indices.clone()).await;

integration_test_manager.send_deploy_and_invoke_txs_and_verify().await;

integration_test_manager.await_block_on_all_running_nodes(BLOCK_PAST_HASH_BUFFER).await;

integration_test_manager
.test_and_verify(ProofFlowTxs::new(), DEFAULT_SENDER_ACCOUNT, BLOCK_TO_WAIT_FOR)
.await;

info!("Shutting down nodes.");
integration_test_manager.shutdown_nodes(node_indices).await;

info!("Proof flow integration test completed successfully!");
flows::proof::run().await;
}
101 changes: 2 additions & 99 deletions ...ests/src/bin/sequencer_node_end_to_end_integration_tests/integration_test_restart_flow.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,103 +1,6 @@
use std::time::Duration;

use apollo_deployments::service::NodeType;
use apollo_infra_utils::test_utils::TestIdentifier;
use apollo_integration_tests::integration_test_manager::{
IntegrationTestManager,
DEFAULT_SENDER_ACCOUNT,
};
use apollo_integration_tests::integration_test_utils::integration_test_setup;
use apollo_integration_tests::utils::NodeDescriptor;
use tracing::info;
use apollo_integration_tests::flows;

#[tokio::main]
async fn main() {
integration_test_setup("restart").await;
const TIMEOUT: Duration = Duration::from_secs(30);
const LONG_TIMEOUT: Duration = Duration::from_secs(90);
// Node layout (index → type): 0 = consolidated, 1 = hybrid, 2 = distributed.
// The test restarts a hybrid node and shuts down a non-consolidated (hybrid/distributed) node.
const RESTART_NODE: usize = 1; // hybrid
const SHUTDOWN_NODE: usize = 2; // distributed

let node_descriptors = vec![
NodeDescriptor::consolidated(),
NodeDescriptor::hybrid(),
NodeDescriptor::distributed(),
];

// Get the sequencer configurations.
let mut integration_test_manager = IntegrationTestManager::new(
node_descriptors,
None,
TestIdentifier::RestartFlowIntegrationTest,
)
.await;

// Assert that RESTART_NODE is a hybrid node.
assert_eq!(integration_test_manager.get_node_type(RESTART_NODE), NodeType::Hybrid);
// Assert that SHUTDOWN_NODE is not a consolidated node.
assert_ne!(integration_test_manager.get_node_type(SHUTDOWN_NODE), NodeType::Consolidated);

let mut node_indices = integration_test_manager.get_node_indices();

info!("Running all nodes.");
integration_test_manager.run_nodes(node_indices.clone()).await;

integration_test_manager.send_deploy_and_invoke_txs_and_verify().await;

integration_test_manager.send_declare_txs_and_verify().await;

// Create a simulator for sustained transaction sending.
let simulator = integration_test_manager.create_simulator();
let mut tx_generator = integration_test_manager.tx_generator().snapshot();

// Task that awaits transactions and restarts nodes in phases.
let await_and_restart_nodes_task = async {
info!("Awaiting transactions while all nodes are up");
integration_test_manager.poll_all_running_nodes_received_more_txs(TIMEOUT).await;

integration_test_manager.shutdown_nodes([RESTART_NODE].into()).await;
info!("Awaiting transactions while node {RESTART_NODE} is down");
integration_test_manager.poll_all_running_nodes_received_more_txs(TIMEOUT).await;

// We want the restarted node to rejoin the network while its building blocks to check the
// catch-up mechanism.
integration_test_manager.run_nodes([RESTART_NODE].into()).await;
info!(
"Awaiting node {RESTART_NODE} to join consensus after it was restarted and before \
node {SHUTDOWN_NODE} is shut down"
);

integration_test_manager
.poll_node_reaches_consensus_decisions_after_restart(RESTART_NODE, LONG_TIMEOUT)
.await;

integration_test_manager.poll_all_running_nodes_received_more_txs(TIMEOUT).await;

// Shutdown a second node to test that the restarted node has joined consensus (the network
// can't reach consensus without the restarted node if the second node is down).
integration_test_manager.shutdown_nodes([SHUTDOWN_NODE].into()).await;
// Shutting down a node that's already down results in an error so we remove it from the set
// here.
node_indices.remove(&SHUTDOWN_NODE);
info!(
"Awaiting transactions while node {RESTART_NODE} is up and node {SHUTDOWN_NODE} is \
down"
);
integration_test_manager.poll_all_running_nodes_received_more_txs(LONG_TIMEOUT).await;
};

simulator
.run_test_with_nonstop_tx_sending(
&mut tx_generator,
DEFAULT_SENDER_ACCOUNT,
await_and_restart_nodes_task,
)
.await;

integration_test_manager.verify_block_hash_across_all_running_nodes(None).await;

integration_test_manager.shutdown_nodes(node_indices).await;
info!("Restart flow integration test completed successfully!");
flows::restart::run().await;
}
84 changes: 2 additions & 82 deletions ...node_end_to_end_integration_tests/integration_test_restart_service_multiple_nodes_flow.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,86 +1,6 @@
use std::collections::HashMap;

use apollo_deployments::deployments::hybrid::HybridNodeServiceName;
use apollo_deployments::service::NodeService;
use apollo_infra_utils::test_utils::TestIdentifier;
use apollo_integration_tests::integration_test_manager::IntegrationTestManager;
use apollo_integration_tests::integration_test_utils::integration_test_setup;
use apollo_integration_tests::utils::NodeDescriptor;
use starknet_api::block::BlockNumber;
use strum::IntoEnumIterator;
use tracing::info;
use apollo_integration_tests::flows;

#[tokio::main]
async fn main() {
integration_test_setup("restart_service_multiple_nodes").await;
const INITIAL_BLOCK_TO_WAIT_FOR: usize = 20;
const BLOCK_TO_WAIT_FOR_INCREMENT: usize = 5;
const N_INVOKE_TXS: usize = 20;
const N_L1_HANDLER_TXS: usize = 1;

let node_descriptors = vec![
NodeDescriptor::hybrid(),
NodeDescriptor::hybrid(),
NodeDescriptor::hybrid(),
NodeDescriptor::hybrid(),
NodeDescriptor::hybrid(),
];

// Get the sequencer configurations.
let mut integration_test_manager = IntegrationTestManager::new(
node_descriptors,
None,
TestIdentifier::RestartServiceMultipleNodesFlowIntegrationTest,
)
.await;

let node_indices = integration_test_manager.get_node_indices();
// Run the nodes.
integration_test_manager.run_nodes(node_indices.clone()).await;

// Run the first block scenario to deploy the accounts.
integration_test_manager.send_deploy_and_invoke_txs_and_verify().await;

integration_test_manager.send_declare_txs_and_verify().await;
let mut block_to_wait_for = INITIAL_BLOCK_TO_WAIT_FOR;
for (i, hybrid_node_service) in HybridNodeServiceName::iter().enumerate() {
// TODO(noamsp): Remove this once the equivocaton feature is merged.
if hybrid_node_service == HybridNodeServiceName::Core {
continue;
}

let node_services_to_shutdown = node_indices
.iter()
.map(|&node_index| (node_index, vec![hybrid_node_service.into()]))
.collect::<HashMap<usize, Vec<NodeService>>>();

info!("Shutting down service {hybrid_node_service:?} for all nodes.");
integration_test_manager.shutdown_node_services(node_services_to_shutdown.clone());
info!("Running service {hybrid_node_service:?} for all nodes.");
integration_test_manager.run_node_services(node_services_to_shutdown).await;

block_to_wait_for = INITIAL_BLOCK_TO_WAIT_FOR + i * BLOCK_TO_WAIT_FOR_INCREMENT;
info!(
"Sending txs and verifying after restarting service {hybrid_node_service:?} for all \
nodes."
);
integration_test_manager
.send_txs_and_verify(
N_INVOKE_TXS,
N_L1_HANDLER_TXS,
BlockNumber(block_to_wait_for.try_into().expect("Failed to convert to u64")),
)
.await;
}

integration_test_manager
.verify_block_hash_across_all_running_nodes(Some(BlockNumber(
(block_to_wait_for + 1).try_into().expect("Failed to convert to u64"),
)))
.await;

info!("Shutting down nodes.");
integration_test_manager.shutdown_nodes(node_indices).await;

info!("Restart service multiple nodes flow integration test completed successfully!");
flows::restart_multiple_nodes::run().await;
}
Loading
Loading