io.reactivex.observers.TestObserver#awaitTerminalEvent ( )源码实例Demo

@Test
public void testFindById() throws TechnicalException {
    // create user
    User user = new User();
    user.setReferenceType(ReferenceType.DOMAIN);
    user.setReferenceId("domainId");
    user.setUsername("testsUsername");
    User userCreated = userRepository.create(user).blockingGet();

    // fetch user
    TestObserver<User> testObserver = userRepository.findById(userCreated.getId()).test();
    testObserver.awaitTerminalEvent();

    testObserver.assertComplete();
    testObserver.assertNoErrors();
    testObserver.assertValue(u -> u.getUsername().equals("testsUsername"));
}
 

@Test
public void testUpdate() throws TechnicalException {
    // create extension grant
    ExtensionGrant extensionGrant = new ExtensionGrant();
    extensionGrant.setName("testName");
    ExtensionGrant extensionGrantCreated = extensionGrantRepository.create(extensionGrant).blockingGet();

    // update extension grant
    ExtensionGrant updatedExtension = new ExtensionGrant();
    updatedExtension.setId(extensionGrantCreated.getId());
    updatedExtension.setName("testUpdatedName");

    TestObserver<ExtensionGrant> testObserver = extensionGrantRepository.update(updatedExtension).test();
    testObserver.awaitTerminalEvent();

    testObserver.assertComplete();
    testObserver.assertNoErrors();
    testObserver.assertValue(e -> e.getName().equals(updatedExtension.getName()));
}
 

@Test
public void testFindByTimeFrame() throws TechnicalException {
    final long from = 1571214259000l;
    final long to =  1571214281000l;
    // create event
    Event event = new Event();
    event.setType(Type.DOMAIN);
    event.setCreatedAt(new Date(from));
    event.setUpdatedAt(event.getCreatedAt());
    eventRepository.create(event).blockingGet();

    // fetch events
    TestObserver<List<Event>> testObserver1 = eventRepository.findByTimeFrame(from, to).test();
    testObserver1.awaitTerminalEvent();

    testObserver1.assertComplete();
    testObserver1.assertNoErrors();
    testObserver1.assertValue(events -> events.size() == 1);
}
 

@Test
public void oneToOne() {
    RxGreeterGrpc.RxGreeterStub stub = RxGreeterGrpc.newRxStub(channel);
    Single<HelloRequest> req = Single.just(HelloRequest.newBuilder().setName("rxjava").build());
    Single<HelloResponse> resp = req.compose(stub::sayHello);

    AtomicReference<String> clientThreadName = new AtomicReference<>();

    TestObserver<String> testObserver = resp
            .map(HelloResponse::getMessage)
            .doOnSuccess(x -> clientThreadName.set(Thread.currentThread().getName()))
            .test();
    testObserver.awaitTerminalEvent(3, TimeUnit.SECONDS);

    assertThat(clientThreadName.get()).isEqualTo("TheGrpcClient");
    assertThat(serverThreadName.get()).isEqualTo("TheGrpcServer");
}
 

@Test
public void shouldNotSearchForAUser_clientCredentials() {
    final String token = "token";
    AccessToken accessToken = new AccessToken(token);
    accessToken.setSubject("client-id");
    accessToken.setClientId("client-id");
    when(tokenService.introspect("token")).thenReturn(Single.just(accessToken));

    IntrospectionRequest introspectionRequest = new IntrospectionRequest(token);
    TestObserver<IntrospectionResponse> testObserver = introspectionService.introspect(introspectionRequest).test();

    testObserver.awaitTerminalEvent();
    testObserver.assertComplete();
    testObserver.assertNoErrors();
    verify(userService, never()).findById(anyString());
}
 

@Test
public void testImportStore() {
    TestObserver<Wallet> subscriber = accountKeystoreService
            .importKeystore(STORE_1, PASS_1)
            .toObservable()
            .test();
    subscriber.awaitTerminalEvent();
    subscriber.assertComplete();
    subscriber.assertNoErrors();

    subscriber.assertOf(accountTestObserver -> {
        assertEquals(accountTestObserver.valueCount(), 1);
        assertEquals(accountTestObserver.values().get(0).getAddress(), ADDRESS_1);
        assertTrue(accountTestObserver.values().get(0).sameAddress(ADDRESS_1));
    });
    deleteAccountStore(ADDRESS_1, PASS_1);
}
 

