/*
 * 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.controller.internal;

import static com.google.common.collect.Iterators.peekingIterator;
import static java.util.Collections.emptyList;
import static java.util.Collections.emptyMap;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toSet;
import static org.apache.ambari.server.state.ConfigHelper.CLUSTER_ENV_STACK_FEATURES_PROPERTY;
import static org.apache.ambari.server.state.ConfigHelper.CLUSTER_ENV_STACK_NAME_PROPERTY;
import static org.apache.ambari.server.state.ConfigHelper.CLUSTER_ENV_STACK_PACKAGES_PROPERTY;
import static org.apache.ambari.server.state.ConfigHelper.CLUSTER_ENV_STACK_ROOT_PROPERTY;
import static org.apache.ambari.server.state.ConfigHelper.CLUSTER_ENV_STACK_TOOLS_PROPERTY;
import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.reset;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariServer;
import org.apache.ambari.server.controller.KerberosHelper;
import org.apache.ambari.server.controller.StackConfigurationResponse;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.PropertyDependencyInfo;
import org.apache.ambari.server.state.PropertyInfo;
import org.apache.ambari.server.state.ServiceInfo;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.ValueAttributesInfo;
import org.apache.ambari.server.state.kerberos.KerberosDescriptor;
import org.apache.ambari.server.topology.AdvisedConfiguration;
import org.apache.ambari.server.topology.AmbariContext;
import org.apache.ambari.server.topology.Blueprint;
import org.apache.ambari.server.topology.Cardinality;
import org.apache.ambari.server.topology.ClusterTopology;
import org.apache.ambari.server.topology.ClusterTopologyImpl;
import org.apache.ambari.server.topology.Component;
import org.apache.ambari.server.topology.ConfigRecommendationStrategy;
import org.apache.ambari.server.topology.Configuration;
import org.apache.ambari.server.topology.HostGroup;
import org.apache.ambari.server.topology.HostGroupImpl;
import org.apache.ambari.server.topology.HostGroupInfo;
import org.apache.ambari.server.topology.InvalidTopologyException;
import org.apache.ambari.server.topology.TopologyRequest;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.easymock.EasyMock;
import org.easymock.EasyMockRule;
import org.easymock.EasyMockSupport;
import org.easymock.Mock;
import org.easymock.MockType;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.api.easymock.PowerMock;
import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.PeekingIterator;
import com.google.common.collect.Sets;

/**
 * BlueprintConfigurationProcessor unit tests.
 */
@RunWith(PowerMockRunner.class)
@PrepareForTest(AmbariServer.class)
@PowerMockIgnore({"org.apache.logging.log4j.*","org.slf4j.*"})
public class BlueprintConfigurationProcessorTest extends EasyMockSupport {

  private static final Configuration EMPTY_CONFIG = new Configuration(emptyMap(), emptyMap());
  private final Map<String, Collection<String>> serviceComponents = new HashMap<>();
  private final Map<String, String> defaultClusterEnvProperties = new HashMap<>();

  private final String STACK_NAME = "testStack";
  private final String STACK_VERSION = "1";

  private final String CLUSTER_ENV_PROP = "cluster-env";

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

  @Rule
  public EasyMockRule mocks = new EasyMockRule(this);

  @Mock(type = MockType.NICE)
  private AmbariContext ambariContext;

  @Mock(type = MockType.NICE)
  private Blueprint bp;

  @Mock(type = MockType.NICE)
  private ServiceInfo serviceInfo;

  @Mock(type = MockType.NICE)
  private Stack stack;

  @Mock(type = MockType.NICE)
  private AmbariManagementController controller;

  @Mock(type = MockType.NICE)
  private KerberosHelper kerberosHelper;

  @Mock(type = MockType.NICE)
  private KerberosDescriptor kerberosDescriptor;

  @Mock(type = MockType.NICE)
  private Clusters clusters;

  @Mock(type = MockType.NICE)
  private Cluster cluster;

  @Mock
  private TopologyRequest topologyRequestMock;

