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

import java.io.IOException;

import org.junit.After;
import org.junit.BeforeClass;
import org.junit.Test;

import org.apache.cassandra.db.Directories;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.tools.ToolRunner.ToolResult;
import org.apache.cassandra.utils.FBUtilities;
import org.assertj.core.api.AbstractStringAssert;
import org.assertj.core.api.Assertions;

import static java.lang.String.format;

/**
 * Tests for {@link SSTablePartitions}.
 */
public class SSTablePartitionsTest extends OfflineToolUtils
{
    private static final String SSTABLE_1 = sstable("legacy_ma_simple");
    private static final String SSTABLE_2 = sstable("legacy_ma_clust");
    private static final String HEADER_1 = "\nProcessing  #306 (big-ma) (173 B uncompressed, 88 B on disk)\n";
    private static final String HEADER_2 = "\nProcessing  #306 (big-ma) (328.145 KiB uncompressed, 5.096 KiB on disk)\n";
    private static final String BACKUPS_HEADER_1 = "\nProcessing Backup:backups  #306 (big-ma) (173 B uncompressed, 88 B on disk)\n";
    private static final String BACKUPS_HEADER_2 = "\nProcessing Backup:backups  #306 (big-ma) (328.145 KiB uncompressed, 5.096 KiB on disk)\n";
    private static final String SNAPSHOTS_HEADER_1 = "\nProcessing Snapshot:snapshot-1  #306 (big-ma) (173 B uncompressed, 88 B on disk)\n";
    private static final String SNAPSHOTS_HEADER_2 = "\nProcessing Snapshot:snapshot-1  #306 (big-ma) (328.145 KiB uncompressed, 5.096 KiB on disk)\n";
    private static final String SUMMARY_1 = "               Partition size            Row count           Cell count      Tombstone count\n" +
                                            "  ~p50                   35 B                    1                    1                    0\n" +
                                            "  ~p75                   35 B                    1                    1                    0\n" +
                                            "  ~p90                   35 B                    1                    1                    0\n" +
                                            "  ~p95                   35 B                    1                    1                    0\n" +
                                            "  ~p99                   35 B                    1                    1                    0\n" +
                                            "  ~p999                  35 B                    1                    1                    0\n" +
                                            "  min                    33 B                    1                    1                    0\n" +
                                            "  max                    35 B                    1                    1                    0\n" +
                                            "  count                     5\n";
    private static final String SUMMARY_2 = "               Partition size            Row count           Cell count      Tombstone count\n" +
                                            "  ~p50             71.735 KiB                   50                   50                    0\n" +
                                            "  ~p75             71.735 KiB                   50                   50                    0\n" +
                                            "  ~p90             71.735 KiB                   50                   50                    0\n" +
                                            "  ~p95             71.735 KiB                   50                   50                    0\n" +
                                            "  ~p99             71.735 KiB                   50                   50                    0\n" +
                                            "  ~p999            71.735 KiB                   50                   50                    0\n" +
                                            "  min              65.625 KiB                   50                   50                    0\n" +
                                            "  max              65.630 KiB                   50                   50                    0\n" +
                                            "  count                     5\n";

    @BeforeClass
    public static void prepareDirectories()
    {
        createBackupsAndSnapshots(SSTABLE_1);
        createBackupsAndSnapshots(SSTABLE_2);
    }

    private static void createBackupsAndSnapshots(String sstable)
    {
        File parentDir = new File(sstable).parent();

        File backupsDir = new File(parentDir, Directories.BACKUPS_SUBDIR);
        backupsDir.tryCreateDirectory();

        File snapshotsDir = new File(parentDir, Directories.SNAPSHOT_SUBDIR);
        snapshotsDir.tryCreateDirectory();

        File snapshotDir = new File(snapshotsDir, "snapshot-1");
        snapshotDir.tryCreateDirectory();

        for (File f : parentDir.tryList(File::isFile))
        {
            FileUtils.copyWithOutConfirm(f, new File(backupsDir, f.name()));
            FileUtils.copyWithOutConfirm(f, new File(snapshotDir, f.name()));
        }
    }

