/*
 * 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.geode.security;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.util.Properties;

import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;

import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionService;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.query.CqAttributesFactory;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.Query;
import org.apache.geode.examples.SimpleSecurityManager;
import org.apache.geode.pdx.JSONFormatter;
import org.apache.geode.pdx.PdxInstance;
import org.apache.geode.test.dunit.rules.ClusterStartupRule;
import org.apache.geode.test.dunit.rules.MemberVM;
import org.apache.geode.test.junit.rules.ClientCacheRule;

public class MultiUserAPIDUnitTest {
  @ClassRule
  public static ClusterStartupRule cluster = new ClusterStartupRule();

  private static MemberVM server;

  @Rule
  public ClientCacheRule client = new ClientCacheRule();

  @BeforeClass
  public static void setUp() throws Exception {
    MemberVM locator =
        cluster.startLocatorVM(0, c -> c.withSecurityManager(SimpleSecurityManager.class));
    server = cluster.startServerVM(1, s -> s.withCredential("cluster", "cluster")
        .withConnectionToLocator(locator.getPort()));

    server.invoke(() -> {
      ClusterStartupRule.memberStarter.createRegion(RegionShortcut.REPLICATE, "region");
    });
  }

  @Test
  public void testSingleUserUnsupportedAPIs() throws Exception {
    client.withCredential("stranger", "stranger").withMultiUser(false)
        .withServerConnection(server.getPort());
    ClientCache clientCache = client.createCache();

    assertThatThrownBy(() -> clientCache.createAuthenticatedView(new Properties()))
        .isInstanceOf(IllegalStateException.class)
        .hasMessageContaining("did not have multiuser-authentication set to true");
  }

  @Test
  public void testMultiUserUnsupportedAPIs() throws Exception {
    client.withCredential("stranger", "stranger")
        .withPoolSubscription(true)
        .withMultiUser(true)
        .withServerConnection(server.getPort());
    client.createCache();
    Region region = client.createProxyRegion("region");
    Pool pool = client.getCache().getDefaultPool();

    RegionService regionService = client.createAuthenticatedView("data", "data");
    Region regionView = regionService.getRegion("region");

    assertThatThrownBy(() -> region.create("key", "value"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> region.put("key", "value"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> region.get("key"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> region.containsKeyOnServer("key"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> region.remove("key"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> region.destroy("key"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> region.destroyRegion())
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> region.registerInterest("key"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> region.clear()).isInstanceOf(UnsupportedOperationException.class);


    assertThatThrownBy(() -> regionView.createSubregion("subRegion", null))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionView.forceRolling())
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionView.getAttributesMutator())
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionView.loadSnapshot(null))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionView.saveSnapshot(null))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionView.registerInterest("key"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionView.setUserAttribute(null))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionView.unregisterInterestRegex("*"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionView.localDestroy("key"))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionView.localInvalidate("key"))
        .isInstanceOf(UnsupportedOperationException.class);

    assertThatThrownBy(() -> FunctionService.onRegion(region))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> FunctionService.onServer(pool))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> FunctionService.onServers(pool))
        .isInstanceOf(UnsupportedOperationException.class);


    assertThatThrownBy(() -> {
      Query query = pool.getQueryService().newQuery("SELECT * FROM /region");
      query.execute();
    }).isInstanceOf(UnsupportedOperationException.class);

    CqQuery cqQuery =
        pool.getQueryService().newCq("SELECT * FROM /region",
            new CqAttributesFactory().create());
    assertThatThrownBy(() -> cqQuery.execute())
        .hasCauseInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> cqQuery.executeWithInitialResults())
        .hasCauseInstanceOf(UnsupportedOperationException.class);

    assertThatThrownBy(() -> regionService.getQueryService().getIndexes())
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionService.getQueryService().getIndexes(null))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionService.getQueryService().createIndex(null, null, null))
        .isInstanceOf(UnsupportedOperationException.class);
    assertThatThrownBy(() -> regionService.getQueryService().removeIndexes())
        .isInstanceOf(UnsupportedOperationException.class);
  }

  @Test
  public void singleUserSubscriptionEnabled() throws Exception {
    ClientCache clientCache = client.withPoolSubscription(true)
        .withServerConnection(server.getPort())
        .createCache();
    assertThat(clientCache).isNotNull();
    assertThatThrownBy(() -> client.createProxyRegion("region"))
        .isInstanceOf(AuthenticationRequiredException.class);
  }

  @Test
  public void singleUserSubscriptionNotEnabled() throws Exception {
    ClientCache clientCache = client.withPoolSubscription(false)
        .withServerConnection(server.getPort())
        .createCache();
    assertThat(clientCache).isNotNull();
    Region region = client.createProxyRegion("region");
    assertThat(region).isNotNull();
    assertThatThrownBy(() -> region.put("key", "value"))
        .hasCauseInstanceOf(AuthenticationRequiredException.class);
  }

  @Test
  public void noCredentialCanCreateCacheWithMultiUser() throws Exception {
    ClientCache clientCache = client.withServerConnection(server.getPort()).withMultiUser(true)
        .createCache();

    // with multiuser, client cache can be created without credential and proxy region can be
    // created. It's the same as single user with no pool subscription case
    Region region = client.createProxyRegion("region");
    assertThat(clientCache).isNotNull();
    assertThat(region).isNotNull();
    assertThatThrownBy(() -> region.put("key", "value"))
        .isInstanceOf(UnsupportedOperationException.class);

    RegionService validView = client.createAuthenticatedView("data", "data");
    Region<String, Object> validRegion = validView.getRegion("region");
    validRegion.put("key", "value");

    RegionService notAuthenticatedView = client.createAuthenticatedView("test", "invalid");
    Region<String, Object> notAuthenticatedRegion = notAuthenticatedView.getRegion("region");
    assertThatThrownBy(() -> notAuthenticatedRegion.put("key", "value"))
        .hasCauseInstanceOf(AuthenticationFailedException.class);

    RegionService notAuthorizedView = client.createAuthenticatedView("test", "test");
    Region<String, Object> notAuthorizedRegion = notAuthorizedView.getRegion("region");
    assertThatThrownBy(() -> notAuthorizedRegion.put("key", "value"))
        .hasCauseInstanceOf(NotAuthorizedException.class);
  }

  @Test
  public void jsonFormatterOnTheClientWithSingleUser() throws Exception {
    client.withCredential("data", "data").withMultiUser(false)
        .withServerConnection(server.getPort()).createCache();
    Region region = client.createProxyRegion("region");

    // with single user, the static method in JSONFormatter can be used
    String json = "{\"key\" : \"value\"}";
    PdxInstance value = JSONFormatter.fromJSON(json);
    region.put("key", value);
  }

  @Test
  public void jsonFormatterOnTheClientWithMultiUser() throws Exception {
    client.withMultiUser(true).withServerConnection(server.getPort()).createCache();
    client.createProxyRegion("region");

    RegionService regionService = client.createAuthenticatedView("data", "data");
    Region<String, Object> regionView = regionService.getRegion("region");

    String json = "{\"key\" : \"value\"}";

    // in multiUser view, can not use the static methods in JSONFormatter directly
    assertThatThrownBy(() -> JSONFormatter.fromJSON(json))
        .hasCauseInstanceOf(UnsupportedOperationException.class);

    // need to get the jsonFormatter from the proxy cache
    PdxInstance value = regionService.getJsonFormatter().toPdxInstance(json);
    regionView.put("key", value);
  }
}