package com.catalina.platform.api.repositories.hbase;

import static org.fest.assertions.api.Assertions.*;
import static org.mockito.Mockito.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Scanner;
import java.util.UUID;

import com.catalina.platform.hadoop.hbase.HBaseClient;
import org.hbase.async.GetRequest;
import org.hbase.async.KeyValue;
import org.hbase.async.PutRequest;
import org.joda.time.DateTime;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.mockito.runners.MockitoJUnitRunner;
import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;

import com.catalina.platform.DateFunctions;
import com.catalina.platform._;
import com.catalina.platform.api.barcodes.BarcodeLease;
import com.catalina.platform.api.barcodes.BarcodeLeases;
import com.catalina.platform.api.models.advertiser.BarcodeType;
import com.catalina.platform.api.models.advertiser.Promotion;
import com.catalina.platform.api.models.publisher.Experience;
import com.stumbleupon.async.Deferred;

@RunWith(MockitoJUnitRunner.class)
public class HbaseBarcodeReservationRepositoryTest {

    @Mock
    private HBaseClient client;
    @InjectMocks
    private HbaseBarcodeReservationRepository repo;
    private Promotion promotion;

    @Before
    public void init() throws IOException {
        //client = mock(HBaseClient.class);

        //Deferred<Object> table_check = new Deferred<>();
        //when(client.ensureTableExists("barcodes")).thenReturn(table_check);
        //table_check.callback(true);

        repo = new HbaseBarcodeReservationRepository("barcodes");

        promotion = new Promotion(Experience.newBuilder().withId("foo").build());
        promotion.id = UUID.randomUUID().toString();
        promotion.start_date = DateFunctions.toISO8601(DateTime.now().withTimeAtStartOfDay());
        promotion.stop_date = DateFunctions.toISO8601(DateTime.now().withTimeAtStartOfDay().plusDays(7));
        MockitoAnnotations.initMocks(this);
    }

    @Test
    public void we_can_claim_a_barcode() {
        BarcodeLeases v = new BarcodeLeases("USA", BarcodeType.EAN13, "1234123412");

        when(client.get("barcodes","1234123412", "c")).thenReturn(_.toJson(v).getBytes());
        //when(hbase_client.get(TABLE_NAME, "689:helvetica", AbstractHbaseRepository.Schema.COLUMN_FAMILY)).thenReturn(hbase.serialize(sample_font));

        //ArrayList<KeyValue> rows = new ArrayList<>();
        //rows.add(new KeyValue("doesntmatter".getBytes(), "c".getBytes(), "doc".getBytes(), _.toJson(v).getBytes()));
        //dfd.callback(rows);

        Boolean success = this.repo.tryLease(BarcodeType.EAN13, "1234123412", promotion, 14);

        assertThat(success).isTrue();

        //verify(this.client, times(1)).put(any(PutRequest.class));
    }


    @Test
    public void when_barcode_already_claimed_we_cant_claim_it_for_ourselves() {

        BarcodeLeases v = new BarcodeLeases("USA", BarcodeType.EAN13, "1234123412");
        BarcodeLease e = new BarcodeLease();
        e.promotion_id = "some other promotion";
        e.expire_date = DateFunctions.toISO8601(DateTime.now().plusDays(7));
        v.leases.add(e);


        when(client.get("barcodes","1234123412", "c")).thenReturn(repo.serialize(v));


        Boolean success = this.repo.tryLease(BarcodeType.EAN13, "1234123412", promotion, 14);

        assertThat(success).isFalse();

        //verify(this.client, times(0)).put(any(PutRequest.class));
    }

    @Test
    public void when_we_are_the_one_with_an_active_lease_we_can_lease_again() {

        BarcodeLeases v = new BarcodeLeases("USA", BarcodeType.EAN13, "1234123412");
        BarcodeLease e = new BarcodeLease();
        e.promotion_id = promotion.id;
        e.expire_date = DateFunctions.toISO8601(DateTime.now().plusDays(7));
        v.leases.add(e);


        when(client.get("barcodes","1234123412", AbstractHbaseRepository.Schema.COLUMN_FAMILY)).thenReturn(_.toJson(v).getBytes());


        Boolean success = this.repo.tryLease(BarcodeType.EAN13, "1234123412", promotion, 14);

        assertThat(success).isTrue();

        //verify(this.client, times(1)).put(any(PutRequest.class));
    }

    @Test
    public void we_can_get_the_active_lease() {

        BarcodeLeases v = new BarcodeLeases("USA", BarcodeType.EAN13, "1234123412");
        BarcodeLease e = new BarcodeLease();
        e.promotion_id = promotion.id;
        e.expire_date = DateFunctions.toISO8601(DateTime.now().plusDays(7));
        v.leases.add(e);


        when(client.get("barcodes","1234123412", "c")).thenReturn(_.toJson(v).getBytes());


        BarcodeLease value = this.repo.getLeaseIfPresent(BarcodeType.EAN13, "1234123412");

        assertThat(value).isNotNull();
    }

    @Test
    public void we_can_release_leases() {
        BarcodeLeases v = new BarcodeLeases("USA", BarcodeType.EAN13, "1234123412");
        BarcodeLease e = new BarcodeLease();
        e.promotion_id = promotion.id;
        e.expire_date = DateFunctions.toISO8601(DateTime.now().plusDays(7));
        v.leases.add(e);

        Deferred<ArrayList<KeyValue>> dfd = new Deferred<>();
        when(client.get("barcodes","1234123412", AbstractHbaseRepository.Schema.COLUMN_FAMILY)).thenReturn(_.toJson(v).getBytes());


        this.repo.release(BarcodeType.EAN13, "1234123412", promotion);

        verify(this.client, times(1)).put("barcodes","1234123412", AbstractHbaseRepository.Schema.COLUMN_FAMILY, AbstractHbaseRepository.Schema.Qualifier, _.toJson(v).getBytes());
    }
}