ruststream 0.7.0-rc.11

//! Assertion builders over what the harness recorded: per-subscriber deliveries
//! ([`SubscriberAssertions`]) and per-channel publishes ([`PublishedAssertions`]).
//!
//! Every method asserts (panicking with a descriptive message on failure) and returns `self`, so
//! checks chain: `subscriber("orders").assert_called_once().with(&Order { id: 1 }).settled(Ack)`.

// These methods run their check eagerly (panicking on failure) and return `self` only for optional
// chaining, so ending a chain on one is fine and none of them is `#[must_use]`.
#![allow(clippy::return_self_not_must_use, clippy::must_use_candidate)]

use std::marker::PhantomData;

use bytes::Bytes;

use crate::RawMessage;
use crate::codec::Codec;
// Only the DEFAULT codec is feature-dependent; the assertions that take a codec are always here.
#[cfg(any(feature = "json", feature = "cbor", feature = "msgpack"))]
use crate::codec::DefaultCodec;
use crate::runtime::HandlerOutcome;
use serde::de::DeserializeOwned;
use std::fmt::Debug;

use super::coordinator::{Coordinator, Delivered, Outcome, Record, RecordedOptions, SlotRecord};

/// Assertions over the deliveries one subscriber received, recorded by the harness.
///
/// The unit these count is the handler CALL, not the message: a single-message handler is called
/// once per delivery, and a batch handler once per batch. So `assert_called_once` on a batch
/// subscription means one batch arrived, whatever its size, while
/// [`received_raw`](Self::received_raw) still lists every element of it. An element the decode
/// policy rejected before the body ran is settled by that policy and is not part of the batch the
/// handler was called with, so it does not appear here.
#[derive(Debug)]
pub struct SubscriberAssertions<'a> {
    coordinator: &'a Coordinator,
    scope_id: usize,
    name: String,
}

impl<'a> SubscriberAssertions<'a> {
    pub(crate) fn new(coordinator: &'a Coordinator, scope_id: usize, name: String) -> Self {
        Self {
            coordinator,
            scope_id,
            name,
        }
    }

    /// Runs `f` over the recorded deliveries to this subscriber, in delivery order.
    fn with_records<R>(&self, f: impl FnOnce(&[&Record]) -> R) -> R {
        self.coordinator.with_records(self.scope_id, &self.name, f)
    }

    /// Runs `f` over the most recent recorded call, panicking if there were none.
    fn with_last<R>(&self, what: &str, f: impl FnOnce(&Record) -> R) -> R {
        self.with_records(|records| {
            let last = records.last().unwrap_or_else(|| {
                panic!(
                    "subscriber {:?} was not called, cannot assert {what}",
                    self.name
                )
            });
            f(last)
        })
    }

    /// Runs `f` over the sole delivery of the most recent call. The assertions that name one
    /// expected payload go through here, so a batch of several reports what it was instead of
    /// silently checking one element of it.
    fn with_sole_delivery<R>(&self, what: &str, f: impl FnOnce(&Delivered) -> R) -> R {
        self.with_last(what, |record| match record.deliveries.as_slice() {
            [only] => f(only),
            batch => panic!(
                "subscriber {:?} last received a batch of {} deliveries, so it has no single \
                 {what}; read the batch with received_raw()",
                self.name,
                batch.len(),
            ),
        })
    }

    /// Asserts this subscriber was called exactly once (one delivery, or one batch).
    ///
    /// # Panics
    ///
    /// Panics if the subscriber was called zero or more than one time.
    pub fn assert_called_once(self) -> Self {
        let count = self.with_records(|records| records.len());
        assert_eq!(
            count, 1,
            "subscriber {:?} was called {count} times, expected exactly once",
            self.name,
        );
        self
    }