  @Mock(type = MockType.NICE)
  private ConfigHelper configHelper;

  @Before
  public void init() throws Exception {
    expect(bp.getStack()).andReturn(stack).anyTimes();
    expect(bp.getName()).andReturn("test-bp").anyTimes();
    expect(bp.getServiceInfos()).andReturn(Collections.emptyList()).anyTimes();

    expect(stack.getName()).andReturn(STACK_NAME).atLeastOnce();
    expect(stack.getVersion()).andReturn(STACK_VERSION).atLeastOnce();
    // return false for all components since for this test we don't care about the value
    expect(stack.isMasterComponent((String) anyObject())).andReturn(false).anyTimes();
    expect(stack.getConfigurationPropertiesWithMetadata(anyObject(String.class), anyObject(String.class))).andReturn(emptyMap()).anyTimes();
    expect(stack.getConfiguration()).andReturn(Configuration.newEmpty()).anyTimes();

    expect(serviceInfo.getRequiredProperties()).andReturn(
      emptyMap()).anyTimes();
    expect(serviceInfo.getRequiredServices()).andReturn(Collections.emptyList()).anyTimes();

    Collection<String> hdfsComponents = new HashSet<>();
    hdfsComponents.add("NAMENODE");
    hdfsComponents.add("SECONDARY_NAMENODE");
    hdfsComponents.add("DATANODE");
    hdfsComponents.add("HDFS_CLIENT");
    serviceComponents.put("HDFS", hdfsComponents);

    Collection<String> yarnComponents = new HashSet<>();
    yarnComponents.add("RESOURCEMANAGER");
    yarnComponents.add("NODEMANAGER");
    yarnComponents.add("YARN_CLIENT");
    yarnComponents.add("APP_TIMELINE_SERVER");
    yarnComponents.add("TIMELINE_READER");
    serviceComponents.put("YARN", yarnComponents);

    Collection<String> mrComponents = new HashSet<>();
    mrComponents.add("MAPREDUCE2_CLIENT");
    mrComponents.add("HISTORY_SERVER");
    serviceComponents.put("MAPREDUCE2", mrComponents);

    Collection<String> zkComponents = new HashSet<>();
    zkComponents.add("ZOOKEEPER_SERVER");
    zkComponents.add("ZOOKEEPER_CLIENT");
    serviceComponents.put("ZOOKEEPER", zkComponents);

    Collection<String> hiveComponents = new HashSet<>();
    hiveComponents.add("MYSQL_SERVER");
    hiveComponents.add("HIVE_METASTORE");
    hiveComponents.add("HIVE_SERVER");
    serviceComponents.put("HIVE", hiveComponents);

    Collection<String> falconComponents = new HashSet<>();
    falconComponents.add("FALCON_SERVER");
    falconComponents.add("FALCON_CLIENT");
    serviceComponents.put("FALCON", falconComponents);

    Collection<String> gangliaComponents = new HashSet<>();
    gangliaComponents.add("GANGLIA_SERVER");
    gangliaComponents.add("GANGLIA_CLIENT");
    serviceComponents.put("GANGLIA", gangliaComponents);

    Collection<String> kafkaComponents = new HashSet<>();
    kafkaComponents.add("KAFKA_BROKER");
    serviceComponents.put("KAFKA", kafkaComponents);

    Collection<String> knoxComponents = new HashSet<>();
    knoxComponents.add("KNOX_GATEWAY");
    serviceComponents.put("KNOX", knoxComponents);

    Collection<String> oozieComponents = new HashSet<>();
    oozieComponents.add("OOZIE_SERVER");
    oozieComponents.add("OOZIE_CLIENT");
    serviceComponents.put("OOZIE", oozieComponents);

    Collection<String> hbaseComponents = new HashSet<>();
    hbaseComponents.add("HBASE_MASTER");
    serviceComponents.put("HBASE", hbaseComponents);

    Collection<String> atlasComponents = new HashSet<>();
    atlasComponents.add("ATLAS_SERVER");
    atlasComponents.add("ATLAS_CLIENT");
    serviceComponents.put("ATLAS", atlasComponents);

    Collection<String> amsComponents = new HashSet<>();
    amsComponents.add("METRICS_COLLECTOR");
    serviceComponents.put("AMBARI_METRICS", amsComponents);

    Collection<String> stormComponents = new HashSet<>();
    stormComponents.add("NIMBUS");
    serviceComponents.put("STORM", stormComponents);

    Collection<String> rangerComponents = new HashSet<>();
    rangerComponents.add("RANGER_ADMIN");
    rangerComponents.add("RANGER_USERSYNC");
    rangerComponents.add("RANGER_TAGSYNC");
    serviceComponents.put("RANGER",rangerComponents);

    for (Map.Entry<String, Collection<String>> entry : serviceComponents.entrySet()) {
      String service = entry.getKey();
      for (String component : entry.getValue()) {
        expect(stack.getServiceForComponent(component)).andReturn(service).anyTimes();
      }
    }

    expect(stack.getCardinality("MYSQL_SERVER")).andReturn(new Cardinality("0-1")).anyTimes();

    Set<String> emptySet = Collections.emptySet();
    expect(stack.getExcludedConfigurationTypes(anyObject(String.class))).andReturn(emptySet).anyTimes();

    expect(ambariContext.getConfigHelper()).andReturn(configHelper).anyTimes();
    expect(configHelper.getDefaultStackProperties(
        EasyMock.eq(new StackId(STACK_NAME, STACK_VERSION)))).andReturn(stackProperties).anyTimes();

   stackProperties.put(ConfigHelper.CLUSTER_ENV, defaultClusterEnvProperties);


    expect(ambariContext.isClusterKerberosEnabled(1)).andReturn(true).once();
    expect(ambariContext.getClusterName(1L)).andReturn("clusterName").anyTimes();
    PowerMock.mockStatic(AmbariServer.class);
    expect(AmbariServer.getController()).andReturn(controller).anyTimes();
    PowerMock.replay(AmbariServer.class);
    expect(clusters.getCluster("clusterName")).andReturn(cluster).anyTimes();
    expect(controller.getKerberosHelper()).andReturn(kerberosHelper).anyTimes();
    expect(controller.getClusters()).andReturn(clusters).anyTimes();
    expect(kerberosHelper.getKerberosDescriptor(cluster, false)).andReturn(kerberosDescriptor).anyTimes();
    Set<String> properties = new HashSet<>();
    properties.add("core-site/hadoop.security.auth_to_local");
    expect(kerberosDescriptor.getAllAuthToLocalProperties()).andReturn(properties).anyTimes();
  }

