### Question:
KeyValueBucketLoader extends BaseBucketLoader { @Override protected Mono<byte[]> discoverConfig(final NodeIdentifier seed, final String bucket) { final CoreContext ctx = core().context(); return Mono.defer(() -> { CarrierBucketConfigRequest request = new CarrierBucketConfigRequest( ctx.environment().timeoutConfig().connectTimeout(), ctx, new CollectionIdentifier(bucket, Optional.empty(), Optional.empty()), BestEffortRetryStrategy.INSTANCE, seed ); core().send(request); return Reactor.wrap(request, request.response(), true); }) .map(response -> { if (response.status().success()) { return response.content(); } else if (response.status() == ResponseStatus.UNSUPPORTED) { throw new UnsupportedConfigMechanismException(); } else { throw new ConfigException("Received error status from KeyValueBucketLoader: " + response); } }); } KeyValueBucketLoader(final Core core);   }### Answer:
@Test void loadsConfigSuccessfully() { byte[] expectedConfig = "config".getBytes(UTF_8); CarrierBucketConfigResponse response = mock(CarrierBucketConfigResponse.class); when(response.status()).thenReturn(ResponseStatus.SUCCESS); when(response.content()).thenReturn(expectedConfig); doAnswer(i -> { ((CarrierBucketConfigRequest) i.getArgument(0)).succeed(response); return null; }).when(core).send(any(CarrierBucketConfigRequest.class)); byte[] config = loader.discoverConfig(SEED, BUCKET).block(); assertArrayEquals(expectedConfig, config); }
@Test void errorsIfNonSuccessful() { CarrierBucketConfigResponse response = mock(CarrierBucketConfigResponse.class); when(response.status()).thenReturn(ResponseStatus.UNKNOWN); doAnswer(i -> { ((CarrierBucketConfigRequest) i.getArgument(0)).succeed(response); return null; }).when(core).send(any(CarrierBucketConfigRequest.class)); assertThrows(ConfigException.class, () -> loader.discoverConfig(SEED, BUCKET).block()); }
@Test void errorsIfUnsupported() { CarrierBucketConfigResponse response = mock(CarrierBucketConfigResponse.class); when(response.status()).thenReturn(ResponseStatus.UNSUPPORTED); doAnswer(i -> { ((CarrierBucketConfigRequest) i.getArgument(0)).succeed(response); return null; }).when(core).send(any(CarrierBucketConfigRequest.class)); assertThrows( UnsupportedConfigMechanismException.class, () -> loader.discoverConfig(SEED, BUCKET).block() ); }
@Test void errorsIfFailedRequest() { doAnswer(i -> { ((CarrierBucketConfigRequest) i.getArgument(0)) .fail(new UnsupportedOperationException()); return null; }).when(core).send(any(CarrierBucketConfigRequest.class)); assertThrows( UnsupportedOperationException.class, () -> loader.discoverConfig(SEED, BUCKET).block() ); }
@Test void cancelRequestOnceUnsubscribed() { final AtomicReference<CarrierBucketConfigRequest> request = new AtomicReference<>(); doAnswer(i -> { request.set(i.getArgument(0)); return null; }).when(core).send(any(CarrierBucketConfigRequest.class)); Disposable disposable = loader.discoverConfig(SEED, BUCKET).subscribe(); disposable.dispose(); assertTrue(request.get().completed()); assertTrue(request.get().cancelled()); assertEquals(CancellationReason.STOPPED_LISTENING, request.get().cancellationReason()); }