@Test
public void testActionSchedule() throws Exception {

  Type type = new TypeToken<Map<String, Set<String>>>() {}.getType();
  Map<String, List<String>> clusterHostInfo = StageUtils.getGson().fromJson(CLUSTER_HOST_INFO, type);
  ActionQueue aq = new ActionQueue();
  Properties properties = new Properties();
  Configuration conf = new Configuration(properties);
  Clusters fsm = mock(Clusters.class);
  Cluster oneClusterMock = mock(Cluster.class);
  Service serviceObj = mock(Service.class);
  ServiceComponent scomp = mock(ServiceComponent.class);
  ServiceComponentHost sch = mock(ServiceComponentHost.class);
  UnitOfWork unitOfWork = mock(UnitOfWork.class);
  when(fsm.getCluster(anyString())).thenReturn(oneClusterMock);
  when(fsm.getClusterById(anyLong())).thenReturn(oneClusterMock);
  when(oneClusterMock.getService(anyString())).thenReturn(serviceObj);
  when(oneClusterMock.getClusterId()).thenReturn(Long.valueOf(1L));
  when(serviceObj.getServiceComponent(anyString())).thenReturn(scomp);
  when(scomp.getServiceComponentHost(anyString())).thenReturn(sch);
  when(serviceObj.getCluster()).thenReturn(oneClusterMock);
  Host host = mock(Host.class);
  HashMap<String, ServiceComponentHost> hosts =
          new HashMap<String, ServiceComponentHost>();
  hosts.put(hostname, sch);
  HostEntity hostEntity = new HostEntity();
  hostEntity.setHostName(hostname);
  hostDAO.merge(hostEntity);
  when(scomp.getServiceComponentHosts()).thenReturn(hosts);
  when(fsm.getHost(anyString())).thenReturn(host);
  when(host.getState()).thenReturn(HostState.HEALTHY);
  when(host.getHostName()).thenReturn(hostname);
  ActionDBAccessor db = mock(ActionDBAccessorImpl.class);
  HostRoleCommandDAO hostRoleCommandDAOMock = mock(HostRoleCommandDAO.class);
  Mockito.doNothing().when(hostRoleCommandDAOMock).publishTaskCreateEvent(anyListOf(HostRoleCommand.class));
  List<Stage> stages = new ArrayList<Stage>();
  Stage s = StageUtils.getATestStage(1, 977, hostname, CLUSTER_HOST_INFO,
    "{\"host_param\":\"param_value\"}", "{\"stage_param\":\"param_value\"}");
  stages.add(s);
  when(db.getCommandsInProgressCount()).thenReturn(stages.size());
  when(db.getStagesInProgress()).thenReturn(stages);
  RequestEntity request = mock(RequestEntity.class);
  when(request.isExclusive()).thenReturn(false);
  when(db.getRequestEntity(anyLong())).thenReturn(request);
  //Keep large number of attempts so that the task is not expired finally
  //Small action timeout to test rescheduling
  ActionScheduler scheduler = new ActionScheduler(100, 5, db, aq, fsm,
      10000, new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock, hostRoleCommandDAOMock, null);
  scheduler.setTaskTimeoutAdjustment(false);
  List<AgentCommand> ac = waitForQueueSize(hostname, aq, 1, scheduler);
  AgentCommand scheduledCommand = ac.get(0);
  assertTrue(scheduledCommand instanceof ExecutionCommand);
  assertEquals("1-977", ((ExecutionCommand) scheduledCommand).getCommandId());
  assertEquals(clusterHostInfo, ((ExecutionCommand) scheduledCommand).getClusterHostInfo());
  //The action status has not changed, it should be queued again.
  ac = waitForQueueSize(hostname, aq, 2, scheduler);
  // first command is cancel for previous
  scheduledCommand = ac.get(1);
  assertTrue(scheduledCommand instanceof ExecutionCommand);
  assertEquals("1-977", ((ExecutionCommand) scheduledCommand).getCommandId());
  assertEquals(clusterHostInfo, ((ExecutionCommand) scheduledCommand).getClusterHostInfo());
  //Now change the action status
  s.setHostRoleStatus(hostname, "NAMENODE", HostRoleStatus.COMPLETED);
  ac = aq.dequeueAll(hostname);
  //Wait for sometime, it shouldn't be scheduled this time.
  ac = waitForQueueSize(hostname, aq, 0, scheduler);
  EasyMock.verify(entityManagerProviderMock);
}