Testkit tests for tx events

core/tests/testkit/src/scenarios.rs

@@ -101,7 +101,7 @@ pub async fn perform_basic_tests() {
     let deposit_amount = parse_ether("1.0").unwrap();
 
     let token = TokenId(1);
-    perform_basic_operations(
+    let executed_blocks = perform_basic_operations(
         token,
         &mut test_setup,
         deposit_amount.clone(),
@@ -110,6 +110,18 @@ pub async fn perform_basic_tests() {
     .await;
     let tokens = vec![token];
 
+    // Verify queued transactions events.
+    let expected_operations_num: usize = executed_blocks
+        .iter()
+        .map(|block| block.block_transactions.len())
+        .sum();
+    let mut operations_num: usize = 0;
+    while let Ok(Some(message)) = test_setup.processed_tx_events_receiver.try_next() {
+        operations_num += message.executed_ops.len();
+    }
+    assert!(operations_num > 0);
+    assert_eq!(operations_num, expected_operations_num);
+
     verify_restore(
         &testkit_config,
         &contracts,

core/tests/testkit/src/state_keeper_utils.rs

@@ -4,9 +4,12 @@ use futures::{
 };
 use std::thread::JoinHandle;
 use tokio::runtime::Runtime;
-use zksync_core::committer::CommitRequest;
-use zksync_core::state_keeper::{
-    start_state_keeper, StateKeeperRequest, ZkSyncStateInitParams, ZkSyncStateKeeper,
+use zksync_core::{
+    committer::CommitRequest,
+    state_keeper::{
+        start_state_keeper, StateKeeperRequest, ZkSyncStateInitParams, ZkSyncStateKeeper,
+    },
+    tx_event_emitter::ProcessedOperations,
 };
 use zksync_types::{
     Account, AccountId, Address, DepositOp, FullExitOp, TransferOp, TransferToNewOp, WithdrawOp,
@@ -29,6 +32,7 @@ pub async fn state_keeper_get_account(
 pub struct StateKeeperChannels {
     pub requests: mpsc::Sender<StateKeeperRequest>,
     pub new_blocks: mpsc::Receiver<CommitRequest>,
+    pub queued_txs_events: mpsc::Receiver<ProcessedOperations>,
 }
 
 // Thread join handle and stop channel sender.
@@ -38,7 +42,7 @@ pub fn spawn_state_keeper(
 ) -> (JoinHandle<()>, oneshot::Sender<()>, StateKeeperChannels) {
     let (proposed_blocks_sender, proposed_blocks_receiver) = mpsc::channel(256);
     let (state_keeper_req_sender, state_keeper_req_receiver) = mpsc::channel(256);
-    let (processed_tx_events_sender, _processed_tx_events_receiver) = mpsc::channel(256);
+    let (processed_tx_events_sender, processed_tx_events_receiver) = mpsc::channel(256);
 
     let max_ops_in_block = 1000;
     let ops_chunks = vec![
@@ -85,6 +89,7 @@ pub fn spawn_state_keeper(
         StateKeeperChannels {
             requests: state_keeper_req_sender,
             new_blocks: proposed_blocks_receiver,
+            queued_txs_events: processed_tx_events_receiver,
         },
     )
 }

core/tests/testkit/src/test_setup.rs

@@ -7,9 +7,12 @@ use futures::{
 };
 use num::{bigint::Sign, BigInt, BigUint, ToPrimitive, Zero};
 use std::collections::HashMap;
-use zksync_core::committer::{BlockCommitRequest, CommitRequest};
-use zksync_core::mempool::ProposedBlock;
-use zksync_core::state_keeper::{StateKeeperRequest, ZkSyncStateInitParams};
+use zksync_core::{
+    committer::{BlockCommitRequest, CommitRequest},
+    mempool::ProposedBlock,
+    state_keeper::{StateKeeperRequest, ZkSyncStateInitParams},
+    tx_event_emitter::ProcessedOperations,
+};
 use zksync_types::{
     aggregated_operations::{BlocksCommitOperation, BlocksExecuteOperation, BlocksProofOperation},
     block::Block,
@@ -41,6 +44,7 @@ use zksync_types::tx::TimeRange;
 pub struct TestSetup {
     pub state_keeper_request_sender: mpsc::Sender<StateKeeperRequest>,
     pub proposed_blocks_receiver: mpsc::Receiver<CommitRequest>,
+    pub processed_tx_events_receiver: mpsc::Receiver<ProcessedOperations>,
 
     pub accounts: AccountSet,
     pub tokens: HashMap<TokenId, Address>,
@@ -87,6 +91,7 @@ impl TestSetup {
         Self {
             state_keeper_request_sender: sk_channels.requests,
             proposed_blocks_receiver: sk_channels.new_blocks,
+            processed_tx_events_receiver: sk_channels.queued_txs_events,
             accounts,
             tokens,
             expected_changes_for_current_block: ExpectedAccountState::default(),