    /**
     * Runs post-test assertions about loaded classed and started threads.
     */
    @After
    public void assertPostTestEnv()
    {
        assertNoUnexpectedThreadsStarted(OPTIONAL_THREADS_WITH_SCHEMA, false);
        assertCLSMNotLoaded();
        assertSystemKSNotLoaded();
        assertKeyspaceNotLoaded();
        assertServerNotLoaded();
    }

    /**
     * Verify that the tool prints help when no arguments are provided.
     */
    @Test
    public void testNoArgsPrintsHelp()
    {
        ToolResult tool = ToolRunner.invokeClass(SSTablePartitions.class);
        Assertions.assertThat(tool.getExitCode()).isOne();
        Assertions.assertThat(tool.getCleanedStderr()).contains("You must supply at least one sstable or directory");
        Assertions.assertThat(tool.getStdout()).contains("usage");
    }

    /**
     * Verify that the tool prints the right help contents.
     * If you added, modified options or help, please update docs if necessary.
     */
    @Test
    public void testMaybeChangeDocs()
    {
        ToolResult tool = ToolRunner.invokeClass(SSTablePartitions.class);
        Assertions.assertThat(tool.getStdout())
                  .isEqualTo("usage: sstablepartitions <options> <sstable files or directories>\n" +
                             "Print partition statistics of one or more sstables.\n" +
                             " -b,--backups                   include backups present in data\n" +
                             "                                directories (recursive scans)\n" +
                             " -c,--min-cells <arg>           partition cell count threshold\n" +
                             " -k,--key <arg>                 Partition keys to include\n" +
                             " -m,--csv                       CSV output (machine readable)\n" +
                             " -o,--min-tombstones <arg>      partition tombstone count threshold\n" +
                             " -r,--recursive                 scan for sstables recursively\n" +
                             " -s,--snapshots                 include snapshots present in data\n" +
                             "                                directories (recursive scans)\n" +
                             " -t,--min-size <arg>            partition size threshold, expressed as\n" +
                             "                                either the number of bytes or a size with\n" +
                             "                                unit of the form 10KiB, 20MiB, 30GiB, etc.\n" +
                             " -u,--current-timestamp <arg>   timestamp (seconds since epoch, unit time)\n" +
                             "                                for TTL expired calculation\n" +
                             " -w,--min-rows <arg>            partition row count threshold\n" +
                             " -x,--exclude-key <arg>         Excluded partition key(s) from partition\n" +
                             "                                detailed row/cell/tombstone information\n" +
                             "                                (irrelevant, if --partitions-only is\n" +
                             "                                given)\n" +
                             " -y,--partitions-only           Do not process per-partition detailed\n" +
                             "                                row/cell/tombstone information, only brief\n" +
                             "                                information\n");
    }

    /**
     * Verify that the tool can select single sstable file.
     */
    @Test
    public void testSingleSSTable()
    {
        assertThatToolSucceds(SSTABLE_1).isEqualTo(HEADER_1 + SUMMARY_1);
        assertThatToolSucceds(SSTABLE_2).isEqualTo(HEADER_2 + SUMMARY_2);
    }

    /**
     * Verify that the tool can select multiple sstable files.
     */
    @Test
    public void testMultipleSSTables()
    {
        assertThatToolSucceds(SSTABLE_1, SSTABLE_2)
                .isEqualTo(HEADER_2 + SUMMARY_2 + HEADER_1 + SUMMARY_1);
    }

    /**
     * Verify that the tool can select all the sstable files in a directory.
     */
    @Test
    public void testDirectory()
    {
        assertThatToolSucceds(new File(SSTABLE_1).parentPath())
                .isEqualTo(HEADER_1 + SUMMARY_1);

        assertThatToolSucceds("-r", new File(SSTABLE_1).parent().parentPath())
                .contains(HEADER_1 + SUMMARY_1)
                .contains(HEADER_2 + SUMMARY_2);

        assertThatToolSucceds("--recursive", new File(SSTABLE_2).parent().parentPath())
                .contains(HEADER_1 + SUMMARY_1)
                .contains(HEADER_2 + SUMMARY_2);
    }

    /**
     * Test the flag for collecting and printing sstable partition sizes only.
     */
    @Test
    public void testPartitionsOnly()
    {
        testPartitionsOnly("-y");
        testPartitionsOnly("--partitions-only");
    }

