KubernetesLander implements Lander { @Override public CompletableFuture<LanderConfiguration> run() { client.pods().withName(podName).delete(); client.pods().create(pod); PodResource<Pod, DoneablePod> resource = client.pods().withName(podName); PodWatcher podWatcher = new PodWatcher(resource, podName); return podWatcher.get().thenApply(c -> config); }  @Override CompletableFuture<LanderConfiguration> run();  }
@Test public void successfulPodCompletion() throws Exception { when(podNameFactory.newName(any(LanderConfiguration.class))).thenReturn("pod-name"); when(podResource.watch(podWatcherCaptor.capture())).thenReturn(watch); when(client.pods()).thenReturn(mixedOperation); when(mixedOperation.withName(anyString())).thenReturn(podResource); Pod pod = new Pod(); when(podResource.get()).thenReturn(pod); KubernetesLander.Factory factory = new KubernetesLander.Factory(argsFactory, podFactory, podNameFactory, client); LanderConfiguration config = new LanderConfiguration("road", "topic", ImmutableMap.of(), "s3Prefix", false, "partitionColumnValue", false); when(argsFactory.newInstance(config)).thenReturn(ARGS); when(podFactory.newInstance(config, ARGS)).thenReturn(pod); Lander lander = factory.newInstance(config); CompletableFuture<LanderConfiguration> future = lander.run(); verify(podResource).delete(); verify(mixedOperation).create(pod); pod.setStatus(terminatedPostStatus(0)); podWatcherCaptor.getValue().eventReceived(null, pod); assertThat(config, is(future.get())); }
@Test public void nonZeroExitCode() throws Exception { when(podNameFactory.newName(any(LanderConfiguration.class))).thenReturn("pod-name"); when(podResource.watch(podWatcherCaptor.capture())).thenReturn(watch); when(client.pods()).thenReturn(mixedOperation); when(mixedOperation.withName(anyString())).thenReturn(podResource); Pod pod = new Pod(); when(podResource.get()).thenReturn(pod); KubernetesLander.Factory factory = new KubernetesLander.Factory(argsFactory, podFactory, podNameFactory, client); LanderConfiguration config = new LanderConfiguration("road", "topic", ImmutableMap.of(), "s3Prefix", false, "partitionColumnValue", false); when(argsFactory.newInstance(config)).thenReturn(ARGS); when(podFactory.newInstance(config, ARGS)).thenReturn(pod); Lander lander = factory.newInstance(config); CompletableFuture<LanderConfiguration> run = lander.run(); verify(podResource).delete(); verify(mixedOperation).create(pod); pod.setStatus(terminatedPostStatus(1)); podWatcherCaptor.getValue().eventReceived(null, pod); assertThat(run.isCompletedExceptionally(), is(true)); }