Add more testcases

Author: szilard-nemeth

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerConfiguration.java

@@ -18,16 +18,31 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.util.Sets;
 import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.junit.Test;
 
+import java.util.Set;
+
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.ROOT;
 import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.getQueuePrefix;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 public class TestCapacitySchedulerConfiguration {
 
+  private static final String ROOT_TEST = ROOT + ".test";
+  private static final String EMPTY_ACL = "";
+  private static final String SPACE_ACL = " ";
+  private static final String USER1 = "user1";
+  private static final String USER2 = "user2";
+  private static final String GROUP1 = "group1";
+  private static final String GROUP2 = "group2";
+  public static final String ONE_USER_ONE_GROUP_ACL = USER1 + " " + GROUP1;
+  public static final String TWO_USERS_TWO_GROUPS_ACL =
+      USER1 + "," + USER2 + " " + GROUP1 + ", " + GROUP2;
+
   private CapacitySchedulerConfiguration createDefaultCsConf() {
     return new CapacitySchedulerConfiguration(new Configuration(false), false);
   }
@@ -44,10 +59,45 @@ private String getSubmitAppsConfigKey(String queue) {
     return getQueuePrefix(queue) + "acl_submit_applications";
   }
 
+  private void testWithGivenAclNoOneHasAccess(String queue, String aclValue) {
+    testWithGivenAclNoOneHasAccessInternal(queue, queue, aclValue);
+  }
+
+  private void testWithGivenAclNoOneHasAccess(String queueToSet, String queueToVerify, String aclValue) {
+    testWithGivenAclNoOneHasAccessInternal(queueToSet, queueToVerify, aclValue);
+  }
+
+  private void testWithGivenAclNoOneHasAccessInternal(String queueToSet, String queueToVerify, String aclValue) {
+    CapacitySchedulerConfiguration csConf = createDefaultCsConf();
+    setSubmitAppsConfig(csConf, queueToSet, aclValue);
+    AccessControlList acl = getSubmitAcl(csConf, queueToVerify);
+    assertTrue(acl.getUsers().isEmpty());
+    assertTrue(acl.getGroups().isEmpty());
+    assertFalse(acl.isAllAllowed());
+  }
+
+  private void testWithGivenAclCorrectUserAndGroupHasAccess(String queue, String aclValue,
+      Set<String> expectedUsers, Set<String> expectedGroups) {
+    testWithGivenAclCorrectUserAndGroupHasAccessInternal(queue, queue, aclValue, expectedUsers, expectedGroups);
+  }
+
+  private void testWithGivenAclCorrectUserAndGroupHasAccessInternal(String queueToSet,
+      String queueToVerify, String aclValue, Set<String> expectedUsers,
+      Set<String> expectedGroups) {
+    CapacitySchedulerConfiguration csConf = createDefaultCsConf();
+    setSubmitAppsConfig(csConf, queueToSet, aclValue);
+    AccessControlList acl = getSubmitAcl(csConf, queueToVerify);
+    assertFalse(acl.getUsers().isEmpty());
+    assertFalse(acl.getGroups().isEmpty());
+    assertEquals(expectedUsers, acl.getUsers());
+    assertEquals(expectedGroups, acl.getGroups());
+    assertFalse(acl.isAllAllowed());
+  }
+
   @Test
   public void testDefaultSubmitACLForRootAllAllowed() {
     CapacitySchedulerConfiguration csConf = createDefaultCsConf();
-    AccessControlList acl = csConf.getAcl(ROOT, QueueACL.SUBMIT_APPLICATIONS);
+    AccessControlList acl = getSubmitAcl(csConf, ROOT);
     assertTrue(acl.getUsers().isEmpty());
     assertTrue(acl.getGroups().isEmpty());
     assertTrue(acl.isAllAllowed());
@@ -56,30 +106,49 @@ public void testDefaultSubmitACLForRootAllAllowed() {
   @Test
   public void testDefaultSubmitACLForRootChildNoneAllowed() {
     CapacitySchedulerConfiguration csConf = createDefaultCsConf();
-    AccessControlList acl = getSubmitAcl(csConf, ROOT + ".test");
+    AccessControlList acl = getSubmitAcl(csConf, ROOT_TEST);
     assertTrue(acl.getUsers().isEmpty());
     assertTrue(acl.getGroups().isEmpty());
     assertFalse(acl.isAllAllowed());
   }
 
   @Test
   public void testSpecifiedEmptySubmitACLForRoot() {
-    CapacitySchedulerConfiguration csConf = createDefaultCsConf();
-    setSubmitAppsConfig(csConf, ROOT, "");
-    AccessControlList acl = csConf.getAcl(ROOT, QueueACL.SUBMIT_APPLICATIONS);
-    assertTrue(acl.getUsers().isEmpty());
-    assertTrue(acl.getGroups().isEmpty());
-    assertFalse(acl.isAllAllowed());
+    testWithGivenAclNoOneHasAccess(ROOT, EMPTY_ACL);
   }
 
   @Test
   public void testSpecifiedEmptySubmitACLForRootIsNotInherited() {
-    CapacitySchedulerConfiguration csConf = createDefaultCsConf();
-    setSubmitAppsConfig(csConf, ROOT, "");
-    AccessControlList acl = getSubmitAcl(csConf, ROOT + ".test");
-    assertTrue(acl.getUsers().isEmpty());
-    assertTrue(acl.getGroups().isEmpty());
-    assertFalse(acl.isAllAllowed());
+    testWithGivenAclNoOneHasAccess(ROOT, ROOT_TEST, EMPTY_ACL);
+  }
+
+  @Test
+  public void testSpecifiedSpaceSubmitACLForRoot() {
+    testWithGivenAclNoOneHasAccess(ROOT, SPACE_ACL);
+  }
+
+  @Test
+  public void testSpecifiedSpaceSubmitACLForRootIsNotInherited() {
+    testWithGivenAclNoOneHasAccess(ROOT, ROOT_TEST, SPACE_ACL);
   }
-  
+
+  @Test
+  public void testSpecifiedSubmitACLForRoot() {
+    Set<String> expectedUsers = Sets.newHashSet(USER1);
+    Set<String> expectedGroups = Sets.newHashSet(GROUP1);
+    testWithGivenAclCorrectUserAndGroupHasAccess(ROOT, ONE_USER_ONE_GROUP_ACL, expectedUsers, expectedGroups);
+  }
+
+  @Test
+  public void testSpecifiedSubmitACLForRootIsNotInherited() {
+    testWithGivenAclNoOneHasAccess(ROOT, ROOT_TEST, ONE_USER_ONE_GROUP_ACL);
+  }
+
+  @Test
+  public void testSpecifiedSubmitACLTwoUsersTwoGroupsForRoot() {
+    Set<String> expectedUsers = Sets.newHashSet(USER1, USER2);
+    Set<String> expectedGroups = Sets.newHashSet(GROUP1, GROUP2);
+    testWithGivenAclCorrectUserAndGroupHasAccess(ROOT, TWO_USERS_TWO_GROUPS_ACL, expectedUsers, expectedGroups);
+  }
+
 }