test: cover bulk-dispatch + null-publication-topic for reply producer lookup

- bulk dispatch reply messages via outbox sweep (firstMessage bag lookup)
- wrap with null publication.Topic skips bag write (guard pin, sync + async)

Co-Authored-By: Claude (claude-opus-4-7) <noreply@anthropic.com>

Author: Jonny-Freemarket

tests/Paramore.Brighter.Core.Tests/CommandProcessors/Post/When_Bulk_Dispatching_Reply_Messages_Async.cs

@@ -0,0 +1,110 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using System.Transactions;
+using Microsoft.Extensions.Time.Testing;
+using Paramore.Brighter.Core.Tests.CommandProcessors.TestDoubles;
+using Paramore.Brighter.Extensions;
+using Paramore.Brighter.Observability;
+using Polly.Registry;
+using Xunit;
+
+namespace Paramore.Brighter.Core.Tests.CommandProcessors.Post
+{
+    // Gap: BulkDispatchAsync groups by Header.Topic (the mapper-set reply topic) but
+    // looks up the producer using the first message's bag. The single-message Dispatch
+    // paths are covered; this pins the bulk-outbox-clear path so a future refactor of
+    // GetProducerLookupTopic or the firstMessage assumption cannot silently regress and
+    // fail to locate the producer when draining the outbox via the bulk sweep.
+    public class CommandProcessorBulkDispatchReplyAsyncTests
+    {
+        private const string ProducerTopic = "Reply";
+        private readonly CommandProcessor _commandProcessor;
+        private readonly IAmAnOutboxProducerMediator _mediator;
+        private readonly MyResponse _replyOne;
+        private readonly MyResponse _replyTwo;
+        private readonly InternalBus _internalBus = new();
+        private readonly RoutingKey _replyTopic;
+
+        public CommandProcessorBulkDispatchReplyAsyncTests()
+        {
+            var timeProvider = new FakeTimeProvider();
+            var producerRoutingKey = new RoutingKey(ProducerTopic);
+
+            _replyTopic = new RoutingKey(Uuid.NewAsString());
+            var replyAddress = new ReplyAddress(_replyTopic, Uuid.NewAsString());
+            _replyOne = new MyResponse(replyAddress) { ReplyValue = "Hello" };
+            _replyTwo = new MyResponse(replyAddress) { ReplyValue = "World" };
+
+            InMemoryMessageProducer messageProducer = new(_internalBus,
+                new Publication
+                {
+                    Topic = producerRoutingKey,
+                    RequestType = typeof(MyResponse)
+                });
+
+            var messageMapperRegistry = new MessageMapperRegistry(
+                null,
+                new SimpleMessageMapperFactoryAsync(_ => new MyResponseMessageMapperAsync()));
+            messageMapperRegistry.RegisterAsync<MyResponse, MyResponseMessageMapperAsync>();
+
+            var resiliencePipelineRegistry = new ResiliencePipelineRegistry<string>()
+                .AddBrighterDefault();
+
+            var producerRegistry = new ProducerRegistry(
+                new Dictionary<RoutingKey, IAmAMessageProducer>
+                {
+                    { producerRoutingKey, messageProducer }
+                });
+
+            var tracer = new BrighterTracer(timeProvider);
+            var outbox = new InMemoryOutbox(timeProvider) { Tracer = tracer };
+
+            _mediator = new OutboxProducerMediator<Message, CommittableTransaction>(
+                producerRegistry,
+                resiliencePipelineRegistry,
+                messageMapperRegistry,
+                new EmptyMessageTransformerFactory(),
+                new EmptyMessageTransformerFactoryAsync(),
+                tracer,
+                new FindPublicationByPublicationTopicOrRequestType(),
+                outbox,
+                maxOutStandingMessages: -1
+            );
+
+            _commandProcessor = new CommandProcessor(
+                new InMemoryRequestContextFactory(),
+                new DefaultPolicy(),
+                resiliencePipelineRegistry,
+                _mediator,
+                new InMemorySchedulerFactory()
+            );
+        }
+
+        [Fact]
+        public async Task When_Bulk_Dispatching_Reply_Messages_Async()
+        {
+            //arrange - deposit two replies whose mapper sets Header.Topic to the reply
+            //address, not the registered producer topic
+            var context = new RequestContext();
+            await _commandProcessor.DepositPostAsync<MyResponse>([_replyOne, _replyTwo], context);
+
+            //act - drain via the bulk path (exercised by background outbox sweeps)
+            await _mediator.ClearOutstandingFromOutboxAsync(
+                amountToClear: 10,
+                minimumAge: TimeSpan.Zero,
+                useBulk: true,
+                requestContext: context);
+
+            //allow background clear to run
+            await Task.Delay(500);
+
+            //assert - messages landed on the reply topic. If producer lookup had used
+            //Header.Topic (reply address) instead of the bag's ProducerTopic, LookupBy
+            //would have thrown and nothing would arrive on the bus.
+            var messages = _internalBus.Stream(_replyTopic).ToArray();
+            Assert.Equal(2, messages.Length);
+        }
+    }
+}

