/*
 * 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
 *
 *   https://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.accumulo.core.iterators.user;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Map.Entry;
import java.util.TreeMap;

import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.Filter;
import org.apache.accumulo.core.iteratorsImpl.system.SortedMapIterator;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.hadoop.io.Text;
import org.junit.jupiter.api.Test;

public class VisibilityFilterTest {

  private static final Collection<ByteSequence> EMPTY_COL_FAMS = new ArrayList<>();

  private static final Text BAD = new Text("bad");
  private static final Text GOOD = new Text("good");
  private static final Text EMPTY_VIS = new Text("");
  private static final Text GOOD_VIS = new Text("abc|def");
  private static final Text HIDDEN_VIS = new Text("abc&def&ghi");
  private static final Text BAD_VIS = new Text("&");
  private static final Value EMPTY_VALUE = new Value();

  private TreeMap<Key,Value> createUnprotectedSource(int numPublic, int numHidden) {
    TreeMap<Key,Value> source = new TreeMap<>();
    for (int i = 0; i < numPublic; i++) {
      source.put(new Key(new Text(String.format("%03d", i)), GOOD, GOOD, EMPTY_VIS), EMPTY_VALUE);
    }
    for (int i = 0; i < numHidden; i++) {
      source.put(new Key(new Text(String.format("%03d", i)), BAD, BAD, GOOD_VIS), EMPTY_VALUE);
    }
    return source;
  }

  private TreeMap<Key,Value> createPollutedSource(int numGood, int numBad) {
    TreeMap<Key,Value> source = new TreeMap<>();
    for (int i = 0; i < numGood; i++) {
      source.put(new Key(new Text(String.format("%03d", i)), GOOD, GOOD, GOOD_VIS), EMPTY_VALUE);
    }
    for (int i = 0; i < numBad; i++) {
      source.put(new Key(new Text(String.format("%03d", i)), BAD, BAD, BAD_VIS), EMPTY_VALUE);
    }
    return source;
  }

  private TreeMap<Key,Value> createSourceWithHiddenData(int numViewable, int numHidden) {
    TreeMap<Key,Value> source = new TreeMap<>();
    for (int i = 0; i < numViewable; i++) {
      source.put(new Key(new Text(String.format("%03d", i)), GOOD, GOOD, GOOD_VIS), EMPTY_VALUE);
    }
    for (int i = 0; i < numHidden; i++) {
      source.put(new Key(new Text(String.format("%03d", i)), BAD, BAD, HIDDEN_VIS), EMPTY_VALUE);
    }
    return source;
  }

  private void verify(TreeMap<Key,Value> source, int expectedSourceSize, Map<String,String> options,
      Text expectedCF, Text expectedCQ, Text expectedCV, int expectedFinalCount)
      throws IOException {
    assertEquals(expectedSourceSize, source.size());

    Filter filter = new VisibilityFilter();
    filter.init(new SortedMapIterator(source), options, null);
    filter.seek(new Range(), EMPTY_COL_FAMS, false);

    int count = 0;
    while (filter.hasTop()) {
      count++;
      // System.out.println(DefaultFormatter.formatEntry(
      // Collections.singletonMap(filter.getTopKey(),
      // filter.getTopValue()).entrySet().iterator().next(),
      // false));
      assertEquals(expectedCF, filter.getTopKey().getColumnFamily());
      assertEquals(expectedCQ, filter.getTopKey().getColumnQualifier());
      assertEquals(expectedCV, filter.getTopKey().getColumnVisibility());
      filter.next();
    }
    assertEquals(expectedFinalCount, count);
  }

  @Test
  public void testAllowValidLabelsOnly() throws IOException {
    IteratorSetting is = new IteratorSetting(1, VisibilityFilter.class);
    VisibilityFilter.filterInvalidLabelsOnly(is, true);

    TreeMap<Key,Value> source = createPollutedSource(1, 2);
    verify(source, 3, is.getOptions(), GOOD, GOOD, GOOD_VIS, 1);

    source = createPollutedSource(30, 500);
    verify(source, 530, is.getOptions(), GOOD, GOOD, GOOD_VIS, 30);

    source = createPollutedSource(1000, 500);
    verify(source, 1500, is.getOptions(), GOOD, GOOD, GOOD_VIS, 1000);
  }

  @Test
  public void testAllowBadLabelsOnly() throws IOException {
    IteratorSetting is = new IteratorSetting(1, VisibilityFilter.class);
    VisibilityFilter.setNegate(is, true);
    VisibilityFilter.filterInvalidLabelsOnly(is, true);

    TreeMap<Key,Value> source = createPollutedSource(1, 2);
    verify(source, 3, is.getOptions(), BAD, BAD, BAD_VIS, 2);

    source = createPollutedSource(30, 500);
    verify(source, 530, is.getOptions(), BAD, BAD, BAD_VIS, 500);

    source = createPollutedSource(1000, 500);
    verify(source, 1500, is.getOptions(), BAD, BAD, BAD_VIS, 500);
  }

  @Test
  public void testAllowAuthorizedLabelsOnly() throws IOException {
    IteratorSetting is = new IteratorSetting(1, VisibilityFilter.class);
    VisibilityFilter.setAuthorizations(is, new Authorizations("def"));

    TreeMap<Key,Value> source = createSourceWithHiddenData(1, 2);
    verify(source, 3, is.getOptions(), GOOD, GOOD, GOOD_VIS, 1);

    source = createSourceWithHiddenData(30, 500);
    verify(source, 530, is.getOptions(), GOOD, GOOD, GOOD_VIS, 30);

    source = createSourceWithHiddenData(1000, 500);
    verify(source, 1500, is.getOptions(), GOOD, GOOD, GOOD_VIS, 1000);
  }

  @Test
  public void testAllowUnauthorizedLabelsOnly() throws IOException {
    IteratorSetting is = new IteratorSetting(1, VisibilityFilter.class);
    VisibilityFilter.setNegate(is, true);
    VisibilityFilter.setAuthorizations(is, new Authorizations("def"));

    TreeMap<Key,Value> source = createSourceWithHiddenData(1, 2);
    verify(source, 3, is.getOptions(), BAD, BAD, HIDDEN_VIS, 2);

    source = createSourceWithHiddenData(30, 500);
    verify(source, 530, is.getOptions(), BAD, BAD, HIDDEN_VIS, 500);

    source = createSourceWithHiddenData(1000, 500);
    verify(source, 1500, is.getOptions(), BAD, BAD, HIDDEN_VIS, 500);
  }

  @Test
  public void testNoLabels() throws IOException {
    IteratorSetting is = new IteratorSetting(1, VisibilityFilter.class);
    VisibilityFilter.setNegate(is, false);
    VisibilityFilter.setAuthorizations(is, new Authorizations());

    TreeMap<Key,Value> source = createUnprotectedSource(5, 2);
    verify(source, 7, is.getOptions(), GOOD, GOOD, EMPTY_VIS, 5);

    VisibilityFilter.setNegate(is, true);
    verify(source, 7, is.getOptions(), BAD, BAD, GOOD_VIS, 2);
  }

  @Test
  public void testFilterUnauthorizedAndBad() throws IOException {
    /*
     * if not explicitly filtering bad labels, they will still be filtered while validating against
     * authorizations, but it will be very verbose in the logs
     */
    IteratorSetting is = new IteratorSetting(1, VisibilityFilter.class);
    VisibilityFilter.setAuthorizations(is, new Authorizations("def"));

    TreeMap<Key,Value> source = createSourceWithHiddenData(1, 5);
    for (Entry<Key,Value> entry : createPollutedSource(0, 1).entrySet()) {
      source.put(entry.getKey(), entry.getValue());
    }

    verify(source, 7, is.getOptions(), GOOD, GOOD, GOOD_VIS, 1);
  }

  @Test
  public void testCommaSeparatedAuthorizations() throws IOException {
    Map<String,String> options = Collections.singletonMap("auths", "x,def,y");

    TreeMap<Key,Value> source = createSourceWithHiddenData(1, 2);
    verify(source, 3, options, GOOD, GOOD, GOOD_VIS, 1);

    source = createSourceWithHiddenData(30, 500);
    verify(source, 530, options, GOOD, GOOD, GOOD_VIS, 30);

    source = createSourceWithHiddenData(1000, 500);
    verify(source, 1500, options, GOOD, GOOD, GOOD_VIS, 1000);
  }

  @Test
  public void testSerializedAuthorizations() throws IOException {
    Map<String,String> options =
        Collections.singletonMap("auths", new Authorizations("x", "def", "y").serialize());

    TreeMap<Key,Value> source = createSourceWithHiddenData(1, 2);
    verify(source, 3, options, GOOD, GOOD, GOOD_VIS, 1);

    source = createSourceWithHiddenData(30, 500);
    verify(source, 530, options, GOOD, GOOD, GOOD_VIS, 30);

    source = createSourceWithHiddenData(1000, 500);
    verify(source, 1500, options, GOOD, GOOD, GOOD_VIS, 1000);
  }

  @Test
  public void testStaticConfigurators() {
    IteratorSetting is = new IteratorSetting(1, VisibilityFilter.class);
    VisibilityFilter.filterInvalidLabelsOnly(is, false);
    VisibilityFilter.setNegate(is, true);
    VisibilityFilter.setAuthorizations(is, new Authorizations("abc", "def"));

    Map<String,String> opts = is.getOptions();
    assertEquals("false", opts.get("filterInvalid"));
    assertEquals("true", opts.get("negate"));
    assertEquals(new Authorizations("abc", "def").serialize(), opts.get("auths"));
  }

  @Test
  public void testDeepCopyAfterInit() throws IOException {
    IteratorSetting is = new IteratorSetting(1, VisibilityFilter.class);
    VisibilityFilter.setAuthorizations(is, new Authorizations("abc"));
    Map<String,String> opts = is.getOptions();
    Filter filter = new VisibilityFilter();
    TreeMap<Key,Value> source = new TreeMap<>();
    filter.init(new SortedMapIterator(source), opts, null);
    Filter copyFilter = (Filter) filter.deepCopy(null);
    Key k = new Key("row", "cf", "cq", "abc");
    assertTrue(copyFilter.accept(k, new Value()));
  }

}
