@Test
public void testGetClusterHostInfo() throws Exception {

  final HashMap<String, String> hostAttributes = new HashMap<String, String>() {{
    put("os_family", "redhat");
    put("os_release_version", "5.9");
  }};
  final Clusters clusters = createNiceMock(Clusters.class);
  List<Host> hosts = new ArrayList<Host>();
  List<String> hostNames = new ArrayList<String>();
  List<Integer> pingPorts = Arrays.asList(StageUtils.DEFAULT_PING_PORT,
      StageUtils.DEFAULT_PING_PORT,
      StageUtils.DEFAULT_PING_PORT,
      8671,
      8671,
      null,
      8672,
      8672,
      null,
      8673);
  for (int i = 0; i < 10; i++) {
    String hostname = String.format("h%d", i);
    Host host = createNiceMock(Host.class);
    expect(host.getHostName()).andReturn(hostname).anyTimes();
    expect(host.getHostAttributes()).andReturn(hostAttributes).anyTimes();
    expect(host.getCurrentPingPort()).andReturn(pingPorts.get(i)).anyTimes();
    hosts.add(host);
    hostNames.add(hostname);
    expect(clusters.getHost(hostname)).andReturn(host).anyTimes();
  }
  final ServiceComponentHost nnh0ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(nnh0ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost snnh1ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(snnh1ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost dnh0ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(dnh0ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost dnh1ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(dnh1ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost dnh2ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(dnh2ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost dnh3ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(dnh3ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost dnh5ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(dnh5ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost dnh7ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(dnh7ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost dnh8ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(dnh8ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost dnh9ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(dnh9ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost hbm5ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(hbm5ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost hbrs1ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(hbrs1ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost hbrs3ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(hbrs3ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost hbrs5ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(hbrs5ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost hbrs8ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(hbrs8ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost hbrs9ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(hbrs9ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost mrjt5ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(mrjt5ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost mrtt1ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(mrtt1ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost mrtt2ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(mrtt2ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.DECOMMISSIONED).anyTimes();
  final ServiceComponentHost mrtt3ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(mrtt3ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.DECOMMISSIONED).anyTimes();
  final ServiceComponentHost mrtt4ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(mrtt4ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost mrtt5ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(mrtt5ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost mrtt7ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(mrtt7ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost mrtt9ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(mrtt9ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  final ServiceComponentHost nns7ServiceComponentHost = createMock(ServiceComponentHost.class);
  expect(nns7ServiceComponentHost.getComponentAdminState()).andReturn(HostComponentAdminState.INSERVICE).anyTimes();
  Map<String, Collection<String>> projectedTopology = new HashMap<String, Collection<String>>();
  final HashMap<String, ServiceComponentHost> nnServiceComponentHosts = new HashMap<String, ServiceComponentHost>() {
    {
      put("h0", nnh0ServiceComponentHost);
    }
  };
  insertTopology(projectedTopology, "NAMENODE", nnServiceComponentHosts.keySet());
  final HashMap<String, ServiceComponentHost> snnServiceComponentHosts = new HashMap<String, ServiceComponentHost>() {
    {
      put("h1", snnh1ServiceComponentHost);
    }
  };
  insertTopology(projectedTopology, "SECONDARY_NAMENODE", snnServiceComponentHosts.keySet());
  final Map<String, ServiceComponentHost> dnServiceComponentHosts = new HashMap<String, ServiceComponentHost>() {
    {
      put("h0", dnh0ServiceComponentHost);
      put("h1", dnh1ServiceComponentHost);
      put("h2", dnh2ServiceComponentHost);
      put("h3", dnh3ServiceComponentHost);
      put("h5", dnh5ServiceComponentHost);
      put("h7", dnh7ServiceComponentHost);
      put("h8", dnh8ServiceComponentHost);
      put("h9", dnh9ServiceComponentHost);
    }
  };
  insertTopology(projectedTopology, "DATANODE", dnServiceComponentHosts.keySet());
  final Map<String, ServiceComponentHost> hbmServiceComponentHosts = new HashMap<String, ServiceComponentHost>() {
    {
      put("h5", hbm5ServiceComponentHost);
    }
  };
  insertTopology(projectedTopology, "HBASE_MASTER", hbmServiceComponentHosts.keySet());
  final Map<String, ServiceComponentHost> hbrsServiceComponentHosts = new HashMap<String, ServiceComponentHost>() {
    {
      put("h1", hbrs1ServiceComponentHost);
      put("h3", hbrs3ServiceComponentHost);
      put("h5", hbrs5ServiceComponentHost);
      put("h8", hbrs8ServiceComponentHost);
      put("h9", hbrs9ServiceComponentHost);
    }
  };
  insertTopology(projectedTopology, "HBASE_REGIONSERVER", hbrsServiceComponentHosts.keySet());
  final Map<String, ServiceComponentHost> mrjtServiceComponentHosts = new HashMap<String, ServiceComponentHost>() {
    {
      put("h5", mrjt5ServiceComponentHost);
    }
  };
  insertTopology(projectedTopology, "JOBTRACKER", mrjtServiceComponentHosts.keySet());
  final Map<String, ServiceComponentHost> mrttServiceComponentHosts = new HashMap<String, ServiceComponentHost>() {
    {
      put("h1", mrtt1ServiceComponentHost);
      put("h2", mrtt2ServiceComponentHost);
      put("h3", mrtt3ServiceComponentHost);
      put("h4", mrtt4ServiceComponentHost);
      put("h5", mrtt5ServiceComponentHost);
      put("h7", mrtt7ServiceComponentHost);
      put("h9", mrtt9ServiceComponentHost);
    }
  };
  insertTopology(projectedTopology, "TASKTRACKER", mrttServiceComponentHosts.keySet());
  final Map<String, ServiceComponentHost> nnsServiceComponentHosts = new HashMap<String, ServiceComponentHost>() {
    {
      put("h7", nns7ServiceComponentHost);
    }
  };
  insertTopology(projectedTopology, "NONAME_SERVER", nnsServiceComponentHosts.keySet());
  final ServiceComponent nnComponent = createMock(ServiceComponent.class);
  expect(nnComponent.getName()).andReturn("NAMENODE").anyTimes();
  expect(nnComponent.getServiceComponentHost(anyObject(String.class)))
      .andAnswer(new IAnswer<ServiceComponentHost>() {
        @Override
        public ServiceComponentHost answer() throws Throwable {
          Object[] args = getCurrentArguments();
          return nnServiceComponentHosts.get(args[0]);
        }
      }).anyTimes();
  expect(nnComponent.getServiceComponentHosts()).andReturn(nnServiceComponentHosts).anyTimes();
  expect(nnComponent.isClientComponent()).andReturn(false).anyTimes();
  final ServiceComponent snnComponent = createMock(ServiceComponent.class);
  expect(snnComponent.getName()).andReturn("SECONDARY_NAMENODE").anyTimes();
  expect(snnComponent.getServiceComponentHost(anyObject(String.class)))
      .andAnswer(new IAnswer<ServiceComponentHost>() {
        @Override
        public ServiceComponentHost answer() throws Throwable {
          Object[] args = getCurrentArguments();
          return snnServiceComponentHosts.get(args[0]);
        }
      }).anyTimes();
  expect(snnComponent.getServiceComponentHosts()).andReturn(snnServiceComponentHosts).anyTimes();
  expect(snnComponent.isClientComponent()).andReturn(false).anyTimes();
  final ServiceComponent dnComponent = createMock(ServiceComponent.class);
  expect(dnComponent.getName()).andReturn("DATANODE").anyTimes();
  expect(dnComponent.getServiceComponentHost(anyObject(String.class)))
      .andAnswer(new IAnswer<ServiceComponentHost>() {
        @Override
        public ServiceComponentHost answer() throws Throwable {
          Object[] args = getCurrentArguments();
          return dnServiceComponentHosts.get(args[0]);
        }
      }).anyTimes();
  expect(dnComponent.getServiceComponentHosts()).andReturn(dnServiceComponentHosts).anyTimes();
  expect(dnComponent.isClientComponent()).andReturn(false).anyTimes();
  final ServiceComponent hbmComponent = createMock(ServiceComponent.class);
  expect(hbmComponent.getName()).andReturn("HBASE_MASTER").anyTimes();
  expect(hbmComponent.getServiceComponentHost(anyObject(String.class)))
      .andAnswer(new IAnswer<ServiceComponentHost>() {
        @Override
        public ServiceComponentHost answer() throws Throwable {
          Object[] args = getCurrentArguments();
          return hbmServiceComponentHosts.get(args[0]);
        }
      }).anyTimes();
  expect(hbmComponent.getServiceComponentHosts()).andReturn(hbmServiceComponentHosts).anyTimes();
  expect(hbmComponent.isClientComponent()).andReturn(false).anyTimes();
  final ServiceComponent hbrsComponent = createMock(ServiceComponent.class);
  expect(hbrsComponent.getName()).andReturn("HBASE_REGIONSERVER").anyTimes();
  expect(hbrsComponent.getServiceComponentHost(anyObject(String.class)))
      .andAnswer(new IAnswer<ServiceComponentHost>() {
        @Override
        public ServiceComponentHost answer() throws Throwable {
          Object[] args = getCurrentArguments();
          return hbrsServiceComponentHosts.get(args[0]);
        }
      }).anyTimes();
  Map<String, ServiceComponentHost> hbrsHosts = Maps.filterKeys(hbrsServiceComponentHosts, new Predicate<String>() {
    @Override
    public boolean apply(String s) {
      return s.equals("h1");
    }
  });
  expect(hbrsComponent.getServiceComponentHosts()).andReturn(hbrsServiceComponentHosts).anyTimes();
  expect(hbrsComponent.isClientComponent()).andReturn(false).anyTimes();
  final ServiceComponent mrjtComponent = createMock(ServiceComponent.class);
  expect(mrjtComponent.getName()).andReturn("JOBTRACKER").anyTimes();
  expect(mrjtComponent.getServiceComponentHost(anyObject(String.class)))
      .andAnswer(new IAnswer<ServiceComponentHost>() {
        @Override
        public ServiceComponentHost answer() throws Throwable {
          Object[] args = getCurrentArguments();
          return mrjtServiceComponentHosts.get(args[0]);
        }
      }).anyTimes();
  expect(mrjtComponent.getServiceComponentHosts()).andReturn(mrjtServiceComponentHosts).anyTimes();
  expect(mrjtComponent.isClientComponent()).andReturn(false).anyTimes();
  final ServiceComponent mrttCompomnent = createMock(ServiceComponent.class);
  expect(mrttCompomnent.getName()).andReturn("TASKTRACKER").anyTimes();
  expect(mrttCompomnent.getServiceComponentHost(anyObject(String.class)))
      .andAnswer(new IAnswer<ServiceComponentHost>() {
        @Override
        public ServiceComponentHost answer() throws Throwable {
          Object[] args = getCurrentArguments();
          return mrttServiceComponentHosts.get(args[0]);
        }
      }).anyTimes();
  expect(mrttCompomnent.getServiceComponentHosts()).andReturn(mrttServiceComponentHosts).anyTimes();
  expect(mrttCompomnent.isClientComponent()).andReturn(false).anyTimes();
  final ServiceComponent nnsComponent = createMock(ServiceComponent.class);
  expect(nnsComponent.getName()).andReturn("NONAME_SERVER").anyTimes();
  expect(nnsComponent.getServiceComponentHost(anyObject(String.class)))
      .andAnswer(new IAnswer<ServiceComponentHost>() {
        @Override
        public ServiceComponentHost answer() throws Throwable {
          Object[] args = getCurrentArguments();
          return nnsServiceComponentHosts.get(args[0]);
        }
      }).anyTimes();
  expect(nnsComponent.getServiceComponentHosts()).andReturn(nnsServiceComponentHosts).anyTimes();
  expect(nnsComponent.isClientComponent()).andReturn(false).anyTimes();
  final Service hdfsService = createMock(Service.class);
  expect(hdfsService.getServiceComponents()).andReturn(new HashMap<String, ServiceComponent>() {{
    put("NAMENODE", nnComponent);
    put("SECONDARY_NAMENODE", snnComponent);
    put("DATANODE", dnComponent);
  }}).anyTimes();
  final Service hbaseService = createMock(Service.class);
  expect(hbaseService.getServiceComponents()).andReturn(new HashMap<String, ServiceComponent>() {{
    put("HBASE_MASTER", hbmComponent);
    put("HBASE_REGIONSERVER", hbrsComponent);
  }}).anyTimes();
  final Service mrService = createMock(Service.class);
  expect(mrService.getServiceComponents()).andReturn(new HashMap<String, ServiceComponent>() {
    {
      put("JOBTRACKER", mrjtComponent);
      put("TASKTRACKER", mrttCompomnent);
    }
  }).anyTimes();
  final Service nnService = createMock(Service.class);
  expect(nnService.getServiceComponents()).andReturn(new HashMap<String, ServiceComponent>() {
    {
      put("NONAME_SERVER", nnsComponent);
    }
  }).anyTimes();
  Cluster cluster = createMock(Cluster.class);
  expect(cluster.getHosts()).andReturn(hosts).anyTimes();
  expect(cluster.getServices()).andReturn(new HashMap<String, Service>() {{
    put("HDFS", hdfsService);
    put("HBASE", hbaseService);
    put("MAPREDUCE", mrService);
    put("NONAME", nnService);
  }}).anyTimes();
  final TopologyManager topologyManager = injector.getInstance(TopologyManager.class);
  topologyManager.getPendingHostComponents();
  expectLastCall().andReturn(projectedTopology).once();
  replayAll();
  // This is required by the infrastructure
  injector.getInstance(AmbariMetaInfo.class).init();
  //Get cluster host info
  Map<String, Set<String>> info = StageUtils.getClusterHostInfo(cluster);
  verifyAll();
  //All hosts present in cluster host info
  Set<String> allHosts = info.get(StageUtils.HOSTS_LIST);
  assertEquals(hosts.size(), allHosts.size());
  for (Host host : hosts) {
    assertTrue(allHosts.contains(host.getHostName()));
  }
  checkServiceHostIndexes(info, "DATANODE", "slave_hosts", projectedTopology, hostNames);
  checkServiceHostIndexes(info, "NAMENODE", "namenode_host", projectedTopology, hostNames);
  checkServiceHostIndexes(info, "SECONDARY_NAMENODE", "snamenode_host", projectedTopology, hostNames);
  checkServiceHostIndexes(info, "HBASE_MASTER", "hbase_master_hosts", projectedTopology, hostNames);
  checkServiceHostIndexes(info, "HBASE_REGIONSERVER", "hbase_rs_hosts", projectedTopology, hostNames);
  checkServiceHostIndexes(info, "JOBTRACKER", "jtnode_host", projectedTopology, hostNames);
  checkServiceHostIndexes(info, "TASKTRACKER", "mapred_tt_hosts", projectedTopology, hostNames);
  checkServiceHostIndexes(info, "NONAME_SERVER", "noname_server_hosts", projectedTopology, hostNames);
  Set<String> actualPingPorts = info.get(StageUtils.PORTS);
  if (pingPorts.contains(null)) {
    assertEquals(new HashSet<Integer>(pingPorts).size(), actualPingPorts.size() + 1);
  } else {
    assertEquals(new HashSet<Integer>(pingPorts).size(), actualPingPorts.size());
  }
  List<Integer> pingPortsActual = getRangeMappedDecompressedSet(actualPingPorts);
  List<Integer> reindexedPorts = getReindexedList(pingPortsActual, new ArrayList<String>(allHosts), hostNames);
  //Treat null values
  List<Integer> expectedPingPorts = new ArrayList<Integer>(pingPorts);
  for (int i = 0; i < expectedPingPorts.size(); i++) {
    if (expectedPingPorts.get(i) == null) {
      expectedPingPorts.set(i, StageUtils.DEFAULT_PING_PORT);
    }
  }
  assertEquals(expectedPingPorts, reindexedPorts);
  assertTrue(info.containsKey("decom_tt_hosts"));
  Set<String> decommissionedHosts = info.get("decom_tt_hosts");
  assertEquals(2, decommissionedHosts.toString().split(",").length);
  // check server hostname field
  assertTrue(info.containsKey(StageUtils.AMBARI_SERVER_HOST));
  Set<String> serverHost = info.get(StageUtils.AMBARI_SERVER_HOST);
  assertEquals(1, serverHost.size());
  assertEquals(StageUtils.getHostName(), serverHost.iterator().next());
  // check host role replacing by the projected topology
  assertTrue(getDecompressedSet(info.get("hbase_rs_hosts")).contains(9));
  // Validate substitutions...
  info = StageUtils.substituteHostIndexes(info);
  checkServiceHostNames(info, "DATANODE", "slave_hosts", projectedTopology);
  checkServiceHostNames(info, "NAMENODE", "namenode_host", projectedTopology);
  checkServiceHostNames(info, "SECONDARY_NAMENODE", "snamenode_host", projectedTopology);
  checkServiceHostNames(info, "HBASE_MASTER", "hbase_master_hosts", projectedTopology);
  checkServiceHostNames(info, "HBASE_REGIONSERVER", "hbase_rs_hosts", projectedTopology);
  checkServiceHostNames(info, "JOBTRACKER", "jtnode_host", projectedTopology);
  checkServiceHostNames(info, "TASKTRACKER", "mapred_tt_hosts", projectedTopology);
  checkServiceHostNames(info, "NONAME_SERVER", "noname_server_hosts", projectedTopology);
}