metron-platform/metron-management/src/test/java/org/apache/metron/management/ConfigurationFunctionsTest.java - metron - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.metron.management;

 import com.google.common.collect.ImmutableMap;
 import java.nio.charset.StandardCharsets;
 import org.adrianwalker.multilinestring.Multiline;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.PosixParser;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.test.TestingServer;
 import org.apache.log4j.Level;
 import org.apache.metron.common.cli.ConfigurationManager;
 import org.apache.metron.common.configuration.ConfigurationsUtils;
 import org.apache.metron.common.configuration.SensorParserConfig;
 import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
 import org.apache.metron.common.configuration.profiler.ProfilerConfig;
 import org.apache.metron.stellar.dsl.Context;
 import org.apache.metron.stellar.dsl.ParseException;
 import org.apache.metron.test.utils.UnitTestHelper;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;

 import java.util.Collections;
 import java.util.Map;

 import static org.apache.metron.TestConstants.PARSER_CONFIGS_PATH;
 import static org.apache.metron.TestConstants.SAMPLE_CONFIG_PATH;
 import static org.apache.metron.common.configuration.ConfigurationType.GLOBAL;
 import static org.apache.metron.common.configuration.ConfigurationType.PROFILER;
 import static org.apache.metron.common.configuration.ConfigurationsUtils.writeProfilerConfigToZookeeper;
 import static org.apache.metron.management.utils.FileUtils.slurp;
 import static org.apache.metron.stellar.common.utils.StellarProcessorUtils.run;
 import static org.junit.jupiter.api.Assertions.*;

 /**
  * Tests the ConfigurationFunctions class.
  */
 public class ConfigurationFunctionsTest {

   private static TestingServer testZkServer;
   private static String zookeeperUrl;
   private static CuratorFramework client;
   private static String goodGlobalConfig = slurp( SAMPLE_CONFIG_PATH+ "/global.json");
   private static String goodTestEnrichmentConfig = slurp( SAMPLE_CONFIG_PATH + "/enrichments/test.json");
   private static String goodBroParserConfig = slurp(PARSER_CONFIGS_PATH + "/parsers/bro.json");
   private static String goodTestIndexingConfig = slurp( SAMPLE_CONFIG_PATH + "/indexing/test.json");

   private Context context;
   private JSONParser parser;

   /**
    * {
    *   "profiles" : [
    *      {
    *        "profile" : "counter",
    *        "foreach" : "ip_src_addr",
    *        "init"    : { "counter" : 0 },
    *        "update"  : { "counter" : "counter + 1" },
    *        "result"  : "counter"
    *      }
    *   ],
    *   "timestampField" : "timestamp"
    * }
    */
   @Multiline
   private static String goodProfilerConfig;

   @BeforeAll
   public static void setupZookeeper() throws Exception {

     // zookeeper server
     testZkServer = new TestingServer(true);
     zookeeperUrl = testZkServer.getConnectString();

     // zookeeper client
     client = ConfigurationsUtils.getClient(zookeeperUrl);
     client.start();
   }

   @BeforeEach
   public void setup() throws Exception {

     context = new Context.Builder()
             .with(Context.Capabilities.ZOOKEEPER_CLIENT, () -> client)
             .build();

     parser = new JSONParser();

     // push configs to zookeeper
     pushConfigs(SAMPLE_CONFIG_PATH, zookeeperUrl);
     pushConfigs(PARSER_CONFIGS_PATH, zookeeperUrl);
     writeProfilerConfigToZookeeper(goodProfilerConfig.getBytes(StandardCharsets.UTF_8), client);
   }

   /**
    * Deletes a path within Zookeeper.
    *
    * @param path The path within Zookeeper to delete.
    * @throws Exception
    */
   private void deletePath(String path) throws Exception {
     client.delete().forPath(path);
   }

   /**
    * Transforms a String to a {@link JSONObject}.
    *
    * @param input The input String to transform
    * @return A {@link JSONObject}.
    * @throws org.json.simple.parser.ParseException
    */
   private JSONObject toJSONObject(String input) throws org.json.simple.parser.ParseException {

     if(input == null) {
       return null;
     }
     return (JSONObject) parser.parse(input.trim());
   }

   /**
    * Push configuration values to Zookeeper.
    *
    * @param inputPath The local filesystem path to the configurations.
    * @param zookeeperUrl The URL of Zookeeper.
    * @throws Exception
    */
   private static void pushConfigs(String inputPath, String zookeeperUrl) throws Exception {

     String[] args = new String[] {
             "-z", zookeeperUrl,
             "--mode", "PUSH",
             "--input_dir", inputPath
     };
     CommandLine cli = ConfigurationManager.ConfigurationOptions.parse(new PosixParser(), args);

     ConfigurationManager manager = new ConfigurationManager();
     manager.run(cli);
   }

   /**
    * The CONFIG_GET function should be able to return the Parser configuration
    * for a given sensor.
    */
   @Test
   public void testGetParser() throws Exception {

     String out = (String) run("CONFIG_GET('PARSER', 'bro')", context);

     SensorParserConfig actual = SensorParserConfig.fromBytes(out.getBytes(StandardCharsets.UTF_8));
     SensorParserConfig expected = SensorParserConfig.fromBytes(goodBroParserConfig.getBytes(
         StandardCharsets.UTF_8));
     assertEquals(expected, actual);
   }

   /**
    * The CONFIG_GET function should NOT return any configuration when the
    * Parser configuration for a given sensor is missing AND emptyIfNotPresent = false.
    */
   @Test
   public void testGetParserMissWithoutDefault() {

     // expect null because emptyIfNotPresent = false
     Object out = run("CONFIG_GET('PARSER', 'sensor', false)", context);
     assertNull(out);
   }

   /**
    * The CONFIG_GET function should return a default configuration when none
    * currently exists.
    */
   @Test
   public void testGetParserMissWithDefault() throws Exception {

     SensorParserConfig expected = new SensorParserConfig();
     {
       Object out = run("CONFIG_GET('PARSER', 'sensor')", context);
       SensorParserConfig actual = SensorParserConfig.fromBytes(out.toString().getBytes(
           StandardCharsets.UTF_8));
       assertEquals(expected, actual);
     }
     {
       Object out = run("CONFIG_GET('PARSER', 'sensor', true)", context);
       SensorParserConfig actual = SensorParserConfig.fromBytes(out.toString().getBytes(
           StandardCharsets.UTF_8));
       assertEquals(expected, actual);
     }
   }

   /**
    * The CONFIG_GET function should be able to return the Enrichment configuration
    * for a given sensor.
    */
   @Test
   public void testGetEnrichment() throws Exception {

     String out = (String) run("CONFIG_GET('ENRICHMENT', 'test')", context);

     SensorEnrichmentConfig actual = SensorEnrichmentConfig.fromBytes(out.getBytes(
         StandardCharsets.UTF_8));
     SensorEnrichmentConfig expected = SensorEnrichmentConfig.fromBytes(goodTestEnrichmentConfig.getBytes(
         StandardCharsets.UTF_8));
     assertEquals(expected, actual);
   }

   /**
    * No default configuration should be provided in this case.
    */
   @Test
   public void testGetEnrichmentMissWithoutDefault() {

     // expect null because emptyIfNotPresent = false
     Object out = run("CONFIG_GET('ENRICHMENT', 'sense', false)", context);
     assertNull(out);
   }

   /**
    * A default empty configuration should be provided, if one does not exist.
    */
   @Test
   public void testGetEnrichmentMissWithDefault() throws Exception {

     // expect an empty configuration to be returned
     SensorEnrichmentConfig expected = new SensorEnrichmentConfig();
     {
       String out = (String) run("CONFIG_GET('ENRICHMENT', 'missing-sensor')", context);
       SensorEnrichmentConfig actual = SensorEnrichmentConfig.fromBytes(out.getBytes(
           StandardCharsets.UTF_8));
       assertEquals(expected, actual);
     }
     {
       String out = (String) run("CONFIG_GET('ENRICHMENT', 'missing-sensor', true)", context);
       SensorEnrichmentConfig actual = SensorEnrichmentConfig.fromBytes(out.getBytes(
           StandardCharsets.UTF_8));
       assertEquals(expected, actual);
     }
   }

   /**
    * The CONFIG_GET function should be able to return the Indexing configuration
    * for a given sensor.
    */
   @Test
   public void testGetIndexing() throws Exception {

     String out = (String) run("CONFIG_GET('INDEXING', 'test')", context);

     Map<String, Object> actual = toJSONObject(out);
     Map<String, Object> expected = toJSONObject(goodTestIndexingConfig);
     assertEquals(expected, actual);
   }

   /**
    * No default configuration should be provided in this case.
    */
   @Test
   public void testGetIndexingMissWithoutDefault() {

     // expect null because emptyIfNotPresent = false
     Object out = run("CONFIG_GET('INDEXING', 'sense', false)", context);
     assertNull(out);
   }

   /**
    * A default empty configuration should be provided, if one does not exist.
    */
   @Test
   public void testGetIndexingtMissWithDefault() throws Exception {

     // expect an empty configuration to be returned
     Map<String, Object> expected = Collections.emptyMap();
     {
       String out = (String) run("CONFIG_GET('INDEXING', 'missing-sensor')", context);
       Map<String, Object> actual = toJSONObject(out);
       assertEquals(expected, actual);
     }
     {
       String out = (String) run("CONFIG_GET('INDEXING', 'missing-sensor', true)", context);
       Map<String, Object> actual = toJSONObject(out);
       assertEquals(expected, actual);
     }
   }

   /**
    * The CONFIG_GET function should be able to return the Profiler configuration.
    */
   @Test
   public void testGetProfiler() throws Exception {

     String out = (String) run("CONFIG_GET('PROFILER')", context);

     ProfilerConfig actual = ProfilerConfig.fromBytes(out.getBytes(StandardCharsets.UTF_8));
     ProfilerConfig expected = ProfilerConfig.fromBytes(goodProfilerConfig.getBytes(
         StandardCharsets.UTF_8));
     assertEquals(expected, actual);
   }

   /**
    * No default configuration should be provided in this case.
    */
   @Test
   public void testGetProfilerMissWithoutDefault() throws Exception {

     deletePath(PROFILER.getZookeeperRoot());

     // expect null because emptyIfNotPresent = false
     String out = (String) run("CONFIG_GET('PROFILER', false)", context);
     assertNull(out);
   }

   /**
    * A default empty configuration should be provided, if one does not exist.
    */
   @Test
   public void testGetProfilerMissWithDefault() throws Exception {

     // there is no profiler config in zookeeper
     deletePath(PROFILER.getZookeeperRoot());

     // expect an empty configuration to be returned
     ProfilerConfig expected = new ProfilerConfig();
     {
       String out = (String) run("CONFIG_GET('PROFILER', true)", context);
       ProfilerConfig actual = ProfilerConfig.fromJSON(out);
       assertEquals(expected, actual);
     }
     {
       String out = (String) run("CONFIG_GET('PROFILER')", context);
       ProfilerConfig actual = ProfilerConfig.fromJSON(out);
       assertEquals(expected, actual);
     }
   }

   @Test
   public void testGetGlobal() throws Exception {

     String out = (String) run("CONFIG_GET('GLOBAL')", context);

     Map<String, Object> actual = toJSONObject(out);
     Map<String, Object> expected = toJSONObject(goodGlobalConfig);
     assertEquals(expected, actual);
   }

   /**
    * No default configuration should be provided in this case.
    */
   @Test
   public void testGetGlobalMissWithoutDefault() throws Exception {

     // there is no global config in zookeeper
     deletePath(GLOBAL.getZookeeperRoot());

     // expect null because emptyIfNotPresent = false
     Object out = run("CONFIG_GET('GLOBAL', false)", context);
     assertNull(out);
   }

   /**
    * A default empty configuration should be provided, if one does not exist.
    */
   @Test
   public void testGetGlobalMissWithDefault() throws Exception {

     // there is no global config in zookeeper
     deletePath(GLOBAL.getZookeeperRoot());

     // expect an empty configuration to be returned
     Map<String, Object> expected = Collections.emptyMap();
     {
       String out = (String) run("CONFIG_GET('GLOBAL')", context);
       Map<String, Object> actual = toJSONObject(out);
       assertEquals(expected, actual);
     }
     {
       String out = (String) run("CONFIG_GET('GLOBAL', true)", context);
       Map<String, Object> actual = toJSONObject(out);
       assertEquals(expected, actual);
     }
   }

   @Test
   public void testPutGlobal() throws Exception {

     String out = (String) run("CONFIG_GET('GLOBAL')", context);

     Map<String, Object> actual = toJSONObject(out);
     Map<String, Object> expected = toJSONObject(goodGlobalConfig);
     assertEquals(expected, actual);
   }

   @Test
   public void testPutGlobalBad() {
     {
       UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.FATAL);
       assertThrows(ParseException.class, () -> run("CONFIG_PUT('GLOBAL', 'foo bar')", context));
       UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.ERROR);
     }
   }

   @Test
   public void testPutIndexing() throws InterruptedException {
     String brop= (String) run("CONFIG_GET('INDEXING', 'testIndexingPut')", context);
     run("CONFIG_PUT('INDEXING', config, 'testIndexingPut')", ImmutableMap.of("config", brop), context);
     boolean foundMatch = false;
     for(int i = 0;i < 10 && !foundMatch;++i) {
       String bropNew = (String) run("CONFIG_GET('INDEXING', 'testIndexingPut', false)", context);
       foundMatch =  brop.equals(bropNew);
       if(foundMatch) {
         break;
       }
       Thread.sleep(2000);
     }
     assertTrue(foundMatch);
   }

   @Test
   public void testPutIndexingBad() {
     {
       {
         UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.FATAL);
         assertThrows(
             ParseException.class,
             () ->
                 run(
                     "CONFIG_PUT('INDEXING', config, 'brop')",
                     ImmutableMap.of("config", "foo bar"),
                     context));
         UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.ERROR);
       }
     }
   }

   @Test
   public void testPutEnrichment() throws InterruptedException {
     String config = (String) run("CONFIG_GET('ENRICHMENT', 'sensor')", context);
     assertNotNull(config);

     run("CONFIG_PUT('ENRICHMENT', config, 'sensor')", ImmutableMap.of("config", config), context);

     boolean foundMatch = false;
     for(int i = 0;i < 10 && !foundMatch;++i) {
       String newConfig = (String) run("CONFIG_GET('ENRICHMENT', 'sensor', false)", context);
       foundMatch = config.equals(newConfig);
       if(foundMatch) {
         break;
       }
       Thread.sleep(2000);
     }
     assertTrue(foundMatch);
   }

   @Test
   public void testPutEnrichmentBad() {
     {
       {
         UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.FATAL);
         assertThrows(
             ParseException.class,
             () ->
                 run(
                     "CONFIG_PUT('ENRICHMENT', config, 'brop')",
                     ImmutableMap.of("config", "foo bar"),
                     context));
         UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.ERROR);
       }
     }
   }

   @Test
   public void testPutParser() throws InterruptedException {
     String brop= (String) run("CONFIG_GET('PARSER', 'testParserPut')", context);
     run("CONFIG_PUT('PARSER', config, 'testParserPut')", ImmutableMap.of("config", brop), context);
     boolean foundMatch = false;
     for(int i = 0;i < 10 && !foundMatch;++i) {
       String bropNew = (String) run("CONFIG_GET('PARSER', 'testParserPut', false)", context);
       foundMatch =  brop.equals(bropNew);
       if(foundMatch) {
         break;
       }
       Thread.sleep(2000);
     }
     assertTrue(foundMatch);
   }

   @Test
   public void testPutParserBad() {
     {
       UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.FATAL);
       assertThrows(
           ParseException.class,
           () ->
               run(
                   "CONFIG_PUT('PARSER', config, 'brop')",
                   ImmutableMap.of("config", "foo bar"),
                   context));
       UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.ERROR);
     }
   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.metron.management;

	import com.google.common.collect.ImmutableMap;
	import java.nio.charset.StandardCharsets;
	import org.adrianwalker.multilinestring.Multiline;
	import org.apache.commons.cli.CommandLine;
	import org.apache.commons.cli.PosixParser;
	import org.apache.curator.framework.CuratorFramework;
	import org.apache.curator.test.TestingServer;
	import org.apache.log4j.Level;
	import org.apache.metron.common.cli.ConfigurationManager;
	import org.apache.metron.common.configuration.ConfigurationsUtils;
	import org.apache.metron.common.configuration.SensorParserConfig;
	import org.apache.metron.common.configuration.enrichment.SensorEnrichmentConfig;
	import org.apache.metron.common.configuration.profiler.ProfilerConfig;
	import org.apache.metron.stellar.dsl.Context;
	import org.apache.metron.stellar.dsl.ParseException;
	import org.apache.metron.test.utils.UnitTestHelper;
	import org.json.simple.JSONObject;
	import org.json.simple.parser.JSONParser;
	import org.junit.jupiter.api.BeforeEach;
	import org.junit.jupiter.api.BeforeAll;
	import org.junit.jupiter.api.Test;

	import java.util.Collections;
	import java.util.Map;

	import static org.apache.metron.TestConstants.PARSER_CONFIGS_PATH;
	import static org.apache.metron.TestConstants.SAMPLE_CONFIG_PATH;
	import static org.apache.metron.common.configuration.ConfigurationType.GLOBAL;
	import static org.apache.metron.common.configuration.ConfigurationType.PROFILER;
	import static org.apache.metron.common.configuration.ConfigurationsUtils.writeProfilerConfigToZookeeper;
	import static org.apache.metron.management.utils.FileUtils.slurp;
	import static org.apache.metron.stellar.common.utils.StellarProcessorUtils.run;
	import static org.junit.jupiter.api.Assertions.*;

	/**
	* Tests the ConfigurationFunctions class.
	*/
	public class ConfigurationFunctionsTest {

	private static TestingServer testZkServer;
	private static String zookeeperUrl;
	private static CuratorFramework client;
	private static String goodGlobalConfig = slurp( SAMPLE_CONFIG_PATH+ "/global.json");
	private static String goodTestEnrichmentConfig = slurp( SAMPLE_CONFIG_PATH + "/enrichments/test.json");
	private static String goodBroParserConfig = slurp(PARSER_CONFIGS_PATH + "/parsers/bro.json");
	private static String goodTestIndexingConfig = slurp( SAMPLE_CONFIG_PATH + "/indexing/test.json");

	private Context context;
	private JSONParser parser;

	/**
	* {
	* "profiles" : [
	* {
	* "profile" : "counter",
	* "foreach" : "ip_src_addr",
	* "init" : { "counter" : 0 },
	* "update" : { "counter" : "counter + 1" },
	* "result" : "counter"
	* }
	* ],
	* "timestampField" : "timestamp"
	* }
	*/
	@Multiline
	private static String goodProfilerConfig;

	@BeforeAll
	public static void setupZookeeper() throws Exception {

	// zookeeper server
	testZkServer = new TestingServer(true);
	zookeeperUrl = testZkServer.getConnectString();

	// zookeeper client
	client = ConfigurationsUtils.getClient(zookeeperUrl);
	client.start();
	}

	@BeforeEach
	public void setup() throws Exception {

	context = new Context.Builder()
	.with(Context.Capabilities.ZOOKEEPER_CLIENT, () -> client)
	.build();

	parser = new JSONParser();

	// push configs to zookeeper
	pushConfigs(SAMPLE_CONFIG_PATH, zookeeperUrl);
	pushConfigs(PARSER_CONFIGS_PATH, zookeeperUrl);
	writeProfilerConfigToZookeeper(goodProfilerConfig.getBytes(StandardCharsets.UTF_8), client);
	}

	/**
	* Deletes a path within Zookeeper.
	*
	* @param path The path within Zookeeper to delete.
	* @throws Exception
	*/
	private void deletePath(String path) throws Exception {
	client.delete().forPath(path);
	}

	/**
	* Transforms a String to a {@link JSONObject}.
	*
	* @param input The input String to transform
	* @return A {@link JSONObject}.
	* @throws org.json.simple.parser.ParseException
	*/
	private JSONObject toJSONObject(String input) throws org.json.simple.parser.ParseException {

	if(input == null) {
	return null;
	}
	return (JSONObject) parser.parse(input.trim());
	}

	/**
	* Push configuration values to Zookeeper.
	*
	* @param inputPath The local filesystem path to the configurations.
	* @param zookeeperUrl The URL of Zookeeper.
	* @throws Exception
	*/
	private static void pushConfigs(String inputPath, String zookeeperUrl) throws Exception {

	String[] args = new String[] {
	"-z", zookeeperUrl,
	"--mode", "PUSH",
	"--input_dir", inputPath
	};
	CommandLine cli = ConfigurationManager.ConfigurationOptions.parse(new PosixParser(), args);

	ConfigurationManager manager = new ConfigurationManager();
	manager.run(cli);
	}

	/**
	* The CONFIG_GET function should be able to return the Parser configuration
	* for a given sensor.
	*/
	@Test
	public void testGetParser() throws Exception {

	String out = (String) run("CONFIG_GET('PARSER', 'bro')", context);

	SensorParserConfig actual = SensorParserConfig.fromBytes(out.getBytes(StandardCharsets.UTF_8));
	SensorParserConfig expected = SensorParserConfig.fromBytes(goodBroParserConfig.getBytes(
	StandardCharsets.UTF_8));
	assertEquals(expected, actual);
	}

	/**
	* The CONFIG_GET function should NOT return any configuration when the
	* Parser configuration for a given sensor is missing AND emptyIfNotPresent = false.
	*/
	@Test
	public void testGetParserMissWithoutDefault() {

	// expect null because emptyIfNotPresent = false
	Object out = run("CONFIG_GET('PARSER', 'sensor', false)", context);
	assertNull(out);
	}

	/**
	* The CONFIG_GET function should return a default configuration when none
	* currently exists.
	*/
	@Test
	public void testGetParserMissWithDefault() throws Exception {

	SensorParserConfig expected = new SensorParserConfig();
	{
	Object out = run("CONFIG_GET('PARSER', 'sensor')", context);
	SensorParserConfig actual = SensorParserConfig.fromBytes(out.toString().getBytes(
	StandardCharsets.UTF_8));
	assertEquals(expected, actual);
	}
	{
	Object out = run("CONFIG_GET('PARSER', 'sensor', true)", context);
	SensorParserConfig actual = SensorParserConfig.fromBytes(out.toString().getBytes(
	StandardCharsets.UTF_8));
	assertEquals(expected, actual);
	}
	}

	/**
	* The CONFIG_GET function should be able to return the Enrichment configuration
	* for a given sensor.
	*/
	@Test
	public void testGetEnrichment() throws Exception {

	String out = (String) run("CONFIG_GET('ENRICHMENT', 'test')", context);

	SensorEnrichmentConfig actual = SensorEnrichmentConfig.fromBytes(out.getBytes(
	StandardCharsets.UTF_8));
	SensorEnrichmentConfig expected = SensorEnrichmentConfig.fromBytes(goodTestEnrichmentConfig.getBytes(
	StandardCharsets.UTF_8));
	assertEquals(expected, actual);
	}

	/**
	* No default configuration should be provided in this case.
	*/
	@Test
	public void testGetEnrichmentMissWithoutDefault() {

	// expect null because emptyIfNotPresent = false
	Object out = run("CONFIG_GET('ENRICHMENT', 'sense', false)", context);
	assertNull(out);
	}

	/**
	* A default empty configuration should be provided, if one does not exist.
	*/
	@Test
	public void testGetEnrichmentMissWithDefault() throws Exception {

	// expect an empty configuration to be returned
	SensorEnrichmentConfig expected = new SensorEnrichmentConfig();
	{
	String out = (String) run("CONFIG_GET('ENRICHMENT', 'missing-sensor')", context);
	SensorEnrichmentConfig actual = SensorEnrichmentConfig.fromBytes(out.getBytes(
	StandardCharsets.UTF_8));
	assertEquals(expected, actual);
	}
	{
	String out = (String) run("CONFIG_GET('ENRICHMENT', 'missing-sensor', true)", context);
	SensorEnrichmentConfig actual = SensorEnrichmentConfig.fromBytes(out.getBytes(
	StandardCharsets.UTF_8));
	assertEquals(expected, actual);
	}
	}

	/**
	* The CONFIG_GET function should be able to return the Indexing configuration
	* for a given sensor.
	*/
	@Test
	public void testGetIndexing() throws Exception {

	String out = (String) run("CONFIG_GET('INDEXING', 'test')", context);

	Map<String, Object> actual = toJSONObject(out);
	Map<String, Object> expected = toJSONObject(goodTestIndexingConfig);
	assertEquals(expected, actual);
	}

	/**
	* No default configuration should be provided in this case.
	*/
	@Test
	public void testGetIndexingMissWithoutDefault() {

	// expect null because emptyIfNotPresent = false
	Object out = run("CONFIG_GET('INDEXING', 'sense', false)", context);
	assertNull(out);
	}

	/**
	* A default empty configuration should be provided, if one does not exist.
	*/
	@Test
	public void testGetIndexingtMissWithDefault() throws Exception {

	// expect an empty configuration to be returned
	Map<String, Object> expected = Collections.emptyMap();
	{
	String out = (String) run("CONFIG_GET('INDEXING', 'missing-sensor')", context);
	Map<String, Object> actual = toJSONObject(out);
	assertEquals(expected, actual);
	}
	{
	String out = (String) run("CONFIG_GET('INDEXING', 'missing-sensor', true)", context);
	Map<String, Object> actual = toJSONObject(out);
	assertEquals(expected, actual);
	}
	}

	/**
	* The CONFIG_GET function should be able to return the Profiler configuration.
	*/
	@Test
	public void testGetProfiler() throws Exception {

	String out = (String) run("CONFIG_GET('PROFILER')", context);

	ProfilerConfig actual = ProfilerConfig.fromBytes(out.getBytes(StandardCharsets.UTF_8));
	ProfilerConfig expected = ProfilerConfig.fromBytes(goodProfilerConfig.getBytes(
	StandardCharsets.UTF_8));
	assertEquals(expected, actual);
	}

	/**
	* No default configuration should be provided in this case.
	*/
	@Test
	public void testGetProfilerMissWithoutDefault() throws Exception {

	deletePath(PROFILER.getZookeeperRoot());

	// expect null because emptyIfNotPresent = false
	String out = (String) run("CONFIG_GET('PROFILER', false)", context);
	assertNull(out);
	}

	/**
	* A default empty configuration should be provided, if one does not exist.
	*/
	@Test
	public void testGetProfilerMissWithDefault() throws Exception {

	// there is no profiler config in zookeeper
	deletePath(PROFILER.getZookeeperRoot());

	// expect an empty configuration to be returned
	ProfilerConfig expected = new ProfilerConfig();
	{
	String out = (String) run("CONFIG_GET('PROFILER', true)", context);
	ProfilerConfig actual = ProfilerConfig.fromJSON(out);
	assertEquals(expected, actual);
	}
	{
	String out = (String) run("CONFIG_GET('PROFILER')", context);
	ProfilerConfig actual = ProfilerConfig.fromJSON(out);
	assertEquals(expected, actual);
	}
	}

	@Test
	public void testGetGlobal() throws Exception {

	String out = (String) run("CONFIG_GET('GLOBAL')", context);

	Map<String, Object> actual = toJSONObject(out);
	Map<String, Object> expected = toJSONObject(goodGlobalConfig);
	assertEquals(expected, actual);
	}

	/**
	* No default configuration should be provided in this case.
	*/
	@Test
	public void testGetGlobalMissWithoutDefault() throws Exception {

	// there is no global config in zookeeper
	deletePath(GLOBAL.getZookeeperRoot());

	// expect null because emptyIfNotPresent = false
	Object out = run("CONFIG_GET('GLOBAL', false)", context);
	assertNull(out);
	}

	/**
	* A default empty configuration should be provided, if one does not exist.
	*/
	@Test
	public void testGetGlobalMissWithDefault() throws Exception {

	// there is no global config in zookeeper
	deletePath(GLOBAL.getZookeeperRoot());

	// expect an empty configuration to be returned
	Map<String, Object> expected = Collections.emptyMap();
	{
	String out = (String) run("CONFIG_GET('GLOBAL')", context);
	Map<String, Object> actual = toJSONObject(out);
	assertEquals(expected, actual);
	}
	{
	String out = (String) run("CONFIG_GET('GLOBAL', true)", context);
	Map<String, Object> actual = toJSONObject(out);
	assertEquals(expected, actual);
	}
	}

	@Test
	public void testPutGlobal() throws Exception {

	String out = (String) run("CONFIG_GET('GLOBAL')", context);

	Map<String, Object> actual = toJSONObject(out);
	Map<String, Object> expected = toJSONObject(goodGlobalConfig);
	assertEquals(expected, actual);
	}

	@Test
	public void testPutGlobalBad() {
	{
	UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.FATAL);
	assertThrows(ParseException.class, () -> run("CONFIG_PUT('GLOBAL', 'foo bar')", context));
	UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.ERROR);
	}
	}

	@Test
	public void testPutIndexing() throws InterruptedException {
	String brop= (String) run("CONFIG_GET('INDEXING', 'testIndexingPut')", context);
	run("CONFIG_PUT('INDEXING', config, 'testIndexingPut')", ImmutableMap.of("config", brop), context);
	boolean foundMatch = false;
	for(int i = 0;i < 10 && !foundMatch;++i) {
	String bropNew = (String) run("CONFIG_GET('INDEXING', 'testIndexingPut', false)", context);
	foundMatch = brop.equals(bropNew);
	if(foundMatch) {
	break;
	}
	Thread.sleep(2000);
	}
	assertTrue(foundMatch);
	}

	@Test
	public void testPutIndexingBad() {
	{
	{
	UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.FATAL);
	assertThrows(
	ParseException.class,
	() ->
	run(
	"CONFIG_PUT('INDEXING', config, 'brop')",
	ImmutableMap.of("config", "foo bar"),
	context));
	UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.ERROR);
	}
	}
	}

	@Test
	public void testPutEnrichment() throws InterruptedException {
	String config = (String) run("CONFIG_GET('ENRICHMENT', 'sensor')", context);
	assertNotNull(config);

	run("CONFIG_PUT('ENRICHMENT', config, 'sensor')", ImmutableMap.of("config", config), context);

	boolean foundMatch = false;
	for(int i = 0;i < 10 && !foundMatch;++i) {
	String newConfig = (String) run("CONFIG_GET('ENRICHMENT', 'sensor', false)", context);
	foundMatch = config.equals(newConfig);
	if(foundMatch) {
	break;
	}
	Thread.sleep(2000);
	}
	assertTrue(foundMatch);
	}

	@Test
	public void testPutEnrichmentBad() {
	{
	{
	UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.FATAL);
	assertThrows(
	ParseException.class,
	() ->
	run(
	"CONFIG_PUT('ENRICHMENT', config, 'brop')",
	ImmutableMap.of("config", "foo bar"),
	context));
	UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.ERROR);
	}
	}
	}

	@Test
	public void testPutParser() throws InterruptedException {
	String brop= (String) run("CONFIG_GET('PARSER', 'testParserPut')", context);
	run("CONFIG_PUT('PARSER', config, 'testParserPut')", ImmutableMap.of("config", brop), context);
	boolean foundMatch = false;
	for(int i = 0;i < 10 && !foundMatch;++i) {
	String bropNew = (String) run("CONFIG_GET('PARSER', 'testParserPut', false)", context);
	foundMatch = brop.equals(bropNew);
	if(foundMatch) {
	break;
	}
	Thread.sleep(2000);
	}
	assertTrue(foundMatch);
	}

	@Test
	public void testPutParserBad() {
	{
	UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.FATAL);
	assertThrows(
	ParseException.class,
	() ->
	run(
	"CONFIG_PUT('PARSER', config, 'brop')",
	ImmutableMap.of("config", "foo bar"),
	context));
	UnitTestHelper.setLog4jLevel(ConfigurationFunctions.class, Level.ERROR);
	}
	}
	}