/*
 * 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.manager.upgrade;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.accumulo.manager.upgrade.Upgrader11to12.UPGRADE_FAMILIES;
import static org.apache.accumulo.manager.upgrade.Upgrader11to12.ZNAMESPACE_NAME;
import static org.apache.accumulo.manager.upgrade.Upgrader11to12.ZTABLE_NAME;
import static org.easymock.EasyMock.aryEq;
import static org.easymock.EasyMock.capture;
import static org.easymock.EasyMock.createMock;
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.isA;
import static org.easymock.EasyMock.mock;
import static org.easymock.EasyMock.newCapture;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.UUID;
import java.util.stream.Collectors;

import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.NamespaceNotFoundException;
import org.apache.accumulo.core.clientImpl.Namespace;
import org.apache.accumulo.core.clientImpl.NamespaceMapping;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.ColumnUpdate;
import org.apache.accumulo.core.data.InstanceId;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.NamespaceId;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.data.constraints.DefaultKeySizeConstraint;
import org.apache.accumulo.core.fate.zookeeper.ZooReader;
import org.apache.accumulo.core.fate.zookeeper.ZooReaderWriter;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.core.fate.zookeeper.ZooUtil.NodeMissingPolicy;
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ChoppedColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ExternalCompactionColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LastLocationColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.RootTabletMetadata;
import org.apache.accumulo.core.zookeeper.ZooSession;
import org.apache.accumulo.manager.WithTestNames;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.conf.codec.VersionedProperties;
import org.apache.accumulo.server.conf.store.NamespacePropKey;
import org.apache.accumulo.server.conf.store.SystemPropKey;
import org.apache.accumulo.server.conf.store.TablePropKey;
import org.apache.accumulo.server.conf.store.impl.ZooPropStore;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.easymock.Capture;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Upgrader11to12Test extends WithTestNames {

  private static final Logger LOG = LoggerFactory.getLogger(Upgrader11to12Test.class);

  @Test
  void upgradeDataFileCF2Test() {
    String fileName = "hdfs://localhost:8020/accumulo/tables/12/default_tablet/A000000v.rf";
    Key k = Key.builder().row(new Text("12;")).family(DataFileColumnFamily.NAME)
        .qualifier(new Text(fileName)).build();
    Value v = new Value("1234,5678");

    Mutation upgrade = new Mutation(k.getRow());
    Upgrader11to12.upgradeDataFileCF(k, v, upgrade);

    var pending = upgrade.getUpdates();
    assertEquals(2, pending.size());
    // leverage sort order for "expected" values
    // check file entry converted is in the mutation
    Iterator<ColumnUpdate> m = pending.iterator();
    var cu1 = m.next();
    assertEquals("file", new Text(cu1.getColumnFamily()).toString());

    StoredTabletFile oldFileEntry = StoredTabletFile.of(new Path(fileName));
    StoredTabletFile updateEnry = StoredTabletFile.of(new String(cu1.getColumnQualifier(), UTF_8));

    assertEquals(oldFileEntry, updateEnry);
    assertFalse(cu1.isDeleted());

    // check old file entry is deleted is in the mutation

    var cu2 = m.next();
    assertEquals("file", new Text(cu1.getColumnFamily()).toString());
    assertEquals(fileName, new String(cu2.getColumnQualifier(), UTF_8));
    assertTrue(cu2.isDeleted());

  }

  @Test
  public void processReferencesTest() {

    // create sample data "served" by the mocked scanner
    TreeMap<Key,Value> scanData = new TreeMap<>();
    Text row1 = new Text("123");

    String fileName1 = "hdfs://localhost:8020/accumulo/tables/12/default_tablet/A000000v.rf";
    Key key1 =
        Key.builder(false).row(row1).family(DataFileColumnFamily.NAME).qualifier(fileName1).build();
    Value value1 = new Value("123,456");
    scanData.put(key1, value1);

    String fileName2 = "hdfs://localhost:8020/accumulo/tables/12/default_tablet/B000000v.rf";
    Key key2 =
        Key.builder(false).row(row1).family(DataFileColumnFamily.NAME).qualifier(fileName2).build();
    Value value2 = new Value("321,654");
    scanData.put(key2, value2);

    @SuppressWarnings("deprecation")
    var chopped = ChoppedColumnFamily.NAME;
    Key chop1 = Key.builder(false).row(row1).family(chopped).qualifier(chopped).build();
    scanData.put(chop1, null);

    Key extern1 = Key.builder(false).row(row1).family(ExternalCompactionColumnFamily.NAME)
        .qualifier(ExternalCompactionColumnFamily.NAME).build();
    scanData.put(extern1, null);

    Text row2 = new Text("234");

    String fileName3 = "hdfs://localhost:8020/accumulo/tables/13/default_tablet/C000000v.rf";
    Key key3 =
        Key.builder(false).row(row2).family(DataFileColumnFamily.NAME).qualifier(fileName3).build();
    Value value3 = new Value("1,2");
    scanData.put(key3, value3);

    String fileName4 = "hdfs://localhost:8020/accumulo/tables/13/default_tablet/C000000x.rf";
    Key key4 =
        Key.builder(false).row(row2).family(ScanFileColumnFamily.NAME).qualifier(fileName4).build();
    scanData.put(key4, new Value());

    ArrayList<Mutation> mutations = new ArrayList<>();

    Upgrader11to12 upgrader = new Upgrader11to12();
    upgrader.processReferences(mutations::add, scanData.entrySet(), "accumulo.metadata");

    assertEquals(2, mutations.size());

    var u1 = mutations.get(0);
    LOG.info("c:{}", u1.prettyPrint());
    // 2 file add, 2 file delete. 1 chop delete, 1 ext comp delete
    assertEquals(6, u1.getUpdates().size());

    var u2 = mutations.get(1);
    LOG.info("c:{}", u2.prettyPrint());
    // 1 add, 2 delete
    assertEquals(3, u2.getUpdates().size());
    assertEquals(2, u2.getUpdates().stream().filter(ColumnUpdate::isDeleted).count());

  }

  @Test
  public void skipConvertedFileTest() {
    // create sample data "served" by the mocked scanner
    TreeMap<Key,Value> scanData = new TreeMap<>();
    Text row1 = new Text("123");

    // reference already in expected form with fence info.
    String fileName1 =
        "{\"path\":\"hdfs://localhost:8020/accumulo/tables/12/default_tablet/A000000v.rf\",\"startRow\":\"\",\"endRow\":\"\"}";
    Key key1 =
        Key.builder(false).row(row1).family(DataFileColumnFamily.NAME).qualifier(fileName1).build();
    Value value1 = new Value("123,456");
    scanData.put(key1, value1);

    String fileName2 = "hdfs://localhost:8020/accumulo/tables/12/default_tablet/B000000v.rf";
    Key key2 =
        Key.builder(false).row(row1).family(DataFileColumnFamily.NAME).qualifier(fileName2).build();
    Value value2 = new Value("321,654");
    scanData.put(key2, value2);

    ArrayList<Mutation> mutations = new ArrayList<>();

    Upgrader11to12 upgrader = new Upgrader11to12();
    upgrader.processReferences(mutations::add, scanData.entrySet(), "accumulo.metadata");

    assertEquals(1, mutations.size());

    var u1 = mutations.get(0);
    LOG.info("c:{}", u1.prettyPrint());
    // 1 add, 1 delete
    assertEquals(2, u1.getUpdates().size());
    assertEquals(1, u1.getUpdates().stream().filter(ColumnUpdate::isDeleted).count());
  }

  @Test
  void failOnMutationErrorTest() throws Exception {

    BatchWriter batchWriter = mock(BatchWriter.class);
    Capture<Mutation> capturedUpdate1 = newCapture();
    batchWriter.addMutation(capture(capturedUpdate1));
    expectLastCall().andThrow(new MutationsRejectedException(null, List.of(), Map.of(), List.of(),
        0, new NullPointerException())).once();

    TreeMap<Key,Value> scanData = new TreeMap<>();
    Text row1 = new Text("123");

    // reference already in expected form with fence info.
    String fileName1 = "hdfs://localhost:8020/accumulo/tables/12/default_tablet/A000000v.rf";
    Key key1 =
        Key.builder(false).row(row1).family(DataFileColumnFamily.NAME).qualifier(fileName1).build();
    Value value1 = new Value("123,456");
    scanData.put(key1, value1);

    replay(batchWriter);
    Upgrader11to12 upgrader = new Upgrader11to12();

    assertThrows(IllegalStateException.class, () -> upgrader
        .processReferences(batchWriter::addMutation, scanData.entrySet(), "accumulo.metadata"));

    verify(batchWriter);
  }

  @Test
  void upgradeDataFileCFInvalidPathTest() throws Exception {

    BatchWriter batchWriter = mock(BatchWriter.class);
    Capture<Mutation> capturedUpdate1 = newCapture();
    batchWriter.addMutation(capture(capturedUpdate1));
    // expecting that exception will be called before mutation is updated.
    expectLastCall().andThrow(new UnsupportedOperationException()).anyTimes();

    // create sample data "served" by the mocked scanner
    TreeMap<Key,Value> scanData = new TreeMap<>();
    Text row1 = new Text("123");

    String fileName1 = "bad path";
    Key key1 =
        Key.builder(false).row(row1).family(DataFileColumnFamily.NAME).qualifier(fileName1).build();
    Value value1 = new Value("123,456");
    scanData.put(key1, value1);

    String fileName2 = "hdfs://localhost:8020/accumulo/tables/12/default_tablet/B000000v.rf";
    Key key2 =
        Key.builder(false).row(row1).family(DataFileColumnFamily.NAME).qualifier(fileName2).build();
    Value value2 = new Value("321,654");
    scanData.put(key2, value2);

    replay(batchWriter);

    Upgrader11to12 upgrader = new Upgrader11to12();
    assertThrows(IllegalArgumentException.class, () -> upgrader
        .processReferences(batchWriter::addMutation, scanData.entrySet(), "accumulo.metadata"));

    verify(batchWriter);
  }

  @Test
  void unexpectedColFailsTest() throws Exception {

    BatchWriter batchWriter = mock(BatchWriter.class);
    Capture<Mutation> capturedUpdate1 = newCapture();
    batchWriter.addMutation(capture(capturedUpdate1));
    // expecting that exception will be called before mutation is updated.
    expectLastCall().andThrow(new UnsupportedOperationException()).anyTimes();

    // create sample data "served" by the mocked scanner
    TreeMap<Key,Value> scanData = new TreeMap<>();
    Text row1 = new Text("123");

    Key key1 = Key.builder(false).row(row1).family(LastLocationColumnFamily.NAME).qualifier("srv1")
        .build();
    Value value1 = new Value("123,456");
    scanData.put(key1, value1);

    replay(batchWriter);

    Upgrader11to12 upgrader = new Upgrader11to12();
    assertThrows(IllegalStateException.class, () -> upgrader
        .processReferences(batchWriter::addMutation, scanData.entrySet(), "accumulo.metadata"));

    verify(batchWriter);
  }

  /**
   * process 3 rows, 2 should result in no mutations and batch writer addMutation should not be
   * called for those rows
   */
  @Test
  public void verifyEmptyMutation() {
    // create sample data "served" by the mocked scanner
    TreeMap<Key,Value> scanData = new TreeMap<>();

    Text row1 = new Text("1");

    String fileName1 = "hdfs://localhost:8020/accumulo/tables/12/default_tablet/1111000v.rf";
    Key key1 =
        Key.builder(false).row(row1).family(DataFileColumnFamily.NAME).qualifier(fileName1).build();
    Value value1 = new Value("111,222");
    scanData.put(key1, value1);

    Text row2 = new Text("a");

    // reference already in expected form with fence info.
    String fileName2 =
        "{\"path\":\"hdfs://localhost:8020/accumulo/tables/12/default_tablet/A000000v.rf\",\"startRow\":\"\",\"endRow\":\"\"}";
    Key key2 =
        Key.builder(false).row(row2).family(DataFileColumnFamily.NAME).qualifier(fileName2).build();
    Value value2 = new Value("222,333");
    scanData.put(key2, value2);

    Text row3 = new Text("b");

    // reference already in expected form with fence info.
    String fileName3 =
        "{\"path\":\"hdfs://localhost:8020/accumulo/tables/12/default_tablet/BBBB000v.rf\",\"startRow\":\"\",\"endRow\":\"\"}";
    Key key3 =
        Key.builder(false).row(row3).family(DataFileColumnFamily.NAME).qualifier(fileName3).build();
    Value value3 = new Value("333,444");
    scanData.put(key3, value3);

    ArrayList<Mutation> mutations = new ArrayList<>();

    Upgrader11to12 upgrader = new Upgrader11to12();
    upgrader.processReferences(mutations::add, scanData.entrySet(), "accumulo.metadata");

    assertEquals(1, mutations.size());
    var u1 = mutations.get(0);
    LOG.info("c:{}", u1.prettyPrint());
    // 1 add, 1 delete
    assertEquals(2, u1.getUpdates().size());
    assertEquals(1, u1.getUpdates().stream().filter(ColumnUpdate::isDeleted).count());
  }

  @SuppressWarnings("unchecked")
  @Test
  public void upgradeZooKeeperTest() throws Exception {

    // taken from an uno instance.
    final byte[] zKRootV1 =
        "{\"version\":1,\"columnValues\":{\"file\":{\"hdfs://localhost:8020/accumulo/tables/+r/root_tablet/A0000030.rf\":\"856,15\",\"hdfs://localhost:8020/accumulo/tables/+r/root_tablet/F000000r.rf\":\"308,2\"},\"last\":{\"100017f46240004\":\"localhost:9997\"},\"loc\":{\"100017f46240004\":\"localhost:9997\"},\"srv\":{\"dir\":\"root_tablet\",\"flush\":\"16\",\"lock\":\"tservers/localhost:9997/zlock#f6a582b9-9583-4553-b179-a7a3852c8332#0000000000$100017f46240004\",\"time\":\"L42\"},\"~tab\":{\"~pr\":\"\\u0000\"}}}\n"
            .getBytes(UTF_8);
    final String zKRootV2 =
        "{\"version\":1,\"columnValues\":{\"file\":{\"{\\\"path\\\":\\\"hdfs://localhost:8020/accumulo/tables/+r/root_tablet/A0000030.rf\\\",\\\"startRow\\\":\\\"\\\",\\\"endRow\\\":\\\"\\\"}\":\"856,15\",\"{\\\"path\\\":\\\"hdfs://localhost:8020/accumulo/tables/+r/root_tablet/F000000r.rf\\\",\\\"startRow\\\":\\\"\\\",\\\"endRow\\\":\\\"\\\"}\":\"308,2\"},\"last\":{\"100017f46240004\":\"localhost:9997\"},\"loc\":{\"100017f46240004\":\"localhost:9997\"},\"srv\":{\"dir\":\"root_tablet\",\"flush\":\"16\",\"lock\":\"tservers/localhost:9997/zlock#f6a582b9-9583-4553-b179-a7a3852c8332#0000000000$100017f46240004\",\"time\":\"L42\"},\"~tab\":{\"~pr\":\"\\u0000\"}}}";

    InstanceId iid = InstanceId.of(UUID.randomUUID());
    Upgrader11to12 upgrader = new Upgrader11to12();

    ServerContext context = createMock(ServerContext.class);
    ZooSession zk = createStrictMock(ZooSession.class);
    ZooReaderWriter zrw = createStrictMock(ZooReaderWriter.class);
    ZooPropStore store = createMock(ZooPropStore.class);

    expect(context.getInstanceID()).andReturn(iid).anyTimes();
    expect(context.getZooSession()).andReturn(zk).anyTimes();
    expect(zk.asReaderWriter()).andReturn(zrw).anyTimes();

    zrw.recursiveDelete("/tracers", ZooUtil.NodeMissingPolicy.SKIP);
    expectLastCall().once();

    Capture<Stat> statCapture = newCapture();
    expect(zrw.getData(eq("/root_tablet"), capture(statCapture))).andAnswer(() -> {
      Stat stat = statCapture.getValue();
      stat.setCtime(System.currentTimeMillis());
      stat.setMtime(System.currentTimeMillis());
      stat.setVersion(123); // default version
      stat.setDataLength(zKRootV1.length);
      statCapture.setValue(stat);
      return zKRootV1;
    }).once();

    Capture<byte[]> byteCapture = newCapture();
    expect(zrw.overwritePersistentData(eq("/root_tablet"), capture(byteCapture), eq(123)))
        .andReturn(true).once();

    expect(zrw.getData(eq(Constants.ZNAMESPACES))).andReturn(new byte[0]).once();
    Map<String,String> mockNamespaces = Map.of("ns1", "ns1name", "ns2", "ns2name");
    expect(zrw.getChildren(eq(Constants.ZNAMESPACES)))
        .andReturn(List.copyOf(mockNamespaces.keySet())).once();
    for (String ns : mockNamespaces.keySet()) {
      expect(zrw.getData(Constants.ZNAMESPACES + "/" + ns + ZNAMESPACE_NAME))
          .andReturn(mockNamespaces.get(ns).getBytes(UTF_8)).once();
    }
    byte[] mapping = NamespaceMapping.serializeMap(mockNamespaces);
    expect(zrw.putPersistentData(eq(Constants.ZNAMESPACES), aryEq(mapping),
        eq(ZooUtil.NodeExistsPolicy.OVERWRITE))).andReturn(true).once();
    for (String ns : mockNamespaces.keySet()) {
      zrw.delete(Constants.ZNAMESPACES + "/" + ns + ZNAMESPACE_NAME);
      expectLastCall().once();
    }

    expect(zrw.exists("/problems")).andReturn(false).atLeastOnce();

    expect(zrw.putPersistentData(isA(String.class), isA(byte[].class), isA(NodeExistsPolicy.class)))
        .andReturn(true).times(7);
    expect(context.getPropStore()).andReturn(store).atLeastOnce();
    expect(store.exists(isA(TablePropKey.class))).andReturn(false).atLeastOnce();
    store.create(isA(TablePropKey.class), isA(Map.class));
    expectLastCall().once();

    replay(context, zk, zrw, store);

    upgrader.removeZTracersNode(context);
    upgrader.updateRootTabletFileReferences(context);
    upgrader.createNamespaceMappings(context);
    upgrader.removeZKProblemReports(context);
    upgrader.initializeScanRefTable(context);

    assertEquals(zKRootV2, new String(byteCapture.getValue(), UTF_8));

    verify(context, zk, zrw, store);
  }

  @Test
  public void convertRoot1File() {
    String root21ZkData =
        "{\"version\":1,\"columnValues\":{\"file\":{\"hdfs://localhost:8020/accumulo/tables/+r/root_tablet/A000000v.rf\":\"1368,61\"},\"last\":{\"100025091780006\":\"localhost:9997\"},\"loc\":{\"100025091780006\":\"localhost:9997\"},\"srv\":{\"dir\":\"root_tablet\",\"flush\":\"3\",\"lock\":\"tservers/localhost:9997/zlock#9db8961a-4ee9-400e-8e80-3353148baadd#0000000000$100025091780006\",\"time\":\"L53\"},\"~tab\":{\"~pr\":\"\\u0000\"}}}";

    RootTabletMetadata rtm = new RootTabletMetadata(root21ZkData);
    ArrayList<Mutation> mutations = new ArrayList<>();
    Upgrader11to12 upgrader = new Upgrader11to12();
    upgrader.processReferences(mutations::add,
        rtm.getKeyValues().filter(e -> UPGRADE_FAMILIES.contains(e.getKey().getColumnFamily()))
            .collect(Collectors.toList()),
        "accumulo.metadata");
    assertEquals(1, mutations.size());
    var mutation = mutations.get(0);
    rtm.update(mutation);

    LOG.debug("converted column values: {}", rtm.toTabletMetadata().getFiles());

    var files = rtm.toTabletMetadata().getFiles();
    LOG.info("FILES: {}", rtm.toTabletMetadata().getFilesMap());

    assertEquals(1, files.size());
    assertTrue(files.contains(StoredTabletFile
        .of(new Path("hdfs://localhost:8020/accumulo/tables/+r/root_tablet/A000000v.rf"))));
  }

  @Test
  public void convertRoot2Files() {
    String root212ZkData2Files =
        "{\"version\":1,\"columnValues\":{\"file\":{\"hdfs://localhost:8020/accumulo/tables/+r/root_tablet/00000_00000.rf\":\"0,0\",\"hdfs://localhost:8020/accumulo/tables/+r/root_tablet/F000000c.rf\":\"926,18\"},\"last\":{\"10001a84d7d0005\":\"localhost:9997\"},\"loc\":{\"10001a84d7d0005\":\"localhost:9997\"},\"srv\":{\"dir\":\"root_tablet\",\"flush\":\"2\",\"lock\":\"tservers/localhost:9997/zlock#d21adaa4-0f97-4004-9ff8-cce9dbb6687f#0000000000$10001a84d7d0005\",\"time\":\"L6\"},\"~tab\":{\"~pr\":\"\\u0000\"}}}\n";

    RootTabletMetadata rtm = new RootTabletMetadata(root212ZkData2Files);
    ArrayList<Mutation> mutations = new ArrayList<>();
    Upgrader11to12 upgrader = new Upgrader11to12();
    upgrader.processReferences(mutations::add,
        rtm.getKeyValues().filter(e -> UPGRADE_FAMILIES.contains(e.getKey().getColumnFamily()))
            .collect(Collectors.toList()),
        "accumulo.metadata");
    assertEquals(1, mutations.size());
    var mutation = mutations.get(0);
    rtm.update(mutation);

    LOG.debug("converted column values: {}", rtm.toTabletMetadata());

    var files = rtm.toTabletMetadata().getFiles();
    LOG.info("FILES: {}", rtm.toTabletMetadata().getFilesMap());

    assertEquals(2, files.size());
    assertTrue(files.contains(StoredTabletFile
        .of(new Path("hdfs://localhost:8020/accumulo/tables/+r/root_tablet/00000_00000.rf"))));
    assertTrue(files.contains(StoredTabletFile
        .of(new Path("hdfs://localhost:8020/accumulo/tables/+r/root_tablet/F000000c.rf"))));
  }

  @Test
  public void testZooKeeperUpgradeFailsServerCheck()
      throws InterruptedException, KeeperException, NamespaceNotFoundException {
    Upgrader11to12 upgrader = new Upgrader11to12();
    InstanceId iid = InstanceId.of(UUID.randomUUID());

    ServerContext context = createMock(ServerContext.class);
    ZooSession zk = createStrictMock(ZooSession.class);
    ZooReader zr = createStrictMock(ZooReader.class);
    ZooReaderWriter zrw = createStrictMock(ZooReaderWriter.class);

    expect(context.getInstanceID()).andReturn(iid).anyTimes();
    expect(context.getZooSession()).andReturn(zk).anyTimes();
    expect(zk.asReader()).andReturn(zr).anyTimes();
    expect(zk.asReaderWriter()).andReturn(zrw).anyTimes();

    expect(zr.getChildren(Constants.ZCOMPACTORS)).andReturn(List.of());
    expect(zr.getChildren(Constants.ZSSERVERS)).andReturn(List.of("localhost:9996"));
    expect(zr.getChildren(Constants.ZSSERVERS + "/localhost:9996")).andReturn(List.of());
    zrw.recursiveDelete(Constants.ZSSERVERS + "/localhost:9996", NodeMissingPolicy.SKIP);
    expect(zr.getChildren(Constants.ZTSERVERS)).andReturn(List.of("localhost:9997"));
    expect(zr.getChildren(Constants.ZTSERVERS + "/localhost:9997"))
        .andReturn(List.of(UUID.randomUUID().toString()));

    replay(context, zk, zr, zrw);
    IllegalStateException e =
        assertThrows(IllegalStateException.class, () -> upgrader.upgradeZookeeper(context));
    assertTrue(e.getMessage()
        .contains("Was expecting either a nothing, a resource group name or an empty directory"));
    verify(context, zk, zr, zrw);

  }

  @Test
  public void testAddingTableMappingToZooKeeper() throws InterruptedException, KeeperException {
    Upgrader11to12 upgrader = new Upgrader11to12();
    InstanceId iid = InstanceId.of(UUID.randomUUID());

    ServerContext context = createMock(ServerContext.class);
    ZooSession zk = createStrictMock(ZooSession.class);
    ZooReaderWriter zrw = createStrictMock(ZooReaderWriter.class);

    expect(context.getInstanceID()).andReturn(iid).anyTimes();
    expect(context.getZooSession()).andReturn(zk).anyTimes();
    expect(zk.asReaderWriter()).andReturn(zrw).anyTimes();

    Map<String,String> mockTables = Map.of("t1Id", "t1", "t2Id", "t2", "t3Id", "t3");
    List<String> mockTableIds = List.copyOf(mockTables.keySet());
    Map<String,String> mockTableToNamespace =
        Map.of("t1Id", "ns1Id", "t2Id", "ns1Id", "t3Id", "ns2Id");

    expect(zrw.getChildren(eq(Constants.ZTABLES))).andReturn(mockTableIds).once();
    for (String tableId : mockTableIds) {
      expect(zrw.getData(Constants.ZTABLES + "/" + tableId + ZTABLE_NAME))
          .andReturn(mockTables.get(tableId).getBytes(UTF_8)).once();
      expect(zrw.getData(Constants.ZTABLES + "/" + tableId + Constants.ZTABLE_NAMESPACE))
          .andReturn(mockTableToNamespace.get(tableId).getBytes(UTF_8)).once();
    }

    Map<String,Map<String,String>> expectedNamespaceMaps = new HashMap<>();
    for (Map.Entry<String,String> entry : mockTables.entrySet()) {
      String tableId = entry.getKey();
      expectedNamespaceMaps.computeIfAbsent(mockTableToNamespace.get(tableId), k -> new HashMap<>())
          .put(tableId, entry.getValue());
    }

    expectedNamespaceMaps.put(Namespace.DEFAULT.id().canonical(), new HashMap<>());
    for (Map.Entry<String,Map<String,String>> entry : expectedNamespaceMaps.entrySet()) {
      expect(zrw.putPersistentData(
          eq(Constants.ZNAMESPACES + "/" + entry.getKey() + Constants.ZTABLES),
          aryEq(NamespaceMapping.serializeMap(entry.getValue())),
          eq(ZooUtil.NodeExistsPolicy.FAIL))).andReturn(true).once();
    }

    for (String table : mockTables.keySet()) {
      zrw.delete(Constants.ZTABLES + "/" + table + ZTABLE_NAME);
      expectLastCall().once();
    }

    replay(context, zk, zrw);
    upgrader.addTableMappingsToZooKeeper(context);
    verify(context, zk, zrw);
  }

  @Test
  public void testMoveTableProperties() throws Exception {

    final ServerContext context = createMock(ServerContext.class);
    final ZooPropStore propStore = createMock(ZooPropStore.class);
    final VersionedProperties sysVerProps = createMock(VersionedProperties.class);
    final VersionedProperties systemNsProps = createMock(VersionedProperties.class);
    final VersionedProperties defaultNsProps = createMock(VersionedProperties.class);
    final VersionedProperties testNsProps = createMock(VersionedProperties.class);
    final NamespaceMapping nsmap = createMock(NamespaceMapping.class);

    var testNsId = NamespaceId.of("nsid_" + testName());
    var testNsName = "nsname_" + testName();
    final TreeMap<NamespaceId,String> namespaces =
        new TreeMap<>(Map.of(Namespace.DEFAULT.id(), Namespace.DEFAULT.name(),
            Namespace.ACCUMULO.id(), Namespace.ACCUMULO.name(), testNsId, testNsName));

    final Map<String,String> sysProps = new HashMap<>();
    sysProps.put(Property.TABLE_BLOOM_ENABLED.getKey(), "true");
    sysProps.put(Property.TABLE_BLOCKCACHE_ENABLED.getKey(), "true");
    sysProps.put(Property.TABLE_CLASSLOADER_CONTEXT.getKey(), "sysContext");
    sysProps.put(Property.TABLE_ITERATOR_PREFIX.getKey() + "test", "iteratorProperty");
    sysProps.put(Property.TABLE_CONSTRAINT_PREFIX.getKey() + "1",
        DefaultKeySizeConstraint.class.getName());

    // Accumulo ns props
    final Map<String,String> accProps = new HashMap<>();

    // iterator property will not get moved
    final Map<String,String> accChanges = new HashMap<>();
    accChanges.put(Property.TABLE_BLOOM_ENABLED.getKey(), "true");
    accChanges.put(Property.TABLE_BLOCKCACHE_ENABLED.getKey(), "true");
    accChanges.put(Property.TABLE_CLASSLOADER_CONTEXT.getKey(), "sysContext");

    // Default ns has one same and one different prop
    final Map<String,String> defProps = new HashMap<>();
    defProps.put(Property.TABLE_BLOOM_ENABLED.getKey(), "true");
    defProps.put(Property.TABLE_CLASSLOADER_CONTEXT.getKey(), "defContext");

    final Map<String,String> defChanges = new HashMap<>();
    defChanges.put(Property.TABLE_BLOCKCACHE_ENABLED.getKey(), "true");
    defChanges.put(Property.TABLE_ITERATOR_PREFIX.getKey() + "test", "iteratorProperty");
    defChanges.put(Property.TABLE_CONSTRAINT_PREFIX.getKey() + "1",
        DefaultKeySizeConstraint.class.getName());

    // Test ns has one different prop
    final Map<String,String> testProps = new HashMap<>();
    testProps.put(Property.TABLE_BLOOM_ENABLED.getKey(), "false");

    final Map<String,String> testChanges = new HashMap<>();
    testChanges.put(Property.TABLE_BLOCKCACHE_ENABLED.getKey(), "true");
    testChanges.put(Property.TABLE_CLASSLOADER_CONTEXT.getKey(), "sysContext");
    testChanges.put(Property.TABLE_ITERATOR_PREFIX.getKey() + "test", "iteratorProperty");
    testChanges.put(Property.TABLE_CONSTRAINT_PREFIX.getKey() + "1",
        DefaultKeySizeConstraint.class.getName());

    expect(context.getPropStore()).andReturn(propStore).anyTimes();
    expect(propStore.get(SystemPropKey.of())).andReturn(sysVerProps).once();
    expect(sysVerProps.asMap()).andReturn(sysProps).once();

    expect(context.getNamespaceMapping()).andReturn(nsmap).once();
    expect(nsmap.getIdToNameMap()).andReturn(namespaces).once();

    final NamespacePropKey apk = NamespacePropKey.of(Namespace.ACCUMULO.id());
    final NamespacePropKey dpk = NamespacePropKey.of(Namespace.DEFAULT.id());
    final NamespacePropKey tpk = NamespacePropKey.of(testNsId);

    expect(propStore.get(apk)).andReturn(systemNsProps).once();
    expect(systemNsProps.asMap()).andReturn(accProps).once();

    propStore.putAll(apk, accChanges);
    expectLastCall().once();

    expect(propStore.get(dpk)).andReturn(defaultNsProps).once();
    expect(defaultNsProps.asMap()).andReturn(defProps).once();

    propStore.putAll(dpk, defChanges);
    expectLastCall().once();

    expect(propStore.get(tpk)).andReturn(testNsProps).once();
    expect(testNsProps.asMap()).andReturn(testProps).once();

    propStore.putAll(tpk, testChanges);
    expectLastCall().once();

    propStore.removeProperties(SystemPropKey.of(), sysProps.keySet());

    replay(context, propStore, sysVerProps, systemNsProps, defaultNsProps, testNsProps, nsmap);

    new Upgrader11to12().moveTableProperties(context);

    verify(context, propStore, sysVerProps, systemNsProps, defaultNsProps, testNsProps, nsmap);

  }
}
