content: /*
   * 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.cassandra.db.guardrails;
  
  import static java.lang.String.format;
  
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Set;
  import java.util.stream.Collectors;
  
  import org.apache.cassandra.config.GuardrailsOptions;
  import org.apache.cassandra.cql3.statements.schema.KeyspaceAttributes;
  import org.apache.cassandra.schema.KeyspaceParams;
  import org.apache.cassandra.utils.LocalizeString;
  import org.junit.Before;
  import org.junit.Test;
  
  import com.google.common.collect.ImmutableSet;
  import com.google.common.collect.Sets;
  
  /**
   * Tests the guardrail for keyspace properties, {@link Guardrails#keyspaceProperties}.
   */
  public class GuardrailKeyspacePropertiesTest extends GuardrailTester
  {
      private static final String CREATE_KEYSPACE = "CREATE KEYSPACE %s %s";
      private static final String ALTER_KEYSPACE = "ALTER KEYSPACE %s WITH %s";
  
      private static final String WARNED_PROPERTY_NAME = "keyspace_properties_warned";
      private static final String IGNORED_PROPERTY_NAME = "keyspace_properties_ignored";
      private static final String DISALLOWED_PROPERTY_NAME = "keyspace_properties_disallowed";
  
      private static final String DURABLE_WRITES = "durable_writes = false";
      private static final String REPLICATION = "WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 1}";
      
      private static final String WARN_MESSAGE = "Provided values [%s] are not recommended for Keyspace Properties";
      private static final String IGNORE_MESSAGE = "Ignoring provided values [%s] as they are not supported for Keyspace Properties";
      private static final String FAIL_MESSAGE = "Provided values [%s] are not allowed for Keyspace Properties";
  
      public GuardrailKeyspacePropertiesTest()
      {
          super(Guardrails.keyspaceProperties);
      }
  
      @Before
      public void before()
      {
          // Set up basic test configuration - only allow "replication" by default
          Set<String> allowed = new HashSet<>(List.of("replication", "durable_writes"));
          Set<String> allKeyspaceProperties = Arrays.stream(KeyspaceParams.Option.values())
                                                    .map(KeyspaceParams.Option::toString)
                                                    .collect(Collectors.toSet());
          guardrails().setKeyspacePropertiesDisallowed(allKeyspaceProperties
                                                       .stream()
                                                       .filter(p -> !allowed.contains(p))
                                                       .map(LocalizeString::toUpperCaseLocalized)
                                                       .collect(Collectors.toSet()));
          // Configure "durable_writes" to be warned about (tests will override as needed)
          guardrails().setKeyspacePropertiesWarned("durable_writes");
          guardrails().setKeyspacePropertiesIgnored(Collections.emptySet());
      }
  
      @Test
      public void testConfigValidation()
      {
          String message = "Invalid value for %s: null is not allowed";
          assertInvalidProperty(Guardrails::setKeyspacePropertiesWarned, (Set<String>) null, message, WARNED_PROPERTY_NAME);
          assertInvalidProperty(Guardrails::setKeyspacePropertiesIgnored, (Set<String>) null, message, IGNORED_PROPERTY_NAME);
          assertInvalidProperty(Guardrails::setKeyspacePropertiesDisallowed, (Set<String>) null, message, DISALLOWED_PROPERTY_NAME);
  
          assertValidProperty(Collections.emptySet());
          assertValidProperty(getValidKeyspaceProperties());
  
          assertValidPropertyCSV("");
          assertValidPropertyCSV(String.join(",", getValidKeyspaceProperties()));
  
          assertInvalidProperty(Collections.singleton("invalid"), Collections.singleton("invalid"));
          assertInvalidProperty(ImmutableSet.of("invalid1", "invalid2"), ImmutableSet.of("invalid1", "invalid2"));
          assertInvalidProperty(ImmutableSet.of("replication", "invalid2"), ImmutableSet.of("replication"));
          assertInvalidProperty(ImmutableSet.of("durable_writes", "invalid2", "replication"), ImmutableSet.of("replication"));
          assertInvalidProperty(ImmutableSet.of("durable_writes", "replication", "fast_path"), ImmutableSet.of("replication"));
  
          assertInvalidPropertyCSV("invalid", "[invalid]");
          assertInvalidPropertyCSV("invalid1,invalid2", "[invalid1, invalid2]");
          assertInvalidPropertyCSV("replication,invalid1", "[replication]");
          assertInvalidPropertyCSV("durable_writes,invalid2,replication", "[replication]");
          assertInvalidPropertyCSV("durable_writes,replication,fast_path", "[replication]");
      }
  
      private void assertValidProperty(Set<String> properties)
      {
          assertValidProperty(Guardrails::setKeyspacePropertiesWarned, Guardrails::getKeyspacePropertiesWarned, properties);
          assertValidProperty(Guardrails::setKeyspacePropertiesIgnored, Guardrails::getKeyspacePropertiesIgnored, properties);
          assertValidProperty(Guardrails::setKeyspacePropertiesDisallowed, Guardrails::getKeyspacePropertiesDisallowed, properties);
      }
  
      private void assertValidPropertyCSV(String csv)
      {
          csv = sortCSV(csv);
          assertValidProperty(Guardrails::setKeyspacePropertiesWarnedCSV, g -> sortCSV(g.getKeyspacePropertiesWarnedCSV()), csv);
          assertValidProperty(Guardrails::setKeyspacePropertiesIgnoredCSV, g -> sortCSV(g.getKeyspacePropertiesIgnoredCSV()), csv);
          assertValidProperty(Guardrails::setKeyspacePropertiesDisallowedCSV, g -> sortCSV(g.getKeyspacePropertiesDisallowedCSV()), csv);
      }
  
      private void assertInvalidProperty(Set<String> properties, Set<String> invalidProperties)
      {
          if (invalidProperties.contains("replication"))
          {
              String replicationMessage = "Invalid value for %s: '%s' is a required keyspace property";
              assertInvalidProperty(Guardrails::setKeyspacePropertiesWarned, properties, format(replicationMessage, WARNED_PROPERTY_NAME, "replication"), WARNED_PROPERTY_NAME);
              assertInvalidProperty(Guardrails::setKeyspacePropertiesIgnored, properties, format(replicationMessage, IGNORED_PROPERTY_NAME, "replication"), IGNORED_PROPERTY_NAME);
              assertInvalidProperty(Guardrails::setKeyspacePropertiesDisallowed, properties, format(replicationMessage, DISALLOWED_PROPERTY_NAME, "replication"), DISALLOWED_PROPERTY_NAME);
          }
          else 
          {
              String message = GuardrailsOptions.invalidValueMessage(WARNED_PROPERTY_NAME, invalidProperties, "keyspace");
              assertInvalidProperty(Guardrails::setKeyspacePropertiesWarned, properties, message, WARNED_PROPERTY_NAME);
  
              message = GuardrailsOptions.invalidValueMessage(IGNORED_PROPERTY_NAME, invalidProperties, "keyspace");
              assertInvalidProperty(Guardrails::setKeyspacePropertiesIgnored, properties, message, IGNORED_PROPERTY_NAME);
  
              message = GuardrailsOptions.invalidValueMessage(DISALLOWED_PROPERTY_NAME, invalidProperties, "keyspace");
              assertInvalidProperty(Guardrails::setKeyspacePropertiesDisallowed, properties, message, DISALLOWED_PROPERTY_NAME);
          }
      }
  
      private void assertInvalidPropertyCSV(String properties, String rejected)
      {
          // Parse the rejected string to extract individual properties
          // Expected format is "[prop1, prop2]" so we need to extract the properties inside brackets
          String cleanRejected = rejected.substring(1, rejected.length() - 1); // Remove brackets
          Set<String> rejectedSet = Arrays.stream(cleanRejected.split(","))
                                          .map(String::trim)
                                          .collect(Collectors.toSet());
  
          if (rejectedSet.contains("replication"))
          {
              String replicationMessage = "Invalid value for %s: '%s' is a required keyspace property";
              assertInvalidProperty(Guardrails::setKeyspacePropertiesWarnedCSV, properties, format(replicationMessage, WARNED_PROPERTY_NAME, "replication"), WARNED_PROPERTY_NAME, rejected);
              assertInvalidProperty(Guardrails::setKeyspacePropertiesIgnoredCSV, properties, format(replicationMessage, IGNORED_PROPERTY_NAME, "replication"), IGNORED_PROPERTY_NAME, rejected);
              assertInvalidProperty(Guardrails::setKeyspacePropertiesDisallowedCSV, properties, format(replicationMessage, DISALLOWED_PROPERTY_NAME, "replication"), DISALLOWED_PROPERTY_NAME, rejected);
          }
          else 
          {
              String message = GuardrailsOptions.invalidValueMessage(WARNED_PROPERTY_NAME, rejectedSet, "keyspace");
              assertInvalidProperty(Guardrails::setKeyspacePropertiesWarnedCSV, properties, message, WARNED_PROPERTY_NAME, rejected);
  
              message = GuardrailsOptions.invalidValueMessage(IGNORED_PROPERTY_NAME, rejectedSet, "keyspace");
              assertInvalidProperty(Guardrails::setKeyspacePropertiesIgnoredCSV, properties, message, IGNORED_PROPERTY_NAME, rejected);
  
              message = GuardrailsOptions.invalidValueMessage(DISALLOWED_PROPERTY_NAME, rejectedSet, "keyspace");
              assertInvalidProperty(Guardrails::setKeyspacePropertiesDisallowedCSV, properties, message, DISALLOWED_PROPERTY_NAME, rejected);
          }
      }
  
      @Test
      public void testCreateKeyspaceWithWarned() throws Throwable
      {
          String ks = createKeyspaceName();
          String statement = format(CREATE_KEYSPACE, ks, REPLICATION + " AND " + DURABLE_WRITES);
          assertWarns(statement, format(WARN_MESSAGE, "durable_writes"));
      }
  
      @Test
      public void testCreateKeyspaceWithIgnored() throws Throwable
      {
          guardrails().setKeyspacePropertiesIgnored("durable_writes");
          guardrails().setKeyspacePropertiesWarned(Collections.emptySet());
          String ks = createKeyspaceName();
          String statement = format(CREATE_KEYSPACE, ks, REPLICATION + " AND " + DURABLE_WRITES);
          assertWarns(statement, format(IGNORE_MESSAGE, "durable_writes"));
      }
  
      @Test
      public void testCreateKeyspaceWithDisallowed() throws Throwable
      {
          guardrails().setKeyspacePropertiesDisallowed("durable_writes");
          String ks = createKeyspaceName();
          String statement = format(CREATE_KEYSPACE, ks, REPLICATION + " AND " + DURABLE_WRITES);
          assertFails(statement, format(FAIL_MESSAGE, "durable_writes"));
      }
  
      @Test
      public void testAlterKeyspaceWithWarned() throws Throwable
      {
          String ks = createKeyspaceName();
          execute(format(CREATE_KEYSPACE, ks, "WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 1}"));
          String statement = format(ALTER_KEYSPACE, ks, DURABLE_WRITES);
          assertWarns(statement, format(WARN_MESSAGE, "durable_writes"));
      }
  
      @Test
      public void testAlterKeyspaceWithDisallowed() throws Throwable
      {
          guardrails().setKeyspacePropertiesDisallowed("durable_writes");
          String ks = createKeyspaceName();
          execute(format(CREATE_KEYSPACE, ks, "WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 1}"));
          String statement = format(ALTER_KEYSPACE, ks, DURABLE_WRITES);
          assertFails(statement, format(FAIL_MESSAGE, "durable_writes"));
      }
  
      @Test
      public void testDisallowedPropertyTakesPrecedenceOverWarned() throws Throwable
      {
          // Set a property as both warned and disallowed - disallowed should take precedence
          guardrails().setKeyspacePropertiesWarned("durable_writes");
          guardrails().setKeyspacePropertiesDisallowed("durable_writes");
          String ks = createKeyspaceName();
          String statement = format(CREATE_KEYSPACE, ks, REPLICATION + " AND " + DURABLE_WRITES);
          assertFails(statement, format(FAIL_MESSAGE, "durable_writes"));
      }
  
      @Test
      public void testValidKeyspaceCreation() throws Throwable
      {
          // Test normal keyspace creation case without any guardrail violations
          guardrails().setKeyspacePropertiesDisallowed(Collections.emptySet());
          guardrails().setKeyspacePropertiesWarned(Collections.emptySet());
          String ks = createKeyspaceName();
          String statement = format(CREATE_KEYSPACE, ks, REPLICATION);
          assertValid(statement);
      }
  
      private Set<String> getValidKeyspaceProperties()
      {
          return Sets.difference(KeyspaceAttributes.allKeywords(), KeyspaceAttributes.requiredKeywords());
      }
  }
  