Unit Tests for Consensus

justfile

@@ -22,6 +22,10 @@ test_async_std_all:
   echo Testing with async std executor
   cargo test  --features=full-ci --lib --bins --tests --benches --workspace --no-fail-fast -- --test-threads=1
 
+test_consensus:
+  echo Testing with async std executor
+  RUST_LOG="error" cargo test  --features=full-ci --lib --bins --tests --benches --workspace --no-fail-fast test_consensus -- --test-threads=1 --nocapture
+
 test_basic:
   echo Testing with async std executor
   RUST_LOG="" cargo test  --features=full-ci --lib --bins --tests --benches --workspace --no-fail-fast test_basic -- --test-threads=1 --nocapture

task-impls/src/consensus.rs

@@ -1095,6 +1095,7 @@ where
                     })
                     .await;
                 if !self.update_view(new_view).await {
+                    error!("view not updated");
                     return;
                 }
 
@@ -1181,7 +1182,6 @@ where
                     data: proposal,
                     signature,
                 };
-                // debug!("Sending proposal for view {:?} \n {:?}", self.cur_view, message.clone());
                 debug!("Sending proposal for view {:?}", message.data.clone());
 
                 self.event_stream
@@ -1416,5 +1416,6 @@ pub fn consensus_event_filter<TYPES: NodeType, I: NodeImplementation<TYPES>>(
             | SequencingHotShotEvent::ViewChange(_)
             | SequencingHotShotEvent::SendDABlockData(_)
             | SequencingHotShotEvent::Timeout(_)
+            | SequencingHotShotEvent::Shutdown,
     )
 }

task-impls/src/harness.rs

@@ -12,11 +12,11 @@ use hotshot_task::{
 
 use hotshot_types::traits::node_implementation::{NodeImplementation, NodeType};
 use snafu::Snafu;
-use std::collections::HashSet;
+use std::collections::HashMap;
+use std::future::Future;
 use std::sync::Arc;
-
 pub struct TestHarnessState<TYPES: NodeType, I: NodeImplementation<TYPES>> {
-    expected_output: HashSet<SequencingHotShotEvent<TYPES, I>>,
+    expected_output: HashMap<SequencingHotShotEvent<TYPES, I>, usize>,
 }
 
 pub struct EventBundle<TYPES: NodeType, I: NodeImplementation<TYPES>>(
@@ -44,11 +44,15 @@ pub type TestHarnessTaskTypes<TYPES, I> = HSTWithEvent<
     TestHarnessState<TYPES, I>,
 >;
 
-pub async fn run_harness<TYPES: NodeType, I: NodeImplementation<TYPES>>(
+pub async fn run_harness<TYPES, I, Fut>(
     input: Vec<SequencingHotShotEvent<TYPES, I>>,
-    expected_output: HashSet<SequencingHotShotEvent<TYPES, I>>,
-    build_fn: fn(TaskRunner, ChannelStream<SequencingHotShotEvent<TYPES, I>>) -> TaskRunner,
-) {
+    expected_output: HashMap<SequencingHotShotEvent<TYPES, I>, usize>,
+    build_fn: impl FnOnce(TaskRunner, ChannelStream<SequencingHotShotEvent<TYPES, I>>) -> Fut,
+) where
+    TYPES: NodeType,
+    I: NodeImplementation<TYPES>,
+    Fut: Future<Output = TaskRunner>,
+{
     let task_runner = TaskRunner::new();
     let registry = task_runner.registry.clone();
     let event_stream = event_stream::ChannelStream::new();
@@ -70,19 +74,14 @@ pub async fn run_harness<TYPES: NodeType, I: NodeImplementation<TYPES>>(
     let task = TestHarnessTaskTypes::build(builder).launch();
 
     let task_runner = task_runner.add_task(id, "test_harness".to_string(), task);
-    let task_runner = build_fn(task_runner, event_stream.clone());
+    let task_runner = build_fn(task_runner, event_stream.clone()).await;
 
-    let _runner = async_spawn(async move { task_runner.launch().await });
+    let runner = async_spawn(async move { task_runner.launch().await });
 
     for event in input {
         let _ = event_stream.publish(event).await;
     }
-    // TODO fix type weirdness btwn tokio and async-std
-    todo!();
-
-    // for (_task_name, result) in runner.await.into_iter() {
-    //     assert!(matches!(result, HotShotTaskCompleted::ShutDown));
-    // }
+    let _ = runner.await;
 }
 
 pub fn handle_event<TYPES: NodeType, I: NodeImplementation<TYPES>>(
@@ -92,10 +91,15 @@ pub fn handle_event<TYPES: NodeType, I: NodeImplementation<TYPES>>(
     std::option::Option<HotShotTaskCompleted>,
     TestHarnessState<TYPES, I>,
 ) {
-    if !state.expected_output.contains(&event) {
+    if !state.expected_output.contains_key(&event) {
         panic!("Got and unexpected event: {:?}", event);
     }
-    state.expected_output.remove(&event);
+    let num_expected = state.expected_output.get_mut(&event).unwrap();
+    if *num_expected == 1 {
+        state.expected_output.remove(&event);
+    } else {
+        *num_expected -= 1;
+    }
 
     if state.expected_output.is_empty() {
         return (Some(HotShotTaskCompleted::ShutDown), state);

testing/src/node_types.rs

@@ -71,9 +71,10 @@ pub struct SequencingWebImpl;
 #[derive(Clone, Debug, Deserialize, Serialize, Hash, Eq, PartialEq)]
 pub struct StaticFallbackImpl;
 
-type StaticMembership = StaticCommittee<SequencingTestTypes, SequencingLeaf<SequencingTestTypes>>;
+pub type StaticMembership =
+    StaticCommittee<SequencingTestTypes, SequencingLeaf<SequencingTestTypes>>;
 
-type StaticMemoryDAComm = MemoryCommChannel<
+pub type StaticMemoryDAComm = MemoryCommChannel<
     SequencingTestTypes,
     SequencingMemoryImpl,
     DAProposal<SequencingTestTypes>,
@@ -100,7 +101,7 @@ type StaticWebDAComm = WebCommChannel<
 type StaticFallbackComm =
     WebServerWithFallbackCommChannel<SequencingTestTypes, StaticFallbackImpl, StaticMembership>;
 
-type StaticMemoryQuorumComm = MemoryCommChannel<
+pub type StaticMemoryQuorumComm = MemoryCommChannel<
     SequencingTestTypes,
     SequencingMemoryImpl,
     QuorumProposal<SequencingTestTypes, SequencingLeaf<SequencingTestTypes>>,
@@ -124,7 +125,7 @@ type StaticWebQuorumComm = WebCommChannel<
     StaticMembership,
 >;
 
-type StaticMemoryViewSyncComm = MemoryCommChannel<
+pub type StaticMemoryViewSyncComm = MemoryCommChannel<
     SequencingTestTypes,
     SequencingMemoryImpl,
     ViewSyncCertificate<SequencingTestTypes>,