ParseSessionDataSource implements SessionDataSource { @NonNull @Override public Observable<BaseUser> register(@NonNull SessionQuery.Register query) { final Map<String, Object> additionalParams = new HashMap<>(); additionalParams.put(UsersTable.Fields.DISPLAY_NAME, query.getDisplayName()); return mParseHelper.signUp(query.getUserName(), query.getPassword(), additionalParams) .map(ParseUtils::fromParseUser) .map(user -> ((BaseUser) user)) .subscribeOn(Schedulers.io()) .doOnCompleted(() -> mBroadcaster.userSignedIn()); } ParseSessionDataSource(@NonNull Broadcaster broadcaster, @NonNull ParseHelper parseHelper); @NonNull @Override Observable<BaseUser> signIn(@NonNull SessionQuery.SignIn query); @Override Observable<Void> signOut(); @NonNull @Override Observable<BaseUser> register(@NonNull SessionQuery.Register query);  }
@Test public void register() { ParseUser currentUser = ModelTestHelper.createParseUser(USER_ID); when(mParseHelper.signUp(eq(USER_NAME), eq(PASSWORD), any())).thenReturn(Observable.just(currentUser)); TestSubscriber<BaseUser> testSubscriber = executeTarget(mTarget.register(new SessionQuery.Register(USER_NAME, DISPLAY_NAME, PASSWORD))); assertThat(testSubscriber.getOnNextEvents().size(), is(1)); BaseUser user = testSubscriber.getOnNextEvents().get(0); assertEquals(USER_ID, user.getUserId()); testSubscriber.assertCompleted(); verify(mBroadcaster).userSignedIn(); }