KafkaModelAuthorizables { public static KafkaAuthorizable from(KeyValue keyValue) throws ConfigurationException { String prefix = keyValue.getKey().toLowerCase(); String name = keyValue.getValue(); for (AuthorizableType type : AuthorizableType.values()) { if (prefix.equalsIgnoreCase(type.name())) { return from(type, name); } } return null; }  static KafkaAuthorizable from(KeyValue keyValue); static KafkaAuthorizable from(String keyValue); static KafkaAuthorizable from(AuthorizableType type, String name);  }
@Test public void testHost() throws Exception { Host host1 = (Host)KafkaModelAuthorizables.from("HOST=host1"); assertEquals("host1", host1.getName()); }
@Test(expected=IllegalArgumentException.class) public void testNoKV() throws Exception { System.out.println(KafkaModelAuthorizables.from("nonsense")); }
@Test(expected=IllegalArgumentException.class) public void testEmptyKey() throws Exception { System.out.println(KafkaModelAuthorizables.from("=host1")); }
@Test(expected=IllegalArgumentException.class) public void testEmptyValue() throws Exception { System.out.println(KafkaModelAuthorizables.from("HOST=")); }
@Test public void testNotAuthorizable() throws Exception { assertNull(KafkaModelAuthorizables.from("k=v")); }
@Test public void testResourceNameIsCaseSensitive() throws Exception { Host host1 = (Host)KafkaModelAuthorizables.from("HOST=Host1"); assertEquals("Host1", host1.getName()); Cluster cluster1 = (Cluster)KafkaModelAuthorizables.from("Cluster=kafka-cluster"); assertEquals("kafka-cluster", cluster1.getName()); Topic topic1 = (Topic)KafkaModelAuthorizables.from("topic=topiC1"); assertEquals("topiC1", topic1.getName()); ConsumerGroup consumergroup1 = (ConsumerGroup)KafkaModelAuthorizables.from("ConsumerGroup=CG1"); assertEquals("CG1", consumergroup1.getName()); }
@Test public void testClusterResourceNameIsRestricted() throws Exception { try { KafkaModelAuthorizables.from("Cluster=cluster1"); fail("Cluster with name other than " + Cluster.NAME + " must not have been created."); } catch (ConfigurationException cex) { assertEquals("Exception message is not as expected.", "Kafka's cluster resource can only have name " + Cluster.NAME, cex.getMessage()); } catch (Exception ex) { fail("Configuration exception was expected for invalid Cluster name."); } }