tests/Paramore.Brighter.Core.Tests/MessageSerialisation/When_Wrapping_With_Null_Publication_Topic.cs

@@ -0,0 +1,55 @@
+using Paramore.Brighter.Core.Tests.CommandProcessors.TestDoubles;
+using Xunit;
+
+namespace Paramore.Brighter.Core.Tests.MessageSerialisation;
+
+// Gap: WrapPipeline.Wrap now writes publication.Topic into Header.Bag when the
+// mapper-set topic differs from the publication topic. The fix guards against a
+// null publication.Topic before writing. Without this test, a future refactor that
+// drops the null-check would NRE at runtime for publications that legitimately
+// have no topic (dynamic routing scenarios).
+public class WrapNullPublicationTopicTests
+{
+    private readonly TransformPipelineBuilder _pipelineBuilder;
+    private readonly MyResponse _myResponse;
+    private readonly Publication _publication;
+
+    public WrapNullPublicationTopicTests()
+    {
+        TransformPipelineBuilder.ClearPipelineCache();
+
+        var mapperRegistry = new MessageMapperRegistry(
+            new SimpleMessageMapperFactory(_ => new MyResponseMessageMapper()),
+            null);
+        mapperRegistry.Register<MyResponse, MyResponseMessageMapper>();
+
+        var replyTopic = new RoutingKey(Uuid.NewAsString());
+        var replyAddress = new ReplyAddress(replyTopic, Uuid.NewAsString());
+        _myResponse = new MyResponse(replyAddress) { ReplyValue = "Hello World" };
+
+        var messageTransformerFactory = new SimpleMessageTransformerFactory(_ => null);
+
+        //publication with no topic — mapper still sets its own topic from the reply address
+        _publication = new Publication
+        {
+            Topic = null,
+            RequestType = typeof(MyResponse)
+        };
+
+        _pipelineBuilder = new TransformPipelineBuilder(mapperRegistry, messageTransformerFactory);
+    }
+
+    [Fact]
+    public void When_Wrapping_With_Null_Publication_Topic()
+    {
+        //act
+        var transformPipeline = _pipelineBuilder.BuildWrapPipeline<MyResponse>();
+        var message = transformPipeline.Wrap(_myResponse, new RequestContext(), _publication);
+
+        //assert - topic came from the mapper
+        Assert.Equal(_myResponse.SendersAddress.Topic, message.Header.Topic);
+
+        //assert - no ProducerTopic entry was written to the bag (the guard held)
+        Assert.False(message.Header.Bag.ContainsKey(Message.ProducerTopicHeaderName));
+    }
+}

tests/Paramore.Brighter.Core.Tests/MessageSerialisation/When_Wrapping_With_Null_Publication_Topic_Async.cs

@@ -0,0 +1,53 @@
+using System.Threading.Tasks;
+using Paramore.Brighter.Core.Tests.CommandProcessors.TestDoubles;
+using Paramore.Brighter.Observability;
+using Xunit;
+
+namespace Paramore.Brighter.Core.Tests.MessageSerialisation;
+
+// Async twin of WrapNullPublicationTopicTests — pins the null-publication-topic
+// guard in WrapPipelineAsync so refactors cannot silently NRE.
+public class AsyncWrapNullPublicationTopicTests
+{
+    private readonly TransformPipelineBuilderAsync _pipelineBuilder;
+    private readonly MyResponse _myResponse;
+    private readonly Publication _publication;
+
+    public AsyncWrapNullPublicationTopicTests()
+    {
+        TransformPipelineBuilder.ClearPipelineCache();
+
+        var mapperRegistry = new MessageMapperRegistry(
+            null,
+            new SimpleMessageMapperFactoryAsync(_ => new MyResponseMessageMapperAsync()));
+        mapperRegistry.RegisterAsync<MyResponse, MyResponseMessageMapperAsync>();
+
+        var replyTopic = new RoutingKey(Uuid.NewAsString());
+        var replyAddress = new ReplyAddress(replyTopic, Uuid.NewAsString());
+        _myResponse = new MyResponse(replyAddress) { ReplyValue = "Hello World" };
+
+        var messageTransformerFactory = new SimpleMessageTransformerFactoryAsync(_ => null);
+
+        _publication = new Publication
+        {
+            Topic = null,
+            RequestType = typeof(MyResponse)
+        };
+
+        _pipelineBuilder = new TransformPipelineBuilderAsync(mapperRegistry, messageTransformerFactory, InstrumentationOptions.All);
+    }
+
+    [Fact]
+    public async Task When_Wrapping_With_Null_Publication_Topic_Async()
+    {
+        //act
+        var transformPipeline = _pipelineBuilder.BuildWrapPipeline<MyResponse>();
+        var message = await transformPipeline.WrapAsync(_myResponse, new RequestContext(), _publication);
+
+        //assert - topic came from the mapper
+        Assert.Equal(_myResponse.SendersAddress.Topic, message.Header.Topic);
+
+        //assert - no ProducerTopic entry was written to the bag (the guard held)
+        Assert.False(message.Header.Bag.ContainsKey(Message.ProducerTopicHeaderName));
+    }
+}