/*
 * 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 org.easymock.EasyMock.anyBoolean;
import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.capture;
import static org.easymock.EasyMock.createMock;
import static org.easymock.EasyMock.createNiceMock;
import static org.easymock.EasyMock.createStrictMock;
import static org.easymock.EasyMock.eq;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.isNull;
import static org.easymock.EasyMock.newCapture;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;

import java.lang.reflect.Field;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.KerberosHelper;
import org.apache.ambari.server.controller.MaintenanceStateHelper;
import org.apache.ambari.server.controller.RequestStatusResponse;
import org.apache.ambari.server.controller.ServiceRequest;
import org.apache.ambari.server.controller.ServiceResponse;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.Request;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.spi.ResourceProvider;
import org.apache.ambari.server.controller.spi.SystemException;
import org.apache.ambari.server.controller.utilities.PredicateBuilder;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
import org.apache.ambari.server.controller.utilities.state.DefaultServiceCalculatedState;
import org.apache.ambari.server.metadata.RoleCommandOrder;
import org.apache.ambari.server.orm.dao.RepositoryVersionDAO;
import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
import org.apache.ambari.server.security.TestAuthenticationFactory;
import org.apache.ambari.server.security.authorization.AuthorizationException;
import org.apache.ambari.server.serveraction.kerberos.KerberosAdminAuthenticationException;
import org.apache.ambari.server.serveraction.kerberos.KerberosMissingAdminCredentialsException;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.ServiceFactory;
import org.apache.ambari.server.state.ServiceInfo;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.State;
import org.apache.ambari.server.topology.STOMPComponentsDeleteHandler;
import org.apache.ambari.spi.RepositoryType;
import org.easymock.Capture;
import org.easymock.EasyMock;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;

import com.google.common.collect.ImmutableMap;

/**
 * ServiceResourceProvider tests.
 */
public class ServiceResourceProviderTest {

  @BeforeClass
  public static void resetDefaultServiceCalculatedState() throws NoSuchFieldException, IllegalAccessException {
    Field clustersProviderField = DefaultServiceCalculatedState.class.getDeclaredField("clustersProvider");
    clustersProviderField.setAccessible(true);
    clustersProviderField.set(null, null);
  }

  @Before
  public void clearAuthentication() {
    SecurityContextHolder.getContext().setAuthentication(null);
  }

  @Test
  public void testCreateResourcesAsAdministrator() throws Exception{
    testCreateResources(TestAuthenticationFactory.createAdministrator());
  }

  @Test
  public void testCreateResourcesAsClusterAdministrator() throws Exception{
    testCreateResources(TestAuthenticationFactory.createClusterAdministrator());
  }

  @Test(expected = AuthorizationException.class)
  public void testCreateResourcesAsServiceAdministrator() throws Exception{
    testCreateResources(TestAuthenticationFactory.createServiceAdministrator());
  }

