fix(dmq): blocking tests missing KES signed DMQ messages

Author: jpraynaud

internal/mithril-dmq/src/consumer/server/pallas.rs

@@ -282,43 +282,27 @@ mod tests {
 
     use pallas_network::{
         facades::DmqClient,
-        miniprotocols::{
-            localmsgnotification,
-            localmsgsubmission::{DmqMsg, DmqMsgPayload},
-        },
+        miniprotocols::{localmsgnotification, localmsgsubmission::DmqMsg},
     };
     use tokio::sync::{mpsc::unbounded_channel, watch};
     use tokio::time::sleep;
 
     use mithril_common::{current_function, test::TempDir};
 
-    use crate::test_tools::TestLogger;
+    use crate::{test::fake_message::compute_fake_msg, test_tools::TestLogger};
 
     use super::*;
 
     fn create_temp_dir(folder_name: &str) -> PathBuf {
         TempDir::create_with_short_path("dmq_consumer_server", folder_name)
     }
 
-    fn fake_msg() -> DmqMsg {
-        DmqMsg {
-            msg_payload: DmqMsgPayload {
-                msg_id: vec![0, 1],
-                msg_body: vec![0, 1, 2],
-                kes_period: 10,
-                expires_at: 100,
-            },
-            kes_signature: vec![0, 1, 2, 3],
-            operational_certificate: vec![0, 1, 2, 3, 4],
-            cold_verification_key: vec![0, 1, 2, 3, 4, 5],
-        }
-    }
-
     #[tokio::test(flavor = "multi_thread")]
     async fn pallas_dmq_consumer_server_non_blocking_success() {
+        let current_function_name = current_function!();
         let (stop_tx, stop_rx) = watch::channel(());
         let (signature_dmq_tx, signature_dmq_rx) = unbounded_channel::<DmqMessage>();
-        let socket_path = create_temp_dir(current_function!()).join("node.socket");
+        let socket_path = create_temp_dir(current_function_name).join("node.socket");
         let cardano_network = CardanoNetwork::TestNet(0);
         let dmq_consumer_server = Arc::new(DmqConsumerServerPallas::new(
             socket_path.to_path_buf(),
@@ -327,7 +311,7 @@ mod tests {
             TestLogger::stdout(),
         ));
         dmq_consumer_server.register_receiver(signature_dmq_rx).await.unwrap();
-        let message = fake_msg();
+        let message: DmqMsg = compute_fake_msg(b"test", current_function_name).await.into();
         let client = tokio::spawn({
             async move {
                 // sleep to avoid refused connection from the server
@@ -375,9 +359,10 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn pallas_dmq_consumer_server_blocking_success() {
+        let current_function_name = current_function!();
         let (stop_tx, stop_rx) = watch::channel(());
         let (signature_dmq_tx, signature_dmq_rx) = unbounded_channel::<DmqMessage>();
-        let socket_path = create_temp_dir(current_function!()).join("node.socket");
+        let socket_path = create_temp_dir(current_function_name).join("node.socket");
         let cardano_network = CardanoNetwork::TestNet(0);
         let dmq_consumer_server = Arc::new(DmqConsumerServerPallas::new(
             socket_path.to_path_buf(),
@@ -386,7 +371,7 @@ mod tests {
             TestLogger::stdout(),
         ));
         dmq_consumer_server.register_receiver(signature_dmq_rx).await.unwrap();
-        let message = fake_msg();
+        let message: DmqMsg = compute_fake_msg(b"test", current_function_name).await.into();
         let client = tokio::spawn({
             async move {
                 // sleep to avoid refused connection from the server

internal/mithril-dmq/tests/consumer_client_server.rs

@@ -3,35 +3,12 @@ use std::sync::Arc;
 
 use tokio::sync::{mpsc::unbounded_channel, watch};
 
-use mithril_cardano_node_chain::test::double::FakeChainObserver;
-use mithril_common::{
-    CardanoNetwork,
-    crypto_helper::TryToBytes,
-    current_function,
-    test::{TempDir, crypto_helper::KesSignerFake},
-};
+use mithril_common::{CardanoNetwork, current_function, test::TempDir};
 use mithril_dmq::{
     DmqConsumerClient, DmqConsumerClientPallas, DmqConsumerServer, DmqConsumerServerPallas,
-    DmqMessage, DmqMessageBuilder, test::payload::DmqMessageTestPayload,
+    DmqMessage, test::fake_message::compute_fake_msg, test::payload::DmqMessageTestPayload,
 };
 
-async fn create_fake_msg(bytes: &[u8], test_directory: &str) -> DmqMessage {
-    let dmq_builder = DmqMessageBuilder::new(
-        {
-            let (kes_signature, operational_certificate) =
-                KesSignerFake::dummy_signature(test_directory);
-            let kes_signer =
-                KesSignerFake::new(vec![Ok((kes_signature, operational_certificate.clone()))]);
-
-            Arc::new(kes_signer)
-        },
-        Arc::new(FakeChainObserver::default()),
-    )
-    .set_ttl(100);
-    let message = DmqMessageTestPayload::new(bytes);
-    dmq_builder.build(&message.to_bytes_vec().unwrap()).await.unwrap()
-}
-
 #[tokio::test(flavor = "multi_thread")]
 async fn dmq_consumer_client_server() {
     let current_function_name = current_function!();
@@ -69,14 +46,14 @@ async fn dmq_consumer_client_server() {
             );
             let mut messages = vec![];
             signature_dmq_tx
-                .send(create_fake_msg(b"msg_1", current_function_name).await)
+                .send(compute_fake_msg(b"msg_1", current_function_name).await)
                 .unwrap();
             signature_dmq_tx
-                .send(create_fake_msg(b"msg_2", current_function_name).await)
+                .send(compute_fake_msg(b"msg_2", current_function_name).await)
                 .unwrap();
             messages.extend_from_slice(&consumer_client.consume_messages().await.unwrap());
             signature_dmq_tx
-                .send(create_fake_msg(b"msg_3", current_function_name).await)
+                .send(compute_fake_msg(b"msg_3", current_function_name).await)
                 .unwrap();
             messages.extend_from_slice(&consumer_client.consume_messages().await.unwrap());
 
@@ -109,7 +86,7 @@ async fn dmq_consumer_client_server() {
             );
             let mut messages = vec![];
             signature_dmq_tx
-                .send(create_fake_msg(b"msg_4", current_function_name).await)
+                .send(compute_fake_msg(b"msg_4", current_function_name).await)
                 .unwrap();
             messages.extend_from_slice(&consumer_client.consume_messages().await.unwrap());
             stop_tx.send(()).unwrap();

internal/mithril-dmq/tests/publisher_client_server.rs

@@ -5,33 +5,15 @@ use tokio::sync::{mpsc::unbounded_channel, watch};
 
 use mithril_cardano_node_chain::test::double::FakeChainObserver;
 use mithril_common::{
-    CardanoNetwork,
-    crypto_helper::TryToBytes,
-    current_function,
+    CardanoNetwork, current_function,
     test::{TempDir, crypto_helper::KesSignerFake},
 };
 use mithril_dmq::{
     DmqMessage, DmqMessageBuilder, DmqPublisherClient, DmqPublisherClientPallas,
-    DmqPublisherServer, DmqPublisherServerPallas, test::payload::DmqMessageTestPayload,
+    DmqPublisherServer, DmqPublisherServerPallas,
+    test::{fake_message::compute_fake_msg, payload::DmqMessageTestPayload},
 };
 
-async fn create_fake_msg(bytes: &[u8], test_directory: &str) -> DmqMessage {
-    let dmq_builder = DmqMessageBuilder::new(
-        {
-            let (kes_signature, operational_certificate) =
-                KesSignerFake::dummy_signature(test_directory);
-            let kes_signer =
-                KesSignerFake::new(vec![Ok((kes_signature, operational_certificate.clone()))]);
-
-            Arc::new(kes_signer)
-        },
-        Arc::new(FakeChainObserver::default()),
-    )
-    .set_ttl(100);
-    let message = DmqMessageTestPayload::new(bytes);
-    dmq_builder.build(&message.to_bytes_vec().unwrap()).await.unwrap()
-}
-
 #[tokio::test]
 async fn dmq_publisher_client_server() {
     let current_function_name = current_function!();
@@ -157,9 +139,9 @@ async fn dmq_publisher_client_server() {
     let (_, _, messages) = tokio::try_join!(server, client, recorder).unwrap();
     assert_eq!(
         vec![
-            create_fake_msg(b"msg_1", current_function_name).await,
-            create_fake_msg(b"msg_2", current_function_name).await,
-            create_fake_msg(b"msg_3", current_function_name).await,
+            compute_fake_msg(b"msg_1", current_function_name).await,
+            compute_fake_msg(b"msg_2", current_function_name).await,
+            compute_fake_msg(b"msg_3", current_function_name).await,
         ],
         messages
     );