RxSwift中使用RxBlocking测试无法结束的问题求助

看起来你在测试RxSwift实现的ViewModel时遇到了RxBlocking卡住的问题，这在处理带有throttle这类延迟操作符的场景里很常见。我来帮你分析可能的原因并给出解决方案：

可能的核心问题 #

你的ViewModel里用到了throttle(0.5, scheduler: MainScheduler.instance)，而测试环境下默认不会自动运行MainScheduler的事件循环，再加上如果你的Stub服务没有正确处理事件终止，就会导致RxBlocking一直等待事件而无法结束。另外还要检查是否测试时发送的查询字符串被filter过滤掉了（比如空字符串），这也会让results序列没有任何输出，导致阻塞。

解决方案步骤 #

1. 让ViewModel的调度器可注入（推荐做法） #

硬编码MainScheduler会让测试变得麻烦，最好把调度器作为初始化参数传入，这样测试时可以用即时调度器跳过延迟：

修改你的ViewModel：

struct SearchViewModelImpl: SearchViewModel {
    let query = PublishSubject<String>()
    let results: Observable<BookResult<[Book]>>
    
    // 新增scheduler参数，默认用MainScheduler
    init(searchService: SearchService, scheduler: SchedulerType = MainScheduler.instance) {
        results = query
            .distinctUntilChanged()
            .throttle(0.5, scheduler: scheduler)
            .filter({ !$0.isEmpty })
            .flatMapLatest({ searchService.search(query: $0) })
    }
}

2. 完善SearchServiceStub的实现 #

确保Stub能正确返回错误事件，并且对应的Observable会终止（错误事件本身会终止序列）：

class SearchServiceStub: SearchService {
    // 用变量来控制返回的结果，方便测试不同场景
    var mockResult: Observable<BookResult<[Book]>>!
    
    func search(query: String) -> Observable<BookResult<[Book]>> {
        return mockResult
    }
}

3. 编写正确的测试用例 #

使用ImmediateScheduler跳过延迟，同时确保发送的查询字符串非空，并且Stub返回错误事件：

import XCTest
import RxSwift
import RxBlocking

class SearchViewModelTests: XCTestCase {
    private let disposeBag = DisposeBag()
    
    func testSearchErrorScenario() {
        // 1. 准备Stub和测试错误
        let searchServiceStub = SearchServiceStub()
        let testError = NSError(domain: "SearchError", code: 404, userInfo: [NSLocalizedDescriptionKey: "No results found"])
        searchServiceStub.mockResult = Observable.error(testError)
        
        // 2. 初始化ViewModel，用ImmediateScheduler避免延迟
        let viewModel = SearchViewModelImpl(searchService: searchServiceStub, scheduler: ImmediateScheduler.instance)
        
        // 3. 发送非空的查询事件（避免被filter过滤）
        viewModel.query.onNext("swift programming")
        
        // 4. 使用RxBlocking验证结果
        do {
            let _ = try viewModel.results.toBlocking().first()
            XCTFail("Expected to receive an error, but got a result instead")
        } catch let error as NSError {
            XCTAssertEqual(error.domain, testError.domain)
            XCTAssertEqual(error.code, testError.code)
        } catch {
            XCTFail("Unexpected error type: \(error)")
        }
    }
}

备选方案：如果不想修改ViewModel调度器 #

如果你暂时不想改动ViewModel的结构，可以在测试中手动运行RunLoop来等待throttle的延迟：

func testSearchErrorWithMainScheduler() {
    let searchServiceStub = SearchServiceStub()
    let testError = NSError(domain: "SearchError", code: 404, userInfo: nil)
    searchServiceStub.mockResult = Observable.error(testError)
    
    let viewModel = SearchViewModelImpl(searchService: searchServiceStub)
    
    viewModel.query.onNext("swift programming")
    
    // 等待throttle的0.5秒，多留一点冗余时间
    RunLoop.current.run(until: Date().addingTimeInterval(0.6))
    
    do {
        let _ = try viewModel.results.toBlocking().first()
        XCTFail("Expected error")
    } catch {
        XCTAssertEqual((error as NSError).code, 404)
    }
}

额外检查点 #

• 确保测试时发送的query不是空字符串，否则会被filter({ !$0.isEmpty })过滤，导致results序列没有任何事件，RxBlocking会一直等待。
• 确认你的BookResult类型是否正确处理了错误场景，避免错误被包装后没有正确传递到Observable的错误事件流中。

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