Test case and fixes for in.partake.model.dao.cassandra.CassandraKeyColumnDataIterator

KDCIterTests.patch

Index: src/main/java/in/partake/model/dao/cassandra/CassandraDao.java
===================================================================
--- src/main/java/in/partake/model/dao/cassandra/CassandraDao.java	(revision 310)
+++ src/main/java/in/partake/model/dao/cassandra/CassandraDao.java	(working copy)
@@ -12,6 +12,7 @@
 
 import org.apache.cassandra.thrift.Column;
 import org.apache.cassandra.thrift.ColumnOrSuperColumn;
+import org.apache.cassandra.thrift.ColumnPath;
 import org.apache.cassandra.thrift.ConsistencyLevel;
 import org.apache.cassandra.thrift.Mutation;
 import org.apache.cassandra.thrift.Cassandra.Client;
@@ -126,11 +127,11 @@
     
     private void removeAllData(CassandraConnection con, String columnFamily, long now) throws Exception {
         // TODO: Since it takes too much time, we do not remove the data for now.
-//        KeyIterator it = new CassandraKeyIterator(con, "Keyspace1", "", columnFamily, ConsistencyLevel.ALL);
-//        while (it.hasNext()) {
-//            String key = it.nextWithPrefix();
-//            ColumnPath columnPath = new ColumnPath(columnFamily);
-//            con.getClient().remove("Keyspace1", key, columnPath, now, ConsistencyLevel.ALL);
-//        }        
+    	CassandraKeyIterator it = new CassandraKeyIterator(con, "Keyspace1", "", columnFamily, ConsistencyLevel.ALL);
+        while (it.hasNext()) {
+            String key = it.nextWithPrefix();
+            ColumnPath columnPath = new ColumnPath(columnFamily);
+            con.getClient().remove("Keyspace1", key, columnPath, now, ConsistencyLevel.ALL);
+        }        
     }
 }
