Add test to beacon node fallback feature

Cargo.toml

@@ -138,6 +138,7 @@ log = "0.4"
 lru = "0.12"
 maplit = "1"
 milhouse = "0.3"
+mockito = "1.5.0"
 num_cpus = "1"
 parking_lot = "0.12"
 paste = "1"
@@ -224,6 +225,7 @@ malloc_utils = { path = "common/malloc_utils" }
 merkle_proof = { path = "consensus/merkle_proof" }
 monitoring_api = { path = "common/monitoring_api" }
 network = { path = "beacon_node/network" }
+node_test_rig = { path = "testing/node_test_rig"}
 operation_pool = { path = "beacon_node/operation_pool" }
 pretty_reqwest_error = { path = "common/pretty_reqwest_error" }
 proto_array = { path = "consensus/proto_array" }

lighthouse/tests/validator_client.rs

@@ -506,7 +506,6 @@ fn monitoring_endpoint() {
             assert_eq!(api_conf.update_period_secs, Some(30));
         });
 }
-
 #[test]
 fn no_broadcast_flag() {
     CommandLineTest::new().run().with_config(|config| {

validator_client/Cargo.toml

@@ -10,6 +10,11 @@ path = "src/lib.rs"
 
 [dev-dependencies]
 tokio = { workspace = true }
+eth2_config = { workspace = true }
+node_test_rig = { workspace = true }
+mockito = { workspace = true }
+regex = { workspace = true }
+env_logger = { workspace = true }
 
 [dependencies]
 tree_hash = { workspace = true }

validator_client/src/beacon_node_fallback.rs

@@ -737,13 +737,21 @@ impl ApiTopic {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use crate::beacon_node_health::BeaconNodeHealthTier;
     use crate::SensitiveUrl;
     use eth2::Timeouts;
+    use logging::test_logger;
     use std::str::FromStr;
     use strum::VariantNames;
-    use types::{MainnetEthSpec, Slot};
+    use types::EmptyBlock;
+
+    use slot_clock::TestingSlotClock;
+    use types::{BeaconBlockDeneb, BlindedBeaconBlock, MainnetEthSpec, Slot};
+
+    use super::*;
+    use crate::block_service::{BlockService, BlockServiceBuilder, UnsignedBlock};
+    use crate::testing::mock_beacon_node::MockBeaconNode;
+    use crate::testing::validator_test_rig::ValidatorTestRig;
 
     type E = MainnetEthSpec;
 
@@ -867,4 +875,147 @@ mod tests {
 
         assert_eq!(candidates, expected_candidates);
     }
+
+    #[tokio::test]
+    async fn update_all_candidates_should_update_sync_status() {
+        let spec = Arc::new(MainnetEthSpec::default_spec());
+        let mut mock_beacon_node_one = MockBeaconNode::<E>::new().await;
+        let mut mock_beacon_node_two = MockBeaconNode::<E>::new().await;
+        let mut mock_beacon_node_three = MockBeaconNode::<E>::new().await;
+
+        let beacon_node_one =
+            CandidateBeaconNode::<E>::new(mock_beacon_node_one.beacon_api_client.clone(), 0);
+        let beacon_node_two =
+            CandidateBeaconNode::<E>::new(mock_beacon_node_two.beacon_api_client.clone(), 1);
+        let beacon_node_three =
+            CandidateBeaconNode::<E>::new(mock_beacon_node_three.beacon_api_client.clone(), 2);
+
+        let mut beacon_node_fallback: BeaconNodeFallback<TestingSlotClock, E> =
+            BeaconNodeFallback::new(
+                // Put this out of order to be sorted later
+                vec![
+                    beacon_node_two.clone(),
+                    beacon_node_three.clone(),
+                    beacon_node_one.clone(),
+                ],
+                Config::default(),
+                vec![], // to broadcast blocks to both bns
+                spec.clone(),
+                test_logger(),
+            );
+
+        beacon_node_fallback.set_slot_clock(TestingSlotClock::new(
+            Slot::new(1),
+            Duration::from_secs(0),
+            Duration::from_secs(12),
+        ));
+
+        mock_beacon_node_one.mock_config_spec(&spec);
+        mock_beacon_node_two.mock_config_spec(&spec);
+        mock_beacon_node_three.mock_config_spec(&spec);
+
+        // BeaconNodeHealthTier 1
+        mock_beacon_node_one.mock_get_node_syncing(eth2::types::SyncingData {
+            is_syncing: false,
+            is_optimistic: false,
+            el_offline: false,
+            head_slot: Slot::new(1),
+            sync_distance: Slot::new(0),
+        });
+        // BeaconNodeHealthTier 3
+        mock_beacon_node_two.mock_get_node_syncing(eth2::types::SyncingData {
+            is_syncing: false,
+            is_optimistic: false,
+            el_offline: true,
+            head_slot: Slot::new(1),
+            sync_distance: Slot::new(0),
+        });
+        // BeaconNodeHealthTier 5
+        mock_beacon_node_three.mock_get_node_syncing(eth2::types::SyncingData {
+            is_syncing: false,
+            is_optimistic: true,
+            el_offline: false,
+            head_slot: Slot::new(1),
+            sync_distance: Slot::new(0),
+        });
+
+        beacon_node_fallback.update_all_candidates().await;
+
+        let candidates = beacon_node_fallback.candidates.read().await;
+        assert_eq!(
+            vec![beacon_node_one, beacon_node_two, beacon_node_three],
+            *candidates
+        );
+    }
+
+    #[tokio::test]
+    async fn broadcast_should_send_to_all_bns() {
+        let test_rig = ValidatorTestRig::new().await;
+        let mut mock_beacon_node_one = MockBeaconNode::<E>::new().await;
+        let mut mock_beacon_node_two = MockBeaconNode::<E>::new().await;
+
+        let beacon_node_one =
+            CandidateBeaconNode::new(mock_beacon_node_one.beacon_api_client.clone(), 0);
+        let beacon_node_two =
+            CandidateBeaconNode::new(mock_beacon_node_two.beacon_api_client.clone(), 1);
+
+        let mut beacon_node_fallback = BeaconNodeFallback::new(
+            vec![beacon_node_one, beacon_node_two],
+            Config::default(),
+            vec![ApiTopic::Blocks], // to broadcast blocks to both bns
+            test_rig.spec.clone(),
+            test_rig.logger.clone(),
+        );
+
+        beacon_node_fallback.set_slot_clock(TestingSlotClock::new(
+            Slot::new(1),
+            Duration::from_secs(0),
+            Duration::from_secs(12),
+        ));
+
+        mock_beacon_node_one.mock_config_spec(&test_rig.spec);
+        mock_beacon_node_two.mock_config_spec(&test_rig.spec);
+
+        let mock1 = mock_beacon_node_one.mock_post_beacon_blinded_blocks_v2(Duration::from_secs(0));
+        let mock2 = mock_beacon_node_two.mock_post_beacon_blinded_blocks_v2(Duration::from_secs(0));
+
+        let beacon_node_fallback = Arc::new(beacon_node_fallback);
+        let block_service: BlockService<TestingSlotClock, MainnetEthSpec> =
+            BlockServiceBuilder::new()
+                .slot_clock(test_rig.slot_clock)
+                .validator_store(test_rig.validator_store.clone())
+                .beacon_nodes(beacon_node_fallback.clone())
+                .runtime_context(test_rig.runtime_context)
+                .build()
+                .unwrap();
+
+        beacon_node_fallback.update_all_candidates().await;
+
+        let validators = test_rig.validator_store.initialized_validators();
+        let first_validator_pubkey = {
+            let validators = validators.read();
+            if let Some(first_validator) = validators.validator_definitions().first() {
+                first_validator.voting_public_key.clone()
+            } else {
+                panic!("No validators found");
+            }
+        };
+        let unsigned_block = UnsignedBlock::Blinded(BlindedBeaconBlock::Deneb(
+            BeaconBlockDeneb::empty(&test_rig.spec),
+        ));
+
+        let result = block_service
+            .publish_block_for_testing(Slot::new(1), &first_validator_pubkey.into(), unsigned_block)
+            .await;
+        assert!(result.is_ok());
+
+        mock1.expect(1).assert();
+        mock2.expect(1).assert();
+
+        let received_blocks_one = mock_beacon_node_one.received_blocks.lock().unwrap();
+        let received_blocks_two = mock_beacon_node_two.received_blocks.lock().unwrap();
+
+        assert_eq!(received_blocks_one.len(), 1);
+        assert_eq!(received_blocks_two.len(), 1);
+    }
 }