@Test
public void shouldCreateDefault() {

    Organization defaultOrganization = new Organization();
    defaultOrganization.setId("DEFAULT");

    when(organizationRepository.count()).thenReturn(Single.just(0L));
    when(organizationRepository.create(argThat(organization -> organization.getId().equals(Organization.DEFAULT)))).thenReturn(Single.just(defaultOrganization));
    when(roleService.createDefaultRoles("DEFAULT")).thenReturn(Completable.complete());
    when(entrypointService.createDefault("DEFAULT")).thenReturn(Single.just(new Entrypoint()));

    TestObserver<Organization> obs = cut.createDefault().test();

    obs.awaitTerminalEvent();
    obs.assertValue(defaultOrganization);

    verify(auditService, times(1)).report(argThat(builder -> {
        Audit audit = builder.build(new ObjectMapper());
        assertEquals(ReferenceType.ORGANIZATION, audit.getReferenceType());
        assertEquals(defaultOrganization.getId(), audit.getReferenceId());
        assertEquals("system", audit.getActor().getId());

        return true;
    }));
}
 

@Test
public void shouldFindAll() {
    when(emailRepository.findAll(ReferenceType.DOMAIN, DOMAIN)).thenReturn(Single.just(Collections.singletonList(new Email())));
    TestObserver testObserver = emailTemplateService.findAll(ReferenceType.DOMAIN, DOMAIN).test();

    testObserver.awaitTerminalEvent();
    testObserver.assertComplete();
    testObserver.assertNoErrors();
    testObserver.assertValueCount(1);
}
 

@Test
public void delete() throws TechnicalException {
    // create resource_set, resource_scopes being the most important field.
    Resource resource = new Resource().setResourceScopes(Arrays.asList("a","b","c"));
    Resource rsCreated = repository.create(resource).blockingGet();

    // fetch resource_set
    TestObserver<Void> testObserver = repository.delete(rsCreated.getId()).test();
    testObserver.awaitTerminalEvent();

    testObserver.assertComplete();
    testObserver.assertNoErrors();
    testObserver.assertNoValues();
}
 

@Test
public void shouldNotCreate_memberNotFound() {

    Membership membership = new Membership();
    membership.setReferenceId(DOMAIN_ID);
    membership.setReferenceType(ReferenceType.DOMAIN);
    membership.setMemberId("user-id");
    membership.setMemberType(MemberType.USER);
    membership.setRoleId("role-id");

    Role role = new Role();
    role.setId("role-id");
    role.setReferenceId(DOMAIN_ID);
    role.setReferenceType(ReferenceType.DOMAIN);
    role.setAssignableType(ReferenceType.DOMAIN);

    when(userService.findById(ReferenceType.ORGANIZATION, ORGANIZATION_ID, membership.getMemberId())).thenReturn(Single.error(new UserNotFoundException("user-id")));
    when(roleService.findById(role.getId())).thenReturn(Maybe.just(role));
    when(membershipRepository.findByReferenceAndMember(membership.getReferenceType(), membership.getReferenceId(), membership.getMemberType(), membership.getMemberId())).thenReturn(Maybe.empty());

    TestObserver testObserver = membershipService.addOrUpdate(ORGANIZATION_ID, membership).test();
    testObserver.awaitTerminalEvent();

    testObserver.assertNotComplete();
    testObserver.assertError(UserNotFoundException.class);

    verify(membershipRepository, never()).create(any());
}
 

@Test
public void testCreate() throws TechnicalException {
    Application application = new Application();
    application.setName("testClientId");

    TestObserver<Application> testObserver = applicationRepository.create(application).test();
    testObserver.awaitTerminalEvent();

    testObserver.assertComplete();
    testObserver.assertNoErrors();
    testObserver.assertValue(a -> a.getName().equals(application.getName()));
}
 

@Test
public void shouldFindByExtensionGrant() {
    when(applicationRepository.findByDomainAndExtensionGrant(DOMAIN, "client-extension-grant")).thenReturn(Single.just(Collections.singleton(new Application())));
    TestObserver<Set<Application>> testObserver = applicationService.findByDomainAndExtensionGrant(DOMAIN, "client-extension-grant").test();
    testObserver.awaitTerminalEvent();

    testObserver.assertComplete();
    testObserver.assertNoErrors();
    testObserver.assertValue(extensionGrants -> extensionGrants.size() == 1);
}
 

