如何为Rx.NET的Catch分支编写单元测试？

问题描述 #

我正在对基于Rx.NET的ProductService做单元测试，目前仅能覆盖正常流程，无法测试代码中的异常分支——即以下未被覆盖的逻辑：

.Catch<string, TimeoutException>(_ => Observable.Return("Timeout"))
.Catch<string, Exception>(ex => Observable.Return(ex.Message))

能否借助TestScheduler完成这些异常路径的测试？以下是服务实现及现有测试代码：

服务实现代码 #

public interface IProductService
{
    Task<string> GetAsync();
}

public sealed class ProductService : IProductService
{
    private readonly HttpClient _httpClient;

    public ProductService(HttpClient httpClient)
    {
        _httpClient = httpClient;
    }

    public Task<string> GetAsync()
    {
        return Observable
            .FromAsync(() => _httpClient.GetAsync("http://www.mocky.io/v2/5e307edf3200005d00858b49"))
            .SubscribeOn(TaskPoolScheduler.Default)
            .Retry(5)
            .Timeout(TimeSpan.FromSeconds(5))
            .Do(x => Console.WriteLine($"Is message successful? {x.IsSuccessStatusCode}"))
            .SelectMany(async responseMessage =>
            {
                var response = await responseMessage.Content.ReadAsStringAsync();
                return response;
            })
            .Catch<string, TimeoutException>(_ => Observable.Return("Timeout"))
            .Catch<string, Exception>(ex => Observable.Return(ex.Message))
            .ToTask();
    }
}

现有测试代码 #

public class ProductServiceTests
{
    [Fact]
    public async Task GetAsync_ShouldReturnText_WhenRequestIsSent()
    {
        // Arrange
        var messageHandler = new MockHttpMessageHandler("Stuff I want to return", HttpStatusCode.OK);
        var httpClient = new HttpClient(messageHandler);
        var sut = new ProductService(httpClient);

        // Act
        var result = await sut.GetAsync();

        // Assert
        result.Should().Be("Stuff I want to return");
    }

    // TODO: Unhappy path
}

解决方案 #

可以用TestScheduler测试异常分支，但需要先对ProductService做小幅改造，让调度器和超时时间可配置，这样测试时才能替换为TestScheduler精确控制流程。

步骤1：改造ProductService，支持注入调度器和超时时间 #

保留原构造函数兼容现有调用，新增内部构造函数用于测试注入依赖：

public sealed class ProductService : IProductService
{
    private readonly HttpClient _httpClient;
    private readonly IScheduler _scheduler;
    private readonly TimeSpan _timeout;

    // 原构造函数，保持对外兼容
    public ProductService(HttpClient httpClient)
        : this(httpClient, TaskPoolScheduler.Default, TimeSpan.FromSeconds(5))
    {
    }

    // 测试专用构造函数，允许注入调度器和超时时间
    internal ProductService(HttpClient httpClient, IScheduler scheduler, TimeSpan timeout)
    {
        _httpClient = httpClient;
        _scheduler = scheduler;
        _timeout = timeout;
    }

    public Task<string> GetAsync()
    {
        return Observable
            .FromAsync(() => _httpClient.GetAsync("http://www.mocky.io/v2/5e307edf3200005d00858b49"))
            .SubscribeOn(_scheduler) // 使用注入的调度器
            .Retry(5)
            .Timeout(_timeout, _scheduler) // 指定调度器控制超时触发
            .Do(x => Console.WriteLine($"Is message successful? {x.IsSuccessStatusCode}"))
            .SelectMany(async responseMessage =>
            {
                var response = await responseMessage.Content.ReadAsStringAsync();
                return response;
            })
            .Catch<string, TimeoutException>(_ => Observable.Return("Timeout"))
            .Catch<string, Exception>(ex => Observable.Return(ex.Message))
            .ToTask(_scheduler); // 用测试调度器完成Task转换
    }
}

步骤2：测试TimeoutException分支 #

通过TestScheduler快进时间，触发Timeout操作符：

[Fact]
public async Task GetAsync_ShouldReturnTimeout_WhenRequestTimesOut()
{
    // Arrange
    var testScheduler = new TestScheduler();
    // 模拟HttpClient请求挂起，不返回结果
    var messageHandler = new MockHttpMessageHandler(async (request, cancellationToken) =>
    {
        await Task.Delay(Timeout.Infinite, cancellationToken);
        return new HttpResponseMessage(HttpStatusCode.OK);
    });
    var httpClient = new HttpClient(messageHandler);
    var timeout = TimeSpan.FromSeconds(5);
    var sut = new ProductService(httpClient, testScheduler, timeout);

    // Act
    var task = sut.GetAsync();
    // 快进时间至超时触发点
    testScheduler.AdvanceBy(timeout.Ticks + 1);

    // Assert
    var result = await task;
    result.Should().Be("Timeout");
}

步骤3：测试普通Exception分支 #

模拟HttpClient抛出异常，验证Catch分支返回异常消息：

[Fact]
public async Task GetAsync_ShouldReturnExceptionMessage_WhenRequestThrows()
{
    // Arrange
    var testScheduler = new TestScheduler();
    var expectedExceptionMsg = "连接失败";
    var messageHandler = new MockHttpMessageHandler((request, cancellationToken) =>
    {
        throw new HttpRequestException(expectedExceptionMsg);
    });
    var httpClient = new HttpClient(messageHandler);
    var sut = new ProductService(httpClient, testScheduler, TimeSpan.FromSeconds(5));

    // Act
    var task = sut.GetAsync();
    testScheduler.Start(); // 触发调度器执行所有操作

    // Assert
    var result = await task;
    result.Should().Be(expectedExceptionMsg);
}

关键注意点 #

• 必须将SubscribeOn、Timeout、ToTask的调度器统一替换为TestScheduler，否则超时逻辑不受测试调度器控制。
• 测试超时场景时，需让HttpClient请求持续挂起，不返回结果，确保Timeout操作符能触发。
• 若要测试Retry(5)逻辑，可调整MockHandler的行为计数重试次数，再通过TestScheduler验证重试流程。

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