    /// Asserts this subscriber received exactly `times` deliveries (counting each redelivery).
    ///
    /// # Panics
    ///
    /// Panics if the delivery count differs from `times`.
    pub fn assert_called(self, times: usize) -> Self {
        let count = self.with_records(|records| records.len());
        assert_eq!(
            count, times,
            "subscriber {:?} was called {count} times, expected {times}",
            self.name,
        );
        self
    }

    /// Every raw payload this subscriber received, in delivery order, for custom inspection beyond
    /// the built-in assertions. A batch contributes its elements, so the list is flat whether the
    /// handler takes one message or a slice.
    #[must_use]
    pub fn received_raw(&self) -> Vec<Bytes> {
        self.with_records(|records| {
            records
                .iter()
                .flat_map(|record| record.deliveries.iter().map(|one| one.raw.clone()))
                .collect()
        })
    }

    /// The raw payloads this subscriber received grouped by handler CALL, in call order: one inner
    /// vector per call, holding what that call carried.
    ///
    /// This is [`received_raw`](Self::received_raw) with the batch boundaries kept. A
    /// single-message handler yields one-element vectors; a batch handler yields one vector per
    /// batch, which is what a test pinning how a stream was cut into batches reads.
    #[must_use]
    pub fn batches_raw(&self) -> Vec<Vec<Bytes>> {
        self.with_records(|records| {
            records
                .iter()
                .map(|record| {
                    record
                        .deliveries
                        .iter()
                        .map(|one| one.raw.clone())
                        .collect()
                })
                .collect()
        })
    }

    /// Decodes [`batches_raw`](Self::batches_raw) with
    /// [`DefaultCodec`](crate::codec::DefaultCodec). Use
    /// [`batches_with`](Self::batches_with) for a non-default codec.
    ///
    /// # Panics
    ///
    /// Panics if any received payload fails to decode as `T`.
    #[cfg(any(feature = "json", feature = "cbor", feature = "msgpack"))]
    #[must_use]
    pub fn batches<T: DeserializeOwned>(&self) -> Vec<Vec<T>> {
        self.batches_with(&DefaultCodec::default())
    }

    /// Like [`batches`](Self::batches), but decodes with `codec` - use it when the handler was
    /// mounted with a non-default codec.
    ///
    /// # Panics
    ///
    /// Panics if any received payload fails to decode as `T`.
    #[must_use]
    pub fn batches_with<T, C>(&self, codec: &C) -> Vec<Vec<T>>
    where
        T: DeserializeOwned,
        C: Codec,
    {
        self.with_records(|records| {
            records
                .iter()
                .map(|record| {
                    record
                        .deliveries
                        .iter()
                        .map(|one| {
                            codec.decode(&one.raw).unwrap_or_else(|err| {
                                panic!(
                                    "subscriber {:?} received a payload that did not decode as \
                                     {}: {err}",
                                    self.name,
                                    std::any::type_name::<T>(),
                                )
                            })
                        })
                        .collect()
                })
                .collect()
        })
    }

    /// The classified [`Outcome`] of every call, in call order.
    ///
    /// [`assert_outcome`](Self::assert_outcome) and [`settled`](Self::settled) read the most recent
    /// call only; this is what a test pinning a SEQUENCE (a nack followed by the redelivery's ack)
    /// compares against.
    #[must_use]
    pub fn outcomes(&self) -> Vec<Outcome> {
        self.with_records(|records| records.iter().map(|record| record.outcome()).collect())
    }

    /// Decodes every payload this subscriber received (with
    /// [`DefaultCodec`](crate::codec::DefaultCodec)), in delivery order, for custom inspection. Use
    /// [`received_with`](Self::received_with) for a non-default codec.
    ///
    /// # Panics
    ///
    /// Panics if any received payload fails to decode as `T` (a delivery the handler rejected as a
    /// decode failure will fail here too).
    #[cfg(any(feature = "json", feature = "cbor", feature = "msgpack"))]
    #[must_use]
    pub fn received<T: DeserializeOwned>(&self) -> Vec<T> {
        self.received_with(&DefaultCodec::default())
    }

