Unit-testing event aggregator

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>();
    }
}