### Question:
Scheduler { void runExternalWorkflows(Workflow wf, List<SlotState> slotStates, StateDatabaseConnection connection) throws Exception { List<SlotState> scheduledSlots = wf.getSchedulingStrategy().getSchedulingCandidates(slotStates); for (SlotState slotState : scheduledSlots) { if (!slotState.getStatus().equals(SlotState.Status.READY)) { throw new IllegalStateException("Scheduling strategy returned non-ready slot: " + slotState); } SlotID slotID = slotState.getSlotID(); LOGGER.info("Submitting slot to external service: " + slotID); String externalID = wf.getExternalService().submit(slotID); connection.putSlotState(slotState.transitionToRunning(externalID)); LOGGER.info("Starting slot: " + slotID + " with external ID: " + externalID); wf.getExternalService().start(slotID, externalID); } } Scheduler(WorkflowConfiguration configuration, int slidingWindowHours); void step(ScheduledTime current, StateDatabaseConnection connection); void step(ScheduledTime current, Set<WorkflowID> workflowIDs, StateDatabaseConnection connection); List<SlotState> getSlotStatesIncludingMarkedForRerun(Workflow wf, ScheduledTime current, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); List<SlotState> getSlotStates(Workflow wf, ScheduledTime start, ScheduledTime end, StateDatabaseConnection connection); ScheduledTime getWorkflowStartTime(Workflow wf, ScheduledTime current); int getSlidingWindowHours(); WorkflowConfiguration getWorkflowConfiguration();  }### Answer:
@Test public void runExternalWorkflowsNoCandidates() throws Exception { List<SlotState> slotStates = new ArrayList<SlotState>(); scheduler.runExternalWorkflows(wf, slotStates, mockedConnection); verifyNoMoreInteractions(mockedConnection); }
@Test public void runExternalWorkflowsReadyCandidate() throws Exception { List<SlotState> slotStates = candidate(SlotState.Status.READY); SlotState nextSlotState = slotStates.get(0).transitionToRunning("externalId"); when(externalService.submit(slotId)).thenReturn("externalId"); scheduler.runExternalWorkflows(wf, slotStates, mockedConnection); verify(mockedConnection).putSlotState(nextSlotState); verifyNoMoreInteractions(mockedConnection); }
@Test public void runExternalWorkflowsCallsSchedulerCorrectly() throws Exception { List<SlotState> slotStates = candidate(SlotState.Status.READY); when(externalService.submit(slotId)).thenReturn("externalId"); scheduler.runExternalWorkflows(wf, slotStates, mockedConnection); InOrder inOrder = inOrder(externalService); inOrder.verify(externalService).submit(slotStates.get(0).getSlotID()); inOrder.verify(externalService).start(slotId, "externalId"); verifyNoMoreInteractions(externalService); }
@Test public void runExternalWorkflowsMultipleReadyCandidates() throws Exception { SlotState slotState1 = makeReadySlotStateForTime("2013-11-26T15:01Z"); SlotState slotState2 = makeReadySlotStateForTime("2013-11-26T15:02Z"); List<SlotState> slotStates = new ArrayList<SlotState>(); slotStates.add(slotState1); slotStates.add(slotState2); stubAsSchedulingCandidates(slotStates); SlotState nextSlotState1 = slotState1.transitionToRunning("externalId1"); SlotState nextSlotState2 = slotState2.transitionToRunning("externalId2"); when(externalService.submit(slotState1.getSlotID())).thenReturn("externalId1"); when(externalService.submit(slotState2.getSlotID())).thenReturn("externalId2"); scheduler.runExternalWorkflows(wf, slotStates, mockedConnection); verify(mockedConnection).putSlotState(nextSlotState1); verify(mockedConnection).putSlotState(nextSlotState2); verifyNoMoreInteractions(mockedConnection); }