@Test
public void testMultiLevelQueues() throws Exception {

  /*
   * Structure of queue:
   *            Root
   *           ____________
   *          /    |   \   \
   *         A     B    C   D
   *       / |   / | \   \
   *      A1 A2 B1 B2 B3  C1
   *                        \
   *                         C11
   *                           \
   *                           C111
   *                             \
   *                              C1111
   */
  
  // Setup queue configs
  setupMultiLevelQueues(csConf);
  
  Map<String, CSQueue> queues = new HashMap<String, CSQueue>();
  CSQueue root = 
      CapacityScheduler.parseQueue(csContext, csConf, null, 
          CapacitySchedulerConfiguration.ROOT, queues, queues, 
          TestUtils.spyHook);
  
  // Setup some nodes
  final int memoryPerNode = 10;
  final int coresPerNode = 16;
  final int numNodes = 3;
  
  FiCaSchedulerNode node_0 = 
      TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB);
  FiCaSchedulerNode node_1 = 
      TestUtils.getMockNode("host_1", DEFAULT_RACK, 0, memoryPerNode*GB);
  FiCaSchedulerNode node_2 = 
      TestUtils.getMockNode("host_2", DEFAULT_RACK, 0, memoryPerNode*GB);
  
  final Resource clusterResource = 
      Resources.createResource(numNodes * (memoryPerNode*GB), 
          numNodes * coresPerNode);
  when(csContext.getNumClusterNodes()).thenReturn(numNodes);
  // Start testing
  CSQueue a = queues.get(A);
  CSQueue b = queues.get(B);
  CSQueue c = queues.get(C);
  CSQueue d = queues.get(D);
  CSQueue a1 = queues.get(A1);
  CSQueue a2 = queues.get(A2);
  CSQueue b1 = queues.get(B1);
  CSQueue b2 = queues.get(B2);
  CSQueue b3 = queues.get(B3);
  // Simulate C returning a container on node_0
  stubQueueAllocation(a, clusterResource, node_0, 0*GB);
  stubQueueAllocation(b, clusterResource, node_0, 0*GB);
  stubQueueAllocation(c, clusterResource, node_0, 1*GB);
  stubQueueAllocation(d, clusterResource, node_0, 0*GB);
  root.assignContainers(clusterResource, node_0, 
      new ResourceLimits(clusterResource));
  verifyQueueMetrics(a, 0*GB, clusterResource);
  verifyQueueMetrics(b, 0*GB, clusterResource);
  verifyQueueMetrics(c, 1*GB, clusterResource);
  verifyQueueMetrics(d, 0*GB, clusterResource);
  reset(a); reset(b); reset(c);
  // Now get B2 to allocate
  // A = 0/3, B = 0/15, C = 1/6, D=0/6
  stubQueueAllocation(a, clusterResource, node_1, 0*GB);
  stubQueueAllocation(b2, clusterResource, node_1, 4*GB);
  stubQueueAllocation(c, clusterResource, node_1, 0*GB);
  root.assignContainers(clusterResource, node_1, 
      new ResourceLimits(clusterResource));
  verifyQueueMetrics(a, 0*GB, clusterResource);
  verifyQueueMetrics(b, 4*GB, clusterResource);
  verifyQueueMetrics(c, 1*GB, clusterResource);
  reset(a); reset(b); reset(c);
  
  // Now get both A1, C & B3 to allocate in right order
  // A = 0/3, B = 4/15, C = 1/6, D=0/6
  stubQueueAllocation(a1, clusterResource, node_0, 1*GB);
  stubQueueAllocation(b3, clusterResource, node_0, 2*GB);
  stubQueueAllocation(c, clusterResource, node_0, 2*GB);
  root.assignContainers(clusterResource, node_0, 
      new ResourceLimits(clusterResource));
  InOrder allocationOrder = inOrder(a, c, b);
  allocationOrder.verify(a).assignContainers(eq(clusterResource), 
      any(FiCaSchedulerNode.class), anyResourceLimits());
  allocationOrder.verify(c).assignContainers(eq(clusterResource), 
      any(FiCaSchedulerNode.class), anyResourceLimits());
  allocationOrder.verify(b).assignContainers(eq(clusterResource), 
      any(FiCaSchedulerNode.class), anyResourceLimits());
  verifyQueueMetrics(a, 1*GB, clusterResource);
  verifyQueueMetrics(b, 6*GB, clusterResource);
  verifyQueueMetrics(c, 3*GB, clusterResource);
  reset(a); reset(b); reset(c);
  
  // Now verify max-capacity
  // A = 1/3, B = 6/15, C = 3/6, D=0/6
  // Ensure a1 won't alloc above max-cap although it should get 
  // scheduling opportunity now, right after a2
  LOG.info("here");
  ((ParentQueue)a).setMaxCapacity(.1f);  // a should be capped at 3/30
  stubQueueAllocation(a1, clusterResource, node_2, 1*GB); // shouldn't be 
                                                          // allocated due 
                                                          // to max-cap
  stubQueueAllocation(a2, clusterResource, node_2, 2*GB);
  stubQueueAllocation(b3, clusterResource, node_2, 1*GB);
  stubQueueAllocation(b1, clusterResource, node_2, 1*GB);
  stubQueueAllocation(c, clusterResource, node_2, 1*GB);
  root.assignContainers(clusterResource, node_2, 
      new ResourceLimits(clusterResource));
  allocationOrder = inOrder(a, a2, a1, b, c);
  allocationOrder.verify(a).assignContainers(eq(clusterResource), 
      any(FiCaSchedulerNode.class), anyResourceLimits());
  allocationOrder.verify(a2).assignContainers(eq(clusterResource), 
      any(FiCaSchedulerNode.class), anyResourceLimits());
  allocationOrder.verify(b).assignContainers(eq(clusterResource), 
      any(FiCaSchedulerNode.class), anyResourceLimits());
  allocationOrder.verify(c).assignContainers(eq(clusterResource), 
      any(FiCaSchedulerNode.class), anyResourceLimits());
  verifyQueueMetrics(a, 3*GB, clusterResource);
  verifyQueueMetrics(b, 8*GB, clusterResource);
  verifyQueueMetrics(c, 4*GB, clusterResource);
  reset(a); reset(b); reset(c);
}