package com.ebuddy.cassandra.dao;

import static org.testng.Assert.assertEquals;

import java.util.Arrays;

import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import com.ebuddy.cassandra.StructuredDataSupport;
import com.ebuddy.cassandra.TypeReference;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.Serializer;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.exceptions.HInvalidRequestException;
import me.prettyprint.hector.api.factory.HFactory;

/**
 * System tests for ThriftStructuredDataSupport.
 * Requires local Cassandra 1.2.x+ to be running with a default cluster ("Test Cluster")
 *
 * @author Eric Zoerner <a href="mailto:ezoerner@ebuddy.com">ezoerner@ebuddy.com</a>
 */
public class ThriftStructuredDataSupportSystemTest {

    private static final String TEST_CLUSTER = "Test Cluster";
    private static final String LOCALHOST_IP = "localhost";
    private static final String CASSANDRA_HOSTS_SYSTEM_PROPERTY = "cassandra.hosts";
    private static final String TEST_KEYSPACE = "ThriftStructuredDataSupportSystemTest";
    private final String columnFamily = "testpojo";

    private Cluster cluster;
    private StructuredDataSupport<String> dao;

    @BeforeMethod(alwaysRun = true)
    public void setUp() throws Exception {
        // default to using cassandra on localhosts, but can be overridden with a system property
        String cassandraHosts = System.getProperty(CASSANDRA_HOSTS_SYSTEM_PROPERTY, LOCALHOST_IP);
        cluster = HFactory.getOrCreateCluster(TEST_CLUSTER, cassandraHosts);
        Keyspace keyspace = HFactory.createKeyspace(TEST_KEYSPACE, cluster);
        Serializer<String> keySerializer = StringSerializer.get();
        Serializer<String> columnNameSerializer = StringSerializer.get();
        Serializer<Object> valueSerializer = StructureSerializer.get();
        ColumnFamilyOperations<String,String,Object> operations = new ColumnFamilyTemplate<String,String,Object>(
                keyspace,
                keySerializer,
                columnNameSerializer,
                valueSerializer);

        dao = new ThriftStructuredDataSupport<String>(operations);

        dropAndCreateSchema();
    }

    @Test(groups = {"system"})
    public void shouldReadAndWriteTestPojo() throws Exception {
        TestPojo testObject = new TestPojo("v1", 42L, true, Arrays.asList("e1", "e2"));
        String rowKey = "pojo0";
        String pathString = "a/b/c";
        TypeReference<TestPojo> typeReference = new TypeReference<TestPojo>() { };

        dao.writeToPath(columnFamily, rowKey, pathString, testObject);
        TestPojo result = dao.readFromPath(columnFamily, rowKey, pathString, typeReference);

        assertEquals(result, testObject);
    }

    private void dropAndCreateSchema() {
        dropKeyspaceIfExists();
        cluster.addKeyspace(HFactory.createKeyspaceDefinition(TEST_KEYSPACE), true);
        cluster.addColumnFamily(HFactory.createColumnFamilyDefinition(TEST_KEYSPACE,
                                                                      columnFamily,
                                                                      ComparatorType.UTF8TYPE));
    }

    private void dropKeyspaceIfExists() {
        try {
            cluster.dropKeyspace(TEST_KEYSPACE, true);
        } catch (HInvalidRequestException ignored) {
            // doesn't exist
        }
    }
}

@@ -709,21 +709,17 @@
 equires 
-local
+a
  Cassand
@@ -732,61 +732,16 @@
 .x+ 
-to be running with a default cluster (%22Test Cluster%22)
+instance
 %0A *%0A
@@ -1176,24 +1176,101 @@
 ystemTest%22;%0A
+    private static final String CLUSTER_NAME_PROPERTY = %22cassandra.cluster%22;%0A
     private 
@@ -1516,17 +1516,16 @@
 ocalhost
-s
 , but ca
@@ -1662,16 +1662,182 @@
 ST_IP);%0A
+%0A        // default cluster name can also be overridden with a system property%0A        String clusterName = System.getProperty(CLUSTER_NAME_PROPERTY, TEST_CLUSTER);%0A%0A
         
@@ -1874,28 +1874,27 @@
 Cluster(
-TEST_CLUSTER
+clusterName
 , cassan

use system property for the cluster name in system test