/*
 * Copyright 2016 Palantir Technologies
 *
 * Licensed under the BSD-3 License (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://opensource.org/licenses/BSD-3-Clause
 *
 * 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 com.palantir.atlasdb.keyvalue.cassandra;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.core.StringEndsWith.endsWith;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.argThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.nio.ByteBuffer;
import java.util.List;

import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.CqlRow;
import org.hamcrest.Matcher;
import org.junit.Test;

import com.google.common.collect.ImmutableList;
import com.palantir.atlasdb.encoding.PtBytes;
import com.palantir.atlasdb.keyvalue.api.TableReference;
import com.palantir.util.Pair;

public class AllCellsPerRowPagerTest {

    CqlExecutor executor = mock(CqlExecutor.class);
    ByteBuffer rowKey = toByteBuffer("row");
    int pageSize = 20;

    AllCellsPerRowPager pager = new AllCellsPerRowPager(
            executor,
            rowKey,
            TableReference.fromString("tr"),
            pageSize
    );

    private static final String DEFAULT_COLUMN_NAME = "col1";

    @Test
    public void testGetFirstPage() {
        long timestamp = 1L;
        CqlRow row = makeCqlRow(DEFAULT_COLUMN_NAME, timestamp);
        allQueriesReturn(ImmutableList.of(row));

        List<ColumnOrSuperColumn> firstPage = pager.getFirstPage();

        assertThat(firstPage, hasSize(1));
        assertColumnOrSuperColumnHasCorrectNameAndTimestamp(firstPage.get(0), DEFAULT_COLUMN_NAME, timestamp);
    }

    @Test
    public void getFirstPageShouldReturnMultipleResults() {
        CqlRow row1 = makeCqlRow(DEFAULT_COLUMN_NAME, 1L);
        CqlRow row2 = makeCqlRow(DEFAULT_COLUMN_NAME, 2L);
        allQueriesReturn(ImmutableList.of(row1, row2));

        List<ColumnOrSuperColumn> firstPage = pager.getFirstPage();

        assertThat(firstPage, hasSize(2));
    }

    @Test
    public void getFirstPageShouldExecuteQueryLimitedToPageSize() {
        verifyFirstPageQueryMatches(endsWith(String.format("LIMIT %s;", pageSize)));
    }

    @Test
    public void getFirstPageShouldFireCqlRequestWithCorrectTableName() {
        verifyFirstPageQueryMatches(containsString(String.format("FROM %s ", "tr")));
    }

    @Test
    public void getFirstPageShouldFireCqlRequestWithCorrectRow() {
        verifyFirstPageQueryMatches(containsString(String.format("WHERE key = %s ", encodeAsHex(rowKey.array()))));
    }

    private void verifyFirstPageQueryMatches(Matcher<String> matcher) {
        allQueriesReturn(ImmutableList.of());

        pager.getFirstPage();

        verify(executor).execute(argThat(matcher));
    }

    private CqlRow makeCqlRow(String columnName, long timestamp) {
        List<Column> columns = ImmutableList.of(
                new Column().setValue(columnName.getBytes()),
                new Column().setValue(PtBytes.toBytes(~timestamp)));
        return new CqlRow(rowKey, columns);
    }

    private void allQueriesReturn(List<CqlRow> rows) {
        CqlResult cqlResult = mock(CqlResult.class);
        when(cqlResult.getRows()).thenReturn(rows);
        when(executor.execute(anyString())).thenReturn(cqlResult);
    }

    private void assertColumnOrSuperColumnHasCorrectNameAndTimestamp(ColumnOrSuperColumn columnOrSuperColumn, String expectedName, long expectedTs) {
        Pair<byte[], Long> nameAndTimestamp = CassandraKeyValueServices.decomposeName(columnOrSuperColumn.getColumn());
        String colName = PtBytes.toString(nameAndTimestamp.getLhSide());
        assertThat(colName, equalTo(expectedName));

        long timestamp = nameAndTimestamp.getRhSide();
        assertThat(timestamp, equalTo(expectedTs));

    }

    private ByteBuffer toByteBuffer(String str) {
        return ByteBuffer.wrap(str.getBytes());
    }

    private String encodeAsHex(byte[] array) {
        return "0x" + PtBytes.encodeHexString(array);
    }


}
@@ -1745,237 +1745,308 @@
     
-AllCellsPerRowPager pager = new AllCellsPerRowPager(%0A            executor,%0A            rowKey,%0A            TableReference.fromString(%22tr%22),%0A            pageSize%0A    );%0A%0A    private static final String DEFAULT_COLUMN_NAME = %22col1%22
+private static final TableReference DEFAULT_TABLE = TableReference.fromString(%22tr%22);%0A    private static final String DEFAULT_COLUMN_NAME = %22col1%22;%0A%0A    AllCellsPerRowPager pager = new AllCellsPerRowPager(%0A            executor,%0A            rowKey,%0A            DEFAULT_TABLE,%0A            pageSize%0A    )
 ;%0A%0A 
@@ -3159,20 +3159,48 @@
 M %25s %22, 
-%22tr%22
+DEFAULT_TABLE.getQualifiedName()
 )));%0A   
@@ -3361,16 +3361,21 @@
 ey = %25s 
+LIMIT
 %22, encod
@@ -4876,7 +4876,6 @@
   %7D%0A
-%0A%0A
 %7D
+%0A

Clean up tests