  private void testCreateResources(Authentication authentication) throws Exception{
    AmbariManagementController managementController = createNiceMock(AmbariManagementController.class);
    Clusters clusters = createNiceMock(Clusters.class);
    Cluster cluster = createNiceMock(Cluster.class);
    Service service = createNiceMock(Service.class);
    StackId stackId = new StackId("HDP-2.5");
    ServiceFactory serviceFactory = createNiceMock(ServiceFactory.class);
    AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
    ServiceInfo serviceInfo = createNiceMock(ServiceInfo.class);

    expect(managementController.getClusters()).andReturn(clusters).anyTimes();
    expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();

    expect(cluster.addService(eq("Service100"),
        EasyMock.anyObject(RepositoryVersionEntity.class))).andReturn(service);

    expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();

    expect(cluster.getService("Service100")).andReturn(null);
    expect(cluster.getDesiredStackVersion()).andReturn(stackId).anyTimes();
    expect(cluster.getClusterId()).andReturn(2L).anyTimes();

    expect(ambariMetaInfo.isValidService( (String) anyObject(), (String) anyObject(), (String) anyObject())).andReturn(true);
    expect(ambariMetaInfo.getService((String)anyObject(), (String)anyObject(), (String)anyObject())).andReturn(serviceInfo).anyTimes();

    // replay
    replay(managementController, clusters, cluster, service,
        ambariMetaInfo, serviceFactory, serviceInfo);

    SecurityContextHolder.getContext().setAuthentication(authentication);

    ResourceProvider provider = getServiceProvider(managementController, true, null);

    // add the property map to a set for the request.  add more maps for multiple creates
    Set<Map<String, Object>> propertySet = new LinkedHashSet<>();

    // Service 1: create a map of properties for the request
    Map<String, Object> properties = new LinkedHashMap<>();

    // add properties to the request map
    properties.put(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID, "Cluster100");
    properties.put(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID, "Service100");
    properties.put(ServiceResourceProvider.SERVICE_SERVICE_STATE_PROPERTY_ID, "INIT");
    properties.put(ServiceResourceProvider.SERVICE_DESIRED_STACK_PROPERTY_ID, "HDP-1.1");
    properties.put(ServiceResourceProvider.SERVICE_DESIRED_REPO_VERSION_ID_PROPERTY_ID, "1");

    propertySet.add(properties);

    // create the request
    Request request = PropertyHelper.getCreateRequest(propertySet, null);

    provider.createResources(request);

    // verify
    verify(managementController, clusters, cluster, service,
        ambariMetaInfo, serviceFactory, serviceInfo);
  }

  @Test
  public void testGetResourcesAsAdministrator() throws Exception{
    testGetResources(TestAuthenticationFactory.createAdministrator());
  }

  @Test
  public void testGetResourcesAsClusterAdministrator() throws Exception{
    testGetResources(TestAuthenticationFactory.createClusterAdministrator());
  }

  @Test
  public void testGetResourcesAsServiceAdministrator() throws Exception{
    testGetResources(TestAuthenticationFactory.createServiceAdministrator());
  }

  private void testGetResources(Authentication authentication) throws Exception{
    AmbariManagementController managementController = createMock(AmbariManagementController.class);
    Clusters clusters = createNiceMock(Clusters.class);
    Cluster cluster = createNiceMock(Cluster.class);
    Service service0 = createNiceMock(Service.class);
    Service service1 = createNiceMock(Service.class);
    Service service2 = createNiceMock(Service.class);
    Service service3 = createNiceMock(Service.class);
    Service service4 = createNiceMock(Service.class);
    ServiceResponse serviceResponse0 = createNiceMock(ServiceResponse.class);
    ServiceResponse serviceResponse1 = createNiceMock(ServiceResponse.class);
    ServiceResponse serviceResponse2 = createNiceMock(ServiceResponse.class);
    ServiceResponse serviceResponse3 = createNiceMock(ServiceResponse.class);
    ServiceResponse serviceResponse4 = createNiceMock(ServiceResponse.class);

    StackId stackId = createNiceMock(StackId.class);
    ServiceFactory serviceFactory = createNiceMock(ServiceFactory.class);
    AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);

    Map<String, Service> allResponseMap = new HashMap<>();
    allResponseMap.put("Service100", service0);
    allResponseMap.put("Service101", service1);
    allResponseMap.put("Service102", service2);
    allResponseMap.put("Service103", service3);
    allResponseMap.put("Service104", service4);

    // set expectations
    expect(managementController.getClusters()).andReturn(clusters).anyTimes();
    expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
    expect(managementController.getHostComponents(EasyMock.anyObject())).
        andReturn(Collections.emptySet()).anyTimes();

    expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();

    expect(cluster.getServices()).andReturn(allResponseMap).anyTimes();
    expect(cluster.getService("Service102")).andReturn(service2);

