+ @Test
+ public void successfulPost() throws IOException {
+ doReturn(okResponse()).when(agentClient).postForEntity(anyString(), anyString());
+ doReturn(1).when(dmaapClient).send(anyString());
+ doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
+
+ StepVerifier //
+ .create(testedObject.createTask(dmaapInputMessage(Operation.POST))) //
+ .expectSubscription() //
+ .expectNext("OK") //
+ .verifyComplete(); //
+
+ verify(agentClient).postForEntity(URL, payloadAsString());
+ verifyNoMoreInteractions(agentClient);
+
+ verify(dmaapClient).send(anyString());
+ verify(dmaapClient).sendBatchWithResponse();
+ verifyNoMoreInteractions(dmaapClient);
+ }
+
+ @Test
+ public void exceptionWhenCallingPolicyAgent_thenNotFoundResponse() throws IOException {
+ WebClientResponseException except = new WebClientResponseException(400, "Refused", null, null, null, null);
+ doReturn(Mono.error(except)).when(agentClient).putForEntity(anyString(), any());
+ doReturn(1).when(dmaapClient).send(anyString());
+ doReturn(new MRPublisherResponse()).when(dmaapClient).sendBatchWithResponse();
+
+ StepVerifier //
+ .create(testedObject.createTask(dmaapInputMessage(Operation.PUT))) //
+ .expectSubscription() //
+ .verifyComplete(); //
+
+ verify(agentClient).putForEntity(anyString(), anyString());
+ verifyNoMoreInteractions(agentClient);
+
+ ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
+ verify(dmaapClient).send(captor.capture());
+ String actualMessage = captor.getValue();
+ assertThat(actualMessage.contains(HttpStatus.BAD_REQUEST.toString())).isTrue();
+
+ verify(dmaapClient).sendBatchWithResponse();
+ verifyNoMoreInteractions(dmaapClient);
+ }
+
+ @Test
+ public void unsupportedOperationInMessage_thenNotFoundResponseWithNotImplementedOperation() throws Exception {
+ String message = dmaapInputMessage(Operation.PUT).toString();
+ String badOperation = "BAD";
+ message = message.replace(Operation.PUT.toString(), badOperation);
+
+ testedObject.handleDmaapMsg(message);
+
+ ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
+ verify(dmaapClient).send(captor.capture());
+ String actualMessage = captor.getValue();
+ assertThat(actualMessage.contains(HttpStatus.NOT_FOUND + "\",\"message\":\"Not implemented operation:"))
+ .isTrue();
+
+ verify(dmaapClient).sendBatchWithResponse();
+ verifyNoMoreInteractions(dmaapClient);
+ }
+
+ @Test
+ public void putWithoutPayload_thenNotFoundResponseWithWarning() throws Exception {
+ String message = dmaapInputMessage(Operation.PUT).toString();
+ message = message.replace(",\"payload\":{\"name\":\"name\",\"schema\":\"schema\"}", "");
+
+ final ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(DmaapMessageHandler.class);
+
+ testedObject.handleDmaapMsg(message);
+
+ assertThat(logAppender.list.get(0).getLevel()).isEqualTo(Level.WARN);
+ assertThat(logAppender.list.toString().contains("Expected payload in message from DMAAP: ")).isTrue();
+ }