### Question:
LocalCLusterService { public void launch(String topologyId) throws TopologyLaunchException { PythianToplogyConfiguration configuration = topologyRepository.findById(topologyId); if (configuration == null) { throw new TopologyLaunchException("No topology found with id " + topologyId); } ValidationResult validationResult = configuration.validate(); if (validationResult.hasErrors()) { throw new TopologyLaunchException("Topology is not valid \n" + validationResult.print()); } StormTopology stormTopology; Config tridentConfig; try { PythianTopology pythianTopology = new PythianTopology(); pythianTopology.build(configuration); stormTopology = pythianTopology.getStormTopology(); tridentConfig = pythianTopology.getTridentConfig(); } catch (Exception ex) { throw new TopologyLaunchException("Topology could not be builded", ex); } try { cluster.submitTopology(topologyId, tridentConfig, stormTopology); } catch (Exception ex) { throw new TopologyLaunchException("Topology could not be launched", ex); } }  void launch(String topologyId); void kill(String topologyId); List<TopologyState> getTopologyStates(); TopologyState getTopologyState(String topologyId);  }### Answer:
@SuppressWarnings("unchecked") @Test public void should_launch_topology() throws Exception { String topologyId = random(6); PythianToplogyConfiguration configuration = mock(PythianToplogyConfiguration.class); when(configuration.getComponents()).thenReturn(EMPTY_LIST); when(configuration.getConnections()).thenReturn(EMPTY_LIST); when(configuration.validate()).thenReturn(new ValidationResult("")); when(topologyRepository.findById(topologyId)).thenReturn(configuration); service.launch(topologyId); verify(cluster).submitTopology(isA(String.class), isA(Config.class), isA(StormTopology.class)); }
@Test(expected = TopologyLaunchException.class) public void should_not_launch_not_existing_topology() throws TopologyLaunchException { String topologyId = random(6); given(topologyRepository.findById(topologyId)).willReturn(null); service.launch(topologyId); }
@SuppressWarnings("unchecked") @Test(expected = TopologyLaunchException.class) public void should_throw_launch_exception_when_storm_submit_fails() throws TopologyLaunchException { String topologyId = random(6); PythianToplogyConfiguration configuration = mock(PythianToplogyConfiguration.class); when(configuration.getComponents()).thenReturn(EMPTY_LIST); when(configuration.getConnections()).thenReturn(EMPTY_LIST); when(configuration.validate()).thenReturn(new ValidationResult("")); given(topologyRepository.findById(topologyId)).willReturn(configuration); doThrow(new RuntimeException()).when(cluster).submitTopology(Mockito.anyString(), Mockito.any(Config.class), Mockito.any(StormTopology.class)); service.launch(topologyId); }
@SuppressWarnings("unchecked") @Test(expected = TopologyLaunchException.class) public void should_throw_launch_exception_when_topology_is_not_valid() throws TopologyLaunchException { String topologyId = random(6); PythianToplogyConfiguration configuration = mock(PythianToplogyConfiguration.class); when(configuration.getComponents()).thenReturn(EMPTY_LIST); when(configuration.getConnections()).thenReturn(EMPTY_LIST); when(configuration.validate()).thenReturn(validationResult("Topology").with("Some error").build()); given(topologyRepository.findById(topologyId)).willReturn(configuration); service.launch(topologyId); }