    /// Like [`received`](Self::received), but decodes with `codec` - use it when the handler was
    /// mounted with a non-default codec.
    ///
    /// # Panics
    ///
    /// Panics if any received payload fails to decode as `T`.
    #[must_use]
    pub fn received_with<T, C>(&self, codec: &C) -> Vec<T>
    where
        T: DeserializeOwned,
        C: Codec,
    {
        self.with_records(|records| {
            records
                .iter()
                .flat_map(|record| record.deliveries.iter())
                .map(|one| {
                    codec.decode(&one.raw).unwrap_or_else(|err| {
                        panic!(
                            "subscriber {:?} received a payload that did not decode as {}: {err}",
                            self.name,
                            std::any::type_name::<T>(),
                        )
                    })
                })
                .collect()
        })
    }

    /// Asserts this subscriber was never called.
    ///
    /// # Panics
    ///
    /// Panics if the subscriber received any delivery.
    pub fn assert_not_called(self) {
        let count = self.with_records(|records| records.len());
        assert_eq!(
            count, 0,
            "subscriber {:?} was called {count} times, expected never",
            self.name,
        );
    }

    /// Asserts the most recent delivery's payload decodes (with
    /// [`DefaultCodec`](crate::codec::DefaultCodec)) to `expected`. If the handler was mounted with a
    /// different codec, use [`with_codec`](Self::with_codec).
    ///
    /// # Panics
    ///
    /// Panics if the subscriber was not called, the payload fails to decode, or the decoded value
    /// differs from `expected`.
    #[cfg(any(feature = "json", feature = "cbor", feature = "msgpack"))]
    pub fn with<T>(self, expected: &T) -> Self
    where
        T: DeserializeOwned + PartialEq + Debug,
    {
        self.with_codec(&DefaultCodec::default(), expected)
    }

    /// Like [`with`](Self::with), but decodes the recorded payload with `codec` - use it when the
    /// handler was mounted with a non-default codec (`include_with` / `with_broker_codec`).
    ///
    /// # Panics
    ///
    /// Panics if the subscriber was not called, the payload fails to decode, or the decoded value
    /// differs from `expected`.
    pub fn with_codec<T, C>(self, codec: &C, expected: &T) -> Self
    where
        T: DeserializeOwned + PartialEq + Debug,
        C: Codec,
    {
        self.with_sole_delivery("received value", |one| {
            let actual: T = codec.decode(&one.raw).unwrap_or_else(|err| {
                panic!(
                    "subscriber {:?} received a payload that did not decode as {}: {err}",
                    self.name,
                    std::any::type_name::<T>(),
                )
            });
            assert_eq!(
                &actual, expected,
                "subscriber {:?} received an unexpected value",
                self.name
            );
        });
        self
    }

    /// Asserts the most recent call carried one delivery whose raw payload equals `bytes`.
    ///
    /// # Panics
    ///
    /// Panics if the subscriber was not called, the raw payload differs, or the most recent call
    /// was a batch of several deliveries (which has no single payload - read it with
    /// [`received_raw`](Self::received_raw)).
    pub fn with_raw(self, bytes: &[u8]) -> Self {
        self.with_sole_delivery("raw payload", |one| {
            assert_eq!(
                one.raw.as_ref(),
                bytes,
                "subscriber {:?} received unexpected raw bytes",
                self.name,
            );
        });
        self
    }

