如何对使用EventDataBatch的Azure EventHub Producer进行单元测试？

问题根源 #

Azure Event Hubs SDK v5把EventDataBatch改成了密封类，还移除了v4版本里的ToEnumerable()方法，导致批次里的具体事件没有公开的访问途径。现在单元测试时，除了验证事件数量，只能靠反射去读取内部数据，测试成本一下子高了不少。

解决方案 #

首选方案：封装批次逻辑，用抽象接口解耦 #

别直接在业务代码里硬写EventDataBatch的创建和添加逻辑，把这部分抽成一个接口，让业务代码依赖接口而非具体的SDK类。这样测试时就能Mock这个接口，返回自己能跟踪数据的批次实现。

举个实际例子：

// 定义抽象的批次工厂接口
public interface IEventBatchFactory
{
    Task<EventDataBatch> CreateBatchAsync(CancellationToken cancellationToken = default);
    bool TryAdd(EventDataBatch batch, EventData eventData);
}

// 生产环境用的实现，直接调用SDK的方法
public class EventBatchFactory : IEventBatchFactory
{
    private readonly EventHubProducerClient _producer;

    public EventBatchFactory(EventHubProducerClient producer)
    {
        _producer = producer;
    }

    public async Task<EventDataBatch> CreateBatchAsync(CancellationToken cancellationToken)
    {
        return await _producer.CreateBatchAsync(cancellationToken);
    }

    public bool TryAdd(EventDataBatch batch, EventData eventData)
    {
        return batch.TryAdd(eventData);
    }
}

// 你的业务代码，现在依赖IEventBatchFactory而不是直接操作SDK类
public class ChangeDataPublisher
{
    private readonly IEventBatchFactory _batchFactory;
    private readonly EventHubProducerClient _producer;

    public ChangeDataPublisher(IEventBatchFactory batchFactory, EventHubProducerClient producer)
    {
        _batchFactory = batchFactory;
        _producer = producer;
    }

    public async Task PublishChangesAsync(IEnumerable<ChangeRecord> candidateData, CancellationToken ct)
    {
        var batch = await _batchFactory.CreateBatchAsync(ct);
        
        foreach (var data in candidateData)
        {
            var eventData = new EventData(JsonSerializer.SerializeToUtf8Bytes(data));
            
            if (!_batchFactory.TryAdd(batch, eventData))
            {
                await _producer.SendAsync(batch, ct);
                batch = await _batchFactory.CreateBatchAsync(ct);
                _batchFactory.TryAdd(batch, eventData);
            }
        }

        if (batch.Count > 0)
        {
            await _producer.SendAsync(batch, ct);
        }
    }
}

测试的时候，自己写个能跟踪事件的TestableEventDataBatch，再MockIEventBatchFactory返回这个测试类：

// 测试专用的批次类，能记录所有添加的事件
public class TestableEventDataBatch : EventDataBatch
{
    private readonly List<EventData> _trackedEvents = new();
    public IReadOnlyList<EventData> TrackedEvents => _trackedEvents;

    // 调用父类构造，参数随便填就行，测试用不上
    public TestableEventDataBatch() : base(null, 0, null) { }

    public override bool TryAdd(EventData eventData)
    {
        _trackedEvents.Add(eventData);
        return true; // 测试时可以忽略大小限制，需要的话也能自定义逻辑
    }
}

// 单元测试代码
[Test]
public async Task PublishChanges_ShouldAddAllRecordsToBatches()
{
    // 准备测试数据
    var testRecords = new List<ChangeRecord> 
    { 
        new ChangeRecord { Id = 1, Content = "test1" },
        new ChangeRecord { Id = 2, Content = "test2" }
    };

    var testBatch = new TestableEventDataBatch();
    var mockBatchFactory = new Mock<IEventBatchFactory>();
    
    // 设置Mock返回测试批次
    mockBatchFactory.Setup(f => f.CreateBatchAsync(It.IsAny<CancellationToken>()))
                   .ReturnsAsync(testBatch);
    // 设置Mock调用测试批次的TryAdd
    mockBatchFactory.Setup(f => f.TryAdd(It.IsAny<EventDataBatch>(), It.IsAny<EventData>()))
                   .Callback<EventDataBatch, EventData>((batch, evt) => ((TestableEventDataBatch)batch).TryAdd(evt))
                   .Returns(true);

    var mockProducer = new Mock<EventHubProducerClient>();
    var publisher = new ChangeDataPublisher(mockBatchFactory.Object, mockProducer.Object);

    // 执行测试方法
    await publisher.PublishChangesAsync(testRecords, CancellationToken.None);

    // 验证结果
    Assert.Multiple(() =>
    {
        Assert.That(testBatch.TrackedEvents.Count, Is.EqualTo(testRecords.Count));
        
        // 验证第一个事件的内容
        var firstRecord = JsonSerializer.Deserialize<ChangeRecord>(testBatch.TrackedEvents[0].Body.ToArray());
        Assert.That(firstRecord.Id, Is.EqualTo(1));
    });
}

临时方案：用反射读取内部事件 #

如果不想重构现有代码，也可以用反射去读取EventDataBatch的内部事件集合。但注意：这种方法依赖SDK的内部实现，哪天SDK更新改了内部字段名，测试就会崩，只能作为临时过渡方案。

代码示例：

public static IEnumerable<EventData> ExtractEventsFromBatch(EventDataBatch batch)
{
    // 查找内部的_events字段
    var eventsField = typeof(EventDataBatch).GetField(
        "_events", 
        BindingFlags.NonPublic | BindingFlags.Instance);
    
    if (eventsField == null)
        throw new InvalidOperationException("无法访问EventDataBatch的内部事件集合");
    
    return (IEnumerable<EventData>)eventsField.GetValue(batch);
}

// 测试里这么用
var capturedBatch = ...; // 从Mock的SendAsync里捕获到的批次
var events = ExtractEventsFromBatch(capturedBatch);
Assert.That(events.Count(), Is.EqualTo(expectedCount));
// 再逐个验证事件内容

备选思路：关注官方测试支持 #

目前Azure Event Hubs SDK v5没有官方提供可测试的EventDataBatch实现，但可以留意Azure SDK for .NET的GitHub仓库的Issue和更新，说不定后续会添加测试相关的工具类。

总结 #

优先选封装抽象接口的方案，既能解决测试问题，还能让业务代码更灵活、易维护。反射方案只能临时用用，不建议长期依赖，毕竟SDK内部结构说变就变。

内容的提问来源于stack exchange，提问作者Chris DaMour