@Test
public void shouldThrowExceptionIfNoTypeMappingWasFoundWithoutAccessingDbWithRawQueryAsSingle() {
    final StorIOSQLite storIOSQLite = mock(StorIOSQLite.class);
    final StorIOSQLite.LowLevel lowLevel = mock(StorIOSQLite.LowLevel.class);

    when(storIOSQLite.get()).thenReturn(new PreparedGet.Builder(storIOSQLite));
    when(storIOSQLite.lowLevel()).thenReturn(lowLevel);

    final TestObserver<Optional<TestItem>> testObserver = new TestObserver<Optional<TestItem>>();

    storIOSQLite
            .get()
            .object(TestItem.class)
            .withQuery(RawQuery.builder().query("test query").build())
            .prepare()
            .asRxSingle()
            .subscribe(testObserver);

    testObserver.awaitTerminalEvent();
    testObserver.assertNoValues();
    Throwable error = testObserver.errors().get(0);

    assertThat(error)
            .isInstanceOf(StorIOException.class)
            .hasCauseInstanceOf(IllegalStateException.class)
            .hasMessage("Error has occurred during Get operation. query = RawQuery{query='test query', args=[], affectsTables=[], affectsTags=[], observesTables=[], observesTags=[]}");

    assertThat(error.getCause())
            .hasMessage("This type does not have type mapping: "
                    + "type = " + TestItem.class + "," +
                    "db was not touched by this operation, please add type mapping for this type");

    verify(storIOSQLite).get();
    verify(storIOSQLite).lowLevel();
    verify(storIOSQLite).defaultRxScheduler();
    verify(storIOSQLite).interceptors();
    verify(lowLevel).typeMapping(TestItem.class);
    verify(lowLevel, never()).rawQuery(any(RawQuery.class));
    verifyNoMoreInteractions(storIOSQLite, lowLevel);
}
 

@Test
public void shouldPatch_mobileApplication() {
    Application client = new Application();
    client.setDomain(DOMAIN);

    PatchApplication patchClient = new PatchApplication();
    PatchApplicationSettings patchApplicationSettings = new PatchApplicationSettings();
    PatchApplicationOAuthSettings patchApplicationOAuthSettings = new PatchApplicationOAuthSettings();
    patchApplicationOAuthSettings.setGrantTypes(Optional.of(Arrays.asList("authorization_code")));
    patchApplicationOAuthSettings.setRedirectUris(Optional.of(Arrays.asList("com.gravitee.app://callback")));
    patchApplicationSettings.setOauth(Optional.of(patchApplicationOAuthSettings));
    patchClient.setSettings(Optional.of(patchApplicationSettings));

    when(applicationRepository.findById("my-client")).thenReturn(Maybe.just(client));
    when(applicationRepository.update(any(Application.class))).thenReturn(Single.just(new Application()));
    when(domainService.findById(DOMAIN)).thenReturn(Maybe.just(new Domain()));
    when(eventService.create(any())).thenReturn(Single.just(new Event()));
    when(scopeService.validateScope(DOMAIN,null)).thenReturn(Single.just(true));

    TestObserver testObserver = applicationService.patch(DOMAIN, "my-client", patchClient).test();
    testObserver.awaitTerminalEvent();

    testObserver.assertComplete();
    testObserver.assertNoErrors();

    verify(applicationRepository, times(1)).findById(anyString());
    verify(applicationRepository, times(1)).update(any(Application.class));
}
 

@Test
public void shouldFindByDomain() {
    when(extensionGrantRepository.findByDomain(DOMAIN)).thenReturn(Single.just(Collections.singleton(new ExtensionGrant())));
    TestObserver<List<ExtensionGrant>> testObserver = extensionGrantService.findByDomain(DOMAIN).test();
    testObserver.awaitTerminalEvent();

    testObserver.assertComplete();
    testObserver.assertNoErrors();
    testObserver.assertValue(extensionGrants -> extensionGrants.size() == 1);
}
 

@Test
public void testCreate() {
    Scope scope = new Scope();
    scope.setName("testName");
    scope.setSystem(true);
    scope.setClaims(Collections.emptyList());
    TestObserver<Scope> testObserver = scopeRepository.create(scope).test();
    testObserver.awaitTerminalEvent();

    testObserver.assertComplete();
    testObserver.assertNoErrors();
    testObserver.assertValue(s -> s.getName().equals(scope.getName()));
}
 

