Unit-testing event aggregator

Question

I have written event aggregator with the following API (just for the fun of it, I am aware that nuget has like 100 similar implementations):

/// <summary>
/// Events aggregator
/// </summary>
public interface IMessenger
{
    /// <summary>
    /// Subscribes an object to all relevant events.
    /// </summary>
    /// <param name="listener">Object, that implements one or more IListener<TMessage> interfaces</param>
    /// <returns>Subscription handle. Dispose to unsubscribe.</returns>
    IDisposable Subscribe(object listener);

    /// <summary>
    /// Subscribes a delegate to TMessage event.
    /// </summary>
    /// <param name="listener">Event handler.</param>
    /// <returns>Subscription handle. Dispose to unsubscribe.</returns>
    IDisposable Subscribe<TMessage>(Action<TMessage> handler)
        where TMessage : IMessage;

    /// <summary>
    /// Sends message to event pipeline.
    /// </summary>
    /// <param name="message">Message to send.</param>
    /// <returns>Awaitable task, that returns true if message was processed successfully. Otherwise - false.</returns>
    Task<bool> PublishAsync(IMessage message);
}

/// <summary>
/// Message that can be delivered to subscribers via IMessenger.
/// </summary>
public interface IMessage
{
    /// <summary>
    /// Whether or not message was handled by subscriber.
    /// </summary>
    bool Handled { get; set; }
}

/// <summary>
/// Implementaions of this interface are recognized by IMessenger as event handlers for TMessage.
/// </summary>
/// <typeparam name="TMessage">Type of message.</typeparam>
public interface IListener<in TMessage> where TMessage : IMessage
{
    /// <summary>
    /// This method is called by IMessenger when TMessage is published.
    /// </summary>
    /// <param name="message">Message received by IMessenger.</param>
    void Handle(TMessage message);
}

And here are some unit tests I have written in order to test my implementation:

[TestFixture]
class MessengerTests
{
    [Test]
    public void Subscribe_OnNullListener_Throws()
    {
        var messenger = new Messenger();
        IListener<MessageA> listener = null;
        Assert.Throws<ArgumentNullException>(() => messenger.Subscribe(listener));
    }

    [Test]
    public void Subscribe_OnNotListener_Throws()
    {
        var messenger = new Messenger();
        var listener = new object();
        Assert.Throws<InvalidOperationException>(() => messenger.Subscribe(listener));
    }

    [Test]
    public void Subscribe_OnNullDelegate_Throws()
    {
        var messenger = new Messenger();
        Action<MessageA> listener = null;
        Assert.Throws<ArgumentNullException>(() => messenger.Subscribe(listener));
    }

    [Test]
    public void Publish_OnNullMessage_Throws()
    {
        var messenger = new Messenger();
        Assert.Throws<AggregateException>(() => messenger.PublishAsync(null).Wait());
    }

    [Test]
    public async Task Publish_OnMessageSent_MessageHandledByApproperiateSubscribers()
    {
        var messenger = new Messenger();
        var listener1 = Mock.Of<IListener<MessageA>>();
        var listener2 = Mock.Of<IListener<MessageA>>();
        messenger.Subscribe(listener1);
        messenger.Subscribe(listener2);
        await messenger.PublishAsync(new MessageA());
        Mock.Get(listener1).Verify(l => l.Handle(It.IsAny<MessageA>()), Times.Once);
        Mock.Get(listener2).Verify(l => l.Handle(It.IsAny<MessageA>()), Times.Once);
    }

    [Test]
    public async Task Publish_OnMessageSent_MessageIgnoredByOtherSubscribers()
    {
        var messenger = new Messenger();
        var listener1 = Mock.Of<IListener<MessageA>>();
        var listener2 = Mock.Of<IListener<MessageB>>();
        messenger.Subscribe(listener1);
        messenger.Subscribe(listener2);
        await messenger.PublishAsync(new MessageA());
        Mock.Get(listener1).Verify(l => l.Handle(It.IsAny<MessageA>()), Times.Once);
        Mock.Get(listener2).Verify(l => l.Handle(It.IsAny<MessageB>()), Times.Never);
    }

    [Test]
    public async Task Publish_OnMessageSent_MessageIgnoredByDisposedListeners()
    {
        var messenger = new Messenger();
        var listener1 = Mock.Of<IListener<MessageA>>();
        var listener2 = Mock.Of<IListener<MessageA>>();
        var subscription = messenger.Subscribe(listener1);
        messenger.Subscribe(listener2);
        await messenger.PublishAsync(new MessageA());
        subscription.Dispose();
        await messenger.PublishAsync(new MessageA());
        Mock.Get(listener1).Verify(l => l.Handle(It.IsAny<MessageA>()), Times.Once);
        Mock.Get(listener2).Verify(l => l.Handle(It.IsAny<MessageA>()), Times.Exactly(2));
    }

    [Test]
    public async Task Publish_OnMessageSent_HandledMessageIsIgnored()
    {
        var messenger = new Messenger();
        var listener1 = Mock.Of<IListener<MessageA>>();
        Mock.Get(listener1).Setup(l => l.Handle(It.IsAny<MessageA>()))
                           .Callback<MessageA>(m => m.Handled = true);
        var listener2 = Mock.Of<IListener<MessageA>>();
        messenger.Subscribe(listener1);
        messenger.Subscribe(listener2);
        await messenger.PublishAsync(new MessageA());
        Mock.Get(listener1).Verify(l => l.Handle(It.IsAny<MessageA>()), Times.Once);
        Mock.Get(listener2).Verify(l => l.Handle(It.IsAny<MessageA>()), Times.Never);
    }
}

