使用 CassandraUnit 进行 Cassandra 单元测试的实践指南

1. 概述

Apache Cassandra 是一款强大的开源 NoSQL 分布式数据库。在之前的教程中，我们探讨了 Cassandra 与 Java 集成的基础知识。本文将在此基础上，重点介绍如何使用 CassandraUnit 编写可靠、自包含的单元测试。

CassandraUnit 的核心价值在于：

提供嵌入式 Cassandra 实例，无需外部数据库服务器
支持多种测试方式（原生 API、JUnit 规则、Spring 集成）
确保测试的独立性和可重复性

提示：在生产环境中使用 Cassandra 时，建议考虑基于 Apache Cassandra 构建的云数据库 Astra，避免自建服务器的运维复杂度。

2. 依赖配置

首先需要在 pom.xml 中添加两个关键依赖：

核心驱动依赖

<dependency>
    <groupId>com.datastax.oss</groupId>
    <artifactId>java-driver-core</artifactId>
    <version>4.13.0</version>
</dependency>

测试框架依赖

<dependency>
    <groupId>org.cassandraunit</groupId>
    <artifactId>cassandra-unit</artifactId>
    <version>4.3.1.0</version>
    <scope>test</scope>
</dependency>

注意：测试依赖的 scope 设置为 test，避免污染生产环境。

3. 快速上手

本文测试基于一个简单的 person 表，通过 CQL 脚本定义：

CREATE TABLE person(
    id varchar,
    name varchar,
    PRIMARY KEY(id));

INSERT INTO person(id, name) values('1234','Eugen');
INSERT INTO person(id, name) values('5678','Michael');

CassandraUnit 的测试模式遵循三个核心步骤：

启动嵌入式服务器：在 JVM 内存中运行 Cassandra
加载测试数据：将预定义数据集注入实例
验证数据完整性：执行查询确认数据正确性

测试原则：单元/集成测试应避免依赖外部服务，防止因服务不可用导致测试结果不稳定。嵌入式方案完美解决了这个问题。

4. 原生 API 测试方式

使用 CassandraUnit 的原生 API 需要手动管理生命周期：

public class NativeEmbeddedCassandraUnitTest {

    private CqlSession session;

    @Before
    public void setUp() throws Exception {
        EmbeddedCassandraServerHelper.startEmbeddedCassandra();
        session = EmbeddedCassandraServerHelper.getSession();
        new CQLDataLoader(session).load(new ClassPathCQLDataSet("people.cql", "people"));
    }
}

关键点解析：

服务器启动：startEmbeddedCassandra() 启动默认端口 9142 的实例

11:13:36.754 [pool-2-thread-1] INFO  o.apache.cassandra.transport.Server
  - Starting listening for CQL clients on localhost/127.0.0.1:9142 (unencrypted)...

端口配置：可使用随机端口替代固定端口

EmbeddedCassandraServerHelper
  .startEmbeddedCassandra(EmbeddedCassandraServerHelper.CASSANDRA_RNDPORT_YML_FILE);

数据加载：ClassPathCQLDataSet 从类路径加载 CQL 文件

测试用例示例：

@Test
public void givenEmbeddedCassandraInstance_whenStarted_thenQuerySuccess() throws Exception {
    ResultSet result = session.execute("select * from person WHERE id=1234");
    assertThat(result.iterator().next().getString("name"), is("Eugen"));
}

清理资源：

@After
public void tearDown() throws Exception {
    EmbeddedCassandraServerHelper.cleanEmbeddedCassandra();
}

踩坑提醒：cleanEmbeddedCassandra() 会删除除 system 外的所有 keyspace，确保测试隔离性。

5. 使用抽象测试类

CassandraUnit 提供了 AbstractCassandraUnit4CQLTestCase 简化测试：

public class AbstractTestCaseWithEmbeddedCassandraUnitTest
  extends AbstractCassandraUnit4CQLTestCase {

    @Override
    public CQLDataSet getDataSet() {
        return new ClassPathCQLDataSet("people.cql", "people");
    }

    @Test
    public void givenEmbeddedCassandraInstance_whenStarted_thenQuerySuccess()
      throws Exception {
        ResultSet result = this.getSession().execute("select * from person WHERE id=1234");
        assertThat(result.iterator().next().getString("name"), is("Eugen"));
    }
}

优势：

自动管理服务器生命周期
只需重写 getDataSet() 指定数据集
通过 getSession() 获取会话连接

6. 使用 JUnit Rule 方式

对于不想继承抽象类的场景，使用 CassandraCQLUnit Rule：

public class JUnitRuleWithEmbeddedCassandraUnitTest {

    @Rule
    public CassandraCQLUnit cassandra = new CassandraCQLUnit(new ClassPathCQLDataSet("people.cql", "people"));

    @Test
    public void givenEmbeddedCassandraInstance_whenStarted_thenQuerySuccess() throws Exception {
        ResultSet result = cassandra.session.execute("select * from person WHERE id=5678");
        assertThat(result.iterator().next().getString("name"), is("Michael"));
    }
}

特点：

通过 @Rule 声明管理生命周期
直接访问 cassandra.session 执行查询
适合需要灵活继承结构的测试类

7. Spring 集成测试

在 Spring 项目中集成 CassandraUnit 需要额外依赖：

<dependency>
    <groupId>org.cassandraunit</groupId>
    <artifactId>cassandra-unit-spring</artifactId>
    <version>4.3.1.0</version>
    <scope>test</scope>
</dependency>

测试类示例：

@RunWith(SpringJUnit4ClassRunner.class)
@TestExecutionListeners({ CassandraUnitTestExecutionListener.class })
@CassandraDataSet(value = "people.cql", keyspace = "people")
@EmbeddedCassandra
public class SpringWithEmbeddedCassandraUnitTest {

    @Test
    public void givenEmbeddedCassandraInstance_whenStarted_thenQuerySuccess() throws Exception {
        CqlSession session = EmbeddedCassandraServerHelper.getSession();

        ResultSet result = session.execute("select * from person WHERE id=1234");
        assertThat(result.iterator().next().getString("name"), is("Eugen"));
    }
}

注解解析：

@RunWith：启用 Spring 测试上下文
@TestExecutionListeners：注册 CassandraUnitTestExecutionListener 管理服务器
@EmbeddedCassandra：注入嵌入式实例，支持配置：
- configuration：自定义配置文件
- clusterName：集群名称
- host：主机地址
- port：端口号
@CassandraDataSet：加载测试数据集

简单粗暴：省略配置项时使用默认值，适合快速原型验证。

8. 总结

本文系统介绍了 CassandraUnit 的三种测试模式：

原生 API：最大控制力，适合复杂场景
抽象测试类：减少样板代码，适合标准测试
JUnit Rule：灵活继承，适合特殊需求
Spring 集成：无缝对接 Spring 生态

核心优势：

消除外部依赖，提升测试稳定性
支持多种测试框架集成
提供完整生命周期管理

完整示例代码可参考 GitHub 仓库

Persistence

REST

Security