package com.continuuity.data2.dataset2;

import com.continuuity.api.common.Bytes;
import com.continuuity.data2.dataset2.lib.table.CoreDatasetsModule;
import com.continuuity.data2.dataset2.lib.table.KeyValueTable;
import com.continuuity.data2.dataset2.module.lib.TableModule;
import com.continuuity.data2.dataset2.module.lib.inmemory.InMemoryOrderedTableModule;
import com.continuuity.data2.transaction.DefaultTransactionExecutor;
import com.continuuity.data2.transaction.TransactionAware;
import com.continuuity.data2.transaction.TransactionExecutor;
import com.continuuity.data2.transaction.inmemory.MinimalTxSystemClient;
import com.continuuity.internal.data.dataset.DatasetAdmin;
import com.continuuity.internal.data.dataset.DatasetProperties;
import com.continuuity.internal.data.dataset.lib.table.OrderedTable;
import com.google.gson.Gson;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 */
public abstract class AbstractDatasetFrameworkTest {

  private static final Logger LOG = LoggerFactory.getLogger(AbstractDatasetFrameworkTest.class);
  private static final Gson GSON = new Gson();

  protected abstract DatasetFramework getFramework();

  @Test
  public void testSimpleDataset() throws Exception {
    // Configuring Dataset types
    DatasetFramework framework = getFramework();
    String moduleName = "inMemory";

    framework.addModule(moduleName, new InMemoryOrderedTableModule());

    // Creating instance
    framework.addInstance("orderedTable", "my_table", DatasetProperties.EMPTY);

    // Doing some admin and data ops
    DatasetAdmin admin = framework.getAdmin("my_table", null);
    Assert.assertNotNull(admin);
    final OrderedTable table = framework.getDataset("my_table", null);
    Assert.assertNotNull(table);

    TransactionExecutor txnl = new DefaultTransactionExecutor(new MinimalTxSystemClient(), (TransactionAware) table);
    txnl.execute(new TransactionExecutor.Subroutine() {
      @Override
      public void apply() throws Exception {
        table.put(Bytes.toBytes("key1"), Bytes.toBytes("column1"), Bytes.toBytes("value1"));
      }
    });
    txnl.execute(new TransactionExecutor.Subroutine() {
      @Override
      public void apply() throws Exception {
        Assert.assertEquals("value1", Bytes.toString(table.get(Bytes.toBytes("key1"), Bytes.toBytes("column1"))));
      }
    });
    admin.truncate();
    txnl.execute(new TransactionExecutor.Subroutine() {
      @Override
      public void apply() throws Exception {
        Assert.assertTrue(table.get(Bytes.toBytes("key1")).isEmpty());
      }
    });

    // cleanup
    framework.deleteInstance("my_table");
    framework.deleteModule("inMemory");
  }

  @Test
  public void testCompositeDatasetWithModule() throws Exception {
    // Configuring Dataset types
    DatasetFramework framework = getFramework();

    framework.addModule("inMemory", new InMemoryOrderedTableModule());
    framework.addModule("default", new CoreDatasetsModule());

    // Creating instance
    framework.addInstance("keyValueTable", "my_table", DatasetProperties.EMPTY);

    testCompositeDataset(framework);

    // cleanup
    framework.deleteInstance("my_table");
    framework.deleteModule("default");
    framework.deleteModule("inMemory");
  }

  @Test
  public void testCompositeDataset() throws Exception {
    // Configuring Dataset types
    DatasetFramework framework = getFramework();

    framework.addModule("inMemory", new InMemoryOrderedTableModule());
    framework.addModule("table", new TableModule());
    framework.addModule("keyValue", new SingleTypeModule(SimpleKVTable.class));

    // Creating instance
    framework.addInstance(SimpleKVTable.class.getName(),
                          "my_table", DatasetProperties.EMPTY);

    testCompositeDataset(framework);

    // cleanup
    framework.deleteInstance("my_table");
    framework.deleteModule("keyValue");
    framework.deleteModule("table");
    framework.deleteModule("inMemory");
  }

