Add test case for business rejects coming from the application

Author: davidsteiner

crates/hotfix/src/session.rs

@@ -20,12 +20,12 @@ use crate::Application;
 use crate::application::{InboundDecision, OutboundDecision};
 use crate::config::SessionConfig;
 use crate::message::OutboundMessage;
+use crate::message::business_reject::BusinessReject;
 use crate::message::generate_message;
 use crate::message::heartbeat::Heartbeat;
 use crate::message::logon::{Logon, ResetSeqNumConfig};
 use crate::message::logout::Logout;
 use crate::message::parser::RawFixMessage;
-use crate::message::business_reject::BusinessReject;
 use crate::message::reject::Reject;
 use crate::message::resend_request::ResendRequest;
 use crate::message::sequence_reset::SequenceReset;
@@ -249,9 +249,8 @@ where
                             .header()
                             .get(MSG_TYPE)
                             .map_err(|_| SessionOperationError::MissingField("MSG_TYPE"))?;
-                        let mut reject =
-                            BusinessReject::new(msg_type, reason)
-                                .ref_seq_num(get_msg_seq_num(message));
+                        let mut reject = BusinessReject::new(msg_type, reason)
+                            .ref_seq_num(get_msg_seq_num(message));
                         if let Some(text) = text {
                             reject = reject.text(&text);
                         }

crates/hotfix/tests/session_test_cases/business_reject_tests.rs

@@ -0,0 +1,96 @@
+use crate::common::actions::when;
+use crate::common::assertions::then;
+use crate::common::cleanup::finally;
+use crate::common::fakes::FakeApplication;
+use crate::common::setup::given_an_active_session_with_app;
+use crate::common::test_messages::TestMessage;
+use hotfix::application::{BusinessRejectReason, InboundDecision};
+use hotfix_message::Part;
+use hotfix_message::fix44::{MSG_TYPE, REF_MSG_TYPE, REF_SEQ_NUM, TEXT};
+
+/// Tests that when the application returns InboundDecision::Reject,
+/// the session sends a Business Message Reject (MsgType "j") back to the counterparty.
+#[tokio::test]
+async fn test_inbound_reject_sends_business_message_reject() {
+    let (message_tx, message_rx) = tokio::sync::mpsc::unbounded_channel();
+    let app = FakeApplication::builder(message_tx)
+        .inbound_decision_fn(|_| InboundDecision::Reject {
+            reason: BusinessRejectReason::NotAuthorized,
+            text: Some("Not authorized for this message".to_string()),
+        })
+        .build();
+    let (session, mut counterparty) = given_an_active_session_with_app(app, message_rx).await;
+
+    // counterparty sends an execution report
+    when(&mut counterparty)
+        .sends_message(TestMessage::dummy_execution_report())
+        .await;
+
+    // session should respond with a Business Message Reject (MsgType "j")
+    then(&mut counterparty)
+        .receives(|msg| {
+            let msg_type: &str = msg.header().get(MSG_TYPE).unwrap();
+            assert_eq!(msg_type, "j");
+
+            // RefMsgType should be the original message type ("8" for ExecutionReport)
+            let ref_msg_type: &str = msg.get(REF_MSG_TYPE).unwrap();
+            assert_eq!(ref_msg_type, "8");
+
+            // BusinessRejectReason (tag 380) should be 6 (NotAuthorized)
+            let reason: u32 = msg.get(BUSINESS_REJECT_REASON).unwrap();
+            assert_eq!(reason, 6);
+
+            // RefSeqNum should be the sequence number of the rejected message
+            let ref_seq_num: u64 = msg.get(REF_SEQ_NUM).unwrap();
+            assert!(ref_seq_num > 0);
+
+            // Text should contain our reject reason
+            let text: &str = msg.get(TEXT).unwrap();
+            assert_eq!(text, "Not authorized for this message");
+        })
+        .await;
+
+    finally(&session, &mut counterparty).disconnect().await;
+}
+
+/// Tests that when the application returns InboundDecision::Reject without text,
+/// the Business Message Reject is sent without the Text field.
+#[tokio::test]
+async fn test_inbound_reject_without_text() {
+    let (message_tx, message_rx) = tokio::sync::mpsc::unbounded_channel();
+    let app = FakeApplication::builder(message_tx)
+        .inbound_decision_fn(|_| InboundDecision::Reject {
+            reason: BusinessRejectReason::UnsupportedMessageType,
+            text: None,
+        })
+        .build();
+    let (session, mut counterparty) = given_an_active_session_with_app(app, message_rx).await;
+
+    when(&mut counterparty)
+        .sends_message(TestMessage::dummy_execution_report())
+        .await;
+
+    then(&mut counterparty)
+        .receives(|msg| {
+            let msg_type: &str = msg.header().get(MSG_TYPE).unwrap();
+            assert_eq!(msg_type, "j");
+
+            let reason: u32 = msg.get(BUSINESS_REJECT_REASON).unwrap();
+            assert_eq!(reason, 3);
+
+            // Text field should not be present
+            assert!(msg.get::<&str>(TEXT).is_err());
+        })
+        .await;
+
+    finally(&session, &mut counterparty).disconnect().await;
+}
+
+/// Field definition for BusinessRejectReason (tag 380), used for assertions only.
+const BUSINESS_REJECT_REASON: &hotfix_message::HardCodedFixFieldDefinition =
+    &hotfix_message::HardCodedFixFieldDefinition {
+        name: "BusinessRejectReason",
+        tag: 380,
+        data_type: hotfix_message::dict::FixDatatype::Int,
+        location: hotfix_message::dict::FieldLocation::Body,
+    };