    private static void testPartitionsOnly(String option)
    {
        assertThatToolSucceds(SSTABLE_1, option)
                .isEqualTo(HEADER_1 +
                           "               Partition size\n" +
                           "  ~p50                   35 B\n" +
                           "  ~p75                   35 B\n" +
                           "  ~p90                   35 B\n" +
                           "  ~p95                   35 B\n" +
                           "  ~p99                   35 B\n" +
                           "  ~p999                  35 B\n" +
                           "  min                    33 B\n" +
                           "  max                    35 B\n" +
                           "  count                     5\n");

        assertThatToolSucceds(SSTABLE_2, "--partitions-only")
                .isEqualTo(HEADER_2 +
                           "               Partition size\n" +
                           "  ~p50             71.735 KiB\n" +
                           "  ~p75             71.735 KiB\n" +
                           "  ~p90             71.735 KiB\n" +
                           "  ~p95             71.735 KiB\n" +
                           "  ~p99             71.735 KiB\n" +
                           "  ~p999            71.735 KiB\n" +
                           "  min              65.625 KiB\n" +
                           "  max              65.630 KiB\n" +
                           "  count                     5\n");
    }

    /**
     * Test the flag for detecting partitions over a certain size threshold.
     */
    @Test
    public void testMinSize()
    {
        testMinSize("-t");
        testMinSize("--min-size");
    }