@Test
public void shouldDelete_technicalException() {
    when(applicationService.delete("my-client", null)).thenReturn(Completable.error(TechnicalManagementException::new));

    TestObserver testObserver = clientService.delete("my-client").test();
    testObserver.awaitTerminalEvent();

    testObserver.assertError(TechnicalManagementException.class);
    testObserver.assertNotComplete();
}
 

@Test
public void shouldCreate() {

    DefaultUser user = new DefaultUser("test");
    user.setId(USER_ID);

    NewEntrypoint newEntrypoint = new NewEntrypoint();
    newEntrypoint.setName("name");
    newEntrypoint.setDescription("description");
    newEntrypoint.setTags(Arrays.asList("tag#1", "tags#2"));
    newEntrypoint.setUrl("https://auth.company.com");

    when(entrypointRepository.create(any(Entrypoint.class))).thenAnswer(i -> Single.just(i.getArgument(0)));

    TestObserver<Entrypoint> obs = cut.create(ORGANIZATION_ID, newEntrypoint, user).test();

    obs.awaitTerminalEvent();
    obs.assertValue(entrypoint -> entrypoint.getId() != null
            && !entrypoint.isDefaultEntrypoint()
            && entrypoint.getOrganizationId().equals(ORGANIZATION_ID)
            && entrypoint.getName().equals(newEntrypoint.getName())
            && entrypoint.getDescription().equals(newEntrypoint.getDescription())
            && entrypoint.getTags().equals(newEntrypoint.getTags())
            && entrypoint.getUrl().equals(newEntrypoint.getUrl()));

    verify(auditService, times(1)).report(argThat(builder -> {
        Audit audit = builder.build(new ObjectMapper());
        assertEquals(EventType.ENTRYPOINT_CREATED, audit.getType());
        assertEquals(ReferenceType.ORGANIZATION, audit.getReferenceType());
        assertEquals(ORGANIZATION_ID, audit.getReferenceId());
        assertEquals(user.getId(), audit.getActor().getId());

        return true;
    }));
}
 

@Test
public void serverPublishShouldWork() {
    RxGreeterGrpc.GreeterImplBase svc = new RxGreeterGrpc.GreeterImplBase() {
        @Override
        public Single<HelloResponse> sayHelloReqStream(Flowable<HelloRequest> rxRequest) {
            return rxRequest
                // a function that can use the multicasted source sequence as many times as needed, without causing
                // multiple subscriptions to the source sequence. Subscribers to the given source will receive all
                // notifications of the source from the time of the subscription forward.
                .publish(shared -> {
                    Single<HelloRequest> first = shared.firstOrError();
                    Flowable<HelloRequest> rest = shared.skip(0);
                    return first
                        .flatMap(firstVal -> rest
                            .map(HelloRequest::getName)
                            .toList()
                            .map(names -> {
                                        ArrayList<String> strings = Lists.newArrayList(firstVal.getName());
                                        strings.addAll(names);
                                        Thread.sleep(1000);
                                        return HelloResponse.newBuilder().setMessage("Hello " + String.join(" and ", strings)).build();
                                    }
                            ).doOnError(System.out::println))
                        .toFlowable();
                })
                .singleOrError();
        }
    };

    serverRule.getServiceRegistry().addService(svc);
    RxGreeterGrpc.RxGreeterStub stub = RxGreeterGrpc.newRxStub(serverRule.getChannel());

    TestObserver<String> resp = Flowable.just("Alpha", "Bravo", "Charlie")
            .map(s -> HelloRequest.newBuilder().setName(s).build())
            .as(stub::sayHelloReqStream)
            .map(HelloResponse::getMessage)
            .test();

    resp.awaitTerminalEvent(5, TimeUnit.SECONDS);
    resp.assertComplete();
    resp.assertValue("Hello Alpha and Bravo and Charlie");
}
 

@Test
public void shouldFindById_technicalException() {

    when(organizationRepository.findById(ORGANIZATION_ID)).thenReturn(Maybe.error(TechnicalException::new));

    TestObserver<Organization> obs = cut.findById(ORGANIZATION_ID).test();

    obs.awaitTerminalEvent();
    obs.assertError(TechnicalException.class);
}