@Test
@SuppressWarnings("unchecked")
public void testCompleteWorkflow() {

    WorkflowDef def = new WorkflowDef();
    def.setName("test");
    Workflow workflow = new Workflow();
    workflow.setWorkflowDefinition(def);
    workflow.setWorkflowId("1");
    workflow.setStatus(Workflow.WorkflowStatus.RUNNING);
    workflow.setOwnerApp("junit_test");
    workflow.setStartTime(10L);
    workflow.setEndTime(100L);
    workflow.setOutput(Collections.EMPTY_MAP);
    when(executionDAOFacade.getWorkflowById(anyString(), anyBoolean())).thenReturn(workflow);
    AtomicInteger updateWorkflowCalledCounter = new AtomicInteger(0);
    doAnswer(invocation -> {
        updateWorkflowCalledCounter.incrementAndGet();
        return null;
    }).when(executionDAOFacade).updateWorkflow(any());
    AtomicInteger updateTasksCalledCounter = new AtomicInteger(0);
    doAnswer(invocation -> {
        updateTasksCalledCounter.incrementAndGet();
        return null;
    }).when(executionDAOFacade).updateTasks(any());
    AtomicInteger removeQueueEntryCalledCounter = new AtomicInteger(0);
    doAnswer(invocation -> {
        removeQueueEntryCalledCounter.incrementAndGet();
        return null;
    }).when(queueDAO).remove(anyString(), anyString());
    workflowExecutor.completeWorkflow(workflow);
    assertEquals(Workflow.WorkflowStatus.COMPLETED, workflow.getStatus());
    assertEquals(1, updateWorkflowCalledCounter.get());
    assertEquals(1, updateTasksCalledCounter.get());
    assertEquals(1, removeQueueEntryCalledCounter.get());
    verify(workflowStatusListener, times(0)).onWorkflowCompleted(any(Workflow.class));
    def.setWorkflowStatusListenerEnabled(true);
    workflow.setStatus(Workflow.WorkflowStatus.RUNNING);
    workflowExecutor.completeWorkflow(workflow);
    verify(workflowStatusListener, times(1)).onWorkflowCompleted(any(Workflow.class));
}