java - Unit testing clients of Observables -

i have following method go() i'd test:

private pair<string, string> mpair;  public void go() {     observable.zip(             mapi.webcall(),             mapi.webcall2(),             new func2<string, string, pair<string, string>>() {                 @override                 public pair<string, string> call(string s, string s2) {                     return new pair(s, s2);                 }             }     )             .subscribeon(schedulers.io())             .observeon(androidschedulers.mainthread())             .subscribe(new action1<pair<string, string>>() {                 @override                 public void call(pair<string, string> pair) {                     mapi.webcall3(pair.first, pair.second);                 }             }); } 

this method uses observable.zip() execute http requests asynchronously, , merge them in 1 pair. in end, http request executed result of these previous requests.

i'd verify calling go() method makes webcall() , webcall2() requests, followed webcall3(string, string) request. therefore, i'd following test pass (using mockito spy api):

@test public void testgo() {     /* given */     api api = spy(new api() {         @override         public observable<string> webcall() {             return observable.just("first");         }          @override         public observable<string> webcall2() {             return observable.just("second");         }          @override         public void webcall3() {         }     });      test test = new test(api);      /* when */     test.go();      /* */     verify(api).webcall();     verify(api).webcall2();     verify(api).webcall3("first", "second"); } 

however when running this, web calls executed asynchronously, , test executes assertion before subscriber done causing test fail.

i have read can use rxjavaschedulershook , rxandroidschedulershook return schedulers.immediate() methods, results in test running indefinitely.

i running unit tests on local jvm.

how can achieve this, preferably without having modify signature of go()?

(lambdas retrolambda)

for starters, rephrase go as:

private pair<string, string> mpair;  public observable<pair<string, string>> go() {     return observable.zip(                 mapi.webcall(),                 mapi.webcall2(),                 (string s, string s2) -> new pair(s, s2)             .subscribeon(schedulers.io())             .observeon(androidschedulers.mainthread())             .doonnext(pair -> mpair = pair); }  public pair<string, string> getpair() {     return mpair; } 

doonnext allows intercept value being processed in chain whenever subscribe observable

then, call test that:

pair result = test.go().toblocking().lastordefault(null); 

then can test result is.