/*
 * 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.ambari.server.serveraction.upgrades;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import jakarta.persistence.EntityManager;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.H2DatabaseCleaner;
import org.apache.ambari.server.ServiceComponentNotFoundException;
import org.apache.ambari.server.ServiceNotFoundException;
import org.apache.ambari.server.actionmanager.ExecutionCommandWrapper;
import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleCommandFactory;
import org.apache.ambari.server.agent.CommandReport;
import org.apache.ambari.server.agent.ExecutionCommand;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
import org.apache.ambari.server.orm.OrmTestHelper;
import org.apache.ambari.server.orm.dao.RequestDAO;
import org.apache.ambari.server.orm.dao.UpgradeDAO;
import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
import org.apache.ambari.server.orm.entities.RequestEntity;
import org.apache.ambari.server.orm.entities.UpgradeEntity;
import org.apache.ambari.server.orm.entities.UpgradeHistoryEntity;
import org.apache.ambari.server.serveraction.ServerAction;
import org.apache.ambari.server.stack.upgrade.ConfigUpgradeChangeDefinition.ConfigurationKeyValue;
import org.apache.ambari.server.stack.upgrade.ConfigUpgradeChangeDefinition.IfValueMatchType;
import org.apache.ambari.server.stack.upgrade.ConfigUpgradeChangeDefinition.Insert;
import org.apache.ambari.server.stack.upgrade.ConfigUpgradeChangeDefinition.InsertType;
import org.apache.ambari.server.stack.upgrade.ConfigUpgradeChangeDefinition.Replace;
import org.apache.ambari.server.stack.upgrade.ConfigUpgradeChangeDefinition.Transfer;
import org.apache.ambari.server.stack.upgrade.ConfigureTask;
import org.apache.ambari.server.stack.upgrade.PropertyKeyState;
import org.apache.ambari.server.stack.upgrade.TransferCoercionType;
import org.apache.ambari.server.stack.upgrade.TransferOperation;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Config;
import org.apache.ambari.server.state.ConfigFactory;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentFactory;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.ServiceComponentHostFactory;
import org.apache.ambari.server.state.ServiceFactory;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.State;
import org.apache.ambari.spi.upgrade.UpgradeType;
import org.apache.commons.lang3.StringUtils;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import com.google.gson.Gson;
import com.google.inject.Guice;
import com.google.inject.Inject;
import com.google.inject.Injector;

/**
 * Tests upgrade-related server side actions
 */
public class ConfigureActionTest {

  @Inject
  private Injector m_injector;

  @Inject
  private OrmTestHelper m_helper;

  @Inject
  private HostRoleCommandFactory hostRoleCommandFactory;

  @Inject
  private ServiceFactory serviceFactory;

  @Inject
  private ConfigHelper m_configHelper;

  @Inject
  private Clusters clusters;

  @Inject
  private ConfigFactory configFactory;

  @Inject
  private ConfigureAction action;

  @Inject
  private RequestDAO requestDAO;

  @Inject
  private UpgradeDAO upgradeDAO;

  @Inject
  private ServiceComponentFactory serviceComponentFactory;

  @Inject
  private ServiceComponentHostFactory serviceComponentHostFactory;

  private RepositoryVersionEntity repoVersion2110;
  private RepositoryVersionEntity repoVersion2111;
  private RepositoryVersionEntity repoVersion2200;

  private final Map<String, Map<String, String>> NO_ATTRIBUTES = new HashMap<>();

  @Before
  public void setup() throws Exception {
    m_injector = Guice.createInjector(new InMemoryDefaultTestModule());
    m_injector.getInstance(GuiceJpaInitializer.class);
    m_injector.injectMembers(this);

    repoVersion2110 = m_helper.getOrCreateRepositoryVersion(new StackId("HDP-2.1.1"), "2.1.1.0-1234");
    repoVersion2111 = m_helper.getOrCreateRepositoryVersion(new StackId("HDP-2.1.1"), "2.1.1.1-5678");
    repoVersion2200 = m_helper.getOrCreateRepositoryVersion(new StackId("HDP-2.2.0"), "2.2.0.0-1234");

    makeUpgradeCluster();
  }

