@Test (timeout = 30000)
public void testValidateResourceRequestWithErrorLabelsPermission()
    throws IOException {

  // mock queue and scheduler
  YarnScheduler scheduler = mock(YarnScheduler.class);
  Set<String> queueAccessibleNodeLabels = Sets.newHashSet();
  QueueInfo queueInfo = mock(QueueInfo.class);
  when(queueInfo.getQueueName()).thenReturn("queue");
  when(queueInfo.getAccessibleNodeLabels()).thenReturn(queueAccessibleNodeLabels);
  when(scheduler.getQueueInfo(any(String.class), anyBoolean(), anyBoolean()))
      .thenReturn(queueInfo);
  
  Resource maxResource = Resources.createResource(
      YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB,
      YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES);
  // queue has labels, success cases
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
        ImmutableSet.of("x","y"));
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    resReq.setNodeLabelExpression("y");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    
    resReq.setNodeLabelExpression("");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    
    resReq.setNodeLabelExpression(" ");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
  } catch (InvalidResourceRequestException e) {
    e.printStackTrace();
    fail("Should be valid when request labels is a subset of queue labels");
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
        Arrays.asList("x", "y"));
  }
  
  // same as above, but cluster node labels don't contains label being
  // requested. should fail
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  }
  
  // queue has labels, failed cases (when ask a label not included by queue)
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
      ImmutableSet.of("x","y"));
    
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("z");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
        Arrays.asList("x", "y"));
  }
  
  // we don't allow specify more than two node labels in a single expression
  // now
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
      ImmutableSet.of("x","y"));
    
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x && y");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
        Arrays.asList("x", "y"));
  }
  
  // queue doesn't have label, succeed (when request no label)
  queueAccessibleNodeLabels.clear();
  try {
    // set queue accessible node labels to empty
    queueAccessibleNodeLabels.clear();
    
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), ResourceRequest.ANY, resource, 1);
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    
    resReq.setNodeLabelExpression("");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    
    resReq.setNodeLabelExpression("  ");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
  } catch (InvalidResourceRequestException e) {
    e.printStackTrace();
    fail("Should be valid when request labels is empty");
  }
  
  // queue doesn't have label, failed (when request any label)
  try {
    // set queue accessible node labels to empty
    queueAccessibleNodeLabels.clear();
    
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
      ImmutableSet.of("x"));
    
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
        Arrays.asList("x"));
  }
  
  // queue is "*", always succeeded
  try {
    // set queue accessible node labels to empty
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.add(RMNodeLabelsManager.ANY);
    
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
      ImmutableSet.of("x","y", "z"));
    
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    
    resReq.setNodeLabelExpression("y");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    
    resReq.setNodeLabelExpression("z");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
  } catch (InvalidResourceRequestException e) {
    e.printStackTrace();
    fail("Should be valid when queue can access any labels");
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
        Arrays.asList("x", "y", "z"));
  }
  
  // same as above, but cluster node labels don't contains label, should fail
  try {
    // set queue accessible node labels to empty
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.add(RMNodeLabelsManager.ANY);
    
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), ResourceRequest.ANY, resource, 1);
    resReq.setNodeLabelExpression("x");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  }
  
  // we don't allow resource name other than ANY and specify label
  try {
    // set queue accessible node labesl to [x, y]
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays.asList("x", "y"));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
      ImmutableSet.of("x", "y"));
    
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), "rack", resource, 1);
    resReq.setNodeLabelExpression("x");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
        Arrays.asList("x", "y"));
  }
  
  // we don't allow resource name other than ANY and specify label even if
  // queue has accessible label = *
  try {
    // set queue accessible node labesl to *
    queueAccessibleNodeLabels.clear();
    queueAccessibleNodeLabels.addAll(Arrays
        .asList(CommonNodeLabelsManager.ANY));
    rmContext.getNodeLabelManager().addToCluserNodeLabels(
      ImmutableSet.of("x"));
    
    Resource resource = Resources.createResource(
        0,
        YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES);
    ResourceRequest resReq = BuilderUtils.newResourceRequest(
        mock(Priority.class), "rack", resource, 1);
    resReq.setNodeLabelExpression("x");
    SchedulerUtils.normalizeAndvalidateRequest(resReq, maxResource, "queue",
        scheduler, rmContext);
    fail("Should fail");
  } catch (InvalidResourceRequestException e) {
  } finally {
    rmContext.getNodeLabelManager().removeFromClusterNodeLabels(
        Arrays.asList("x"));
  }
}