  @After
  public void tearDown() {
    reset(bp, serviceInfo, stack, ambariContext, configHelper);
  }

  @Test
  public void testDoUpdateForBlueprintExport_SingleHostProperty() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();
    Map<String, String> typeProps = new HashMap<>();
    typeProps.put("yarn.resourcemanager.hostname", "testhost");
    properties.put("yarn-site", typeProps);

    Configuration clusterConfig = new Configuration(properties,
      emptyMap());

    Collection<String> hgComponents = new HashSet<>();
    hgComponents.add("NAMENODE");
    hgComponents.add("SECONDARY_NAMENODE");
    hgComponents.add("RESOURCEMANAGER");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = new HashSet<>();
    hgComponents2.add("DATANODE");
    hgComponents2.add("HDFS_CLIENT");
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, Collections.singleton("testhost2"));

    Collection<TestHostGroup> hostGroups = new HashSet<>();
    hostGroups.add(group1);
    hostGroups.add(group2);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    String updatedVal = properties.get("yarn-site").get("yarn.resourcemanager.hostname");
    assertEquals("%HOSTGROUP::group1%", updatedVal);
  }

  @Test
  public void testDoUpdateForBlueprintExport_FilterProperties() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();

    Map<String, String> kerberosEnvProps = new HashMap<>();
    kerberosEnvProps.put("admin_server_host", "test");
    kerberosEnvProps.put("kdc_hosts", "test");
    kerberosEnvProps.put("realm", "test");
    kerberosEnvProps.put("kdc_type", "test");
    kerberosEnvProps.put("ldap-url", "test");
    kerberosEnvProps.put("container_dn", "test");
    properties.put("kerberos-env", kerberosEnvProps);

    Map<String, String> krb5ConfProps = new HashMap<>();
    krb5ConfProps.put("domains", "test");
    properties.put("krb5-conf", krb5ConfProps);

    Map<String, String> tezSiteConfProps = new HashMap<>();
    tezSiteConfProps.put("tez.tez-ui.history-url.base", "test");
    properties.put("tez-site", tezSiteConfProps);


    Configuration clusterConfig = new Configuration(properties,
      emptyMap());

    Collection<String> hgComponents = new HashSet<>();
    hgComponents.add("NAMENODE");
    hgComponents.add("SECONDARY_NAMENODE");
    hgComponents.add("RESOURCEMANAGER");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = new HashSet<>();
    hgComponents2.add("DATANODE");
    hgComponents2.add("HDFS_CLIENT");
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, Collections.singleton("testhost2"));

    Collection<TestHostGroup> hostGroups = new HashSet<>();
    hostGroups.add(group1);
    hostGroups.add(group2);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    assertEquals(properties.size(), 3);
    assertEquals(((Map) properties.get("kerberos-env")).size(), 0);
    assertEquals(((Map) properties.get("krb5-conf")).size(), 0);
    assertEquals(((Map) properties.get("tez-site")).size(), 0);
  }

  @Test
  public void testDoUpdateForBlueprintExportRangerHAPolicyMgrExternalUrlProperty() throws Exception {
    // Given
    Map<String, String> rangerAdminProperties = Maps.newHashMap();
    rangerAdminProperties.put("DB_FLAVOR", "test_db_flavor");
    rangerAdminProperties.put("policymgr_external_url", "test_policymgr_external_url");


    Map<String, Map<String, String>> properties =
      ImmutableMap.of("admin-properties", rangerAdminProperties);


    Configuration clusterConfig = new Configuration(properties, ImmutableMap.of());

    Collection<String> hostGroup1Components = ImmutableSet.of("RANGER_ADMIN");
    TestHostGroup group1 = new TestHostGroup("group1", hostGroup1Components, Collections.singleton("testhost1"));

    Collection<String> hostGroup2Components = ImmutableSet.of("RANGER_ADMIN");
    TestHostGroup group2 = new TestHostGroup("group2", hostGroup2Components, Collections.singleton("testhost2"));


    Collection<TestHostGroup> hostGroups = ImmutableSet.of(group1, group2);


    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);

    // When
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    // Then
    assertEquals("policymgr_external_url property's original value should be exported when Ranger Admin is deployed to multiple hosts.", "test_policymgr_external_url", properties.get("admin-properties").get("policymgr_external_url"));

  }

  @Test
  public void testDoUpdateForBlueprintExport_SingleHostProperty_specifiedInParentConfig() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();
    Map<String, String> yarnSiteProps = new HashMap<>();
    yarnSiteProps.put("yarn.resourcemanager.hostname", "testhost");
    properties.put("yarn-site", yarnSiteProps);

    Map<String, Map<String, String>> parentProperties = new HashMap<>();
    Map<String, String> parentYarnSiteProps = new HashMap<>();
    parentYarnSiteProps.put("yarn.resourcemanager.resource-tracker.address", "testhost");
    parentProperties.put("yarn-site", parentYarnSiteProps);

    Configuration parentClusterConfig = new Configuration(parentProperties,
      emptyMap());

    Configuration clusterConfig = new Configuration(properties,
      emptyMap(), parentClusterConfig);

    Collection<String> hgComponents = new HashSet<>();
    hgComponents.add("NAMENODE");
    hgComponents.add("SECONDARY_NAMENODE");
    hgComponents.add("RESOURCEMANAGER");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = new HashSet<>();
    hgComponents2.add("DATANODE");
    hgComponents2.add("HDFS_CLIENT");
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, Collections.singleton("testhost2"));

    Collection<TestHostGroup> hostGroups = new HashSet<>();
    hostGroups.add(group1);
    hostGroups.add(group2);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    assertEquals("%HOSTGROUP::group1%", clusterConfig.getPropertyValue("yarn-site", "yarn.resourcemanager.hostname"));
    assertEquals("%HOSTGROUP::group1%", clusterConfig.getPropertyValue("yarn-site", "yarn.resourcemanager.resource-tracker.address"));
  }

  @Test
  public void testDoUpdateForBlueprintExport_SingleHostProperty_hostGroupConfiguration() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();
    Map<String, String> typeProps = new HashMap<>();
    typeProps.put("yarn.resourcemanager.hostname", "testhost");
    properties.put("yarn-site", typeProps);

    Configuration clusterConfig = new Configuration(properties,
      emptyMap());

    Collection<String> hgComponents = new HashSet<>();
    hgComponents.add("NAMENODE");
    hgComponents.add("SECONDARY_NAMENODE");
    hgComponents.add("RESOURCEMANAGER");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = new HashSet<>();
    hgComponents2.add("DATANODE");
    hgComponents2.add("HDFS_CLIENT");

    Map<String, Map<String, String>> group2Properties = new HashMap<>();
    Map<String, String> group2YarnSiteProps = new HashMap<>();
    group2YarnSiteProps.put("yarn.resourcemanager.resource-tracker.address", "testhost");
    group2YarnSiteProps.put("yarn.resourcemanager.webapp.https.address", "{{rm_host}}");
    group2Properties.put("yarn-site", group2YarnSiteProps);
    // host group config -> BP config -> cluster scoped config
    Configuration group2BPConfiguration = new Configuration(emptyMap(),
      emptyMap(), clusterConfig);

    Configuration group2Configuration = new Configuration(group2Properties,
      emptyMap(), group2BPConfiguration);

    // set config on hostgroup
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2,
      Collections.singleton("testhost2"), group2Configuration);

    Collection<TestHostGroup> hostGroups = new HashSet<>();
    hostGroups.add(group1);
    hostGroups.add(group2);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    assertEquals("%HOSTGROUP::group1%", properties.get("yarn-site").get("yarn.resourcemanager.hostname"));
    assertEquals("%HOSTGROUP::group1%",
      group2Configuration.getPropertyValue("yarn-site", "yarn.resourcemanager.resource-tracker.address"));
    assertNotNull("Placeholder property should not have been removed.",
      group2Configuration.getPropertyValue("yarn-site", "yarn.resourcemanager.webapp.https.address"));
  }

  @Test
  public void testDoUpdateForBlueprintExport_SingleHostProperty__withPort() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();
    Map<String, String> typeProps = new HashMap<>();
    typeProps.put("fs.defaultFS", "testhost:8020");
    properties.put("core-site", typeProps);

    Configuration clusterConfig = new Configuration(properties,
      emptyMap());

    Collection<String> hgComponents = new HashSet<>();
    hgComponents.add("NAMENODE");
    hgComponents.add("SECONDARY_NAMENODE");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = new HashSet<>();
    hgComponents2.add("DATANODE");
    hgComponents2.add("HDFS_CLIENT");
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, Collections.singleton("testhost2"));

    Collection<TestHostGroup> hostGroups = new HashSet<>();
    hostGroups.add(group1);
    hostGroups.add(group2);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    String updatedVal = properties.get("core-site").get("fs.defaultFS");
    assertEquals("%HOSTGROUP::group1%:8020", updatedVal);
  }

  @Test
  public void testDoUpdateForBlueprintExport_SingleHostProperty__ExternalReference() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();
    Map<String, String> typeProps = new HashMap<>();
    typeProps.put("yarn.resourcemanager.hostname", "external-host");
    properties.put("yarn-site", typeProps);

    Configuration clusterConfig = new Configuration(properties,
      emptyMap());

    Collection<String> hgComponents = new HashSet<>();
    hgComponents.add("NAMENODE");
    hgComponents.add("SECONDARY_NAMENODE");
    hgComponents.add("RESOURCEMANAGER");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = new HashSet<>();
    hgComponents2.add("DATANODE");
    hgComponents2.add("HDFS_CLIENT");
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, Collections.singleton("testhost2"));

    Collection<TestHostGroup> hostGroups = new HashSet<>();
    hostGroups.add(group1);
    hostGroups.add(group2);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    assertFalse(properties.get("yarn-site").containsKey("yarn.resourcemanager.hostname"));
  }

  @Test
  public void testDoUpdateForBlueprintExport_MultiHostProperty() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();
    Map<String, String> typeProps = new HashMap<>();
    typeProps.put("hbase.zookeeper.quorum", "testhost,testhost2,testhost2a,testhost2b");
    properties.put("hbase-site", typeProps);

    Configuration clusterConfig = new Configuration(properties,
      emptyMap());

    Collection<String> hgComponents = new HashSet<>();
    hgComponents.add("NAMENODE");
    hgComponents.add("SECONDARY_NAMENODE");
    hgComponents.add("ZOOKEEPER_SERVER");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = new HashSet<>();
    hgComponents2.add("DATANODE");
    hgComponents2.add("HDFS_CLIENT");
    hgComponents2.add("ZOOKEEPER_SERVER");
    Set<String> hosts2 = new HashSet<>();
    hosts2.add("testhost2");
    hosts2.add("testhost2a");
    hosts2.add("testhost2b");
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, hosts2);

    Collection<String> hgComponents3 = new HashSet<>();
    hgComponents2.add("HDFS_CLIENT");
    hgComponents2.add("ZOOKEEPER_CLIENT");
    Set<String> hosts3 = new HashSet<>();
    hosts3.add("testhost3");
    hosts3.add("testhost3a");
    TestHostGroup group3 = new TestHostGroup("group3", hgComponents3, hosts3);

    Collection<TestHostGroup> hostGroups = new HashSet<>();
    hostGroups.add(group1);
    hostGroups.add(group2);
    hostGroups.add(group3);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    String updatedVal = properties.get("hbase-site").get("hbase.zookeeper.quorum");
    assertEquals("%HOSTGROUP::group1%,%HOSTGROUP::group2%", updatedVal);
  }

  @Test
  public void testDoUpdateForBlueprintExport_MultiHostProperty__WithPorts() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();
    Map<String, String> typeProps = new HashMap<>();
    typeProps.put("templeton.zookeeper.hosts", "testhost:5050,testhost2:9090,testhost2a:9090,testhost2b:9090");
    properties.put("webhcat-site", typeProps);

    Configuration clusterConfig = new Configuration(properties,
      emptyMap());

    Collection<String> hgComponents = new HashSet<>();
    hgComponents.add("NAMENODE");
    hgComponents.add("SECONDARY_NAMENODE");
    hgComponents.add("ZOOKEEPER_SERVER");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = new HashSet<>();
    hgComponents2.add("DATANODE");
    hgComponents2.add("HDFS_CLIENT");
    hgComponents2.add("ZOOKEEPER_SERVER");
    Set<String> hosts2 = new HashSet<>();
    hosts2.add("testhost2");
    hosts2.add("testhost2a");
    hosts2.add("testhost2b");
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, hosts2);

    Collection<String> hgComponents3 = new HashSet<>();
    hgComponents2.add("HDFS_CLIENT");
    hgComponents2.add("ZOOKEEPER_CLIENT");
    Set<String> hosts3 = new HashSet<>();
    hosts3.add("testhost3");
    hosts3.add("testhost3a");
    TestHostGroup group3 = new TestHostGroup("group3", hgComponents3, hosts3);

    Collection<TestHostGroup> hostGroups = new HashSet<>();
    hostGroups.add(group1);
    hostGroups.add(group2);
    hostGroups.add(group3);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    String updatedVal = properties.get("webhcat-site").get("templeton.zookeeper.hosts");
    assertEquals("%HOSTGROUP::group1%:5050,%HOSTGROUP::group2%:9090", updatedVal);
  }

  @Test
  public void testDoUpdateForBlueprintExport_MultiHostProperty__WithPrefixAndPorts() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();
    Map<String, String> typeProps = new HashMap<>();
    typeProps.put("atlas.server.bind.address",
      "http://testhost:21000,http://testhost2:21000,http://testhost2a:21000,http://testhost2b:21000");
    properties.put("application-properties", typeProps);

    Configuration clusterConfig = new Configuration(properties, emptyMap());

    Collection<String> hgComponents = Sets.newHashSet("NAMENODE", "SECONDARY_NAMENODE", "ZOOKEEPER_SERVER");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = Sets.newHashSet("DATANODE", "HDFS_CLIENT", "ZOOKEEPER_SERVER");
    Set<String> hosts2 = Sets.newHashSet("testhost2", "testhost2a", "testhost2b");
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, hosts2);

    Collection<String> hgComponents3 = Sets.newHashSet("HDFS_CLIENT", "ZOOKEEPER_CLIENT");
    Set<String> hosts3 = Sets.newHashSet("testhost3", "testhost3a");
    TestHostGroup group3 = new TestHostGroup("group3", hgComponents3, hosts3);

    Collection<TestHostGroup> hostGroups = Sets.newHashSet(group1, group2, group3);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    String updatedVal = properties.get("application-properties").get("atlas.server.bind.address");
    assertEquals("http://%HOSTGROUP::group1%:21000,http://%HOSTGROUP::group2%:21000", updatedVal);
  }

  @Test
  public void testDoUpdateForBlueprintExport_MultiHostProperty__YAML() throws Exception {
    Map<String, Map<String, String>> properties = new HashMap<>();
    Map<String, String> typeProps = new HashMap<>();
    typeProps.put("storm.zookeeper.servers", "['testhost:5050','testhost2:9090','testhost2a:9090','testhost2b:9090']");
    typeProps.put("drpc_server_host", "['testhost:5050']");
    typeProps.put("storm_ui_server_host", "['testhost:5050']");
    typeProps.put("supervisor_hosts", "['testhost:5050','testhost2:9090']");

    properties.put("storm-site", typeProps);

    Configuration clusterConfig = new Configuration(properties,
      emptyMap());

    Collection<String> hgComponents = new HashSet<>();
    hgComponents.add("NAMENODE");
    hgComponents.add("SECONDARY_NAMENODE");
    hgComponents.add("ZOOKEEPER_SERVER");
    hgComponents.add("DRPC_SERVER");
    hgComponents.add("STORM_UI_SERVER");
    hgComponents.add("SUPERVISOR");
    TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost"));

    Collection<String> hgComponents2 = new HashSet<>();
    hgComponents2.add("DATANODE");
    hgComponents2.add("HDFS_CLIENT");
    hgComponents2.add("ZOOKEEPER_SERVER");
    hgComponents2.add("SUPERVISOR");
    Set<String> hosts2 = new HashSet<>();
    hosts2.add("testhost2");
    hosts2.add("testhost2a");
    hosts2.add("testhost2b");
    TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, hosts2);

    Collection<String> hgComponents3 = new HashSet<>();
    hgComponents2.add("HDFS_CLIENT");
    hgComponents2.add("ZOOKEEPER_CLIENT");
    Set<String> hosts3 = new HashSet<>();
    hosts3.add("testhost3");
    hosts3.add("testhost3a");
    TestHostGroup group3 = new TestHostGroup("group3", hgComponents3, hosts3);

    Collection<TestHostGroup> hostGroups = new HashSet<>();
    hostGroups.add(group1);
    hostGroups.add(group2);
    hostGroups.add(group3);

    ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
    BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology);
    configProcessor.doUpdateForBlueprintExport(BlueprintExportType.FULL);

    String updatedVal = properties.get("storm-site").get("storm.zookeeper.servers");
    assertEquals("['%HOSTGROUP::group1%:5050','%HOSTGROUP::group2%:9090']", updatedVal);

