单元测试 - 验证 Observable 是否已订阅

Question

我有这样的java代码

 mDataManager.getObservable("hello").subscribe( subscriber );

我想verify 以下 Observable 是 .subscribe()

我试图模拟 getObservable() 和 verify

 Observable<Response> res = mock(Observable.class);
 when(mDataManager.getObservable("hello")).thenReturn(res);
 verify(res).subscribe();

但是有一个错误

Caused by: java.lang.IllegalStateException: onSubscribe function can not be null.
at rx.Observable.subscribe(Observable.java:8167)
at rx.Observable.subscribe(Observable.java:8158)
at rx.Observable.subscribe(Observable.java:7962)
....
Caused by: rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: omni.neo.hk.omniapiservice.v4.model.external.UserLoginBean.class
at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:109)
at rx.exceptions.Exceptions.throwOrReport(Exceptions.java:187)
at rx.internal.operators.OperatorDoOnEach$1.onNext(OperatorDoOnEach.java:82)
... 48 more

我认为 mock 不可能在这里使用 Observable，但如果没有模拟的 Observable 我不能这样做verify(res).subscribe()

在这种情况下有什么建议吗？

Answer 1

我今天花了几个小时才意识到这是一个愚蠢的错误。 如果您使用的是 PowerMockito，请检查您的 @PrepareForTest。

@PrepareForTest({Observable.class})

也不要忘记模拟它。我在@before 上做到了：

PowerMockito.mockStatic(Observable.class);

我希望它可以帮助。谢谢

Answer 2

我不确定我是否在回答所提出的问题，但我认为是这样......

可以创建单元测试来检查是否订阅了 observable。请注意，在 RxJava 1.x 中，TestSubject 支持此功能，但 Subject 实现可以促进该功能。 TestSubject 是dropped for RxJava 2。

不要像原始发布者那样尝试模拟 observable，而是使用 PublishSubject，运行业务逻辑，然后使用发布主题来查看 Observable 是否正在被观察。下面是一些验证此测试是否可行的测试代码。当我运行它们时，测试通过了。

有关代码的说明，请参见代码 cmets。就个人而言，该测试对于验证控制器如何处理数据的其他测试似乎是多余的。如果控制器从未订阅和接收来自数据服务的数据，那么其他测试将失败。另一方面，测试取消订阅有助于避免内存泄漏，这可能具有很高的价值。

import org.junit.Test;

import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.subjects.PublishSubject;

import static junit.framework.Assert.assertFalse;
import static junit.framework.Assert.assertTrue;

/**
 * Tests whether it is possible to test whether a controller properly unsubscribes from an Observable.
 */
public class SubscribeUnsubscribeTests {

    /**
     * A service that returns a stream of strings.
     */
    interface StringService {
        Observable<String> getStrings();
    }

    /**
     * A controller that requests strings from the {@link StringService} and logs them
     * to system out as they are received.
     */
    class StringLoggingController {

        private StringService stringService;

        private Disposable stringSubscriptionDisposable;

        public StringLoggingController(StringService stringService) {
            this.stringService = stringService;
        }

        /**
         * Causese the controller to begin strings request and logging.
         */
        public void start() {
            stringSubscriptionDisposable = stringService.getStrings()
                    .subscribe(string -> System.out.print(string));
        }

        public void stop() {
            if (stringSubscriptionDisposable != null) {
                if (!stringSubscriptionDisposable.isDisposed()) {
                    stringSubscriptionDisposable.dispose();
                    stringSubscriptionDisposable = null;
                }
            }
        }
    }

    /**
     * A {@link StringService} that can report whether {@link StringService#getStrings()}
     * has observers.
     */
    class ReportIsSubscribedStringService implements StringService {

        private PublishSubject<String> publishSubject = PublishSubject.create();

        public Observable<String> getStrings() {
            return publishSubject;
        }

        /**
         * @return true if the {@link #getStrings()} observable has observers.
         */
        public boolean stringsAreBeingObserved() {
            return publishSubject.hasObservers();
        }
    }

    /**
     * Verifies that the {@link StringLoggingController} is subscribing to the service.
     */
    @Test
    public void stringsLoggingControllerSubscribesToStringService() {
        ReportIsSubscribedStringService service = new ReportIsSubscribedStringService();
        StringLoggingController controller = new StringLoggingController(service);

        controller.start();
        assertTrue(service.stringsAreBeingObserved());
    }

    /**
     * Verifies that the {@link StringLoggingController} is unsubscribing from the service.
     */
    @Test
    public void stringsLoggingControllerUnsubscribesFromStringService() {
        ReportIsSubscribedStringService service = new ReportIsSubscribedStringService();
        StringLoggingController controller = new StringLoggingController(service);

        controller.start();
        controller.stop();

        assertFalse(service.stringsAreBeingObserved());
    }
}

-- 编辑-- 我之前写道，我无法测试退订。我确实想出了如何测试它。事实证明我的测试失败了，因为测试的代码没有正确取消订阅（哈哈 - 测试有效，去看看）。我更新了上面的代码来说明如何测试订阅和取消订阅。

Answer 3

也许您可以将Observable.onSubscribe 方法与RunTestOnContext 规则一起使用？ TestContext 可以为您提供Async 对象，该对象仅在测试完成后终止。我认为，如果将其与 Observable#doOnSubscribe 结合使用，您可以实现所需的行为。

但是，使用Async 有时可能会有点混乱。在下面的示例中，如果 observable 从未被订阅，doOnSubscribe 函数将永远不会被评估，您的测试也不会终止。

例子：

@RunWith(VertxUnitRunner.class)
public class SubscriptionTest {

  @Rule
  public RunTestOnContext vertxRule = new RunTestOnContext();

  @Test
  public void observableMustBeSubscribed(final TestContext context) {
    final Async async = context.async();
    final Observable<String> observable = Observable.just("hello").doOnSubscribe(async::complete);
    final Manager mock = mock(Manager.class);
    when(mock.getObservable()).thenReturn(observable);

    mock.getObservable().subscribe();
  }

  interface Manager {
    Observable<String> getObservable();
  }
}

Answer 4

我发现RxJava提供了一个叫TestSubject的类

你可以这样创建

private TestScheduler eventsScheduler = new TestScheduler();
private TestSubject<MyEvent> eventObservable = TestSubject.create(eventsScheduler);

这将为您提供返回布尔值的方法hasObservers()。

@Test
public void testSubscription(){
    myTestClass.init(eventObservable);

    assertTrue(eventObservable.hasObservers());
}

TestSubject 还允许您完美地确定应该发送事件的时间。

eventObservable.onNext(new MyEvent());
eventsScheduler.triggerActions();