    /// Asserts the most recent call settled with `outcome`'s status (any continuation on the
    /// expected value is ignored: the harness compares the settlement itself).
    ///
    /// The settlement recorded is the one the dispatcher applied to the delivery, which is what
    /// the handler answered. A transport that cannot acknowledge refuses it with
    /// [`AckError::Unsupported`](crate::AckError::Unsupported) and the dispatcher logs the refusal,
    /// so a requeue such a transport cannot perform still reads back here as a requeue; what shows
    /// the difference is the redelivery it does not produce, counted by
    /// [`assert_called`](Self::assert_called).
    ///
    /// A batch settles per element, so this asserts EVERY element of the most recent batch settled
    /// that way - which is what a uniform answer (`HandlerOutcome::ack()` for the whole slice)
    /// produces. A batch that answered element by element with differing outcomes matches none.
    ///
    /// # Panics
    ///
    /// Panics if the subscriber was not called, or anything it last received settled differently
    /// (a fail-fast panic leaves the message unsettled, which never matches).
    // The expectation is a just-built outcome token (`settled(HandlerOutcome::ack())`); a
    // reference parameter would force `&` noise at every call site.
    #[allow(clippy::needless_pass_by_value)]
    pub fn settled(self, outcome: HandlerOutcome) -> Self {
        let expected = Some(outcome.outcome());
        self.with_last("the settlement", |record| {
            let mismatched = record
                .deliveries
                .iter()
                .filter(|one| one.settle != expected)
                .count();
            assert_eq!(
                mismatched,
                0,
                "subscriber {:?} settled {mismatched} of its last {} deliveries differently \
                 than expected",
                self.name,
                record.deliveries.len(),
            );
        });
        self
    }

    /// Asserts the body was handed batches of exactly `sizes`, in the order they arrived.
    ///
    /// The batch the body sees is the batch the broker delivered: the mount site names the size
    /// with [`batch`](crate::runtime::SubscriberSettings::batch) and the broker builds the batches
    /// to it, so this is where a test reads back that it did. One call is one batch, which is
    /// what [`assert_called`](Self::assert_called) counts, while
    /// [`received_raw`](Self::received_raw) still lists the elements flat. A single-message
    /// handler is called per delivery, so a run of three reports `[1, 1, 1]`.
    ///
    /// # Panics
    ///
    /// Panics if the batches differ from `sizes`.
    pub fn assert_batch_sizes(self, sizes: &[usize]) -> Self {
        self.with_records(|records| {
            let batches: Vec<usize> = records.iter().map(|r| r.deliveries.len()).collect();
            assert_eq!(
                batches.as_slice(),
                sizes,
                "subscriber {:?} was handed batches of {batches:?}",
                self.name,
            );
        });
        self
    }

    /// Asserts the handler panicked on the most recent delivery.
    ///
    /// # Panics
    ///
    /// Panics if the subscriber was not called, or the handler did not panic.
    pub fn panicked(self) -> Self {
        self.with_last("a panic", |record| {
            assert!(
                record.panicked,
                "subscriber {:?} did not panic on its last delivery",
                self.name,
            );
        });
        self
    }

    /// Asserts the most recent delivery's classified [`Outcome`] equals `expected` (a single check
    /// covering ack / nack / drop / decode-failure / panic).
    ///
    /// # Panics
    ///
    /// Panics if the subscriber was not called, or its last outcome differs.
    pub fn assert_outcome(self, expected: Outcome) -> Self {
        self.with_last("the outcome", |record| {
            assert_eq!(
                record.outcome(),
                expected,
                "subscriber {:?} had an unexpected outcome",
                self.name,
            );
        });
        self
    }

    /// Asserts the most recent delivery's payload failed to decode into the handler's input type.
    ///
    /// # Panics
    ///
    /// Panics if the subscriber was not called, or the payload decoded successfully.
    pub fn assert_last_failed_to_decode(self) {
        self.with_last("a decode failure", |record| {
            assert!(
                record.decode_failed,
                "subscriber {:?} decoded its last delivery successfully",
                self.name,
            );
        });
    }
}

