/*
 * 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.accumulo.examples.dirlist;

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

import java.util.ArrayList;
import java.util.Map.Entry;

import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.examples.cli.BatchWriterOpts;
import org.apache.accumulo.examples.cli.ScannerOpts;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
import org.apache.accumulo.test.functional.ConfigurableMacBase;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

public class CountIT extends ConfigurableMacBase {

  private AccumuloClient client;
  private String tableName;

  @Override
  protected void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
    cfg.setProperty(Property.TSERV_NATIVEMAP_ENABLED, "false");
  }

  @BeforeEach
  public void setupInstance() throws Exception {
    tableName = getUniqueNames(1)[0];
    client = Accumulo.newClient().from(getClientProperties()).build();
    client.tableOperations().create(tableName);
    try (BatchWriter bw = client.createBatchWriter(tableName, new BatchWriterConfig())) {
      ColumnVisibility cv = new ColumnVisibility();
      // / has 1 dir
      // /local has 2 dirs 1 file
      // /local/user1 has 2 files
      bw.addMutation(Ingest.buildMutation(cv, "/local", true, false, true, 272, 12345, null));
      bw.addMutation(Ingest.buildMutation(cv, "/local/user1", true, false, true, 272, 12345, null));
      bw.addMutation(Ingest.buildMutation(cv, "/local/user2", true, false, true, 272, 12345, null));
      bw.addMutation(
          Ingest.buildMutation(cv, "/local/file", false, false, false, 1024, 12345, null));
      bw.addMutation(
          Ingest.buildMutation(cv, "/local/file", false, false, false, 1024, 23456, null));
      bw.addMutation(
          Ingest.buildMutation(cv, "/local/user1/file1", false, false, false, 2024, 12345, null));
      bw.addMutation(
          Ingest.buildMutation(cv, "/local/user1/file2", false, false, false, 1028, 23456, null));
    }
  }

  @AfterEach
  public void teardown() {
    client.close();
  }

  @Test
  public void test() throws Exception {

    ScannerOpts scanOpts = new ScannerOpts();
    BatchWriterOpts bwOpts = new BatchWriterOpts();
    FileCount fc = new FileCount(client, tableName, Authorizations.EMPTY, new ColumnVisibility(),
        scanOpts, bwOpts);

    ArrayList<Pair<String,String>> expected = new ArrayList<>();
    expected.add(new Pair<>(QueryUtil.getRow("").toString(), "1,0,3,3"));
    expected.add(new Pair<>(QueryUtil.getRow("/local").toString(), "2,1,2,3"));
    expected.add(new Pair<>(QueryUtil.getRow("/local/user1").toString(), "0,2,0,2"));
    expected.add(new Pair<>(QueryUtil.getRow("/local/user2").toString(), "0,0,0,0"));

    int actualCount = 0;
    try (Scanner scanner = client.createScanner(tableName, new Authorizations())) {
      scanner.fetchColumn("dir", "counts");
      assertFalse(scanner.iterator().hasNext());

      fc.run();

      for (Entry<Key,Value> e : scanner) {
        assertEquals(e.getKey().getRow().toString(), expected.get(actualCount).getFirst());
        assertEquals(e.getValue().toString(), expected.get(actualCount).getSecond());
        actualCount++;
      }
    }
    assertEquals(expected.size(), actualCount);
  }
}