### Question:
ClusterManagerBucketLoader extends BaseBucketLoader { @Override protected Mono<byte[]> discoverConfig(final NodeIdentifier seed, final String bucket) { final CoreContext ctx = core().context(); return Mono.defer(() -> { BucketConfigRequest request = new BucketConfigRequest( ctx.environment().timeoutConfig().connectTimeout(), ctx, BestEffortRetryStrategy.INSTANCE, bucket, ctx.authenticator(), seed ); core().send(request); return Reactor.wrap(request, request.response(), true); }).map(response -> { if (response.status().success()) { return response.config(); } else { throw new ConfigException("Received error status from ClusterManagerBucketLoader: " + response); } }); } ClusterManagerBucketLoader(final Core core);   }### Answer:
@Test void loadsConfigSuccessfully() { byte[] expectedConfig = "config".getBytes(UTF_8); BucketConfigResponse response = mock(BucketConfigResponse.class); when(response.status()).thenReturn(ResponseStatus.SUCCESS); when(response.config()).thenReturn(expectedConfig); doAnswer(i -> { ((BucketConfigRequest) i.getArgument(0)).succeed(response); return null; }).when(core).send(any(BucketConfigRequest.class)); byte[] config = loader.discoverConfig(SEED, BUCKET).block(); assertArrayEquals(expectedConfig, config); }
@Test void errorsIfNonSuccessful() { BucketConfigResponse response = mock(BucketConfigResponse.class); when(response.status()).thenReturn(ResponseStatus.UNKNOWN); doAnswer(i -> { ((BucketConfigRequest) i.getArgument(0)).succeed(response); return null; }).when(core).send(any(BucketConfigRequest.class)); assertThrows(ConfigException.class, () -> loader.discoverConfig(SEED, BUCKET).block()); }
@Test void errorsIfFailedRequest() { doAnswer(i -> { ((BucketConfigRequest) i.getArgument(0)) .fail(new UnsupportedOperationException()); return null; }).when(core).send(any(BucketConfigRequest.class)); assertThrows( UnsupportedOperationException.class, () -> loader.discoverConfig(SEED, BUCKET).block() ); }
@Test void cancelRequestOnceUnsubscribed() { final AtomicReference<BucketConfigRequest> request = new AtomicReference<>(); doAnswer(i -> { request.set(i.getArgument(0)); return null; }).when(core).send(any(BucketConfigRequest.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()); }