crates/hotfix/tests/session_test_cases/common/fakes/fake_application.rs

@@ -2,27 +2,57 @@ use crate::common::test_messages::TestMessage;
 use hotfix::Application;
 use hotfix::application::{InboundDecision, OutboundDecision};
 use hotfix_message::message::Message;
+use std::sync::Mutex;
+
+type OutboundDecisionFn = Box<dyn Fn(&TestMessage) -> OutboundDecision + Send>;
+type InboundDecisionFn = Box<dyn Fn(&Message) -> InboundDecision + Send>;
 
 pub struct FakeApplication {
     message_sender: tokio::sync::mpsc::UnboundedSender<Message>,
-    outbound_decision: OutboundDecision,
+    outbound_decision_fn: Mutex<OutboundDecisionFn>,
+    inbound_decision_fn: Mutex<InboundDecisionFn>,
 }
 
 impl FakeApplication {
-    pub fn new(message_sender: tokio::sync::mpsc::UnboundedSender<Message>) -> Self {
-        Self {
+    pub fn builder(
+        message_sender: tokio::sync::mpsc::UnboundedSender<Message>,
+    ) -> FakeApplicationBuilder {
+        FakeApplicationBuilder {
             message_sender,
-            outbound_decision: OutboundDecision::Send,
+            outbound_decision_fn: Box::new(|_| OutboundDecision::Send),
+            inbound_decision_fn: Box::new(|_| InboundDecision::Accept),
         }
     }
+}
 
-    pub fn with_outbound_decision(
-        message_sender: tokio::sync::mpsc::UnboundedSender<Message>,
-        decision: OutboundDecision,
+pub struct FakeApplicationBuilder {
+    message_sender: tokio::sync::mpsc::UnboundedSender<Message>,
+    outbound_decision_fn: OutboundDecisionFn,
+    inbound_decision_fn: InboundDecisionFn,
+}
+
+impl FakeApplicationBuilder {
+    pub fn outbound_decision_fn(
+        mut self,
+        f: impl Fn(&TestMessage) -> OutboundDecision + Send + 'static,
     ) -> Self {
-        Self {
-            message_sender,
-            outbound_decision: decision,
+        self.outbound_decision_fn = Box::new(f);
+        self
+    }
+
+    pub fn inbound_decision_fn(
+        mut self,
+        f: impl Fn(&Message) -> InboundDecision + Send + 'static,
+    ) -> Self {
+        self.inbound_decision_fn = Box::new(f);
+        self
+    }
+
+    pub fn build(self) -> FakeApplication {
+        FakeApplication {
+            message_sender: self.message_sender,
+            outbound_decision_fn: Mutex::new(self.outbound_decision_fn),
+            inbound_decision_fn: Mutex::new(self.inbound_decision_fn),
         }
     }
 }
@@ -31,13 +61,15 @@ impl FakeApplication {
 impl Application for FakeApplication {
     type Outbound = TestMessage;
 
-    async fn on_outbound_message(&self, _msg: &TestMessage) -> OutboundDecision {
-        self.outbound_decision
+    async fn on_outbound_message(&self, msg: &TestMessage) -> OutboundDecision {
+        let decision_fn = self.outbound_decision_fn.lock().unwrap();
+        (decision_fn)(msg)
     }
 
     async fn on_inbound_message(&self, msg: &Message) -> InboundDecision {
         self.message_sender.send(msg.clone()).unwrap();
-        InboundDecision::Accept
+        let decision_fn = self.inbound_decision_fn.lock().unwrap();
+        (decision_fn)(msg)
     }
 
     async fn on_logout(&mut self, _reason: &str) {}

crates/hotfix/tests/session_test_cases/common/setup.rs

@@ -3,13 +3,13 @@ use crate::common::assertions::then;
 use crate::common::fakes::{FakeApplication, FakeCounterparty, SessionSpy};
 use crate::common::test_messages::TestMessage;
 use crate::session_test_cases::common::fakes::DisconnectedSession;
-use hotfix::application::OutboundDecision;
 use hotfix::config::SessionConfig;
 use hotfix::session::InternalSessionRef;
 use hotfix::session::Status;
 use hotfix::store::in_memory::InMemoryMessageStore;
 use hotfix_message::Part;
 use hotfix_message::fix44::MSG_TYPE;
+use hotfix_message::message::Message;
 
 pub const HEARTBEAT_INTERVAL: u64 = 30;
 pub const LOGON_TIMEOUT: u64 = 10;
@@ -30,8 +30,12 @@ pub async fn given_a_connected_session_with_store(
     let counterparty_config = create_counterparty_session_config(config.clone());
 
     let (message_tx, message_rx) = tokio::sync::mpsc::unbounded_channel();
-    let session = InternalSessionRef::new(config, FakeApplication::new(message_tx), message_store)
-        .expect("session to be created successfully");
+    let session = InternalSessionRef::new(
+        config,
+        FakeApplication::builder(message_tx).build(),
+        message_store,
+    )
+    .expect("session to be created successfully");
 
     let session_spy = SessionSpy::new(session.clone().into(), message_rx);
     let mock_counterparty = FakeCounterparty::start(session.clone(), counterparty_config)
@@ -42,15 +46,14 @@ pub async fn given_a_connected_session_with_store(
 }
 
 /// Creates an active session with a configurable application.
-pub async fn given_an_active_session_with_outbound_decision(
-    decision: OutboundDecision,
+pub async fn given_an_active_session_with_app(
+    app: FakeApplication,
+    message_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
 ) -> (SessionSpy, FakeCounterparty<TestMessage>) {
     let config = create_session_config();
     let counterparty_config = create_counterparty_session_config(config.clone());
     let message_store = InMemoryMessageStore::default();
 
-    let (message_tx, message_rx) = tokio::sync::mpsc::unbounded_channel();
-    let app = FakeApplication::with_outbound_decision(message_tx, decision);
     let session = InternalSessionRef::new(config, app, message_store)
         .expect("session to be created successfully");
 
@@ -77,9 +80,12 @@ pub fn given_a_disconnected_session() -> DisconnectedSession {
     let message_store = InMemoryMessageStore::default();
 
     let (message_tx, _message_rx) = tokio::sync::mpsc::unbounded_channel();
-    let session_ref =
-        InternalSessionRef::new(config, FakeApplication::new(message_tx), message_store)
-            .expect("session to be created successfully");
+    let session_ref = InternalSessionRef::new(
+        config,
+        FakeApplication::builder(message_tx).build(),
+        message_store,
+    )
+    .expect("session to be created successfully");
 
     let session_handle = session_ref.clone().into();
 

crates/hotfix/tests/session_test_cases/mod.rs

@@ -1,4 +1,5 @@
 mod admin_request_tests;
+mod business_reject_tests;
 mod business_tests;
 pub(crate) mod common;
 mod heartbeat_tests;

crates/hotfix/tests/session_test_cases/send_confirmation_tests.rs

@@ -1,9 +1,9 @@
 use crate::common::actions::when;
 use crate::common::assertions::then;
 use crate::common::cleanup::finally;
+use crate::common::fakes::FakeApplication;
 use crate::common::setup::{
-    given_a_disconnected_session, given_an_active_session,
-    given_an_active_session_with_outbound_decision,
+    given_a_disconnected_session, given_an_active_session, given_an_active_session_with_app,
 };
 use crate::common::test_messages::TestMessage;
 use hotfix::application::OutboundDecision;
@@ -129,8 +129,11 @@ async fn test_send_returns_disconnected_when_not_connected() {
 #[tokio::test]
 async fn test_send_returns_dropped_when_app_drops_message() {
     // Create an active session with an application configured to drop messages
-    let (session, mut counterparty) =
-        given_an_active_session_with_outbound_decision(OutboundDecision::Drop).await;
+    let (message_tx, message_rx) = tokio::sync::mpsc::unbounded_channel();
+    let app = FakeApplication::builder(message_tx)
+        .outbound_decision_fn(|_| OutboundDecision::Drop)
+        .build();
+    let (session, mut counterparty) = given_an_active_session_with_app(app, message_rx).await;
 
     // Send a message - should be dropped by the application
     let result = when(&session)
@@ -153,8 +156,11 @@ async fn test_send_returns_dropped_when_app_drops_message() {
 #[tokio::test]
 async fn test_send_returns_session_terminated_when_app_terminates() {
     // Create an active session with an application configured to terminate session
-    let (session, mut counterparty) =
-        given_an_active_session_with_outbound_decision(OutboundDecision::TerminateSession).await;
+    let (message_tx, message_rx) = tokio::sync::mpsc::unbounded_channel();
+    let app = FakeApplication::builder(message_tx)
+        .outbound_decision_fn(|_| OutboundDecision::TerminateSession)
+        .build();
+    let (session, mut counterparty) = given_an_active_session_with_app(app, message_rx).await;
 
     // Send a message - should cause session termination
     let result = when(&session)