  @Test
  public void testDoubleCompositeDataset() throws Exception {
    // Configuring Dataset types
    DatasetFramework framework = getFramework();

    framework.addModule("inMemory", new InMemoryOrderedTableModule());
    framework.addModule("table", new TableModule());
    framework.addModule("keyValue", new SingleTypeModule(SimpleKVTable.class));
    framework.addModule("doubleKeyValue", new SingleTypeModule(DoubleWrappedKVTable.class));

    // Creating instance
    framework.addInstance(DoubleWrappedKVTable.class.getName(), "my_table", DatasetProperties.EMPTY);

    testCompositeDataset(framework);

    // cleanup
    framework.deleteInstance("my_table");
    framework.deleteModule("doubleKeyValue");
    framework.deleteModule("keyValue");
    framework.deleteModule("table");
    framework.deleteModule("inMemory");
  }

  private void testCompositeDataset(DatasetFramework framework) throws Exception {

    // Doing some admin and data ops
    DatasetAdmin admin = framework.getAdmin("my_table", null);
    Assert.assertNotNull(admin);
    final KeyValueTable table = framework.getDataset("my_table", null);
    Assert.assertNotNull(table);

    TransactionExecutor txnl = new DefaultTransactionExecutor(new MinimalTxSystemClient(), (TransactionAware) table);
    txnl.execute(new TransactionExecutor.Subroutine() {
      @Override
      public void apply() throws Exception {
        table.write("key1", "value1");
      }
    });
    txnl.execute(new TransactionExecutor.Subroutine() {
      @Override
      public void apply() throws Exception {
        Assert.assertEquals("value1", Bytes.toString(table.read("key1")));
      }
    });
    admin.truncate();
    txnl.execute(new TransactionExecutor.Subroutine() {
      @Override
      public void apply() throws Exception {
        Assert.assertEquals(null, Bytes.toString(table.read("key1")));
      }
    });
  }

}

@@ -79,139 +79,8 @@
 es;%0A
-import com.continuuity.data2.dataset2.lib.table.CoreDatasetsModule;%0Aimport com.continuuity.data2.dataset2.lib.table.KeyValueTable;%0A
 impo
@@ -129,32 +129,32 @@
 ib.TableModule;%0A
+
 import com.conti
@@ -682,37 +682,8 @@
 le;%0A
-import com.google.gson.Gson;%0A
 impo
@@ -729,65 +729,8 @@
 est;
-%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0A/*
@@ -796,153 +796,8 @@
  %7B%0A%0A
-  private static final Logger LOG = LoggerFactory.getLogger(AbstractDatasetFrameworkTest.class);%0A  private static final Gson GSON = new Gson();%0A%0A
   pr
@@ -2368,585 +2368,8 @@
  %7D%0A%0A
-  @Test%0A  public void testCompositeDatasetWithModule() throws Exception %7B%0A    // Configuring Dataset types%0A    DatasetFramework framework = getFramework();%0A%0A    framework.addModule(%22inMemory%22, new InMemoryOrderedTableModule());%0A    framework.addModule(%22default%22, new CoreDatasetsModule());%0A%0A    // Creating instance%0A    framework.addInstance(%22keyValueTable%22, %22my_table%22, DatasetProperties.EMPTY);%0A%0A    testCompositeDataset(framework);%0A%0A    // cleanup%0A    framework.deleteInstance(%22my_table%22);%0A    framework.deleteModule(%22default%22);%0A    framework.deleteModule(%22inMemory%22);%0A  %7D%0A%0A
   @T
@@ -4498,13 +4498,11 @@
 ble.
-write
+put
 (%22ke
@@ -4692,33 +4692,17 @@
 1%22, 
-Bytes.toString(
 table.
-read
+get
 (%22ke
@@ -4698,33 +4698,32 @@
 ble.get(%22key1%22))
-)
 ;%0A      %7D%0A    %7D)
@@ -4901,33 +4901,17 @@
 ll, 
-Bytes.toString(
 table.
-read
+get
 (%22ke
@@ -4915,17 +4915,16 @@
 %22key1%22))
-)
 ;%0A      

fix unit-test