arminius_lib_message_bus/mock/
command.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Mock implementations of [`Requester`](crate::Requester) and
6//! [`Responder`](crate::Responder) for unit testing without a NATS server.
7
8use std::{
9	ops::Deref,
10	sync::{Arc, Mutex},
11};
12
13use crate::{Command, Error, ReceivedCommand};
14
15/// Mock implementation of [`ReceivedCommand`] for unit tests.
16///
17/// Captures the reply sent by [`ReceivedCommand::reply`] in a channel
18/// so tests can assert on what the handler replied without a running NATS
19/// server. Obtain a `(MockReceivedCommand<C>, ReplyChecker<C::Reply>)` pair
20/// from [`MockReceivedCommand::new`].
21pub struct MockReceivedCommand<C: Command> {
22	/// The wrapped command.
23	cmd: C,
24	/// The OTel context associated with this command.
25	context: opentelemetry::Context,
26	/// Channel the handler's reply is sent on; taken on first reply.
27	reply_tx: Arc<Mutex<Option<futures::channel::oneshot::Sender<Vec<u8>>>>>,
28}
29
30impl<C: Command + std::fmt::Debug> std::fmt::Debug for MockReceivedCommand<C> {
31	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
32		f.debug_struct("MockReceivedCommand").field("cmd", &self.cmd).finish_non_exhaustive()
33	}
34}
35
36impl<C: Command> MockReceivedCommand<C> {
37	/// Creates a `MockReceivedCommand` paired with a [`ReplyChecker`] that
38	/// receives the reply sent by
39	/// [`reply`](ReceivedCommand::reply).
40	pub fn new(cmd: C) -> (Self, ReplyChecker<C::Reply>) {
41		let (tx, rx) = futures::channel::oneshot::channel();
42		let mock = Self {
43			cmd,
44			context: opentelemetry::Context::new(),
45			reply_tx: Arc::new(Mutex::new(Some(tx))),
46		};
47		(mock, ReplyChecker { rx, _phantom: std::marker::PhantomData })
48	}
49
50	/// Constructs a received command whose reply is delivered through
51	/// `reply_tx`.
52	///
53	/// Used by the in-memory bus to route a handler's reply back to the
54	/// originating [`Requester::request`](crate::Requester::request) call,
55	/// instead of capturing it in a [`ReplyChecker`].
56	pub(crate) fn with_reply_channel(
57		cmd: C,
58		reply_tx: futures::channel::oneshot::Sender<Vec<u8>>,
59	) -> Self {
60		Self {
61			cmd,
62			context: opentelemetry::Context::new(),
63			reply_tx: Arc::new(Mutex::new(Some(reply_tx))),
64		}
65	}
66}
67
68impl<C: Command> Deref for MockReceivedCommand<C> {
69	type Target = C;
70
71	fn deref(&self) -> &C {
72		&self.cmd
73	}
74}
75
76impl<C: Command> ReceivedCommand<C> for MockReceivedCommand<C> {
77	/// Does nothing: there is no broker here to acknowledge to. Tests that need
78	/// to observe acknowledgement order need a double of their own.
79	async fn ack(&self) -> Result<(), Error> {
80		Ok(())
81	}
82
83	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
84	async fn reply(&self, response: C::Reply) -> Result<(), Error> {
85		let bytes = serde_json::to_vec(&response)
86			.map_err(|error| Error::SendFailed { error: error.to_string() })?;
87		if let Some(tx) = self
88			.reply_tx
89			.lock()
90			.expect("lock should not be poisoned: no holder should panic while the mutex is held")
91			.take()
92		{
93			let _ = tx.send(bytes);
94		}
95		Ok(())
96	}
97
98	fn context(&self) -> &opentelemetry::Context {
99		&self.context
100	}
101}
102
103/// Receives the reply captured by a [`MockReceivedCommand`].
104///
105/// Returned alongside the mock by [`MockReceivedCommand::new`]. Call
106/// [`into_reply`](ReplyChecker::into_reply) after the handler under test has
107/// run to get the deserialized reply.
108pub struct ReplyChecker<R> {
109	/// Receives the reply bytes sent by the handler under test.
110	rx: futures::channel::oneshot::Receiver<Vec<u8>>,
111	/// Binds the checker to the reply type `R`.
112	_phantom: std::marker::PhantomData<R>,
113}
114
115impl<R> std::fmt::Debug for ReplyChecker<R> {
116	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
117		f.debug_struct("ReplyChecker").finish_non_exhaustive()
118	}
119}
120
121impl<R: serde::de::DeserializeOwned> ReplyChecker<R> {
122	/// Waits for and deserializes the reply sent by
123	/// [`reply`](ReceivedCommand::reply).
124	///
125	/// Returns `None` if the handler dropped the command without replying.
126	pub async fn into_reply(self) -> Option<R> {
127		self.rx.await.ok().and_then(|bytes| serde_json::from_slice(&bytes).ok())
128	}
129}
arminius_lib_message_bus/mock/command.rs

arminius_lib_message_bus/mock/
command.rs