/// Assertions over what was published to one channel, read from the broker's publish log or from
/// an `Out` slot's view. The type parameter `T` is the expected payload type for
/// [`with`](Self::with).
#[derive(Debug)]
pub struct PublishedAssertions<T> {
    name: String,
    messages: Vec<RawMessage>,
    // One entry per message on a slot view; on a channel view, one per reply the runtime
    // published there. The broker's publish log records none of its own: by the time it sees a
    // message the broker has resolved the options into its protocol.
    options: Vec<RecordedOptions>,
    _payload: PhantomData<fn() -> T>,
}

impl<T> PublishedAssertions<T> {
    /// Assertions over the broker's publish log, with the options of whatever replies the
    /// runtime published to the same channel.
    pub(crate) fn new(
        name: String,
        messages: Vec<RawMessage>,
        options: Vec<RecordedOptions>,
    ) -> Self {
        Self {
            name,
            messages,
            options,
            _payload: PhantomData,
        }
    }

    /// Assertions over an `Out` slot's view, where every publish also carries the options.
    pub(crate) fn captured(name: String, records: Vec<SlotRecord>) -> Self {
        let (messages, options) = records
            .into_iter()
            .map(|record| (record.message, record.options))
            .unzip();
        Self {
            name,
            messages,
            options,
            _payload: PhantomData,
        }
    }

    /// Asserts exactly one message was published to this channel.
    ///
    /// # Panics
    ///
    /// Panics if zero or more than one message was published.
    pub fn assert_called_once(self) -> Self {
        let count = self.messages.len();
        assert_eq!(
            count, 1,
            "channel {:?} was published to {count} times, expected exactly once",
            self.name,
        );
        self
    }

    /// Asserts exactly `times` messages were published to this channel.
    ///
    /// # Panics
    ///
    /// Panics if the published count differs from `times`.
    pub fn assert_called(self, times: usize) -> Self {
        let count = self.messages.len();
        assert_eq!(
            count, times,
            "channel {:?} was published to {count} times, expected {times}",
            self.name,
        );
        self
    }

    /// Asserts nothing was published to this channel.
    ///
    /// # Panics
    ///
    /// Panics if any message was published.
    pub fn assert_not_called(self) {
        let count = self.messages.len();
        assert_eq!(
            count, 0,
            "channel {:?} was published to {count} times, expected never",
            self.name,
        );
    }

    /// Every message published to this channel, in publish order, for custom inspection beyond the
    /// built-in assertions. Bind the assertions to a variable to borrow them:
    /// `let pubs = tb.broker::<B>().published::<T>("x"); for m in pubs.messages() { .. }`.
    #[must_use]
    pub fn messages(&self) -> &[RawMessage] {
        &self.messages
    }

    /// Re-types the expected payload, for assertion sources that do not name one up front
    /// (`tb.out::<Slot>().decoded_as::<Reply>().with(&expected)`).
    #[must_use]
    pub fn decoded_as<U>(self) -> PublishedAssertions<U> {
        PublishedAssertions {
            name: self.name,
            messages: self.messages,
            options: self.options,
            _payload: PhantomData,
        }
    }

    /// The options of the most recent recorded publish, panicking if there was none.
    fn last_options(&self, what: &str) -> &RecordedOptions {
        self.last(what);
        self.options.last().unwrap_or_else(|| {
            panic!(
                "nothing the runtime published carries per-message options on {:?}, cannot \
                 assert {what}: they are recorded for a reply and for a slot publish \
                 (`tb.out::<Marker>()`), never for a message the test published itself",
                self.name,
            )
        })
    }

