com.strapdata.elassandraunit:elassandra-unit

public class SimpleTest {

    static final String KEYSPACE = "ks";

    @ClassRule
    public static CassandraCQLUnit cassandraCQLUnit =
        new CassandraCQLUnit(new SimpleCQLDataSet(
            "CREATE TABLE users (email text PRIMARY KEY, firstname text, lastname text, es_query text, es_options text);", KEYSPACE));

    private static Mapper<User> userMapper;
    private static RestHighLevelClient client;

    @Before
    public void setup() throws IOException {
        userMapper = new MappingManager(cassandraCQLUnit.session).mapper(User.class);
        client = new RestHighLevelClient(RestClient.builder(new HttpHost("localhost", 9200, "http")));

        // create an elasticsearch index on table users
        CreateIndexRequest request = new CreateIndexRequest("users");
        request.mapping("users", XContentFactory.jsonBuilder()
            .startObject()
                .startObject("users")
                    .field("discover", ".*")
                .endObject()
            .endObject());
        request.settings(Settings.builder()
            .put("keyspace", KEYSPACE)                          // map index users to our keyspace.
            .put("index.synchronous_refresh",true)// synchronous elasticsearch refresh
            .build());
        CreateIndexResponse createIndexResponse = client.indices().create(request);
    }

    @Test
    public void testMapper() throws Exception {
        User user1 = new User().withEmail("[email protected]").withFirstname("Bob").withLastname("Smith");
        User user2 = new User().withEmail("[email protected]").withFirstname("Alice").withLastname("Smith");
        User user3 = new User().withEmail("[email protected]").withFirstname("Paul").withLastname("Dupont");
        userMapper.save(user1);
        userMapper.save(user2);
        userMapper.save(user3);

        User user = userMapper.get("[email protected]");
        assertThat(user, is(user1));

        // Elasticsearch search through the REST API
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().query(QueryBuilders.termQuery("lastname", "Smith"));
        SearchRequest searchRequest = new SearchRequest().indices("users").source(sourceBuilder);
        SearchResponse searchResponse = client.search(searchRequest);
        assertThat(searchResponse.getHits().totalHits, is(2L));

        // Elasticsearch search through CQL
        String esQuery = new SearchSourceBuilder().query(QueryBuilders.termQuery("lastname", "Smith")).toString(ToXContent.EMPTY_PARAMS);
        ResultSet results = cassandraCQLUnit.session.execute(
            "SELECT * FROM users WHERE es_options = ? AND es_query = ? ALLOW FILTERING", "indices=users", esQuery);
        Result<User> users = userMapper.map(results);
        assertThat(users.all().size(), is(2));
    }
}