/*
 * 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;

import static org.easymock.EasyMock.createNiceMock;
import static org.easymock.EasyMock.createStrictMock;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;
import static org.hamcrest.CoreMatchers.is;
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.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.File;
import java.io.StringReader;
import java.lang.reflect.Type;
import java.text.MessageFormat;
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.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

import jakarta.persistence.EntityManager;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.ClusterNotFoundException;
import org.apache.ambari.server.DuplicateResourceException;
import org.apache.ambari.server.H2DatabaseCleaner;
import org.apache.ambari.server.HostNotFoundException;
import org.apache.ambari.server.ObjectNotFoundException;
import org.apache.ambari.server.ParentObjectNotFoundException;
import org.apache.ambari.server.Role;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.ServiceNotFoundException;
import org.apache.ambari.server.StackAccessException;
import org.apache.ambari.server.actionmanager.ActionDBAccessor;
import org.apache.ambari.server.actionmanager.ActionManager;
import org.apache.ambari.server.actionmanager.ActionType;
import org.apache.ambari.server.actionmanager.ExecutionCommandWrapper;
import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.actionmanager.Request;
import org.apache.ambari.server.actionmanager.Stage;
import org.apache.ambari.server.actionmanager.StageFactory;
import org.apache.ambari.server.actionmanager.TargetHostType;
import org.apache.ambari.server.agent.ExecutionCommand;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.audit.AuditLogger;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.internal.ClusterStackVersionResourceProviderTest;
import org.apache.ambari.server.controller.internal.ComponentResourceProviderTest;
import org.apache.ambari.server.controller.internal.DeleteStatusMetaData;
import org.apache.ambari.server.controller.internal.HostComponentResourceProviderTest;
import org.apache.ambari.server.controller.internal.HostResourceProviderTest;
import org.apache.ambari.server.controller.internal.RequestOperationLevel;
import org.apache.ambari.server.controller.internal.RequestResourceFilter;
import org.apache.ambari.server.controller.internal.ServiceResourceProviderTest;
import org.apache.ambari.server.controller.internal.TaskResourceProvider;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.utilities.PredicateBuilder;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
import org.apache.ambari.server.customactions.ActionDefinition;
import org.apache.ambari.server.metadata.ActionMetadata;
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.ExecutionCommandDAO;
import org.apache.ambari.server.orm.dao.HostDAO;
import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
import org.apache.ambari.server.orm.dao.RepositoryVersionDAO;
import org.apache.ambari.server.orm.dao.StackDAO;
import org.apache.ambari.server.orm.dao.TopologyHostInfoDAO;
import org.apache.ambari.server.orm.dao.WidgetDAO;
import org.apache.ambari.server.orm.dao.WidgetLayoutDAO;
import org.apache.ambari.server.orm.entities.ExecutionCommandEntity;
import org.apache.ambari.server.orm.entities.HostEntity;
import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
import org.apache.ambari.server.orm.entities.StackEntity;
import org.apache.ambari.server.orm.entities.WidgetEntity;
import org.apache.ambari.server.orm.entities.WidgetLayoutEntity;
import org.apache.ambari.server.orm.entities.WidgetLayoutUserWidgetEntity;
import org.apache.ambari.server.security.TestAuthenticationFactory;
import org.apache.ambari.server.security.authorization.AuthorizationException;
import org.apache.ambari.server.serveraction.ServerAction;
import org.apache.ambari.server.stack.StackManagerMock;
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.HostComponentAdminState;
import org.apache.ambari.server.state.HostState;
import org.apache.ambari.server.state.MaintenanceState;
import org.apache.ambari.server.state.RepositoryInfo;
import org.apache.ambari.server.state.SecurityType;
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.ServiceInfo;
import org.apache.ambari.server.state.ServiceOsSpecific;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.StackInfo;
import org.apache.ambari.server.state.State;
import org.apache.ambari.server.state.configgroup.ConfigGroup;
import org.apache.ambari.server.state.configgroup.ConfigGroupFactory;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostInstallEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpSucceededEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostServerActionEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStartEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStartedEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStopEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStoppedEvent;
import org.apache.ambari.server.topology.TopologyManager;
import org.apache.ambari.server.utils.EventBusSynchronizer;
import org.apache.ambari.server.utils.StageUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.easymock.Capture;
import org.easymock.EasyMock;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.core.context.SecurityContextHolder;

import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.google.inject.Guice;
import com.google.inject.Injector;

public class AmbariManagementControllerTest {

  private static final Logger LOG =
      LoggerFactory.getLogger(AmbariManagementControllerTest.class);

  private static final String STACK_NAME = "HDP";

  private static final String SERVICE_NAME_YARN = "YARN";
  private static final String COMPONENT_NAME_NODEMANAGER = "NODEMANAGER";
  private static final String SERVICE_NAME_HBASE = "HBASE";
  private static final String COMPONENT_NAME_REGIONSERVER = "HBASE_REGIONSERVER";
  private static final String COMPONENT_NAME_DATANODE = "DATANODE";
  private static final String SERVICE_NAME_HIVE = "HIVE";
  private static final String COMPONENT_NAME_HIVE_METASTORE = "HIVE_METASTORE";
  private static final String COMPONENT_NAME_HIVE_SERVER = "HIVE_SERVER";
  private static final String STACK_VERSION = "0.2";
  private static final String NEW_STACK_VERSION = "2.0.6";
  private static final String OS_TYPE = "centos5";
  private static final String REPO_ID = "HDP-1.1.1.16";
  private static final String REPO_NAME = "HDP";
  private static final String PROPERTY_NAME = "hbase.regionserver.msginterval";
  private static final String SERVICE_NAME = "HDFS";
  private static final String FAKE_SERVICE_NAME = "FAKENAGIOS";
  private static final int STACK_VERSIONS_CNT = 17;
  private static final int REPOS_CNT = 3;
  private static final int STACK_PROPERTIES_CNT = 103;
  private static final int STACK_COMPONENTS_CNT = 5;
  private static final int OS_CNT = 2;

  private static final String NON_EXT_VALUE = "XXX";
  private static final String INCORRECT_BASE_URL = "http://incorrect.url";

  private static final String COMPONENT_NAME = "NAMENODE";

  private static final String REQUEST_CONTEXT_PROPERTY = "context";

  private static AmbariManagementController controller;
  private static Clusters clusters;
  private ActionDBAccessor actionDB;
  private static Injector injector;
  private ServiceFactory serviceFactory;
  private ServiceComponentFactory serviceComponentFactory;
  private ServiceComponentHostFactory serviceComponentHostFactory;
  private static AmbariMetaInfo ambariMetaInfo;
  private EntityManager entityManager;
  private ConfigHelper configHelper;
  private ConfigGroupFactory configGroupFactory;
  private OrmTestHelper helper;
  private StageFactory stageFactory;
  private HostDAO hostDAO;
  private TopologyHostInfoDAO topologyHostInfoDAO;
  private HostRoleCommandDAO hostRoleCommandDAO;
  private StackManagerMock stackManagerMock;
  private RepositoryVersionDAO repositoryVersionDAO;

  RepositoryVersionEntity repositoryVersion01;
  RepositoryVersionEntity repositoryVersion02;
  RepositoryVersionEntity repositoryVersion120;
  RepositoryVersionEntity repositoryVersion201;
  RepositoryVersionEntity repositoryVersion206;
  RepositoryVersionEntity repositoryVersion207;
  RepositoryVersionEntity repositoryVersion208;
  RepositoryVersionEntity repositoryVersion220;

  @Rule
  public ExpectedException expectedException = ExpectedException.none();

  @BeforeClass
  public static void beforeClass() throws Exception {
    InMemoryDefaultTestModule module = new InMemoryDefaultTestModule();
    injector = Guice.createInjector(module);
    H2DatabaseCleaner.resetSequences(injector);
    injector.getInstance(GuiceJpaInitializer.class);
    ambariMetaInfo = injector.getInstance(AmbariMetaInfo.class);
    ambariMetaInfo.init();
    clusters = injector.getInstance(Clusters.class);
    controller = injector.getInstance(AmbariManagementController.class);
    TopologyManager topologyManager = injector.getInstance(TopologyManager.class);
    StageUtils.setTopologyManager(topologyManager);
    Configuration configuration = injector.getInstance(Configuration.class);
    StageUtils.setConfiguration(configuration);
    ActionManager.setTopologyManager(topologyManager);

    SecurityContextHolder.getContext().setAuthentication(TestAuthenticationFactory.createAdministrator());
  }

  @Before
  public void setup() throws Exception {
    EventBusSynchronizer.synchronizeAmbariEventPublisher(injector);

    entityManager = injector.getProvider(EntityManager.class).get();
    actionDB = injector.getInstance(ActionDBAccessor.class);
    serviceFactory = injector.getInstance(ServiceFactory.class);
    serviceComponentFactory = injector.getInstance(
        ServiceComponentFactory.class);
    serviceComponentHostFactory = injector.getInstance(
        ServiceComponentHostFactory.class);
    configHelper = injector.getInstance(ConfigHelper.class);
    configGroupFactory = injector.getInstance(ConfigGroupFactory.class);
    helper = injector.getInstance(OrmTestHelper.class);
    stageFactory = injector.getInstance(StageFactory.class);
    hostDAO = injector.getInstance(HostDAO.class);
    topologyHostInfoDAO = injector.getInstance(TopologyHostInfoDAO.class);
    hostRoleCommandDAO = injector.getInstance(HostRoleCommandDAO.class);
    stackManagerMock = (StackManagerMock) ambariMetaInfo.getStackManager();
    EasyMock.replay(injector.getInstance(AuditLogger.class));

    repositoryVersion01 = helper.getOrCreateRepositoryVersion(
        new StackId("HDP-0.1"), "0.1-1234");

    repositoryVersion02 = helper.getOrCreateRepositoryVersion(
        new StackId("HDP-0.2"), "0.2-1234");

    repositoryVersion120 = helper.getOrCreateRepositoryVersion(
        new StackId("HDP-1.2.0"), "1.2.0-1234");

    repositoryVersion201 = helper.getOrCreateRepositoryVersion(
        new StackId("HDP-2.0.1"), "2.0.1-1234");

    repositoryVersion206 = helper.getOrCreateRepositoryVersion(
        new StackId("HDP-2.0.6"), "2.0.6-1234");

    repositoryVersion207 = helper.getOrCreateRepositoryVersion(
        new StackId("HDP-2.0.7"), "2.0.7-1234");

    repositoryVersion208 = helper.getOrCreateRepositoryVersion(
        new StackId("HDP-2.0.8"), "2.0.8-1234");

    repositoryVersion220 = helper.getOrCreateRepositoryVersion(
        new StackId("HDP-2.2.0"), "2.2.0-1234");

    repositoryVersionDAO = injector.getInstance(RepositoryVersionDAO.class);

    for (Host host : clusters.getHosts()) {
      clusters.updateHostMappings(host);
    }
  }

  @After
  public void teardown() {
    actionDB = null;
    EasyMock.reset(injector.getInstance(AuditLogger.class));
  }

  @AfterClass
  public static void afterClass() throws Exception {
    H2DatabaseCleaner.clearDatabaseAndStopPersistenceService(injector);
  }

  private static String getUniqueName() {
    return UUID.randomUUID().toString();
  }


  private void setOsFamily(Host host, String osFamily, String osVersion) {
    Map<String, String> hostAttributes = new HashMap<>();
    hostAttributes.put("os_family", osFamily);
    hostAttributes.put("os_release_version", osVersion);

    host.setHostAttributes(hostAttributes);
  }

  private void addHost(String hostname) throws Exception {
    addHostToCluster(hostname, null);
  }

  private void addHostToCluster(String hostname, String clusterName) throws Exception {

    if (!clusters.hostExists(hostname)) {
      clusters.addHost(hostname);
      setOsFamily(clusters.getHost(hostname), "redhat", "6.3");
      clusters.getHost(hostname).setState(HostState.HEALTHY);
    }

    if (null != clusterName) {
      clusters.mapHostToCluster(hostname, clusterName);
    }
    clusters.updateHostMappings(clusters.getHost(hostname));
  }

  private void deleteHost(String hostname) throws Exception {
    clusters.deleteHost(hostname);
  }



  /**
   * Creates a Cluster object, along with its corresponding ClusterVersion based on the stack.
   * @param clusterName Cluster name
   * @throws Exception
   */
  private void createCluster(String clusterName) throws Exception{
    RepositoryVersionDAO repoDAO = injector.getInstance(RepositoryVersionDAO.class);
    ClusterRequest r = new ClusterRequest(null, clusterName, State.INSTALLED.name(), SecurityType.NONE, "HDP-0.1", null);
    controller.createCluster(r);
  }

  private void createService(String clusterName, String serviceName, State desiredState) throws Exception, AuthorizationException {
    createService(clusterName, serviceName, repositoryVersion02, desiredState);
  }

  private void createService(String clusterName, String serviceName,
      RepositoryVersionEntity repositoryVersion, State desiredState)
      throws Exception, AuthorizationException {
    String dStateStr = null;
    if (desiredState != null) {
      dStateStr = desiredState.toString();
    }

    ServiceRequest r1 = new ServiceRequest(clusterName, serviceName,
        repositoryVersion.getId(), dStateStr,
        null);

    Set<ServiceRequest> requests = new HashSet<>();
    requests.add(r1);

    ServiceResourceProviderTest.createServices(controller, repositoryVersionDAO, requests);
  }

  private void createServiceComponent(String clusterName,
      String serviceName, String componentName, State desiredState)
      throws Exception, AuthorizationException {
    String dStateStr = null;
    if (desiredState != null) {
      dStateStr = desiredState.toString();
    }
    ServiceComponentRequest r = new ServiceComponentRequest(clusterName,
        serviceName, componentName, dStateStr);
    Set<ServiceComponentRequest> requests =
      new HashSet<>();
    requests.add(r);
    ComponentResourceProviderTest.createComponents(controller, requests);
  }

  private void createServiceComponentHost(String clusterName,
      String serviceName, String componentName, String hostname,
      State desiredState) throws Exception, AuthorizationException {
    String dStateStr = null;
    if (desiredState != null) {
      dStateStr = desiredState.toString();
    }
    ServiceComponentHostRequest r = new ServiceComponentHostRequest(clusterName,
        serviceName, componentName, hostname, dStateStr);
    Set<ServiceComponentHostRequest> requests =
      new HashSet<>();
    requests.add(r);
    controller.createHostComponents(requests);
  }

  private void deleteServiceComponentHost(String clusterName,
                                          String serviceName, String componentName, String hostname,
                                          State desiredState) throws Exception, AuthorizationException {
    String dStateStr = null;
    if (desiredState != null) {
      dStateStr = desiredState.toString();
    }
    ServiceComponentHostRequest r = new ServiceComponentHostRequest(clusterName,
        serviceName, componentName, hostname, dStateStr);
    Set<ServiceComponentHostRequest> requests =
      new HashSet<>();
    requests.add(r);
    controller.deleteHostComponents(requests);
  }

  private Long createConfigGroup(Cluster cluster, String serviceName, String name, String tag,
                              List<String> hosts, List<Config> configs)
                              throws Exception {

    Map<Long, Host> hostMap = new HashMap<>();
    Map<String, Config> configMap = new HashMap<>();

    for (String hostname : hosts) {
      Host host = clusters.getHost(hostname);
      HostEntity hostEntity = hostDAO.findByName(hostname);
      hostMap.put(hostEntity.getHostId(), host);
    }

    for (Config config : configs) {
      configMap.put(config.getType(), config);
    }

    ConfigGroup configGroup = configGroupFactory.createNew(cluster, serviceName, name,
      tag, "", configMap, hostMap);

    configGroup.setServiceName(serviceName);

    cluster.addConfigGroup(configGroup);

    return configGroup.getId();
  }

  private long stopService(String clusterName, String serviceName,
      boolean runSmokeTests, boolean reconfigureClients) throws
      Exception, AuthorizationException {
    ServiceRequest r = new ServiceRequest(clusterName, serviceName, null, State.INSTALLED.toString(), null);
    Set<ServiceRequest> requests = new HashSet<>();
    requests.add(r);
    Map<String, String> mapRequestProps = new HashMap<>();
    mapRequestProps.put("context", "Called from a test");
    RequestStatusResponse resp = ServiceResourceProviderTest.updateServices(controller, requests,
      mapRequestProps, runSmokeTests, reconfigureClients);

    Assert.assertEquals(State.INSTALLED,
      clusters.getCluster(clusterName).getService(serviceName)
        .getDesiredState());

    // manually change live state to stopped as no running action manager
    for (ServiceComponent sc :
      clusters.getCluster(clusterName).getService(serviceName)
        .getServiceComponents().values()) {
      for (ServiceComponentHost sch : sc.getServiceComponentHosts().values()) {
        sch.setState(State.INSTALLED);
      }
    }

    return resp.getRequestId();
  }

  private long stopServiceComponentHosts(String clusterName,
      String serviceName) throws Exception {
    Cluster c = clusters.getCluster(clusterName);
    Service s = c.getService(serviceName);
    Set<ServiceComponentHostRequest> requests = new
      HashSet<>();
    for (ServiceComponent sc : s.getServiceComponents().values()) {
      for (ServiceComponentHost sch : sc.getServiceComponentHosts().values()) {
        ServiceComponentHostRequest schr = new ServiceComponentHostRequest
          (clusterName, serviceName, sc.getName(),
            sch.getHostName(), State.INSTALLED.name());
        requests.add(schr);
      }
    }
    Map<String, String> mapRequestProps = new HashMap<>();
    mapRequestProps.put("context", "Called from a test");
    RequestStatusResponse resp = HostComponentResourceProviderTest.updateHostComponents(controller, injector, requests,
        mapRequestProps, false);

    // manually change live state to started as no running action manager
    for (ServiceComponent sc :
      clusters.getCluster(clusterName).getService(serviceName)
        .getServiceComponents().values()) {
      for (ServiceComponentHost sch : sc.getServiceComponentHosts().values()) {
        sch.setState(State.INSTALLED);
      }
    }
    return resp.getRequestId();
  }

  private long startService(String clusterName, String serviceName,
                            boolean runSmokeTests, boolean reconfigureClients) throws
      Exception, AuthorizationException {
    return startService(clusterName, serviceName, runSmokeTests, reconfigureClients, null);
  }


  private long startService(String clusterName, String serviceName,
                            boolean runSmokeTests, boolean reconfigureClients,
                            MaintenanceStateHelper maintenanceStateHelper) throws
      Exception, AuthorizationException {
    ServiceRequest r = new ServiceRequest(clusterName, serviceName, repositoryVersion02.getId(),
        State.STARTED.toString(), null);
    Set<ServiceRequest> requests = new HashSet<>();
    requests.add(r);
    Map<String, String> mapRequestProps = new HashMap<>();
    mapRequestProps.put("context", "Called from a test");
    RequestStatusResponse resp = ServiceResourceProviderTest.updateServices(controller, requests,
        mapRequestProps, runSmokeTests, reconfigureClients, maintenanceStateHelper);

    Assert.assertEquals(State.STARTED,
        clusters.getCluster(clusterName).getService(serviceName)
            .getDesiredState());

    if (resp != null) {
      // manually change live state to stopped as no running action manager
      List<HostRoleCommand> commands = actionDB.getRequestTasks(resp.getRequestId());
      for (HostRoleCommand cmd : commands) {
        String scName = cmd.getRole().toString();
        if (!scName.endsWith("CHECK")) {
          Cluster cluster = clusters.getCluster(clusterName);
          String hostname = cmd.getHostName();
          for (Service s : cluster.getServices().values()) {
            if (s.getServiceComponents().containsKey(scName) &&
              !s.getServiceComponent(scName).isClientComponent()) {
              s.getServiceComponent(scName).getServiceComponentHost(hostname).
                setState(State.STARTED);
              break;
            }
          }
        }
      }
      return resp.getRequestId();
    } else {
      return -1;
    }
  }


  private long installService(String clusterName, String serviceName,
                              boolean runSmokeTests, boolean reconfigureClients)
      throws Exception, AuthorizationException {
    return installService(clusterName, serviceName, runSmokeTests, reconfigureClients, null, null);
  }


  /**
   * Allows to set maintenanceStateHelper. For use when there is anything to test
   * with maintenance mode.
   */
  private long installService(String clusterName, String serviceName,
                              boolean runSmokeTests, boolean reconfigureClients,
                              MaintenanceStateHelper maintenanceStateHelper,
                              Map<String, String> mapRequestPropsInput)
      throws Exception, AuthorizationException {

    ServiceRequest r = new ServiceRequest(clusterName, serviceName, repositoryVersion02.getId(),
        State.INSTALLED.toString(), null);

    Set<ServiceRequest> requests = new HashSet<>();
    requests.add(r);

    Map<String, String> mapRequestProps = new HashMap<>();
    mapRequestProps.put("context", "Called from a test");
    if(mapRequestPropsInput != null) {
      mapRequestProps.putAll(mapRequestPropsInput);
    }

    RequestStatusResponse resp = ServiceResourceProviderTest.updateServices(controller, requests,
        mapRequestProps, runSmokeTests, reconfigureClients, maintenanceStateHelper);

    Assert.assertEquals(State.INSTALLED,
        clusters.getCluster(clusterName).getService(serviceName)
            .getDesiredState());

    if (resp != null) {
      // manually change live state to stopped as no running action manager
      List<HostRoleCommand> commands = actionDB.getRequestTasks(resp.getRequestId());
      for (HostRoleCommand cmd : commands) {
        clusters.getCluster(clusterName).getService(serviceName).getServiceComponent(cmd.getRole().name())
            .getServiceComponentHost(cmd.getHostName()).setState(State.INSTALLED);
      }
      return resp.getRequestId();
    } else {
      return -1;
    }
  }


  private boolean checkExceptionType(Throwable e, Class<? extends Exception> exceptionClass) {
    return e != null && (exceptionClass.isAssignableFrom(e.getClass()) || checkExceptionType(e.getCause(), exceptionClass));
  }

  @Test
  public void testCreateClusterSimple() throws Exception, AuthorizationException {
    String cluster1 = getUniqueName();
    createCluster(cluster1);
    Set<ClusterResponse> r =
        controller.getClusters(Collections.singleton(
            new ClusterRequest(null, cluster1, null, null)));
    Assert.assertEquals(1, r.size());
    ClusterResponse c = r.iterator().next();
    Assert.assertEquals(cluster1, c.getClusterName());

    try {
      createCluster(cluster1);
      fail("Duplicate cluster creation should fail");
    } catch (Exception e) {
      // Expected
    }
  }

  @Test
  public void testCreateClusterWithHostMapping() throws Exception, AuthorizationException {
    String cluster1 = getUniqueName();

    String host1 = getUniqueName();
    String host2 = getUniqueName();

    Set<String> hostNames = new HashSet<>();
    hostNames.add(host1);
    hostNames.add(host2);
    ClusterRequest r = new ClusterRequest(null, cluster1, "HDP-0.1", hostNames);

    try {
      controller.createCluster(r);
      fail("Expected create cluster to fail for invalid hosts");
    } catch (Exception e) {
      // Expected
    }

    try {
      clusters.getCluster(cluster1);
      fail("Expected to fail for non created cluster");
    } catch (ClusterNotFoundException e) {
      // Expected
    }

    clusters.addHost(host1);
    clusters.addHost(host2);
    setOsFamily(clusters.getHost(host1), "redhat", "6.3");
    setOsFamily(clusters.getHost(host2), "redhat", "6.3");

    controller.createCluster(r);
    Assert.assertNotNull(clusters.getCluster(cluster1));
  }

  @Test(expected = IllegalArgumentException.class)
  public void testCreateClusterWithInvalidRequest1() throws Exception {
    ClusterRequest r = new ClusterRequest(null, null, null, null);
    controller.createCluster(r);
  }

  @Test(expected = IllegalArgumentException.class)
  public void testCreateClusterWithInvalidRequest2() throws Exception {
    ClusterRequest r = new ClusterRequest(1L, null, null, null);
    controller.createCluster(r);
  }

  @Test(expected = IllegalArgumentException.class)
  public void testCreateClusterWithInvalidRequest3() throws Exception {
    ClusterRequest r = new ClusterRequest(null, getUniqueName(), null, null);
    controller.createCluster(r);
  }

  @Test(expected = IllegalArgumentException.class)
  public void testCreateClusterWithInvalidRequest4() throws Exception {
    ClusterRequest r = new ClusterRequest(null, null, State.INSTALLING.name(), null, "HDP-1.2.0", null);
    controller.createCluster(r);
    controller.updateClusters(Collections.singleton(r), null);
  }

  @Test
  public void testCreateServicesSimple() throws Exception, AuthorizationException {
    String cluster1 = getUniqueName();
    createCluster(cluster1);
    String serviceName = "HDFS";
    createService(cluster1, serviceName, repositoryVersion02, State.INIT);

    Service s =
        clusters.getCluster(cluster1).getService(serviceName);
    Assert.assertNotNull(s);
    Assert.assertEquals(serviceName, s.getName());
    Assert.assertEquals(cluster1, s.getCluster().getClusterName());

    ServiceRequest req = new ServiceRequest(cluster1, "HDFS", repositoryVersion02.getId(), null, null);

    Set<ServiceResponse> r =
        ServiceResourceProviderTest.getServices(controller, Collections.singleton(req));
    Assert.assertEquals(1, r.size());
    ServiceResponse resp = r.iterator().next();
    Assert.assertEquals(serviceName, resp.getServiceName());
    Assert.assertEquals(cluster1, resp.getClusterName());
    Assert.assertEquals(State.INIT.toString(), resp.getDesiredState());
    Assert.assertEquals("HDP-0.2", resp.getDesiredStackId());
  }

  @Test
  public void testCreateServicesWithInvalidRequest() throws Exception, AuthorizationException {
    // invalid request
    // dups in requests
    // multi cluster updates

    Set<ServiceRequest> set1 = new HashSet<>();

    try {
      set1.clear();
      ServiceRequest rInvalid = new ServiceRequest(null, null, null, null, null);
      set1.add(rInvalid);
      ServiceResourceProviderTest.createServices(controller, repositoryVersionDAO, set1);
      fail("Expected failure for invalid requests");
    } catch (Exception e) {
      // Expected
    }

    try {
      set1.clear();
      ServiceRequest rInvalid = new ServiceRequest("foo", null, null, null, null);
      set1.add(rInvalid);
      ServiceResourceProviderTest.createServices(controller, repositoryVersionDAO, set1);
      fail("Expected failure for invalid requests");
    } catch (Exception e) {
      // Expected
    }

    try {
      set1.clear();
      ServiceRequest rInvalid = new ServiceRequest("foo", "bar", null, null, null);
      set1.add(rInvalid);
      ServiceResourceProviderTest.createServices(controller, repositoryVersionDAO, set1);
      fail("Expected failure for invalid cluster");
    } catch (Exception e) {
      // Expected
      Assert.assertTrue(checkExceptionType(e, ClusterNotFoundException.class));
    }

    String cluster1 = getUniqueName();
    String cluster2 = getUniqueName();


    clusters.addCluster(cluster1, new StackId("HDP-0.1"));
    clusters.addCluster(cluster2, new StackId("HDP-0.1"));

    try {
      set1.clear();
      ServiceRequest valid1 = new ServiceRequest(cluster1, "HDFS", null, null, null);
      ServiceRequest valid2 = new ServiceRequest(cluster1, "HDFS", null, null, null);
      set1.add(valid1);
      set1.add(valid2);
      ServiceResourceProviderTest.createServices(controller, repositoryVersionDAO, set1);
      fail("Expected failure for invalid requests");
    } catch (Exception e) {
      // Expected
    }

    try {
      set1.clear();
      ServiceRequest valid1 = new ServiceRequest(cluster1, "bar", repositoryVersion02.getId(), State.STARTED.toString(), null);
      set1.add(valid1);
      ServiceResourceProviderTest.createServices(controller, repositoryVersionDAO, set1);
      fail("Expected failure for invalid service");
    } catch (Exception e) {
      // Expected
    }


    try {
      set1.clear();