    /// Asserts the most recent publish carried exactly `expected` as the broker's per-message
    /// options.
    ///
    /// The options are what the broker's builder steps set on that one publish (see
    /// [`PublishBuilder::options_mut`](crate::runtime::PublishBuilder::options_mut)); a publish
    /// no step touched carried none, which
    /// [`assert_options_default`](Self::assert_options_default) asserts. They are recorded on
    /// the slot view only (`tb.out::<Marker>()`): the broker's publish log sees the message after
    /// the broker resolved them, and a bare publisher no slot attributes records nothing.
    ///
    /// # Panics
    ///
    /// Panics if nothing was published, these assertions read the broker's publish log rather
    /// than a slot view, the publish carried no options, the broker's options type is not
    /// `Options`, or the value differs from `expected`.
    pub fn with_options<Options>(self, expected: &Options) -> Self
    where
        Options: PartialEq + Debug + 'static,
    {
        let recorded = self.last_options("the per-message options");
        let actual = recorded.downcast::<Options>(&self.name).unwrap_or_else(|| {
            panic!(
                "channel {:?} published with the policy's default options, expected \
                     {expected:?}",
                self.name,
            )
        });
        assert_eq!(
            actual, expected,
            "channel {:?} published with unexpected per-message options",
            self.name,
        );
        self
    }

    /// Asserts the most recent publish carried no per-message options: no builder step ran on
    /// it, and the publish policy's own settings applied.
    ///
    /// Recorded on the slot view only, like [`with_options`](Self::with_options).
    ///
    /// # Panics
    ///
    /// Panics if nothing was published, these assertions read the broker's publish log rather
    /// than a slot view, or a step did set options on the publish.
    pub fn assert_options_default(self) -> Self {
        let recorded = self.last_options("the per-message options");
        assert!(
            recorded.value.is_none(),
            "channel {:?} published with per-message options set (`{}`), expected the policy's \
             defaults",
            self.name,
            recorded.type_name,
        );
        self
    }

    /// The most recent published message, panicking if there were none.
    fn last(&self, what: &str) -> &RawMessage {
        self.messages.last().unwrap_or_else(|| {
            panic!(
                "nothing was published to {:?}, cannot assert {what}",
                self.name
            )
        })
    }

    /// Asserts the most recent published payload equals `bytes`.
    ///
    /// # Panics
    ///
    /// Panics if nothing was published, or the raw payload differs.
    pub fn with_raw(self, bytes: &[u8]) -> Self {
        assert_eq!(
            self.last("the raw payload").payload(),
            bytes,
            "channel {:?} published unexpected raw bytes",
            self.name,
        );
        self
    }

    /// Asserts the most recent published message carries the header `name` with `value`.
    ///
    /// What a publish transform or an app-wide
    /// [`publish_layer`](crate::runtime::RustStream::publish_layer) put on the message is read
    /// here: the assertion sees the message as it left, so a stamp added on the way out shows up
    /// on the reply channel and on `tb.out::<Marker>()` alike.
    ///
    /// # Panics
    ///
    /// Panics if nothing was published, the header is absent, or its value differs.
    pub fn with_header(self, name: &str, value: impl AsRef<[u8]>) -> Self {
        let message = self.last("the published headers");
        let actual = message.headers().get(name).unwrap_or_else(|| {
            panic!(
                "channel {:?} published a message without the {name:?} header",
                self.name,
            )
        });
        assert_eq!(
            actual,
            value.as_ref(),
            "channel {:?} published an unexpected {name:?} header",
            self.name,
        );
        self
    }
}

#[cfg(any(feature = "json", feature = "cbor", feature = "msgpack"))]
impl<T> PublishedAssertions<T>
where
    T: DeserializeOwned + PartialEq + Debug,
{
    /// Asserts the most recent published payload decodes (with
    /// [`DefaultCodec`](crate::codec::DefaultCodec)) to `expected`. If the publisher uses a different
    /// codec, use [`with_codec`](Self::with_codec).
    ///
    /// # Panics
    ///
    /// Panics if nothing was published, the payload fails to decode, or it differs from `expected`.
    pub fn with(self, expected: &T) -> Self {
        self.with_codec(&DefaultCodec::default(), expected)
    }
}

