Add tests for edge cases

Author: davidsteiner

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

@@ -31,6 +31,13 @@ impl When<&SessionSpy> {
     ) -> Result<SendOutcome, SendError> {
         self.target.session_handle().send(message).await
     }
+
+    pub async fn sends_message_without_confirmation(
+        self,
+        message: TestMessage,
+    ) -> Result<(), SendError> {
+        self.target.session_handle().send_forget(message).await
+    }
 }
 
 impl When<&mut FakeCounterparty<TestMessage>> {

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

@@ -1,21 +1,99 @@
 use crate::common::test_messages::TestMessage;
 use hotfix::Application;
 use hotfix::application::{InboundDecision, OutboundDecision};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU8, Ordering};
+
+/// Represents the decision to make for outbound messages.
+/// Uses u8 for atomic storage.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[repr(u8)]
+pub enum OutboundDecisionKind {
+    Send = 0,
+    Drop = 1,
+    TerminateSession = 2,
+}
+
+impl From<OutboundDecision> for OutboundDecisionKind {
+    fn from(decision: OutboundDecision) -> Self {
+        match decision {
+            OutboundDecision::Send => OutboundDecisionKind::Send,
+            OutboundDecision::Drop => OutboundDecisionKind::Drop,
+            OutboundDecision::TerminateSession => OutboundDecisionKind::TerminateSession,
+        }
+    }
+}
+
+impl From<OutboundDecisionKind> for OutboundDecision {
+    fn from(kind: OutboundDecisionKind) -> Self {
+        match kind {
+            OutboundDecisionKind::Send => OutboundDecision::Send,
+            OutboundDecisionKind::Drop => OutboundDecision::Drop,
+            OutboundDecisionKind::TerminateSession => OutboundDecision::TerminateSession,
+        }
+    }
+}
+
+/// A handle to change the outbound decision at runtime.
+#[derive(Clone)]
+pub struct OutboundDecisionHandle {
+    inner: Arc<AtomicU8>,
+}
+
+impl OutboundDecisionHandle {
+    fn new(decision: OutboundDecisionKind) -> Self {
+        Self {
+            inner: Arc::new(AtomicU8::new(decision as u8)),
+        }
+    }
+
+    #[allow(dead_code)]
+    pub fn set(&self, decision: OutboundDecisionKind) {
+        self.inner.store(decision as u8, Ordering::SeqCst);
+    }
+
+    fn get(&self) -> OutboundDecisionKind {
+        match self.inner.load(Ordering::SeqCst) {
+            0 => OutboundDecisionKind::Send,
+            1 => OutboundDecisionKind::Drop,
+            2 => OutboundDecisionKind::TerminateSession,
+            _ => OutboundDecisionKind::Send,
+        }
+    }
+}
 
 pub struct FakeApplication {
     message_sender: tokio::sync::mpsc::UnboundedSender<TestMessage>,
+    outbound_decision: OutboundDecisionHandle,
 }
 
 impl FakeApplication {
     pub fn new(message_sender: tokio::sync::mpsc::UnboundedSender<TestMessage>) -> Self {
-        Self { message_sender }
+        Self {
+            message_sender,
+            outbound_decision: OutboundDecisionHandle::new(OutboundDecisionKind::Send),
+        }
+    }
+
+    pub fn with_outbound_decision(
+        message_sender: tokio::sync::mpsc::UnboundedSender<TestMessage>,
+        decision: OutboundDecision,
+    ) -> Self {
+        Self {
+            message_sender,
+            outbound_decision: OutboundDecisionHandle::new(decision.into()),
+        }
+    }
+
+    pub fn outbound_decision_handle(&self) -> OutboundDecisionHandle {
+        self.outbound_decision.clone()
     }
 }
 
 #[async_trait::async_trait]
 impl Application<TestMessage, TestMessage> for FakeApplication {
     async fn on_outbound_message(&self, _msg: &TestMessage) -> OutboundDecision {
-        OutboundDecision::Send
+        self.outbound_decision.get().into()
     }
 
     async fn on_inbound_message(&self, msg: TestMessage) -> InboundDecision {

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

@@ -2,6 +2,6 @@ mod fake_application;
 mod fake_counterparty;
 mod session_spy;
 
-pub use fake_application::FakeApplication;
+pub use fake_application::{FakeApplication, OutboundDecisionHandle};
 pub use fake_counterparty::FakeCounterparty;
 pub use session_spy::SessionSpy;

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

@@ -1,10 +1,11 @@
 use crate::common::actions::when;
 use crate::common::assertions::then;
-use crate::common::fakes::{FakeApplication, FakeCounterparty, SessionSpy};
+use crate::common::fakes::{FakeApplication, FakeCounterparty, OutboundDecisionHandle, SessionSpy};
 use crate::common::test_messages::TestMessage;
+use hotfix::application::OutboundDecision;
 use hotfix::config::SessionConfig;
-use hotfix::session::InternalSessionRef;
 use hotfix::session::Status;
+use hotfix::session::{InternalSessionRef, SessionHandle};
 use hotfix::store::in_memory::InMemoryMessageStore;
 use hotfix_message::Part;
 use hotfix_message::fix44::MSG_TYPE;
@@ -39,6 +40,87 @@ pub async fn given_a_connected_session_with_store(
     (session_spy, mock_counterparty)
 }
 
+/// Creates a connected session with a configurable application.
+/// Returns the session spy, counterparty, and a handle to change the outbound decision at runtime.
+pub async fn given_a_connected_session_with_configurable_app(
+    initial_decision: OutboundDecision,
+) -> (
+    SessionSpy,
+    FakeCounterparty<TestMessage>,
+    OutboundDecisionHandle,
+) {
+    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, initial_decision);
+    let decision_handle = app.outbound_decision_handle();
+    let session = InternalSessionRef::new(config, app, 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)
+        .await
+        .expect("failed to start FakeCounterparty");
+
+    (session_spy, mock_counterparty, decision_handle)
+}
+
+/// Creates an active session with a configurable application.
+/// Returns the session spy, counterparty, and a handle to change the outbound decision at runtime.
+pub async fn given_an_active_session_with_configurable_app(
+    initial_decision: OutboundDecision,
+) -> (
+    SessionSpy,
+    FakeCounterparty<TestMessage>,
+    OutboundDecisionHandle,
+) {
+    let (mut session, mut mock_counterparty, decision_handle) =
+        given_a_connected_session_with_configurable_app(initial_decision).await;
+
+    then(&mut mock_counterparty)
+        .receives(|msg| assert_eq!(msg.header().get::<&str>(MSG_TYPE).unwrap(), "A"))
+        .await;
+    when(&mut mock_counterparty).sends_logon().await;
+    then(&mut session).status_changes_to(Status::Active).await;
+
+    (session, mock_counterparty, decision_handle)
+}
+
+/// A disconnected session - has no transport connection established.
+/// Holds onto the InternalSessionRef to keep channels alive for the session task.
+pub struct DisconnectedSession {
+    #[allow(dead_code)]
+    session_ref: InternalSessionRef<TestMessage>,
+    session_handle: SessionHandle<TestMessage>,
+}
+
+impl DisconnectedSession {
+    pub fn session_handle(&self) -> &SessionHandle<TestMessage> {
+        &self.session_handle
+    }
+}
+
+/// Creates a session that has not yet established a transport connection.
+/// This is useful for testing the Disconnected error case.
+pub fn given_a_disconnected_session() -> DisconnectedSession {
+    let config = create_session_config();
+    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_handle = session_ref.clone().into();
+
+    DisconnectedSession {
+        session_ref,
+        session_handle,
+    }
+}
+
 pub async fn given_an_active_session() -> (SessionSpy, FakeCounterparty<TestMessage>) {
     let (mut session, mut mock_counterparty) = given_a_connected_session().await;
 

crates/hotfix/tests/session_test_cases/send_confirmation_tests.rs

@@ -1,10 +1,14 @@
 use crate::common::actions::when;
 use crate::common::assertions::then;
 use crate::common::cleanup::finally;
-use crate::common::setup::given_an_active_session;
+use crate::common::setup::{
+    given_a_disconnected_session, given_an_active_session,
+    given_an_active_session_with_configurable_app,
+};
 use crate::common::test_messages::TestMessage;
+use hotfix::application::OutboundDecision;
 use hotfix::message::{InboundMessage, OutboundMessage};
-use hotfix::session::SendOutcome;
+use hotfix::session::{SendError, SendOutcome};
 
 #[tokio::test]
 async fn test_send_returns_sequence_number() {
@@ -78,3 +82,97 @@ async fn test_send_multiple_messages_returns_sequential_sequence_numbers() {
 
     finally(&session, &mut counterparty).disconnect().await;
 }
+
+#[tokio::test]
+async fn test_send_forget_queues_message() {
+    let (session, mut counterparty) = given_an_active_session().await;
+
+    // Send a message using send_forget (no confirmation)
+    when(&session)
+        .sends_message_without_confirmation(TestMessage::dummy_new_order_single())
+        .await
+        .expect("message should be queued successfully");
+
+    // Verify counterparty received the message
+    then(&mut counterparty)
+        .receives(|msg| {
+            let parsed = TestMessage::parse(msg);
+            assert_eq!(parsed.message_type(), "D");
+        })
+        .await;
+
+    finally(&session, &mut counterparty).disconnect().await;
+}
+
+#[tokio::test]
+async fn test_send_returns_disconnected_when_not_connected() {
+    // Create a session without establishing a transport connection
+    let disconnected_session = given_a_disconnected_session();
+
+    // Try to send a message before any connection is established
+    let result = disconnected_session
+        .session_handle()
+        .send(TestMessage::dummy_new_order_single())
+        .await;
+
+    // Should return Disconnected error since no transport connection exists
+    match result {
+        Err(SendError::Disconnected) => {
+            // Expected - session has no transport connection
+        }
+        other => {
+            panic!("Expected SendError::Disconnected, got {:?}", other);
+        }
+    }
+}
+
+#[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, _decision_handle) =
+        given_an_active_session_with_configurable_app(OutboundDecision::Drop).await;
+
+    // Send a message - should be dropped by the application
+    let result = when(&session)
+        .sends_message_with_confirmation(TestMessage::dummy_new_order_single())
+        .await;
+
+    // Verify we get SendOutcome::Dropped
+    match result {
+        Ok(SendOutcome::Dropped) => {
+            // Expected - application chose to drop the message
+        }
+        other => {
+            panic!("Expected SendOutcome::Dropped, got {:?}", other);
+        }
+    }
+
+    finally(&session, &mut counterparty).disconnect().await;
+}
+
+#[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, _decision_handle) =
+        given_an_active_session_with_configurable_app(OutboundDecision::TerminateSession).await;
+
+    // Send a message - should cause session termination
+    let result = when(&session)
+        .sends_message_with_confirmation(TestMessage::dummy_new_order_single())
+        .await;
+
+    // Verify we get SendError::SessionTerminated
+    match result {
+        Err(SendError::SessionTerminated) => {
+            // Expected - application chose to terminate the session
+        }
+        other => {
+            panic!("Expected SendError::SessionTerminated, got {:?}", other);
+        }
+    }
+
+    // Session is terminated, so we just need to wait for disconnect
+    counterparty
+        .assert_disconnected_with_timeout(std::time::Duration::from_secs(5))
+        .await;
+}