SPQRNodeClient { public MicroPipelineShutdownResponse shutdown(final String pipelineId) throws RemoteClientConnectionFailedException { if(StringUtils.isBlank(pipelineId)) return new MicroPipelineShutdownResponse(pipelineId, MicroPipelineShutdownState.PIPELINE_ID_MISSING, "Missing required pipeline identifier"); StringBuffer url = new StringBuffer(this.processingNodeServiceBaseUrl).append("/pipelines/").append(pipelineId); if(logger.isDebugEnabled()) logger.debug("Deleting pipeline [id="+pipelineId+"] on processing node " + url.toString()); try { final WebTarget webTarget = this.restClient.target(url.toString()); return webTarget.request(MediaType.APPLICATION_JSON).accept(MediaType.APPLICATION_JSON).delete(MicroPipelineShutdownResponse.class); } catch(Exception e) { throw new RemoteClientConnectionFailedException("Failed to establish a connection with the remote resource manager [url="+url.toString()+"]. Error: " + e.getMessage()); } } SPQRNodeClient(final String protocol, final String remoteHost, final int servicePort, final int adminPort, final Client client); MicroPipelineInstantiationResponse instantiatePipeline(final MicroPipelineConfiguration pipelineConfiguration); MicroPipelineInstantiationResponse updatePipeline(final MicroPipelineConfiguration pipelineConfiguration); MicroPipelineShutdownResponse shutdown(final String pipelineId); void shutdown();  }
@Test public void testShutdownPipeline_withNullInput() throws Exception { Client clientMock = Mockito.mock(Client.class); String processingNodeHost = "processing-node-localhost"; int processingNodeServicePort = 7070; int processingNodeAdminPort = 7071; SPQRNodeClient client = new SPQRNodeClient("http", processingNodeHost, processingNodeServicePort, processingNodeAdminPort, clientMock); MicroPipelineShutdownResponse response = client.shutdown(null); Assert.assertEquals("Invalid input", MicroPipelineShutdownState.PIPELINE_ID_MISSING, response.getState()); }
@Test public void testShutdownPipeline_withValidInput() throws Exception { String processingNodeHost = "processing-node-localhost"; int processingNodeServicePort = 7070; int processingNodeAdminPort = 7071; String pipelineId = "test-pipeline-id"; MicroPipelineShutdownResponse pipelineShutdownResponse = new MicroPipelineShutdownResponse(pipelineId, MicroPipelineShutdownState.OK, ""); String url = new StringBuffer("http: Client clientMock = Mockito.mock(Client.class); WebTarget webTargetMock = Mockito.mock(WebTarget.class); Builder b1Mock = Mockito.mock(Builder.class); Builder b2Mock = Mockito.mock(Builder.class); Mockito.when(clientMock.target(url)).thenReturn(webTargetMock); Mockito.when(webTargetMock.request(MediaType.APPLICATION_JSON)).thenReturn(b1Mock); Mockito.when(b1Mock.accept(MediaType.APPLICATION_JSON)).thenReturn(b2Mock); Mockito.when(b2Mock.delete(Mockito.eq(MicroPipelineShutdownResponse.class))).thenReturn(pipelineShutdownResponse); SPQRNodeClient client = new SPQRNodeClient("http", processingNodeHost, processingNodeServicePort, processingNodeAdminPort, clientMock); MicroPipelineShutdownResponse response = client.shutdown("test-pipeline-id"); Assert.assertNotNull("Response must not be null", response); Assert.assertEquals("State must be 'OK'", MicroPipelineShutdownState.OK, response.getState()); Mockito.verify(clientMock).target(url); Mockito.verify(webTargetMock).request(MediaType.APPLICATION_JSON); Mockito.verify(b1Mock).accept(MediaType.APPLICATION_JSON); Mockito.verify(b2Mock).delete(Mockito.eq(MicroPipelineShutdownResponse.class)); }