Index: src/main/java/in/partake/model/dao/cassandra/CassandraKeyColumnDataIterator.java
===================================================================
--- src/main/java/in/partake/model/dao/cassandra/CassandraKeyColumnDataIterator.java	(revision 310)
+++ src/main/java/in/partake/model/dao/cassandra/CassandraKeyColumnDataIterator.java	(working copy)
@@ -33,8 +33,8 @@
     public boolean hasNext() throws DAOException {
         while (true) {
             if (inner == null) {
-                if (outer.hasNext()) { return false; }
-                currentKey = outer.next();
+                if (!outer.hasNext()) { return false; }
+                currentKey = outer.nextWithPrefix();
                 inner = new ColumnIterator(con, desc.keyspace, currentKey, desc.columnFamily, false, desc.readConsistency, desc.writeConsistency);
             }
 
Index: src/main/java/in/partake/model/dao/cassandra/UserCassandraDao.java
===================================================================
--- src/main/java/in/partake/model/dao/cassandra/UserCassandraDao.java	(revision 310)
+++ src/main/java/in/partake/model/dao/cassandra/UserCassandraDao.java	(working copy)
@@ -39,11 +39,11 @@
 class UserCassandraDao extends CassandraDao implements IUserAccess {
     
 	// USER MASTER TABLE
-    private static final String USERS_PREFIX = "users:id:";
-    private static final String USERS_KEYSPACE = "Keyspace1";
-    private static final String USERS_COLUMNFAMILY = "Standard2";
-    private static final ConsistencyLevel USERS_CL_R = ConsistencyLevel.ONE;
-    private static final ConsistencyLevel USERS_CL_W = ConsistencyLevel.ALL;
+    static final String USERS_PREFIX = "users:id:";
+    static final String USERS_KEYSPACE = "Keyspace1";
+    static final String USERS_COLUMNFAMILY = "Standard2";
+    static final ConsistencyLevel USERS_CL_R = ConsistencyLevel.ONE;
+    static final ConsistencyLevel USERS_CL_W = ConsistencyLevel.ALL;
     
     // ----------------------------------------------------------------------
     
Index: src/test/java/in/partake/model/dao/cassandra/CassandraKeyColumnDataIteratorTest.java
===================================================================
--- src/test/java/in/partake/model/dao/cassandra/CassandraKeyColumnDataIteratorTest.java	(revision 0)
+++ src/test/java/in/partake/model/dao/cassandra/CassandraKeyColumnDataIteratorTest.java	(revision 0)
@@ -0,0 +1,123 @@
+package in.partake.model.dao.cassandra;
+
+import in.partake.model.dao.AbstractDaoTestCaseBase;
+import in.partake.model.dao.DAOException;
+import in.partake.model.dao.IUserAccess;
+import in.partake.model.dao.PartakeDAOFactory;
+import in.partake.model.dto.User;
+import in.partake.resource.PartakeProperties;
+
+import java.lang.reflect.Field;
+import java.util.Date;
+
+import org.apache.cassandra.thrift.ColumnOrSuperColumn;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class CassandraKeyColumnDataIteratorTest extends AbstractDaoTestCaseBase {
+
+	@BeforeClass
+	public static void setUpOnce() {
+		PartakeProperties.get().reset("cassandra");
+		reset();
+	}
+
+	@AfterClass
+	public static void tearDownOnce() {
+		PartakeProperties.get().reset();
+		reset();
+	}
+
+	@Before
+	public void setup() throws DAOException {
+		// truncate all Users for testing.
+		super.setup(getFactory().getUserAccess()); 
+	}
+
+	@Test
+	public void iterateEmpty() throws DAOException {
+		CassandraTableDescription desc = new CassandraTableDescription(
+				UserCassandraDao.USERS_PREFIX,
+				UserCassandraDao.USERS_KEYSPACE,
+				UserCassandraDao.USERS_COLUMNFAMILY,
+				UserCassandraDao.USERS_CL_R,
+				UserCassandraDao.USERS_CL_W);
+		CassandraConnection con = (CassandraConnection) getPool().getConnection();
+		ColumnOrSuperColumnKeyMapper<Object> mapper = new FakeMapper(con, getFactory());
+
+		try {
+			con.beginTransaction();
+			CassandraKeyColumnDataIterator<Object> iter;
+			for (iter = new CassandraKeyColumnDataIterator<Object>(con, desc, mapper); iter.hasNext();) {
+				Assert.fail();
+			}
+			con.commit();
+		} finally {
+			con.invalidate();
+		}
+	}
+
+	@Test
+	public void iterateOne() throws DAOException {
+		CassandraTableDescription desc = new CassandraTableDescription(
+				UserCassandraDao.USERS_PREFIX,
+				UserCassandraDao.USERS_KEYSPACE,
+				UserCassandraDao.USERS_COLUMNFAMILY,
+				UserCassandraDao.USERS_CL_R,
+				UserCassandraDao.USERS_CL_W);
+		IUserAccess userDao = getFactory().getUserAccess();
+		CassandraConnection con = (CassandraConnection) getPool().getConnection();
+		ColumnOrSuperColumnKeyMapper<Object> mapper = new FakeMapper(con, getFactory());
+
+		try {
+			con.beginTransaction();
+			CassandraKeyColumnDataIterator<Object> iter;
+			for (iter = new CassandraKeyColumnDataIterator<Object>(con, desc, mapper); iter.hasNext();) {
+				Assert.fail();
+			}
+
+			String userId = userDao.getFreshId(con);
+			User user = new User(userId, 0, new Date(), "calendarId");
+			userDao.createUser(con, user);
+
+			int found = 0;
+			for (iter = new CassandraKeyColumnDataIterator<Object>(con, desc, mapper); iter.hasNext();) {
+				iter.next();
+				++found;
+			}
+			con.commit();
+			Assert.assertEquals(counfColumnsOf(User.class), found);
+		} finally {
+			con.invalidate();
+		}
+	}
+
+	private int counfColumnsOf(Class<?> clazz) {
+		int count = 0;
+		for (Field f : clazz.getDeclaredFields()) {
+			if (f.isAnnotationPresent(javax.persistence.Column.class)) {
+				++count;
+			}
+		}
+		return count;
+	}
+
+	private static class FakeMapper extends ColumnOrSuperColumnKeyMapper<Object> {
+		public FakeMapper(CassandraConnection connection, PartakeDAOFactory factory) {
+			super(connection, factory);
+		}
+
+		@Override
+		public Object map(ColumnOrSuperColumn cosc, String key) throws DAOException {
+			return null;
+		}
+
+		@Override
+		public ColumnOrSuperColumn unmap(Object t) throws DAOException {
+			return null;
+		}
+	}
+}