    private static void testMinSize(String option)
    {
        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, option, "35")
                .isEqualTo(HEADER_2 +
                           "  Partition: '0' (30) live, size: 65.625 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_2 +
                           HEADER_1 +
                           "  Partition: '1' (31) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_1 + "\n" +
                           "  4 partitions match\n" +
                           "  Keys: 1 2 3 4\n" +
                           SUMMARY_1);

        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, "--min-size", "36")
                .isEqualTo(HEADER_2 +
                           "  Partition: '0' (30) live, size: 65.625 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_2 + HEADER_1 + SUMMARY_1);

        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, "--min-size", "67201")
                .isEqualTo(HEADER_2 +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  4 partitions match\n" +
                           "  Keys: 1 2 3 4\n" +
                           SUMMARY_2 + HEADER_1 + SUMMARY_1);

        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, "--min-size", "67201B")
                .isEqualTo(HEADER_2 +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  4 partitions match\n" +
                           "  Keys: 1 2 3 4\n" +
                           SUMMARY_2 + HEADER_1 + SUMMARY_1);

        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, "--min-size", "65KiB")
                .isEqualTo(HEADER_2 +
                           "  Partition: '0' (30) live, size: 65.625 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_2 + HEADER_1 + SUMMARY_1);
    }

    /**
     * Test the flag for detecting partitions with more cells than a certain threshold.
     */
    @Test
    public void testMinCells()
    {
        testMinCells("-c");
        testMinCells("--min-cells");
    }

    private static void testMinCells(String option)
    {
        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, option, "0")
                .isEqualTo(HEADER_2 +
                           "  Partition: '0' (30) live, size: 65.625 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_2 +
                           HEADER_1 +
                           "  Partition: '0' (30) live, size: 33 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_1 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_1);

        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, option, "2")
                .isEqualTo(HEADER_2 +
                           "  Partition: '0' (30) live, size: 65.625 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_2 + HEADER_1 + SUMMARY_1);

        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, option, "51")
                .isEqualTo(HEADER_2 + SUMMARY_2 +
                           HEADER_1 + SUMMARY_1);
    }

    /**
     * Test the flag for detecting partitions with more rows than a certain threshold.
     */
    @Test
    public void testMinRows()
    {
        testMinRows("-w");
        testMinRows("--min-rows");
    }

    private static void testMinRows(String option)
    {
        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, option, "0")
                .isEqualTo(HEADER_2 +
                           "  Partition: '0' (30) live, size: 65.625 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_2 +
                           HEADER_1 +
                           "  Partition: '0' (30) live, size: 33 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_1 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_1);

        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, option, "50")
                .isEqualTo(HEADER_2 +
                           "  Partition: '0' (30) live, size: 65.625 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_2 + HEADER_1 + SUMMARY_1);

        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, option, "51")
                .isEqualTo(HEADER_2 + SUMMARY_2 +
                           HEADER_1 + SUMMARY_1);
    }

    /**
     * Test the flag for detecting partitions with more tombstones than a certain threshold.
     */
    @Test
    public void testMinTombstones()
    {
        testMinTombstones("-o");
        testMinTombstones("--min-tombstones");
    }

    private static void testMinTombstones(String option)
    {
        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, option, "0")
                .isEqualTo(HEADER_2 +
                           "  Partition: '0' (30) live, size: 65.625 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 65.630 KiB, rows: 50, cells: 50, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_2 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_2 +
                           HEADER_1 +
                           "  Partition: '0' (30) live, size: 33 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '1' (31) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_1 + "\n" +
                           "  5 partitions match\n" +
                           "  Keys: 0 1 2 3 4\n" +
                           SUMMARY_1);

        assertThatToolSucceds(SSTABLE_1, SSTABLE_2, "--min-tombstones", "1")
                .isEqualTo(HEADER_2 + SUMMARY_2 + HEADER_1 + SUMMARY_1);
    }

    /**
     * Test the flag for providing a current time.
     */
    @Test
    public void testCurrentTimestamp()
    {
        testCurrentTimestamp("-u");
        testCurrentTimestamp("--current-timestamp");
    }

    private static void testCurrentTimestamp(String option)
    {
        String now = String.valueOf(FBUtilities.nowInSeconds());
        assertThatToolSucceds(SSTABLE_1, option, now).isEqualTo(HEADER_1 + SUMMARY_1);
    }

    /**
     * Test the flag for including backup sstables.
     */
    @Test
    public void testBackups()
    {
        testBackups("-b");
        testBackups("--backups");
    }

    private static void testBackups(String option)
    {
        assertThatToolSucceds(new File(SSTABLE_1).parentPath(), "-r", option)
                .isEqualTo(HEADER_1 + SUMMARY_1 + BACKUPS_HEADER_1 + SUMMARY_1);

        assertThatToolSucceds(new File(SSTABLE_2).parentPath(), "-r", option)
                .isEqualTo(HEADER_2 + SUMMARY_2 + BACKUPS_HEADER_2 + SUMMARY_2);
    }

    /**
     * Test the flag for including snapshot sstables.
     */
    @Test
    public void testSnapshots()
    {
        testSnapshots("-s");
        testSnapshots("--snapshots");
    }

    private static void testSnapshots(String option)
    {
        assertThatToolSucceds(new File(SSTABLE_1).parentPath(), "-r", option)
                .isEqualTo(HEADER_1 + SUMMARY_1 + SNAPSHOTS_HEADER_1 + SUMMARY_1);

        assertThatToolSucceds(new File(SSTABLE_2).parentPath(), "-r", option)
                .isEqualTo(HEADER_2 + SUMMARY_2 + SNAPSHOTS_HEADER_2 + SUMMARY_2);
    }

    /**
     * Test the flag for specifying partiton keys to be considered.
     */
    @Test
    public void testIncludedKeys()
    {
        testIncludedKeys("-k");
        testIncludedKeys("--key");
    }

    private static void testIncludedKeys(String option)
    {
        assertThatToolSucceds(SSTABLE_1, "--min-size", "0", option, "1", option, "3")
                .contains(HEADER_1 +
                           "  Partition: '1' (31) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_1 + "\n" +
                           "  2 partitions match\n" +
                           "  Keys: 1 3\n")
                .contains("count                     2\n");

        assertThatToolSucceds(SSTABLE_1, "--min-size", "0", option, "0", option, "2", option, "4")
                .contains(HEADER_1 +
                          "  Partition: '0' (30) live, size: 33 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                          "  Partition: '2' (32) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                          "  Partition: '4' (34) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                          "Summary of  #306 (big-ma):\n" +
                          "  File: " + SSTABLE_1 + "\n" +
                          "  3 partitions match\n" +
                          "  Keys: 0 2 4\n")
                .contains("count                     3\n");

        assertThatToolSucceds(SSTABLE_1, "-y","--min-size", "0", option, "0", option, "2", option, "4")
                .contains(HEADER_1 +
                          "  Partition: '0' (30) live, size: 33 B\n" +
                          "  Partition: '2' (32) live, size: 35 B\n" +
                          "  Partition: '4' (34) live, size: 35 B\n" +
                          "Summary of  #306 (big-ma):\n" +
                          "  File: " + SSTABLE_1 + "\n" +
                          "  3 partitions match\n" +
                          "  Keys: 0 2 4\n")
                .contains("count                     3\n");
    }

    /**
     * Test the flag for specifying partiton keys to be excluded.
     */
    @Test
    public void testExcludedKeys()
    {
        testExcludedKeys("-x");
        testExcludedKeys("--exclude-key");
    }

    private static void testExcludedKeys(String option)
    {
        assertThatToolSucceds(SSTABLE_1, "--min-size", "0", option, "1", option, "3")
                .contains(HEADER_1 +
                           "  Partition: '0' (30) live, size: 33 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '2' (32) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '4' (34) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_1 + "\n" +
                           "  3 partitions match\n" +
                           "  Keys: 0 2 4\n")
                .contains("count                     3\n");

        assertThatToolSucceds(SSTABLE_1, "--min-size", "0", option, "0", option, "2", option, "4")
                .contains(HEADER_1 +
                           "  Partition: '1' (31) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "  Partition: '3' (33) live, size: 35 B, rows: 1, cells: 1, tombstones: 0 (row:0, range:0, complex:0, cell:0, row-TTLd:0, cell-TTLd:0)\n" +
                           "Summary of  #306 (big-ma):\n" +
                           "  File: " + SSTABLE_1 + "\n" +
                           "  2 partitions match\n" +
                           "  Keys: 1 3\n")
                .contains("count                     2\n");
    }

    /**
     * Test the flag for producing machine-readable CSV output.
     */
    @Test
    public void testCSV()
    {
        testCSV("-m");
        testCSV("--csv");
    }

    private static void testCSV(String option)
    {
        assertThatToolSucceds(option, "--min-size", "35", SSTABLE_1, SSTABLE_2)
                .isEqualTo(format("key,keyBinary,live,offset,size,rowCount,cellCount,tombstoneCount," +
                                  "rowTombstoneCount,rangeTombstoneCount,complexTombstoneCount,cellTombstoneCount," +
                                  "rowTtlExpired,cellTtlExpired,directory,keyspace,table,index,snapshot,backup," +
                                  "generation,format,version\n" +
                                  "\"0\",30,true,0,67200,50,50,0,0,0,0,0,0,0,%s,,,,,,306,big,ma\n" +
                                  "\"1\",31,true,67200,67205,50,50,0,0,0,0,0,0,0,%<s,,,,,,306,big,ma\n" +
                                  "\"2\",32,true,134405,67205,50,50,0,0,0,0,0,0,0,%<s,,,,,,306,big,ma\n" +
                                  "\"3\",33,true,201610,67205,50,50,0,0,0,0,0,0,0,%<s,,,,,,306,big,ma\n" +
                                  "\"4\",34,true,268815,67205,50,50,0,0,0,0,0,0,0,%<s,,,,,,306,big,ma\n" +
                                  "\"1\",31,true,33,35,1,1,0,0,0,0,0,0,0,%s,,,,,,306,big,ma\n" +
                                  "\"2\",32,true,68,35,1,1,0,0,0,0,0,0,0,%<s,,,,,,306,big,ma\n" +
                                  "\"3\",33,true,103,35,1,1,0,0,0,0,0,0,0,%<s,,,,,,306,big,ma\n" +
                                  "\"4\",34,true,138,35,1,1,0,0,0,0,0,0,0,%<s,,,,,,306,big,ma\n",
                                  SSTABLE_2, SSTABLE_1));
    }

    private static AbstractStringAssert<?> assertThatToolSucceds(String... args)
    {
        ToolResult tool = invokeTool(args);
        Assertions.assertThat(tool.getExitCode()).isZero();
        tool.assertOnCleanExit();
        return Assertions.assertThat(tool.getStdout());
    }

    private static String sstable(String table)
    {
        try
        {
            return findOneSSTable("legacy_sstables", table);
        }
        catch (IOException e)
        {
            throw new RuntimeException(e);
        }
    }

    private static ToolResult invokeTool(String... args)
    {
        return ToolRunner.invokeClass(SSTablePartitions.class, args);
    }
}
