StormAdminServiceImpl implements StormAdminService { @Override public TopologyResponse stopParserTopology(String name, boolean stopNow) throws RestException { TopologyStatus topologyStatus = stormStatusService.getTopologyStatus(name); String jobName = topologyStatus != null ? topologyStatus.getName() : name; return createResponse(stormCLIClientWrapper.stopParserTopology(jobName, stopNow), TopologyStatusCode.STOPPED, TopologyStatusCode.STOP_ERROR); } @Autowired  StormAdminServiceImpl(StormCLIWrapper stormCLIClientWrapper,
                                 GlobalConfigService globalConfigService,
                                 SensorParserConfigService sensorParserConfigService,
                                 SensorParserGroupService sensorParserGroupService,
                                 StormStatusService stormStatusService); @Override TopologyResponse startParserTopology(String name); @Override TopologyResponse stopParserTopology(String name, boolean stopNow); @Override TopologyResponse startEnrichmentTopology(); @Override TopologyResponse stopEnrichmentTopology(boolean stopNow); @Override TopologyResponse startIndexingTopology(String scriptPath); @Override TopologyResponse stopIndexingTopology(String name, boolean stopNow); @Override Map<String, String> getStormClientStatus();  }
@Test public void stopParserTopologyShouldProperlyReturnErrorTopologyResponse() throws Exception { TopologyStatus topologyStatus = new TopologyStatus(); topologyStatus.setName("bro"); when(stormCLIClientWrapper.stopParserTopology("bro", false)).thenReturn(1); when(stormStatusService.getTopologyStatus("bro")).thenReturn(topologyStatus); TopologyResponse expected = new TopologyResponse(); expected.setErrorMessage(TopologyStatusCode.STOP_ERROR.toString()); assertEquals(expected, stormAdminService.stopParserTopology("bro", false)); }