@Test
public void testBolt() throws Exception {

    TupleMapper<String, String, String> tupleMapper = new DefaultTupleMapper(KEYSPACE, "users", "VALUE");
    String configKey = "cassandra-config";
    CassandraBatchingBolt<String, String, String> bolt = new CassandraBatchingBolt<String, String, String>(configKey, tupleMapper);
    TopologyBuilder builder = new TopologyBuilder();
    builder.setBolt("TEST_BOLT", bolt);
    Fields fields = new Fields("VALUE");
    TopologyContext context = new MockTopologyContext(builder.createTopology(), fields);
    Config config = new Config();
    config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 5000);
    
    Map<String, Object> clientConfig = new HashMap<String, Object>();
    clientConfig.put(StormCassandraConstants.CASSANDRA_HOST, "localhost:9160");
    clientConfig.put(StormCassandraConstants.CASSANDRA_KEYSPACE, Arrays.asList(new String [] {KEYSPACE}));
    config.put(configKey, clientConfig);
    bolt.prepare(config, context, null);
    System.out.println("Bolt Preparation Complete.");
    Values values = new Values(42);
    Tuple tuple = new TupleImpl(context, values, 5, "test");
    bolt.execute(tuple);
    // wait very briefly for the batch to complete
    Thread.sleep(250);
    AstyanaxContext<Keyspace> astyContext = newContext("localhost:9160", KEYSPACE);
    Keyspace ks = astyContext.getEntity();
    Column<String> result = ks
            .prepareQuery(new ColumnFamily<String, String>("users", StringSerializer.get(), StringSerializer.get()))
            .getKey("42").getColumn("VALUE").execute().getResult();
    assertEquals("42", result.getStringValue());
}