I want those reviewed. :) My concerns are:

1. "Publish" tests have a lot of similar initialization logic, where I create and register listeners. Should I try to re-use it? Or is it ok to copy-paste unit tests. :)
2. "Assert once per unit test" they say, but I break this rule on multiple occasions. For example, I call Verify multiple times to verify, that event handler was called for every relevant subscriber.
3. "Test one thing at a time" they say. I feel like I am not following this rule either. For example, it looks like Publish_OnMessageSent_MessageHandledByApproperiateSubscribers tests that: (1) subscribers are correctly added to subscription list (2) message reaches registered subscribers (3) this whole thing works for multiple subscribers. Does this qualify as "testing one thing"?
4. Do I test async method correctly or is there a better way? Tests that return async Task look weird to me.

Edit. Here is Messenger's implementation, just to give you some idea on what I am testing. Feel free to review it as well, however it is not complete (some trivial extension methods and helper classes are missing) and I am more interested in feedback on unit tests.

/// <summary>
/// TPL.Dataflow-based implementation of IMessenger.
/// Guarantees that messages are processed with a degree of parallelism of 1 in FIFO order.
/// </summary>
public class Messenger : IMessenger, IDisposable
{
    public Messenger(int bufferSize = 1000)
    {
        if (bufferSize < 1) throw new ArgumentException("Buffer size can not be less than 1.", nameof(bufferSize));

        _pipeLine = new ActionBlock<Job>((Action<Job>)Handle, new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 1,
            BoundedCapacity = bufferSize,
        });
    }

    public void Dispose()
    {
        if (_disposed) return;
        _pipeLine.Complete();
        _pipeLine.Completion.Wait();
        _disposed = true;
    }

    public IDisposable Subscribe(object listener)
    {
        if (listener == null) throw new ArgumentNullException(nameof(listener));

        var subscriptions = listener.GetType()
                                    .GetInterfaces()
                                    .Where(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IListener<>))
                                    .Select(listenerType => listenerType.GetGenericArguments()[0])
                                    .Select(messageType => Subscribe(messageType, listener))
                                    .ToArray();
        if (!subscriptions.Any())
            throw new InvalidOperationException(
                $"{listener.GetType().Name} does not implement IListener<T>. Use IMessenger.Subscribe<TMessage> instead.");

        return subscriptions.Combine();
    }

    public IDisposable Subscribe<TMessage>(Action<TMessage> handler) where TMessage : IMessage
    {
        if (handler == null) throw new ArgumentNullException(nameof(handler));

        return Subscribe(typeof(TMessage), new DelegateListener<TMessage>(handler));
    }

    public async Task<bool> PublishAsync(IMessage message)
    {
        if (message == null) throw new ArgumentNullException(nameof(message));
        var job = new Job(message);
        var enqueued = await _pipeLine.SendAsync(job);
        if (!enqueued)
        {
            job.Result.SetResult(false);
        }
        return await job.Result.Task;
    }

    private readonly Dictionary<Type, IListenerCollection> _subscribers = new Dictionary<Type, IListenerCollection>();
    private readonly ActionBlock<Job> _pipeLine;
    private bool _disposed;

    private IDisposable Subscribe(Type messageType, object listener)
    {
        IListenerCollection list;

        lock (_subscribers)
        {
            if (!_subscribers.TryGetValue(messageType, out list))
            {
                var collectionType = typeof(ListenerCollection<>).MakeGenericType(messageType);
                list = (IListenerCollection)Activator.CreateInstance(collectionType);
                _subscribers[messageType] = list;
            }
        }

        return list.Add(listener);
    }

    private void Handle(Job job)
    {
        IListenerCollection listeners;
        lock (_subscribers)
        {
            _subscribers.TryGetValue(job.Message.GetType(), out listeners);
        }
        try
        {
            listeners?.Handle(job.Message);
            job.Result.SetResult(true);
        }
        catch (Exception ex)
        {
            job.Result.SetException(ex);
        }
    }

    private class Job
    {
        public Job(IMessage message)
        {
            Message = message;
        }
        public IMessage Message { get; }
        public TaskCompletionSource<bool> Result { get; } = new TaskCompletionSource<bool>();
    }
}

Answer 1

The tests should be easy to maintain and understand. That is the tool that drives development. Sometimes it may contradict with the widely known rules you mentioned, but it does not mean we should be fanatic about that (as soon as the tests drive the development and work as the safety net).

1. That's totally fine and even more - introducing 'common' initialization logic increases coupling of your unit tests. Imagine you need to add a new 'common' line to setup things common for a few tests. You never know for sure if there are any other tests impacted; you may make your tests false positive. It's more safe to modify your test 1 by 1 as necessary.

2. Tests like Publish_OnMessageSent_HandledMessageIsIgnored have about 6-7 arrange lines and 2 assertions. IMO, it is easier to read, understand and (!) maintain this single test than splitting it into two (or more) just because of the 'one assertion per test' rule.

3. That looks like Facade tests. That might be fine to have them at some level, but it's important to understand that tests like that may be a bit more hard to maintain since there are more parts that might be broken during refactoring.

4. Don't have any oppinion on that... Looks fine to me.