  @Test
  public void testGetTasksToLaunchForOneTopologyWithMultipleOffersOnSameHost() {

    TopologyDetails t1 = TestUtils.constructTopologyDetails("t1",
                                                            MesosNimbusTest.DEFAULT_WORKER_COUNT,
                                                            MesosCommon.DEFAULT_WORKER_CPU,
                                                            MesosCommon.DEFAULT_WORKER_MEM_MB);
    Map<String, TopologyDetails> topologyDetailsMap = new HashMap<>();
    topologyDetailsMap.put("t1", t1);

    Topologies topologies = new Topologies(topologyDetailsMap);

    Offer offer = TestUtils.buildOffer("O-1", "h1", 0, 40000);
    map.put(offer.getId(), offer);
    offer = TestUtils.buildOffer("O-2", "h1", 24, 0);
    map.put(offer.getId(), offer);
    offer = TestUtils.buildOfferWithPorts("O-3", "h1", 0, 0, 3100, 3200);
    map.put(offer.getId(), offer);

    Map<String, AggregatedOffers> aggregatedOffersPerNode = MesosCommon.getAggregatedOffersPerNode(map);
    Map<String, Collection<WorkerSlot>> workerSlotsMap = new HashMap<>();
    Collection<WorkerSlot> workerSlots = new ArrayList<>();
    workerSlots.add(new WorkerSlot("h1", 3100));
    workerSlotsMap.put("t1", workerSlots);
    Map<String,List<Protos.TaskInfo>> tasksToLaunch = mesosNimbus.getTasksToLaunch(topologies, workerSlotsMap, aggregatedOffersPerNode);
    assertTrue((tasksToLaunch.size() == 1));
    assertTrue((tasksToLaunch.get("h1").size() == 1));
    List<Protos.TaskInfo> taskInfoList = tasksToLaunch.get("h1");
    assertTrue(hasResources("*", taskInfoList.get(0), MesosCommon.DEFAULT_WORKER_CPU, MesosCommon.DEFAULT_WORKER_MEM_MB, 3100l));
    assertTrue(hasCorrectExecutorResources(taskInfoList));

    aggregatedOffersPerNode = MesosCommon.getAggregatedOffersPerNode(map);
    workerSlots.add(new WorkerSlot("h1", 3101));
    workerSlots.add(new WorkerSlot("h1", 3102));
    tasksToLaunch = mesosNimbus.getTasksToLaunch(topologies, workerSlotsMap, aggregatedOffersPerNode);
    taskInfoList = tasksToLaunch.get("h1");
    assertTrue(taskInfoList.size() == 3);
    assertTrue(hasResources("*", taskInfoList.get(0), MesosCommon.DEFAULT_WORKER_CPU, MesosCommon.DEFAULT_WORKER_MEM_MB, 3100l));
    assertTrue(hasResources("*", taskInfoList.get(1), MesosCommon.DEFAULT_WORKER_CPU, MesosCommon.DEFAULT_WORKER_MEM_MB, 3101l));
    assertTrue(hasResources("*", taskInfoList.get(2), MesosCommon.DEFAULT_WORKER_CPU, MesosCommon.DEFAULT_WORKER_MEM_MB, 3102l));
    assertTrue(hasCorrectExecutorResources(taskInfoList));

    TopologyDetails t2 = TestUtils.constructTopologyDetails("t2",
                                                            MesosNimbusTest.DEFAULT_WORKER_COUNT,
                                                            MesosCommon.DEFAULT_WORKER_CPU,
                                                            MesosCommon.DEFAULT_WORKER_MEM_MB);
    workerSlots = new ArrayList<>();
    workerSlots.add(new WorkerSlot("h1", 3103));
    workerSlots.add(new WorkerSlot("h1", 3104));
    workerSlotsMap.put("t2", workerSlots);

    topologyDetailsMap = new HashMap<>();
    topologyDetailsMap.put("t1", t1);
    topologyDetailsMap.put("t2", t2);
    topologies = new Topologies(topologyDetailsMap);

    aggregatedOffersPerNode = MesosCommon.getAggregatedOffersPerNode(map);

    tasksToLaunch = mesosNimbus.getTasksToLaunch(topologies, workerSlotsMap, aggregatedOffersPerNode);
    Map<String, List<Protos.TaskInfo>> topologyIDtoTaskInfoMap = getTopologyIDtoTaskInfoMap(tasksToLaunch.get("h1"));
    taskInfoList = topologyIDtoTaskInfoMap.get("t1");
    assertTrue(hasResources("*", taskInfoList.get(0), MesosCommon.DEFAULT_WORKER_CPU, MesosCommon.DEFAULT_WORKER_MEM_MB, 3100l));
    assertTrue(hasResources("*", taskInfoList.get(1), MesosCommon.DEFAULT_WORKER_CPU, MesosCommon.DEFAULT_WORKER_MEM_MB, 3101l));
    assertTrue(hasResources("*", taskInfoList.get(2), MesosCommon.DEFAULT_WORKER_CPU, MesosCommon.DEFAULT_WORKER_MEM_MB, 3102l));
    assertTrue(hasCorrectExecutorResources(taskInfoList));

    taskInfoList = topologyIDtoTaskInfoMap.get("t2");
    assertTrue(hasResources("*", taskInfoList.get(0), MesosCommon.DEFAULT_WORKER_CPU, MesosCommon.DEFAULT_WORKER_MEM_MB, 3103l));
    assertTrue(hasResources("*", taskInfoList.get(1), MesosCommon.DEFAULT_WORKER_CPU, MesosCommon.DEFAULT_WORKER_MEM_MB, 3104l));
    assertTrue(hasCorrectExecutorResources(taskInfoList));
  }