我之前做Clean Architecture/MVP的Android项目时，也碰到过一模一样的单元测试困境！咱们一步步拆解解决它～

先明确场景（我猜你的代码结构大概是这样） #

假设你的UseCase基类和实现是这类结构：

// 基类UseCase
abstract class UseCase<T, Params> {
    open fun execute(observer: DisposableObserver<T>, params: Params) {
        val observable = buildObservable(params)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
        observable.subscribe(observer)
    }

    protected abstract fun buildObservable(params: Params): Observable<T>
}

// 具体UseCase
class FetchDataUseCase(private val repo: DataRepo) : UseCase<Data, Unit>() {
    override fun buildObservable(params: Unit): Observable<Data> {
        return repo.fetchRemoteData() // 这里是你不想在测试中执行的真实逻辑
    }
}

Presenter里的调用逻辑大概是：

class DataPresenter(private val useCase: FetchDataUseCase) : BasePresenter<DataView>() {
    fun loadData() {
        view?.showLoading()
        useCase.execute(object : DisposableObserver<Data>() {
            override fun onNext(data: Data) {
                view?.hideLoading()
                view?.renderData(data)
            }

            override fun onError(e: Throwable) {
                view?.hideLoading()
                view?.showErrorMsg(e.message)
            }

            override fun onComplete() {}
        }, Unit)
    }
}

核心解决思路：拦截UseCase的execute方法，手动触发回调 #

我们不需要真的让UseCase执行buildObservable里的逻辑，而是通过Mock框架（比如Mockito）拦截execute调用，拿到传入的DisposableObserver，然后手动调用它的onNext/onError方法来模拟场景。

1. 测试成功场景（OnNext触发）

import org.junit.Before
import org.junit.Test
import org.mockito.Mock
import org.mockito.Mockito.*
import org.mockito.MockitoAnnotations

class DataPresenterTest {
    @Mock
    private lateinit var mockView: DataView

    @Mock
    private lateinit var mockUseCase: FetchDataUseCase

    private lateinit var presenter: DataPresenter

    @Before
    fun setup() {
        MockitoAnnotations.openMocks(this)
        presenter = DataPresenter(mockUseCase)
        presenter.attachView(mockView)
        
        // 避免RxJava线程问题，设置测试调度器
        RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }
        RxAndroidPlugins.setMainThreadSchedulerHandler { Schedulers.trampoline() }
    }

    @Test
    fun `loadData should hide loading and show data when use case succeeds`() {
        // 准备测试数据
        val testData = Data("mock content")

        // 拦截UseCase的execute方法，手动触发OnNext
        doAnswer { invocation ->
            // 取出传入的DisposableObserver参数
            val observer = invocation.getArgument<DisposableObserver<Data>>(0)
            observer.onNext(testData)
            observer.onComplete()
            null
        }.`when`(mockUseCase).execute(any(), eq(Unit))

        // 触发Presenter的业务方法
        presenter.loadData()

        // 验证View的行为是否符合预期
        verify(mockView).showLoading()
        verify(mockView).hideLoading()
        verify(mockView).renderData(testData)
    }
}

2. 测试错误场景（OnError触发）

只需要修改doAnswer里的逻辑，手动调用onError即可：

@Test
fun `loadData should hide loading and show error when use case fails`() {
    val testError = RuntimeException("mock error message")

    doAnswer { invocation ->
        val observer = invocation.getArgument<DisposableObserver<Data>>(0)
        observer.onError(testError)
        null
    }.`when`(mockUseCase).execute(any(), eq(Unit))

    presenter.loadData()

    verify(mockView).showLoading()
    verify(mockView).hideLoading()
    verify(mockView).showErrorMsg(testError.message)
}

关键注意点 #

• 确保基类UseCase的execute方法是open的：Mockito只能拦截可重写的方法，要是基类里的execute是final的，就没法拦截了。
• 重置RxJava调度器：测试完成后记得重置调度器，避免影响其他测试：

  @After
  fun tearDown() {
      RxJavaPlugins.reset()
      RxAndroidPlugins.reset()
  }
  

这样就能完全控制UseCase的回调行为，精准验证Presenter在不同场景下的响应逻辑啦～

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