    expect(service0.convertToResponse()).andReturn(serviceResponse0).anyTimes();
    expect(service1.convertToResponse()).andReturn(serviceResponse1).anyTimes();
    expect(service2.convertToResponse()).andReturn(serviceResponse2).anyTimes();
    expect(service3.convertToResponse()).andReturn(serviceResponse3).anyTimes();
    expect(service4.convertToResponse()).andReturn(serviceResponse4).anyTimes();

    expect(service0.getName()).andReturn("Service100").anyTimes();
    expect(service1.getName()).andReturn("Service101").anyTimes();
    expect(service2.getName()).andReturn("Service102").anyTimes();
    expect(service3.getName()).andReturn("Service103").anyTimes();
    expect(service4.getName()).andReturn("Service104").anyTimes();

    expect(service0.getDesiredState()).andReturn(State.INIT);
    expect(service1.getDesiredState()).andReturn(State.INSTALLED);
    expect(service2.getDesiredState()).andReturn(State.INIT);
    expect(service3.getDesiredState()).andReturn(State.INSTALLED);
    expect(service4.getDesiredState()).andReturn(State.INIT);

    expect(serviceResponse0.getClusterName()).andReturn("Cluster100").anyTimes();
    expect(serviceResponse0.getServiceName()).andReturn("Service100").anyTimes();
    expect(serviceResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
    expect(serviceResponse1.getServiceName()).andReturn("Service101").anyTimes();
    expect(serviceResponse2.getClusterName()).andReturn("Cluster100").anyTimes();
    expect(serviceResponse2.getServiceName()).andReturn("Service102").anyTimes();
    expect(serviceResponse3.getClusterName()).andReturn("Cluster100").anyTimes();
    expect(serviceResponse3.getServiceName()).andReturn("Service103").anyTimes();
    expect(serviceResponse4.getClusterName()).andReturn("Cluster100").anyTimes();
    expect(serviceResponse4.getServiceName()).andReturn("Service104").anyTimes();

    // replay
    replay(managementController, clusters, cluster,
        service0, service1, service2, service3, service4,
        serviceResponse0, serviceResponse1, serviceResponse2, serviceResponse3, serviceResponse4,
        ambariMetaInfo, stackId, serviceFactory);

    SecurityContextHolder.getContext().setAuthentication(authentication);

    ResourceProvider provider = getServiceProvider(managementController);

    Set<String> propertyIds = new HashSet<>();

    propertyIds.add(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID);
    propertyIds.add(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID);

    // create the request
    Predicate predicate = new PredicateBuilder().property(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").toPredicate();
    Request request = PropertyHelper.getReadRequest("ServiceInfo");
    Set<Resource> resources = provider.getResources(request, predicate);

    Assert.assertEquals(5, resources.size());
    Set<String> names = new HashSet<>();
    for (Resource resource : resources) {
      String clusterName = (String) resource.getPropertyValue(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID);
      Assert.assertEquals("Cluster100", clusterName);
      names.add((String) resource.getPropertyValue(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID));
    }
    // Make sure that all of the response objects got moved into resources
    for (Service service : allResponseMap.values() ) {
      Assert.assertTrue(names.contains(service.getName()));
    }

    // get service named Service102
    predicate = new PredicateBuilder().property(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").and().
        property(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID).equals("Service102").toPredicate();
    request = PropertyHelper.getReadRequest("ServiceInfo");
    resources = provider.getResources(request, predicate);

    Assert.assertEquals(1, resources.size());
    Assert.assertEquals("Service102", resources.iterator().next().getPropertyValue(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID));

    // get services where state == "INIT"
    predicate = new PredicateBuilder().property(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").and().
        property(ServiceResourceProvider.SERVICE_SERVICE_STATE_PROPERTY_ID).equals("INIT").toPredicate();
    request = PropertyHelper.getReadRequest(propertyIds);
    resources = provider.getResources(request, predicate);

    Assert.assertEquals(3, resources.size());
    names = new HashSet<>();
    for (Resource resource : resources) {
      String clusterName = (String) resource.getPropertyValue(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID);
      Assert.assertEquals("Cluster100", clusterName);
      names.add((String) resource.getPropertyValue(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID));
    }

    // verify
    verify(managementController, clusters, cluster,
        service0, service1, service2, service3, service4,
        serviceResponse0, serviceResponse1, serviceResponse2, serviceResponse3, serviceResponse4,
        ambariMetaInfo, stackId, serviceFactory);
  }

  @Test
  public void testGetResources_KerberosSpecificProperties() throws Exception{
    AmbariManagementController managementController = createMock(AmbariManagementController.class);
    Clusters clusters = createNiceMock(Clusters.class);
    Cluster cluster = createNiceMock(Cluster.class);
    Service service0 = createNiceMock(Service.class);
    ServiceResponse serviceResponse0 = createNiceMock(ServiceResponse.class);

    StackId stackId = createNiceMock(StackId.class);
    ServiceFactory serviceFactory = createNiceMock(ServiceFactory.class);
    AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
    KerberosHelper kerberosHeper = createStrictMock(KerberosHelper.class);

    Map<String, Service> allResponseMap = new HashMap<>();
    allResponseMap.put("KERBEROS", service0);

    // set expectations
    expect(managementController.getClusters()).andReturn(clusters).anyTimes();
    expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
    expect(managementController.getHostComponents(EasyMock.anyObject())).
        andReturn(Collections.emptySet()).anyTimes();

    expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();

    expect(cluster.getServices()).andReturn(allResponseMap).anyTimes();
    expect(cluster.getService("KERBEROS")).andReturn(service0);

    expect(service0.convertToResponse()).andReturn(serviceResponse0).anyTimes();

    expect(service0.getName()).andReturn("Service100").anyTimes();

    expect(serviceResponse0.getClusterName()).andReturn("Cluster100").anyTimes();
    expect(serviceResponse0.getServiceName()).andReturn("KERBEROS").anyTimes();

    kerberosHeper.validateKDCCredentials(cluster);

    // replay
    replay(managementController, clusters, cluster, service0, serviceResponse0,
        ambariMetaInfo, stackId, serviceFactory, kerberosHeper);

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

    ResourceProvider provider = getServiceProvider(managementController);
    // set kerberos helper on provider
    Class<?> c = provider.getClass();
    Field f = c.getDeclaredField("kerberosHelper");
    f.setAccessible(true);
    f.set(provider, kerberosHeper);

    // create the request
    Predicate predicate = new PredicateBuilder().property(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").and().
        property(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID).equals("KERBEROS").toPredicate();
    Request request = PropertyHelper.getReadRequest("ServiceInfo", "Services");
    Set<Resource> resources = provider.getResources(request, predicate);

    Assert.assertEquals(1, resources.size());
    for (Resource resource : resources) {
      Assert.assertEquals("Cluster100", resource.getPropertyValue(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID));
      Assert.assertEquals("KERBEROS", resource.getPropertyValue(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID));
      Assert.assertEquals("OK", resource.getPropertyValue("Services/attributes/kdc_validation_result"));
      Assert.assertEquals("", resource.getPropertyValue("Services/attributes/kdc_validation_failure_details"));
    }

    // verify
    verify(managementController, clusters, cluster, service0, serviceResponse0,
        ambariMetaInfo, stackId, serviceFactory, kerberosHeper);
  }

  @Test
  public void testGetResources_KerberosSpecificProperties_NoKDCValidation() throws Exception{
    AmbariManagementController managementController = createMock(AmbariManagementController.class);
    Clusters clusters = createNiceMock(Clusters.class);
    Cluster cluster = createNiceMock(Cluster.class);
    Service service0 = createNiceMock(Service.class);
    ServiceResponse serviceResponse0 = createNiceMock(ServiceResponse.class);

    StackId stackId = createNiceMock(StackId.class);
    ServiceFactory serviceFactory = createNiceMock(ServiceFactory.class);
    AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
    KerberosHelper kerberosHelper = createStrictMock(KerberosHelper.class);

    Map<String, Service> allResponseMap = new HashMap<>();
    allResponseMap.put("KERBEROS", service0);

    // set expectations
    expect(managementController.getClusters()).andReturn(clusters).anyTimes();
    expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
    expect(managementController.getHostComponents(EasyMock.anyObject())).
        andReturn(Collections.emptySet()).anyTimes();

    expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();

    expect(cluster.getServices()).andReturn(allResponseMap).anyTimes();
    expect(cluster.getService("KERBEROS")).andReturn(service0);

    expect(service0.convertToResponse()).andReturn(serviceResponse0).anyTimes();

    expect(service0.getName()).andReturn("Service100").anyTimes();

    expect(serviceResponse0.getClusterName()).andReturn("Cluster100").anyTimes();
    expect(serviceResponse0.getServiceName()).andReturn("KERBEROS").anyTimes();

    // The following call should NOT be made
    // kerberosHelper.validateKDCCredentials(cluster);

    // replay
    replay(managementController, clusters, cluster, service0, serviceResponse0,
        ambariMetaInfo, stackId, serviceFactory, kerberosHelper);

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

    ResourceProvider provider = getServiceProvider(managementController);
    // set kerberos helper on provider
    Class<?> c = provider.getClass();
    Field f = c.getDeclaredField("kerberosHelper");
    f.setAccessible(true);
    f.set(provider, kerberosHelper);

    // create the request
    Predicate predicate = new PredicateBuilder().property(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").and().
        property(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID).equals("KERBEROS").toPredicate();
    Request request = PropertyHelper.getReadRequest("ServiceInfo");
    Set<Resource> resources = provider.getResources(request, predicate);

    Assert.assertEquals(1, resources.size());
    for (Resource resource : resources) {
      Assert.assertEquals("Cluster100", resource.getPropertyValue(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID));
      Assert.assertEquals("KERBEROS", resource.getPropertyValue(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID));
    }

    // verify
    verify(managementController, clusters, cluster, service0, serviceResponse0,
        ambariMetaInfo, stackId, serviceFactory, kerberosHelper);
  }

  @Test
  public void testGetResources_KerberosSpecificProperties_KDCInvalidCredentials() throws Exception{
    AmbariManagementController managementController = createMock(AmbariManagementController.class);
    Clusters clusters = createNiceMock(Clusters.class);
    Cluster cluster = createNiceMock(Cluster.class);
    Service service0 = createNiceMock(Service.class);
    ServiceResponse serviceResponse0 = createNiceMock(ServiceResponse.class);

    StackId stackId = createNiceMock(StackId.class);
    ServiceFactory serviceFactory = createNiceMock(ServiceFactory.class);
    AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
    KerberosHelper kerberosHeper = createStrictMock(KerberosHelper.class);

    Map<String, Service> allResponseMap = new HashMap<>();
    allResponseMap.put("KERBEROS", service0);

    // set expectations
    expect(managementController.getClusters()).andReturn(clusters).anyTimes();
    expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
    expect(managementController.getHostComponents(EasyMock.anyObject())).
        andReturn(Collections.emptySet()).anyTimes();

    expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();

    expect(cluster.getServices()).andReturn(allResponseMap).anyTimes();
    expect(cluster.getService("KERBEROS")).andReturn(service0);

    expect(service0.convertToResponse()).andReturn(serviceResponse0).anyTimes();

    expect(service0.getName()).andReturn("Service100").anyTimes();

    expect(serviceResponse0.getClusterName()).andReturn("Cluster100").anyTimes();
    expect(serviceResponse0.getServiceName()).andReturn("KERBEROS").anyTimes();

    kerberosHeper.validateKDCCredentials(cluster);
    expectLastCall().andThrow(new KerberosAdminAuthenticationException("Invalid KDC administrator credentials."));

    // replay
    replay(managementController, clusters, cluster, service0, serviceResponse0,
        ambariMetaInfo, stackId, serviceFactory, kerberosHeper);

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

    ResourceProvider provider = getServiceProvider(managementController);
    // set kerberos helper on provider
    Class<?> c = provider.getClass();
    Field f = c.getDeclaredField("kerberosHelper");
    f.setAccessible(true);
    f.set(provider, kerberosHeper);

    // create the request
    Predicate predicate = new PredicateBuilder().property(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").and().
        property(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID).equals("KERBEROS").toPredicate();
    Request request = PropertyHelper.getReadRequest("ServiceInfo", "Services");
    Set<Resource> resources = provider.getResources(request, predicate);

    Assert.assertEquals(1, resources.size());
    for (Resource resource : resources) {
      Assert.assertEquals("Cluster100", resource.getPropertyValue(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID));
      Assert.assertEquals("KERBEROS", resource.getPropertyValue(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID));
      Assert.assertEquals("INVALID_CREDENTIALS", resource.getPropertyValue("Services/attributes/kdc_validation_result"));
      Assert.assertEquals("Invalid KDC administrator credentials.", resource.getPropertyValue("Services/attributes/kdc_validation_failure_details"));
    }

    // verify
    verify(managementController, clusters, cluster, service0, serviceResponse0,
        ambariMetaInfo, stackId, serviceFactory, kerberosHeper);
  }

  @Test
  public void testGetResources_KerberosSpecificProperties_KDCMissingCredentials() throws Exception{
    AmbariManagementController managementController = createMock(AmbariManagementController.class);
    Clusters clusters = createNiceMock(Clusters.class);
    Cluster cluster = createNiceMock(Cluster.class);
    Service service0 = createNiceMock(Service.class);
    ServiceResponse serviceResponse0 = createNiceMock(ServiceResponse.class);

    StackId stackId = createNiceMock(StackId.class);
    ServiceFactory serviceFactory = createNiceMock(ServiceFactory.class);
    AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
    KerberosHelper kerberosHeper = createStrictMock(KerberosHelper.class);

    Map<String, Service> allResponseMap = new HashMap<>();
    allResponseMap.put("KERBEROS", service0);

    // set expectations
    expect(managementController.getClusters()).andReturn(clusters).anyTimes();
    expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
    expect(managementController.getHostComponents(EasyMock.anyObject())).
        andReturn(Collections.emptySet()).anyTimes();

    expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();

    expect(cluster.getServices()).andReturn(allResponseMap).anyTimes();
    expect(cluster.getService("KERBEROS")).andReturn(service0);

    expect(service0.convertToResponse()).andReturn(serviceResponse0).anyTimes();

    expect(service0.getName()).andReturn("Service100").anyTimes();

    expect(serviceResponse0.getClusterName()).andReturn("Cluster100").anyTimes();
    expect(serviceResponse0.getServiceName()).andReturn("KERBEROS").anyTimes();

    kerberosHeper.validateKDCCredentials(cluster);
    expectLastCall().andThrow(new KerberosMissingAdminCredentialsException("Missing KDC administrator credentials."));

    // replay
    replay(managementController, clusters, cluster, service0, serviceResponse0,
        ambariMetaInfo, stackId, serviceFactory, kerberosHeper);

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

    ResourceProvider provider = getServiceProvider(managementController);
    // set kerberos helper on provider
    Class<?> c = provider.getClass();
    Field f = c.getDeclaredField("kerberosHelper");
    f.setAccessible(true);
    f.set(provider, kerberosHeper);

    // create the request
    Predicate predicate = new PredicateBuilder().property(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").and().
        property(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID).equals("KERBEROS").toPredicate();
    Request request = PropertyHelper.getReadRequest("ServiceInfo", "Services");
    Set<Resource> resources = provider.getResources(request, predicate);

    Assert.assertEquals(1, resources.size());
    for (Resource resource : resources) {
      Assert.assertEquals("Cluster100", resource.getPropertyValue(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID));
      Assert.assertEquals("KERBEROS", resource.getPropertyValue(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID));
      Assert.assertEquals("MISSING_CREDENTIALS", resource.getPropertyValue("Services/attributes/kdc_validation_result"));
      Assert.assertEquals("Missing KDC administrator credentials.", resource.getPropertyValue("Services/attributes/kdc_validation_failure_details"));
    }

    // verify
    verify(managementController, clusters, cluster, service0, serviceResponse0,
        ambariMetaInfo, stackId, serviceFactory, kerberosHeper);
  }

  @Test
  public void testUpdateResourcesAsAdministrator() throws Exception{
    testUpdateResources(TestAuthenticationFactory.createAdministrator());
  }

  @Test
  public void testUpdateResourcesAsClusterAdministrator() throws Exception{
    testUpdateResources(TestAuthenticationFactory.createClusterAdministrator());
  }

  @Test
  public void testUpdateResourcesAsServiceAdministrator() throws Exception{
    testUpdateResources(TestAuthenticationFactory.createServiceAdministrator());
  }

  private void testUpdateResources(Authentication authentication) throws Exception{
    MaintenanceStateHelper maintenanceStateHelper = createNiceMock(MaintenanceStateHelper.class);
    RepositoryVersionDAO repositoryVersionDAO = createNiceMock(RepositoryVersionDAO.class);
    AmbariManagementController managementController = createMock(AmbariManagementController.class);
    Clusters clusters = createNiceMock(Clusters.class);
    Cluster cluster = createNiceMock(Cluster.class);
    Service service0 = createNiceMock(Service.class);
    ServiceFactory serviceFactory = createNiceMock(ServiceFactory.class);
    AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
    RequestStageContainer requestStages = createNiceMock(RequestStageContainer.class);
    RequestStatusResponse requestStatusResponse = createNiceMock(RequestStatusResponse.class);
    RoleCommandOrder rco = createNiceMock(RoleCommandOrder.class);
    StackId stackId = createNiceMock(StackId.class);
    ServiceInfo serviceInfo = createNiceMock(ServiceInfo.class);

    Map<String, String> mapRequestProps = new HashMap<>();
    mapRequestProps.put("context", "Called from a test");

    // set expectations
    expect(managementController.getClusters()).andReturn(clusters).anyTimes();
    expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();

    expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();

    expect(cluster.getClusterId()).andReturn(2L).anyTimes();
    expect(cluster.getService("Service102")).andReturn(service0);

    expect(service0.getDesiredState()).andReturn(State.INSTALLED).anyTimes();
    expect(service0.getServiceComponents()).andReturn(Collections.emptyMap()).anyTimes();

    expect(stackId.getStackId()).andReturn("HDP-2.5").anyTimes();
    expect(stackId.getStackName()).andReturn("HDP").anyTimes();
    expect(stackId.getStackVersion()).andReturn("2.5").anyTimes();
    expect(service0.getDesiredStackId()).andReturn(stackId).anyTimes();
    expect(service0.getName()).andReturn("Service102").anyTimes();
    expect(serviceInfo.isCredentialStoreSupported()).andReturn(true).anyTimes();
    expect(serviceInfo.isCredentialStoreEnabled()).andReturn(false).anyTimes();
    expect(ambariMetaInfo.getService("HDP", "2.5", "Service102")).andReturn(serviceInfo).anyTimes();

    Capture<Map<String, String>> requestPropertiesCapture = newCapture();
    Capture<Map<State, List<Service>>> changedServicesCapture = newCapture();
    Capture<Map<State, List<ServiceComponent>>> changedCompsCapture = newCapture();