impl<T> PublishedAssertions<T>
where
    T: DeserializeOwned + PartialEq + Debug,
{
    /// Like [`with`](Self::with), but decodes the published payload with `codec` - use it when the
    /// publisher was built with a non-default codec.
    ///
    /// # Panics
    ///
    /// Panics if nothing was published, the payload fails to decode, or it differs from `expected`.
    pub fn with_codec<C: Codec>(self, codec: &C, expected: &T) -> Self {
        let actual: T = codec
            .decode(self.last("the published value").payload())
            .unwrap_or_else(|err| {
                panic!(
                    "channel {:?} published a payload that did not decode as {}: {err}",
                    self.name,
                    std::any::type_name::<T>(),
                )
            });
        assert_eq!(
            &actual, expected,
            "channel {:?} published an unexpected value",
            self.name
        );
        self
    }
}

#[cfg(any(feature = "json", feature = "cbor", feature = "msgpack"))]
impl<T: DeserializeOwned> PublishedAssertions<T> {
    /// Decodes every message published to this channel (with
    /// [`DefaultCodec`](crate::codec::DefaultCodec)), in publish order, for custom inspection. Use
    /// [`decoded_with`](Self::decoded_with) for a non-default codec.
    ///
    /// # Panics
    ///
    /// Panics if any published payload fails to decode as `T`.
    #[must_use]
    pub fn decoded(&self) -> Vec<T> {
        self.decoded_with(&DefaultCodec::default())
    }
}

impl<T: DeserializeOwned> PublishedAssertions<T> {
    /// Like [`decoded`](Self::decoded), but decodes with `codec` - use it when the publisher was
    /// built with a non-default codec.
    ///
    /// # Panics
    ///
    /// Panics if any published payload fails to decode as `T`.
    #[must_use]
    pub fn decoded_with<C: Codec>(&self, codec: &C) -> Vec<T> {
        self.messages
            .iter()
            .map(|message| {
                codec.decode(message.payload()).unwrap_or_else(|err| {
                    panic!(
                        "channel {:?} published a payload that did not decode as {}: {err}",
                        self.name,
                        std::any::type_name::<T>(),
                    )
                })
            })
            .collect()
    }
}

#[cfg(all(test, feature = "json"))]
mod tests {
    use super::*;
    use crate::codec::JsonCodec;

    #[derive(Debug, PartialEq, serde::Deserialize)]
    struct Order {
        id: u64,
    }

    fn undecodable() -> PublishedAssertions<Order> {
        PublishedAssertions::new(
            "orders".to_owned(),
            vec![RawMessage::new("orders", b"not json".as_slice())],
            Vec::new(),
        )
    }

    // A decode failure inside an assertion is a test-authoring mistake, so the panic has to name
    // the channel and the type it could not produce.
    #[test]
    #[should_panic(expected = "channel \"orders\" published a payload that did not decode")]
    fn a_published_payload_that_does_not_decode_names_the_channel_and_type() {
        undecodable().with_codec(&JsonCodec, &Order { id: 7 });
    }

    #[test]
    #[should_panic(expected = "did not decode as")]
    fn decoding_every_published_payload_reports_the_first_failure() {
        let _ = undecodable().decoded_with(&JsonCodec);
    }

    // A header assertion that fails has to name the header, or a stamping middleware's absence
    // reads as "some header differs".
    #[test]
    #[should_panic(expected = "without the \"x-app\" header")]
    fn a_missing_published_header_names_itself() {
        let bare = PublishedAssertions::<Order>::new(
            "orders".to_owned(),
            vec![RawMessage::new("orders", br#"{"id":7}"#.as_slice())],
            Vec::new(),
        );
        bare.with_header("x-app", b"1");
    }

    #[test]
    #[should_panic(expected = "was published to 1 times, expected 2")]
    fn a_wrong_published_count_reports_both_numbers() {
        let one = PublishedAssertions::<Order>::new(
            "orders".to_owned(),
            vec![RawMessage::new("orders", br#"{"id":7}"#.as_slice())],
            Vec::new(),
        );
        one.assert_called(2);
    }
}