RxJava+Retrofit+Mockito测试异常排查：特定接口请求失败

我来帮你梳理下这个测试场景里的常见卡点和修正方案，应该能解决你遇到的接口不工作问题：

核心问题排查方向 #

• 你的Retrofit实例是否正确关联了MockWebServer的BaseUrl？这是MockWebServer能拦截请求的前提
• RxJava的线程调度在测试环境下是否适配？异步线程会导致测试还没等到响应就提前结束
• Mockito的doReturn用法是否正确？如果是测试网络交互逻辑，用真实Retrofit+MockWebServer比Mock接口更贴近真实场景

修正后的完整测试代码示例 #

先确认你的ApiService接口定义：

interface ApiService {
    @POST("v2/current-time")
    fun utc_time_from_server(): Observable<Response<Void>>
}

对应的测试代码应该这样写：

class TimeApiTest {
    private lateinit var mockWebServer: MockWebServer
    private lateinit var apiService: ApiService

    @Before
    fun setUp() {
        // 初始化MockWebServer
        mockWebServer = MockWebServer()
        mockWebServer.start()

        // 构建关联MockWebServer的Retrofit实例
        val retrofit = Retrofit.Builder()
            .baseUrl(mockWebServer.url("/"))
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
            .build()

        // 创建真实的ApiService实例（无需Mock，请求会被MockWebServer拦截）
        apiService = retrofit.create(ApiService::class.java)

        // 配置RxJava测试线程，避免异步导致测试提前结束
        RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }
        RxJavaPlugins.setComputationSchedulerHandler { Schedulers.trampoline() }
        RxJavaPlugins.setNewThreadSchedulerHandler { Schedulers.trampoline() }
    }

    @After
    fun tearDown() {
        // 清理资源
        mockWebServer.shutdown()
        RxJavaPlugins.reset()
    }

    @Test
    fun test_call_request_time_success_on_first_header_response() {
        // given
        val testMillis = 1620000000000L
        // 配置MockWebServer的响应：Header值必须是字符串类型
        mockWebServer.enqueue(
            MockResponse()
                .addHeader("millis", testMillis.toString())
                .setResponseCode(200)
                .setBody("")
        )

        // when & then
        val testObserver = apiService.utc_time_from_server().test()

        // 断言响应完成且无错误
        testObserver.assertComplete()
        testObserver.assertNoErrors()

        // 验证响应头的millis值正确
        val response = testObserver.values()[0]
        val responseMillis = response.headers()["millis"]?.toLong()
        assertEquals(testMillis, responseMillis)

        // 验证MockWebServer确实收到了目标请求
        val recordedRequest = mockWebServer.takeRequest()
        assertEquals("POST", recordedRequest.method)
        assertEquals("/v2/current-time", recordedRequest.path)
    }
}

关键修正点说明 #

1. MockWebServer与Retrofit的绑定：必须让Retrofit的baseUrl指向MockWebServer的地址，否则请求不会被拦截
2. RxJava线程适配：测试时用Schedulers.trampoline()让所有操作在当前线程执行，避免异步线程导致测试提前结束
3. Header值类型：HTTP Header的值都是字符串，设置millis时必须转成String，否则会出现解析异常
4. 用TestObserver验证Observable：RxJava的test()方法返回的TestObserver可以方便地断言响应状态、错误和返回值
5. 避免不必要的Mock：如果是测试网络交互逻辑，用真实Retrofit+MockWebServer比Mock ApiService更能模拟真实场景

若需Mock ApiService的场景（测试上层逻辑） #

如果你的测试目标是验证Repository等上层逻辑，而非网络交互，Mock ApiService的正确写法如下：

@Test
fun test_repository_time_success() {
    // given
    val mockApiService = mock(ApiService::class.java)
    val testMillis = 1620000000000L
    val mockResponse = Response.success<Void>(null, Headers.of("millis", testMillis.toString()))
    // 注意Mockito的doReturn语法：doReturn(返回值).when(被Mock对象).方法名()
    doReturn(Observable.just(mockResponse)).`when`(mockApiService).utc_time_from_server()

    // 假设你的Repository依赖ApiService
    val repository = TimeRepository(mockApiService)

    // when
    val testObserver = repository.getServerTime().test()

    // then
    testObserver.assertComplete()
    testObserver.assertValue(testMillis)
}

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