  @After
  public void teardown() throws Exception {
    H2DatabaseCleaner.clearDatabase(m_injector.getProvider(EntityManager.class).get());
  }


  /**
   * Tests that a new configuration is created when upgrading across stack when
   * there is no existing configuration with the correct target stack.
   *
   * @throws Exception
   */
  @Test
  public void testNewConfigCreatedWhenUpgradingAcrossStacks() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("initLimit", "10");
      }
    };

    Config config = createConfig(c, repoVersion2110, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    List<ConfigurationKeyValue> configurations = new ArrayList<>();
    ConfigurationKeyValue keyValue = new ConfigurationKeyValue();
    configurations.add(keyValue);
    keyValue.key = "initLimit";
    keyValue.value = "11";

    createUpgrade(c, repoVersion2200);

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");
    commandParams.put(ConfigureTask.PARAMETER_KEY_VALUE_PAIRS, new Gson().toJson(configurations));

    ExecutionCommand executionCommand = getExecutionCommand(commandParams);
    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null,
        null, null);

    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(
        executionCommand));

    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    assertEquals(3, c.getConfigsByType("zoo.cfg").size());

    config = c.getDesiredConfigByType("zoo.cfg");
    assertNotNull(config);
    assertFalse(StringUtils.equals("version2", config.getTag()));
    assertEquals("11", config.getProperties().get("initLimit"));
  }

  /**
   * Tests that if a configuration with the target stack already exists, then it
   * will be re-used instead of a new one created.
   *
   * @throws Exception
   */
  @Test
  public void testConfigurationWithTargetStackUsed() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("initLimit", "10");
      }
    };

    Config config = createConfig(c, repoVersion2200, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    List<ConfigurationKeyValue> configurations = new ArrayList<>();
    ConfigurationKeyValue keyValue = new ConfigurationKeyValue();
    configurations.add(keyValue);
    keyValue.key = "initLimit";
    keyValue.value = "11";

    createUpgrade(c, repoVersion2200);

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");
    commandParams.put(ConfigureTask.PARAMETER_KEY_VALUE_PAIRS, new Gson().toJson(configurations));

    ExecutionCommand executionCommand = getExecutionCommand(commandParams);
    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null, null, null);

    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(executionCommand));

    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    config = c.getDesiredConfigByType("zoo.cfg");
    assertNotNull(config);
    assertEquals("version2", config.getTag());
    assertEquals("11", config.getProperties().get("initLimit"));
  }

  /**
   * Tests that DELETE "*" with edit preserving works correctly.
   *
   * @throws Exception
   */
  @Test
  public void testDeletePreserveChanges() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    // create a config for zoo.cfg with two values; one is a stack value and the
    // other is custom
    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("tickTime", "2000");
        put("foo", "bar");
      }
    };

    Config config = createConfig(c, repoVersion2110, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");

    createUpgrade(c, repoVersion2111);

    // delete all keys, preserving edits or additions
    List<Transfer> transfers = new ArrayList<>();
    Transfer transfer = new Transfer();
    transfer.operation = TransferOperation.DELETE;
    transfer.deleteKey = "*";
    transfer.preserveEdits = true;
    transfers.add(transfer);

    commandParams.put(ConfigureTask.PARAMETER_TRANSFERS, new Gson().toJson(transfers));

    ExecutionCommand executionCommand = getExecutionCommand(commandParams);
    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null, null, null);
    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(executionCommand));

    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    // make sure there are now 3 versions after the merge
    assertEquals(3, c.getConfigsByType("zoo.cfg").size());
    config = c.getDesiredConfigByType("zoo.cfg");
    assertNotNull(config);
    assertFalse("version2".equals(config.getTag()));

    // time to check our values; there should only be 1 left since tickTime was
    // removed
    Map<String, String> map = config.getProperties();
    assertEquals("bar", map.get("foo"));
    assertFalse(map.containsKey("tickTime"));
  }

  @Test
  public void testConfigTransferCopy() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("initLimit", "10");
        put("copyIt", "10");
        put("moveIt", "10");
        put("deleteIt", "10");
      }
    };

    Config config = createConfig(c, repoVersion2110, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    List<ConfigurationKeyValue> configurations = new ArrayList<>();
    ConfigurationKeyValue keyValue = new ConfigurationKeyValue();
    configurations.add(keyValue);
    keyValue.key = "initLimit";
    keyValue.value = "11";

    createUpgrade(c, repoVersion2111);

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");
    commandParams.put(ConfigureTask.PARAMETER_KEY_VALUE_PAIRS, new Gson().toJson(configurations));

    // normal copy
    List<Transfer> transfers = new ArrayList<>();
    Transfer transfer = new Transfer();
    transfer.operation = TransferOperation.COPY;
    transfer.fromKey = "copyIt";
    transfer.toKey = "copyKey";
    transfers.add(transfer);

    // copy with default
    transfer = new Transfer();
    transfer.operation = TransferOperation.COPY;
    transfer.fromKey = "copiedFromMissingKeyWithDefault";
    transfer.toKey = "copiedToMissingKeyWithDefault";
    transfer.defaultValue = "defaultValue";
    transfers.add(transfer);

    // normal move
    transfer = new Transfer();
    transfer.operation = TransferOperation.MOVE;
    transfer.fromKey = "moveIt";
    transfer.toKey = "movedKey";
    transfers.add(transfer);

    // move with default
    transfer = new Transfer();
    transfer.operation = TransferOperation.MOVE;
    transfer.fromKey = "movedFromKeyMissingWithDefault";
    transfer.toKey = "movedToMissingWithDefault";
    transfer.defaultValue = "defaultValue2";
    transfer.mask = true;
    transfers.add(transfer);

    transfer = new Transfer();
    transfer.operation = TransferOperation.DELETE;
    transfer.deleteKey = "deleteIt";
    transfers.add(transfer);

    commandParams.put(ConfigureTask.PARAMETER_TRANSFERS, new Gson().toJson(transfers));

    ExecutionCommand executionCommand = getExecutionCommand(commandParams);
    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null,
        null, null);

    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(
        executionCommand));


    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    assertEquals(3, c.getConfigsByType("zoo.cfg").size());

    config = c.getDesiredConfigByType("zoo.cfg");
    assertNotNull(config);
    assertFalse("version2".equals(config.getTag()));

    Map<String, String> map = config.getProperties();
    assertEquals("11", map.get("initLimit"));
    assertEquals("10", map.get("copyIt"));
    assertTrue(map.containsKey("copyKey"));
    assertEquals(map.get("copyIt"), map.get("copyKey"));
    assertFalse(map.containsKey("moveIt"));
    assertTrue(map.containsKey("movedKey"));
    assertFalse(map.containsKey("deletedKey"));
    assertTrue(map.containsKey("copiedToMissingKeyWithDefault"));
    assertEquals("defaultValue", map.get("copiedToMissingKeyWithDefault"));
    assertTrue(map.containsKey("movedToMissingWithDefault"));
    assertEquals("defaultValue2", map.get("movedToMissingWithDefault"));

    transfers.clear();
    transfer = new Transfer();
    transfer.operation = TransferOperation.DELETE;
    transfer.deleteKey = "*";
    transfer.preserveEdits = true;
    transfer.keepKeys.add("copyKey");
    // The below key should be ignored/not added as it doesn't exist originally as part of transfer.
    transfer.keepKeys.add("keyNotExisting");
    // The 'null' passed as part of key should be ignored as part of transfer operation.
    transfer.keepKeys.add(null);


    transfers.add(transfer);
    commandParams.put(ConfigureTask.PARAMETER_TRANSFERS, new Gson().toJson(transfers));

    report = action.execute(null);
    assertNotNull(report);

    assertEquals(4, c.getConfigsByType("zoo.cfg").size());
    config = c.getDesiredConfigByType("zoo.cfg");
    map = config.getProperties();
    assertEquals(6, map.size());
    assertTrue(map.containsKey("initLimit")); // it just changed to 11 from 10
    assertTrue(map.containsKey("copyKey")); // is new
    // Below two keys should not have been added in the map.
    assertFalse(map.containsKey("keyNotExisting"));
    assertFalse(map.containsKey(null));
  }

  @Test
  public void testCoerceValueOnCopy() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("zoo.server.csv", "c6401,c6402,  c6403");
      }
    };

    Config config = createConfig(c, repoVersion2110, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    createUpgrade(c, repoVersion2111);

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");

    // copy with coerce
    List<Transfer> transfers = new ArrayList<>();
    Transfer transfer = new Transfer();
    transfer.operation = TransferOperation.COPY;
    transfer.coerceTo = TransferCoercionType.YAML_ARRAY;
    transfer.fromKey = "zoo.server.csv";
    transfer.toKey = "zoo.server.array";
    transfer.defaultValue = "['foo','bar']";
    transfers.add(transfer);

    commandParams.put(ConfigureTask.PARAMETER_TRANSFERS, new Gson().toJson(transfers));

    ExecutionCommand executionCommand = getExecutionCommand(commandParams);
    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null, null, null);

    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(executionCommand));


    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    assertEquals(3, c.getConfigsByType("zoo.cfg").size());

    config = c.getDesiredConfigByType("zoo.cfg");
    assertNotNull(config);
    assertFalse("version2".equals(config.getTag()));

    Map<String, String> map = config.getProperties();
    assertEquals("c6401,c6402,  c6403", map.get("zoo.server.csv"));
    assertEquals("['c6401','c6402','c6403']", map.get("zoo.server.array"));
  }

  @Test
  public void testValueReplacement() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("key_to_replace", "My New Cat");
        put("key_with_no_match", "WxyAndZ");
      }
    };

    Config config = createConfig(c, repoVersion2110, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    createUpgrade(c, repoVersion2111);

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");

    // Replacement task
    List<Replace> replacements = new ArrayList<>();
    Replace replace = new Replace();
    replace.key = "key_to_replace";
    replace.find = "New Cat";
    replace.replaceWith = "Wet Dog";
    replacements.add(replace);

    replace = new Replace();
    replace.key = "key_with_no_match";
    replace.find = "abc";
    replace.replaceWith = "def";
    replacements.add(replace);

    commandParams.put(ConfigureTask.PARAMETER_REPLACEMENTS, new Gson().toJson(replacements));

    ExecutionCommand executionCommand = getExecutionCommand(commandParams);
    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null, null, null);

    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(executionCommand));


    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    assertEquals(3, c.getConfigsByType("zoo.cfg").size());

    config = c.getDesiredConfigByType("zoo.cfg");
    assertNotNull(config);
    assertFalse("version2".equals(config.getTag()));

    assertEquals("My Wet Dog", config.getProperties().get("key_to_replace"));
    assertEquals("WxyAndZ", config.getProperties().get("key_with_no_match"));
  }

  /**
   * Tests that replacing a {@code null} value works.
   *
   * @throws Exception
   */
  @Test
  public void testValueReplacementWithMissingConfigurations() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("existing", "This exists!");
        put("missing", null);
      }
    };

    Config config = createConfig(c, repoVersion2110, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    createUpgrade(c, repoVersion2111);

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");

    // Replacement task
    List<Replace> replacements = new ArrayList<>();
    Replace replace = new Replace();
    replace.key = "missing";
    replace.find = "foo";
    replace.replaceWith = "bar";
    replacements.add(replace);

    commandParams.put(ConfigureTask.PARAMETER_REPLACEMENTS, new Gson().toJson(replacements));

    ExecutionCommand executionCommand = new ExecutionCommand();
    executionCommand.setCommandParams(commandParams);
    executionCommand.setClusterName("c1");
    executionCommand.setRoleParams(new HashMap<>());
    executionCommand.getRoleParams().put(ServerAction.ACTION_USER_NAME, "username");

    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null, null, null);

    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(executionCommand));


    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    assertEquals(3, c.getConfigsByType("zoo.cfg").size());

    config = c.getDesiredConfigByType("zoo.cfg");
    assertEquals(null, config.getProperties().get("missing"));
  }

  @Test
  public void testMultipleKeyValuesPerTask() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("fooKey", "barValue");
      }
    };

    Config config = createConfig(c, repoVersion2200, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    // create several configurations
    List<ConfigurationKeyValue> configurations = new ArrayList<>();
    ConfigurationKeyValue fooKey2 = new ConfigurationKeyValue();
    configurations.add(fooKey2);
    fooKey2.key = "fooKey2";
    fooKey2.value = "barValue2";

    ConfigurationKeyValue fooKey3 = new ConfigurationKeyValue();
    configurations.add(fooKey3);
    fooKey3.key = "fooKey3";
    fooKey3.value = "barValue3";
    fooKey3.mask = true;

    createUpgrade(c, repoVersion2200);

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");
    commandParams.put(ConfigureTask.PARAMETER_KEY_VALUE_PAIRS, new Gson().toJson(configurations));

    ExecutionCommand executionCommand = getExecutionCommand(commandParams);
    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null, null, null);
    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(executionCommand));


    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    config = c.getDesiredConfigByType("zoo.cfg");
    assertNotNull(config);
    assertEquals("barValue", config.getProperties().get("fooKey"));
    assertEquals("barValue2", config.getProperties().get("fooKey2"));
    assertEquals("barValue3", config.getProperties().get("fooKey3"));
    assertTrue(report.getStdOut().contains("******"));

  }

  @Test
  public void testAllowedSet() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("set.key.1", "s1");
        put("set.key.2", "s2");
        put("set.key.3", "s3");
        put("set.key.4", "s4");
      }
    };

    Config config = createConfig(c, repoVersion2200, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    // create several configurations
    List<ConfigurationKeyValue> configurations = new ArrayList<>();
    ConfigurationKeyValue fooKey1 = new ConfigurationKeyValue();
    configurations.add(fooKey1);
    fooKey1.key = "fooKey1";
    fooKey1.value = "barValue1";

    ConfigurationKeyValue fooKey2 = new ConfigurationKeyValue();
    configurations.add(fooKey2);
    fooKey2.key = "fooKey2";
    fooKey2.value = "barValue2";

    ConfigurationKeyValue fooKey3 = new ConfigurationKeyValue();
    configurations.add(fooKey3);
    fooKey3.key = "fooKey3";
    fooKey3.value = "barValue3";
    fooKey3.ifKey = "set.key.1";
    fooKey3.ifType = "zoo.cfg";
    fooKey3.ifValue = "s1";

    ConfigurationKeyValue fooKey4 = new ConfigurationKeyValue();
    configurations.add(fooKey4);
    fooKey4.key = "fooKey4";
    fooKey4.value = "barValue4";
    fooKey4.ifKey = "set.key.2";
    fooKey4.ifType = "zoo.cfg";
    fooKey4.ifKeyState= PropertyKeyState.PRESENT;

    ConfigurationKeyValue fooKey5 = new ConfigurationKeyValue();
    configurations.add(fooKey5);
    fooKey5.key = "fooKey5";
    fooKey5.value = "barValue5";
    fooKey5.ifKey = "abc";
    fooKey5.ifType = "zoo.cfg";
    fooKey5.ifKeyState= PropertyKeyState.ABSENT;

    createUpgrade(c, repoVersion2200);

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");
    commandParams.put(ConfigureTask.PARAMETER_KEY_VALUE_PAIRS, new Gson().toJson(configurations));

    ExecutionCommand executionCommand = getExecutionCommand(commandParams);
    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null, null, null);
    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(executionCommand));


    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    config = c.getDesiredConfigByType("zoo.cfg");
    assertNotNull(config);
    assertEquals("barValue1", config.getProperties().get("fooKey1"));
    assertEquals("barValue2", config.getProperties().get("fooKey2"));
    assertEquals("barValue3", config.getProperties().get("fooKey3"));
    assertEquals("barValue4", config.getProperties().get("fooKey4"));
    assertEquals("barValue5", config.getProperties().get("fooKey5"));
    assertEquals("s1", config.getProperties().get("set.key.1"));
    assertEquals("s2", config.getProperties().get("set.key.2"));
    assertEquals("s3", config.getProperties().get("set.key.3"));
    assertEquals("s4", config.getProperties().get("set.key.4"));
  }

  @Test
  public void testDisallowedSet() throws Exception {
    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("set.key.1", "s1");
        put("set.key.2", "s2");
        put("set.key.3", "s3");
        put("set.key.4", "s4");
      }
    };

    Config config = createConfig(c, repoVersion2200, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    // create several configurations
    List<ConfigurationKeyValue> configurations = new ArrayList<>();
    ConfigurationKeyValue fooKey3 = new ConfigurationKeyValue();
    configurations.add(fooKey3);
    fooKey3.key = "fooKey3";
    fooKey3.value = "barValue3";
    fooKey3.ifKey = "set.key.1";
    fooKey3.ifType = "zoo.cfg";
    fooKey3.ifValue = "no-such-value";

    ConfigurationKeyValue fooKey4 = new ConfigurationKeyValue();
    configurations.add(fooKey4);
    fooKey4.key = "fooKey4";
    fooKey4.value = "barValue4";
    fooKey4.ifKey = "set.key.2";
    fooKey4.ifType = "zoo.cfg";
    fooKey4.ifKeyState= PropertyKeyState.ABSENT;

    ConfigurationKeyValue fooKey5 = new ConfigurationKeyValue();
    configurations.add(fooKey5);
    fooKey5.key = "fooKey5";
    fooKey5.value = "barValue5";
    fooKey5.ifKey = "abc";
    fooKey5.ifType = "zoo.cfg";
    fooKey5.ifKeyState= PropertyKeyState.PRESENT;

    createUpgrade(c, repoVersion2200);

    Map<String, String> commandParams = new HashMap<>();
    commandParams.put("clusterName", "c1");
    commandParams.put(ConfigureTask.PARAMETER_CONFIG_TYPE, "zoo.cfg");
    commandParams.put(ConfigureTask.PARAMETER_KEY_VALUE_PAIRS, new Gson().toJson(configurations));

    ExecutionCommand executionCommand = getExecutionCommand(commandParams);
    HostRoleCommand hostRoleCommand = hostRoleCommandFactory.create(null, null, null, null);
    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(executionCommand));


    action.setExecutionCommand(executionCommand);
    action.setHostRoleCommand(hostRoleCommand);

    CommandReport report = action.execute(null);
    assertNotNull(report);

    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    config = c.getDesiredConfigByType("zoo.cfg");
    assertNotNull(config);
    assertEquals("s1", config.getProperties().get("set.key.1"));
    assertEquals("s2", config.getProperties().get("set.key.2"));
    assertEquals("s3", config.getProperties().get("set.key.3"));
    assertEquals("s4", config.getProperties().get("set.key.4"));
    assertFalse(config.getProperties().containsKey("fooKey3"));
    assertFalse(config.getProperties().containsKey("fooKey4"));
    assertFalse(config.getProperties().containsKey("fooKey5"));
  }

  @Test
  public void testAllowedReplacment() throws Exception {


    Cluster c = clusters.getCluster("c1");
    assertEquals(1, c.getConfigsByType("zoo.cfg").size());

    Map<String, String> properties = new HashMap<String, String>() {
      {
        put("replace.key.1", "r1");
        put("replace.key.2", "r2");
        put("replace.key.3", "r3a1");
        put("replace.key.4", "r4");
        put("replace.key.5", "r5");
      }
    };

    Config config = createConfig(c, repoVersion2200, "zoo.cfg", "version2", properties);

    c.addDesiredConfig("user", Collections.singleton(config));
    assertEquals(2, c.getConfigsByType("zoo.cfg").size());

    // create several configurations
    List<Replace> replacements = new ArrayList<>();
    Replace replace = new Replace();
    replace.key = "replace.key.3";
    replace.find = "a";
    replace.replaceWith = "A";
    replacements.add(replace);

    Replace replace2 = new Replace();
    replacements.add(replace2);
    replace2.key = "replace.key.4";
    replace2.find = "r";
    replace2.replaceWith = "R";
    replace2.ifKey = "replace.key.1";
    replace2.ifType = "zoo.cfg";
    replace2.ifValue = "r1";
    replacements.add(replace2);

    Replace replace3 = new Replace();
    replacements.add(replace3);
    replace3.key = "replace.key.2";
    replace3.find = "r";
    replace3.replaceWith = "R";
