### Question:
CassandraTable { public void save(final List<Event> events) { final BatchStatement batch = new BatchStatement(); for (final Event event : events) { final Map<String, Object> parsedEvent = parse(event); if (parsedEvent.isEmpty()) { log.warn("Event {} could not be mapped. Suggestion: Cassandra is case sensitive, so maybe you can check field names.", event); continue; } if (!hasPrimaryKey(parsedEvent)) { break; } final Insert insert = QueryBuilder.insertInto(table); for (final Map.Entry<String, Object> entry : parsedEvent.entrySet()) { insert.value(entry.getKey(), entry.getValue()); } if (log.isTraceEnabled()) { log.trace("Preparing insert for table {}: {}", table.getName(), insert.getQueryString()); } batch.add(insert); } if (batch.getStatements().isEmpty()) { log.warn("No event produced insert query for table {}", table.getName()); return; } batch.setConsistencyLevel(consistencyLevel); session.execute(batch); } CassandraTable(
      final Session session,
      final TableMetadata table,
      final ConsistencyLevel consistencyLevel,
      final String bodyColumn);  CassandraTable(
      final Session session,
      final TableMetadata table,
      final ConsistencyLevel consistencyLevel,
      final String bodyColumn,
      final boolean ignoreCase); void save(final List<Event> events); Map<String, Object> parse(final Event event);  }### Answer:
@Test public void save() { mockTableMetadata(); final CassandraTable table = new CassandraTable(session, tableMetadata, ConsistencyLevel.QUORUM, null); final Event event = EventBuilder .withBody("my_body".getBytes(), ImmutableMap.of("id", "1", "text_col", "my_text", "other", "foo")); table.save(ImmutableList.of(event)); verify(session).execute(argThat(new ArgumentMatcher<BatchStatement>() { @Override public boolean matches(Object o) { if (!(o instanceof BatchStatement)) { return false; } final BatchStatement batch = (BatchStatement) o; final ConsistencyLevel consistencyLevel = batch.getConsistencyLevel(); final Statement statement = batch.getStatements().iterator().next(); log.debug("Batch got consistency: {}", consistencyLevel); log.debug("Batch got insert: {}", statement); return consistencyLevel == ConsistencyLevel.QUORUM && "INSERT INTO my_keyspace.my_table(text_col,id) VALUES ('my_text',1);".equals(statement.toString()); } })); verifyNoMoreInteractions(session); }
@Test public void saveUnmappable() { mockTableMetadata(); final CassandraTable table = new CassandraTable(session, tableMetadata, ConsistencyLevel.QUORUM, null); final Event event = EventBuilder .withBody("my_body".getBytes(), ImmutableMap.of("XXXid", "1", "XXXtext_col", "my_text")); table.save(ImmutableList.of(event)); verifyZeroInteractions(session); }
@Test public void saveWithNoPrimaryKey() { mockTableMetadata(); final CassandraTable table = new CassandraTable(session, tableMetadata, ConsistencyLevel.QUORUM, null); final Event event = EventBuilder .withBody("my_body".getBytes(), ImmutableMap.of("XXXid", "1", "text_col", "my_text")); table.save(ImmutableList.of(event)); verifyZeroInteractions(session); }
@Test public void saveEmpty() { mockTableMetadata(); final CassandraTable table = new CassandraTable(session, tableMetadata, ConsistencyLevel.QUORUM, null); table.save(ImmutableList.<Event>of()); verifyZeroInteractions(session); }