如何在Ktor+Exposed+Hikari+JUnit5的E2E测试中回滚数据库

基于Ktor的多模块后端应用，使用Exposed DAO+Hikari连接Docker部署的MySQL，已通过Flyway完成测试前的表结构和初始数据迁移。当前集成测试存在问题：修改特定表数据的测试用例会影响其他测试，导致CI运行不稳定，需要实现每个E2E测试后自动回滚数据库数据。

相关代码示例 #

E2E测试用例示例 #

class ExampleTest {

    @Test
    fun `e2e Test for the Example endpoints`() = testApplication {
        environment {
           config = ApplicationConfig(application_integration_test.conf)
        }
        val response = client.post("/path/example/test") {
             header("Authorization", "token")
             header(contentType, APPLICATION_JSON)
             setBody("body")
        }
        assertEquals(201, response.status)
        assertEquals("[]", response.bodyAsText())
        // 其他断言逻辑
    }
}

测试环境数据库配置（application_integration_test.conf） #

database {
    driverClassName = com.mysql.cj.jdbc.Driver
    jdbcUrl = "jdbc:mysql://localhost:3307/test?useSSL=false"
    username = testuser
    password = password
    maximumPoolSize = 4
    isAutoCommit = false
    transactionIsolation = TRANSACTION_REPEATABLE_READ
    connectionTimeout = 60000
    maxLifetime = 20000000
    idleTimeout = 10000000
    leakDetectionThreshold = 8000
}

应用启动相关代码 #

testMain方法

@KtorExperimentalLocationsAPI
fun Application.testMain() {
    setupIntegrationTests(ConfigurationSetUpper)
    installAuthentication()
    install(RoleBasedAuthorization)
    install(CallLogging)

    routing {
        testRoute()
    }
}

setupIntegrationTests方法

@KtorExperimentalLocationsAPI
fun Application.setupIntegrationTests(configurationSetUpper: ConfigurationSetUpper) {
    TimeZone.setDefault(TimeZone.getTimeZone("UTC"))

    val env = Environment.getBy(environment.config.property("ktor.deployment.environment"))
    configurationSetUpper.setUp(environment.config)

    DatabaseFactory.init()
    installIntegrationTestKoin()
    // 其他初始化逻辑
}

DatabaseFactory实现 #

object DatabaseFactory {

    fun init() {
        Database.connect(hikari())
    }

    fun hikari(): HikariDataSource {
        val config = HikariConfig()
        config.driverClassName = DatabaseConfig.driverClassName
        config.jdbcUrl = DatabaseConfig.jdbcUrl
        config.username = DatabaseConfig.userName
        config.password = DatabaseConfig.password
        config.maximumPoolSize = DatabaseConfig.maximumPoolSize
        config.isAutoCommit = DatabaseConfig.isAutoCommit
        config.transactionIsolation = DatabaseConfig.transactionIsolation
        DatabaseConfig.maxLifetime?.let {
            config.maxLifetime = it
        }
        DatabaseConfig.leakDetectionThreshold?.let {
            config.leakDetectionThreshold = it
        }
        config.validate()
        return HikariDataSource(config)
    }
}

已尝试的无效方案 #

在测试客户端代码外层包裹Exposed事务并手动回滚，由于应用内部的数据库操作运行在独立线程，和测试线程的事务无关，无法生效：

class ExampleTest {

    @Test
    fun `e2e Test for the Example endpoints`() = testApplication {
        environment {
           config = ApplicationConfig(application_integration_test.conf)
        }
        transaction {
            try {
                val response = client.post("/path/example/test") {
                     header("Authorization", "token")
                     header(contentType, APPLICATION_JSON)
                     setBody("body")
                }
                assertEquals(201, response.status)
                assertEquals("[]", response.bodyAsText())
                // 其他断言
            } finally {
                TransactionManager.current().rollback()
            }
        }
    }
}

可行解决方案 #

方案1：全局事务包裹整个测试生命周期 #

利用Ktor测试的回调机制，在应用初始化后开启一个全局数据库事务，所有测试请求的数据库操作都在该事务中执行，测试结束后统一回滚：

class ExampleTest {

    @Test
    fun `e2e Test for the Example endpoints`() = testApplication {
        environment {
            config = ApplicationConfig(application_integration_test.conf)
        }

        // 初始化后开启全局事务
        val globalTransaction = transaction(
            db = Database.connect(DatabaseFactory.hikari()),
            transactionIsolation = TransactionIsolationLevel.REPEATABLE_READ
        ) {
            // 禁止自动提交
            TransactionManager.current().connection.autoCommit = false
            this
        }

        try {
            val response = client.post("/path/example/test") {
                 header("Authorization", "token")
                 header(contentType, APPLICATION_JSON)
                 setBody("body")
            }
            assertEquals(201, response.status)
            assertEquals("[]", response.bodyAsText())
            // 其他断言逻辑
        } finally {
            // 回滚全局事务
            globalTransaction.rollback()
            globalTransaction.close()
        }
    }
}

方案2：自定义测试扩展（复用逻辑） #

如果多个测试需要复用该逻辑，可以封装成JUnit 5扩展：

class DatabaseRollbackExtension : BeforeEachCallback, AfterEachCallback {
    private lateinit var transaction: Transaction

    override fun beforeEach(context: ExtensionContext?) {
        // 开启全局事务
        transaction = transaction(Database.connect(DatabaseFactory.hikari())) {
            TransactionManager.current().connection.autoCommit = false
            this
        }
    }

    override fun afterEach(context: ExtensionContext?) {
        // 回滚并关闭事务
        transaction.rollback()
        transaction.close()
    }
}

然后在测试类上添加：

@ExtendWith(DatabaseRollbackExtension::class)
class ExampleTest {
    // 测试用例
}

方案3：重置数据库快照（适合小型测试库） #

如果事务回滚有兼容性问题，可以在每个测试前用Flyway重新执行迁移，或者加载预定义的数据库快照：

class ExampleTest {

    @BeforeEach
    fun resetDatabase() {
        // 重新执行Flyway迁移，重置到初始状态
        val flyway = Flyway.configure()
            .dataSource(DatabaseConfig.jdbcUrl, DatabaseConfig.userName, DatabaseConfig.password)
            .cleanDisabled(false)
            .load()
        flyway.clean()
        flyway.migrate()
    }

    @Test
    fun `e2e Test for the Example endpoints`() = testApplication {
        // 测试逻辑
    }
}

注意：该方案每次测试都会清理并重建数据库，适合数据量小的测试场景，否则会影响测试执行速度。

关键注意事项 #

• 确保测试环境的isAutoCommit配置为false，否则事务无法回滚。
• 全局事务需要覆盖应用所有数据库操作的线程，Exposed的TransactionManager会自动绑定当前线程的事务，因此应用的请求线程会复用该事务。
• 如果使用连接池，需要确保连接池的maximumPoolSize不要过大，避免出现多个连接不在同一事务中的情况。

内容的提问来源于stack exchange，提问作者shokdjsa-dkfdk.ss