HarmonyOS开发：数据库测试——数据层测试

📌 核心要点：数据层是应用的根基，CRUD操作、事务、迁移——任何一环出问题都是数据灾难。数据库测试不是"跑一下SQL看看有没有报错"，而是系统性地验证每个数据操作的输入输出、边界条件和异常恢复。

一、背景与动机

你有没有在生产环境遇到过这种事？用户说"我的数据没了"，你查了半天，发现是事务没提交就崩溃了。或者"数据不对"，查出来是迁移脚本把字段类型改了但没处理旧数据。

数据库操作有个特点——出了问题往往不是"报错"，而是"数据悄悄错了"。查询少返回了一条记录，更新影响了多余的行，事务回滚不彻底——这些都不会抛异常，但数据已经脏了。等你发现的时候，可能已经影响了一大片用户。

手动验证数据库操作？每改一个SQL就手动去数据库里查一遍？十个表、五十个查询、二十个事务呢？你根本验证不过来。数据库测试就是用代码系统性地验证这些操作，确保CRUD正确、事务可靠、迁移安全。

二、核心原理

2.1 数据库测试体系

flowchart TB
    A[数据库测试体系] --> B[CRUD操作测试]
    A --> C[事务测试]
    A --> D[迁移测试]
    A --> E[数据隔离策略]

    B --> B1[插入验证]
    B --> B2[查询验证]
    B --> B3[更新验证]
    B --> B4[删除验证]

    C --> C1[提交验证]
    C --> C2[回滚验证]
    C --> C3[嵌套事务]
    C --> C4[并发事务冲突]

    D --> D1[版本升级迁移]
    D --> D2[版本降级回退]
    D --> D3[数据兼容性]
    D --> D4[迁移幂等性]

    E --> E1[测试前清理]
    E --> E2[测试后清理]
    E --> E3[独立数据库]
    E --> E4[事务回滚隔离]

    classDef mainStyle fill:#4CAF50,stroke:#388E3C,color:#fff,font-weight:bold
    classDef crudStyle fill:#3498DB,stroke:#2980B9,color:#fff
    classDef txStyle fill:#2ECC71,stroke:#27AE60,color:#fff
    classDef migStyle fill:#F39C12,stroke:#E67E22,color:#fff
    classDef isoStyle fill:#9B59B6,stroke:#8E44AD,color:#fff

    class A mainStyle
    class B,B1,B2,B3,B4 crudStyle
    class C,C1,C2,C3,C4 txStyle
    class D,D1,D2,D3,D4 migStyle
    class E,E1,E2,E3,E4 isoStyle

2.2 HarmonyOS RDB数据库核心API

HarmonyOS的关系型数据库（RDB）核心API：

import relationalStore from '@ohos.data.relationalStore'

// 创建/打开数据库
const store = await relationalStore.getRdbStore(context, {
  name: 'app.db',
  securityLevel: relationalStore.SecurityLevel.S1
})

// 建表
await store.executeSql('CREATE TABLE IF NOT EXISTS users (id TEXT PRIMARY KEY, name TEXT, age INTEGER)')

// 插入
const rowId = await store.insert('users', { id: '001', name: '张三', age: 25 })

// 查询
const predicates = new relationalStore.RdbPredicates('users')
predicates.equalTo('id', '001')
const resultSet = await store.query(predicates)

// 更新
const changedRows = await store.update(values, predicates)

// 删除
const deletedRows = await store.delete(predicates)

2.3 测试数据隔离策略

数据库测试最大的挑战是数据隔离——测试用例之间不能互相影响。有四种策略：

推荐用事务回滚作为默认策略，兼顾隔离性和性能。

三、代码实战

3.1 基础用法：CRUD操作验证

// TestDatabase.ets - 测试数据库辅助工具
import relationalStore from '@ohos.data.relationalStore'
import { Context } from '@ohos.abilityAccessCtrl'

export class TestDatabase {
  private store: relationalStore.RdbStore | null = null
  private context: Context

  constructor(context: Context) {
    this.context = context
  }

  // 初始化测试数据库
  async init(): Promise<void> {
    this.store = await relationalStore.getRdbStore(this.context, {
      name: 'test_app.db',
      securityLevel: relationalStore.SecurityLevel.S1
    })
    // 创建测试表
    await this.store.executeSql(`
      CREATE TABLE IF NOT EXISTS users (
        id TEXT PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT,
        age INTEGER,
        created_at TEXT DEFAULT (datetime('now'))
      )
    `)
    await this.store.executeSql(`
      CREATE TABLE IF NOT EXISTS orders (
        id TEXT PRIMARY KEY,
        user_id TEXT NOT NULL,
        amount REAL NOT NULL,
        status TEXT DEFAULT 'pending',
        FOREIGN KEY (user_id) REFERENCES users(id)
      )
    `)
  }

  // 清空所有表
  async cleanAll(): Promise<void> {
    if (!this.store) return
    await this.store.executeSql('DELETE FROM orders')
    await this.store.executeSql('DELETE FROM users')
  }

  // 获取Store实例
  getStore(): relationalStore.RdbStore {
    if (!this.store) throw new Error('数据库未初始化')
    return this.store
  }

  // 关闭数据库
  async close(): Promise<void> {
    if (this.store) {
      await relationalStore.deleteRdbStore(this.context, 'test_app.db')
      this.store = null
    }
  }
}

CRUD基础测试：

// CrudTest.ets - CRUD操作测试
import { describe, it, beforeAll, afterAll, beforeEach, assertEqual, assertNotNull, assertTrue } from '@ohos/hypium'
import { TestDatabase } from '../TestDatabase'

export default function crudTest() {
  describe('CRUD操作测试', () => {
    let testDb: TestDatabase

    beforeAll(async () => {
      testDb = new TestDatabase(getContext(this))
      await testDb.init()
    })

    afterAll(async () => {
      await testDb.close()
    })

    beforeEach(async () => {
      // 每个用例前清空数据
      await testDb.cleanAll()
    })

    it('insert_插入数据后可查询', 0, async () => {
      const store = testDb.getStore()
      const rowId = await store.insert('users', {
        id: 'u-001',
        name: '张三',
        email: 'zhang@test.com',
        age: 25
      })
      assertEqual(rowId > 0, true, '插入应返回正数行ID')

      // 查询验证
      const predicates = new relationalStore.RdbPredicates('users')
      predicates.equalTo('id', 'u-001')
      const resultSet = await store.query(predicates)
      assertEqual(resultSet.rowCount, 1, '应查询到1条记录')
      resultSet.goToFirstRow()
      assertEqual(resultSet.getString(resultSet.getColumnIndex('name')), '张三')
      resultSet.close()
    })

    it('insert_重复主键插入失败', 0, async () => {
      const store = testDb.getStore()
      await store.insert('users', { id: 'u-001', name: '张三', age: 25 })
      try {
        await store.insert('users', { id: 'u-001', name: '李四', age: 30 })
        assertTrue(false, '重复主键应抛出异常')
      } catch (e) {
        assertNotNull(e, '重复主键应报错')
      }
    })

    it('update_更新指定记录', 0, async () => {
      const store = testDb.getStore()
      await store.insert('users', { id: 'u-001', name: '张三', age: 25 })

      // 更新年龄
      const predicates = new relationalStore.RdbPredicates('users')
      predicates.equalTo('id', 'u-001')
      const changedRows = await store.update({ age: 26 }, predicates)
      assertEqual(changedRows, 1, '应更新1行')

      // 验证更新结果
      const resultSet = await store.query(predicates)
      resultSet.goToFirstRow()
      assertEqual(resultSet.getInt(resultSet.getColumnIndex('age')), 26)
      resultSet.close()
    })

    it('update_条件不匹配时更新0行', 0, async () => {
      const store = testDb.getStore()
      const predicates = new relationalStore.RdbPredicates('users')
      predicates.equalTo('id', 'non-exist')
      const changedRows = await store.update({ age: 99 }, predicates)
      assertEqual(changedRows, 0, '不存在的记录应更新0行')
    })

    it('delete_删除指定记录', 0, async () => {
      const store = testDb.getStore()
      await store.insert('users', { id: 'u-001', name: '张三', age: 25 })

      const predicates = new relationalStore.RdbPredicates('users')
      predicates.equalTo('id', 'u-001')
      const deletedRows = await store.delete(predicates)
      assertEqual(deletedRows, 1, '应删除1行')

      // 验证删除结果
      const resultSet = await store.query(predicates)
      assertEqual(resultSet.rowCount, 0, '删除后应查询不到')
      resultSet.close()
    })

    it('query_多条件查询', 0, async () => {
      const store = testDb.getStore()
      await store.insert('users', { id: 'u-001', name: '张三', age: 25 })
      await store.insert('users', { id: 'u-002', name: '李四', age: 30 })
      await store.insert('users', { id: 'u-003', name: '王五', age: 25 })

      // 查询年龄25的用户
      const predicates = new relationalStore.RdbPredicates('users')
      predicates.equalTo('age', 25)
      const resultSet = await store.query(predicates)
      assertEqual(resultSet.rowCount, 2, '年龄25的用户应有2个')
      resultSet.close()
    })
  })
}

3.2 进阶用法：事务测试

// TransactionTest.ets - 事务测试
import { describe, it, beforeAll, afterAll, beforeEach, assertEqual, assertTrue } from '@ohos/hypium'
import { TestDatabase } from '../TestDatabase'

export default function transactionTest() {
  describe('事务测试', () => {
    let testDb: TestDatabase

    beforeAll(async () => {
      testDb = new TestDatabase(getContext(this))
      await testDb.init()
    })

    afterAll(async () => {
      await testDb.close()
    })

    beforeEach(async () => {
      await testDb.cleanAll()
    })

    it('事务提交_所有操作生效', 0, async () => {
      const store = testDb.getStore()

      await store.beginTransaction()
      try {
        await store.insert('users', { id: 'u-001', name: '张三', age: 25 })
        await store.insert('orders', { id: 'o-001', user_id: 'u-001', amount: 99.9, status: 'paid' })
        await store.commit()
      } catch (e) {
        await store.rollBack()
      }

      // 验证两个表的数据都已提交
      const userPred = new relationalStore.RdbPredicates('users')
      const userResult = await store.query(userPred)
      assertEqual(userResult.rowCount, 1, '用户数据应已提交')
      userResult.close()

      const orderPred = new relationalStore.RdbPredicates('orders')
      const orderResult = await store.query(orderPred)
      assertEqual(orderResult.rowCount, 1, '订单数据应已提交')
      orderResult.close()
    })

    it('事务回滚_所有操作不生效', 0, async () => {
      const store = testDb.getStore()

      await store.beginTransaction()
      try {
        await store.insert('users', { id: 'u-001', name: '张三', age: 25 })
        await store.insert('orders', { id: 'o-001', user_id: 'u-001', amount: 99.9, status: 'paid' })
        // 模拟异常，触发回滚
        throw new Error('模拟业务异常')
      } catch (e) {
        await store.rollBack()
      }

      // 验证两个表的数据都未提交
      const userPred = new relationalStore.RdbPredicates('users')
      const userResult = await store.query(userPred)
      assertEqual(userResult.rowCount, 0, '回滚后用户数据不应存在')
      userResult.close()

      const orderPred = new relationalStore.RdbPredicates('orders')
      const orderResult = await store.query(orderPred)
      assertEqual(orderResult.rowCount, 0, '回滚后订单数据不应存在')
      orderResult.close()
    })

    it('事务部分失败_关联数据一致性', 0, async () => {
      const store = testDb.getStore()

      // 场景：转账，A扣钱和B加钱必须在同一个事务中
      await store.insert('users', { id: 'u-A', name: '用户A', age: 25 })
      await store.insert('users', { id: 'u-B', name: '用户B', age: 30 })

      await store.beginTransaction()
      try {
        // A扣100
        const predA = new relationalStore.RdbPredicates('users')
        predA.equalTo('id', 'u-A')
        await store.update({ age: 24 }, predA)  // 用age模拟余额

        // 模拟B加钱失败
        throw new Error('B加钱失败')

        // await store.commit()  // 不会执行
      } catch (e) {
        await store.rollBack()
      }

      // 验证A的余额没有变（事务回滚保护了数据一致性）
      const predA = new relationalStore.RdbPredicates('users')
      predA.equalTo('id', 'u-A')
      const resultA = await store.query(predA)
      resultA.goToFirstRow()
      assertEqual(resultA.getInt(resultA.getColumnIndex('age')), 25, '回滚后A的余额不应变化')
      resultA.close()
    })

    it('事务回滚隔离_用事务回滚实现测试数据隔离', 0, async () => {
      const store = testDb.getStore()

      // 用事务包裹整个测试
      await store.beginTransaction()
      try {
        await store.insert('users', { id: 'u-001', name: '张三', age: 25 })
        // ... 执行各种测试断言 ...
        const pred = new relationalStore.RdbPredicates('users')
        const result = await store.query(pred)
        assertEqual(result.rowCount, 1)
        result.close()
      } finally {
        // 无论测试成功失败，都回滚——数据不会留在数据库中
        await store.rollBack()
      }

      // 验证数据已被回滚
      const pred = new relationalStore.RdbPredicates('users')
      const result = await store.query(pred)
      assertEqual(result.rowCount, 0, '事务回滚后数据不应存在')
      result.close()
    })
  })
}

3.3 完整示例：数据库迁移测试

// MigrationManager.ets - 数据库迁移管理器
import relationalStore from '@ohos.data.relationalStore'

export class MigrationManager {
  private store: relationalStore.RdbStore
  private migrations: Map<number, Array<string>> = new Map()

  constructor(store: relationalStore.RdbStore) {
    this.store = store
  }

  // 注册迁移脚本
  registerMigration(version: number, sqls: Array<string>): void {
    this.migrations.set(version, sqls)
  }

  // 获取当前数据库版本
  async getCurrentVersion(): Promise<number> {
    try {
      const resultSet = await this.store.querySql(
        'SELECT version FROM schema_version ORDER BY version DESC LIMIT 1'
      )
      if (resultSet.rowCount > 0) {
        resultSet.goToFirstRow()
        const version = resultSet.getInt(0)
        resultSet.close()
        return version
      }
      resultSet.close()
    } catch (e) {
      // schema_version表不存在，说明是初始版本
    }
    return 0
  }

  // 执行迁移
  async migrate(targetVersion: number): Promise<void> {
    const currentVersion = await this.getCurrentVersion()

    if (currentVersion >= targetVersion) {
      return  // 无需迁移
    }

    // 创建版本记录表（如果不存在）
    await this.store.executeSql(`
      CREATE TABLE IF NOT EXISTS schema_version (
        version INTEGER PRIMARY KEY,
        applied_at TEXT DEFAULT (datetime('now'))
      )
    `)

    // 按版本顺序执行迁移
    for (let v = currentVersion + 1; v <= targetVersion; v++) {
      const sqls = this.migrations.get(v)
      if (!sqls) {
        throw new Error(`未找到版本${v}的迁移脚本`)
      }

      await this.store.beginTransaction()
      try {
        for (const sql of sqls) {
          await this.store.executeSql(sql)
        }
        await this.store.executeSql(
          'INSERT INTO schema_version (version) VALUES (?)',
          [v]
        )
        await store.commit()
      } catch (e) {
        await this.store.rollBack()
        throw new Error(`版本${v}迁移失败: ${(e as Error).message}`)
      }
    }
  }
}

// ==================== 迁移测试 ====================
import { describe, it, beforeAll, afterAll, assertEqual, assertTrue, assertNotNull } from '@ohos/hypium'
import { TestDatabase } from '../TestDatabase'
import { MigrationManager } from '../MigrationManager'

export default function migrationTest() {
  describe('数据库迁移测试', () => {
    let testDb: TestDatabase
    let migrationManager: MigrationManager

    beforeAll(async () => {
      testDb = new TestDatabase(getContext(this))
      await testDb.init()
      const store = testDb.getStore()
      migrationManager = new MigrationManager(store)

      // 注册V1迁移：创建users表
      migrationManager.registerMigration(1, [
        `CREATE TABLE IF NOT EXISTS users (
          id TEXT PRIMARY KEY,
          name TEXT NOT NULL,
          email TEXT
        )`
      ])

      // 注册V2迁移：给users表添加age字段
      migrationManager.registerMigration(2, [
        `ALTER TABLE users ADD COLUMN age INTEGER DEFAULT 0`
      ])

      // 注册V3迁移：创建orders表，添加外键
      migrationManager.registerMigration(3, [
        `CREATE TABLE IF NOT EXISTS orders (
          id TEXT PRIMARY KEY,
          user_id TEXT NOT NULL,
          amount REAL NOT NULL,
          status TEXT DEFAULT 'pending',
          FOREIGN KEY (user_id) REFERENCES users(id)
        )`,
        `CREATE INDEX IF NOT EXISTS idx_orders_user_id ON orders(user_id)`
      ])
    })

    afterAll(async () => {
      await testDb.close()
    })

    it('V1迁移_创建users表', 0, async () => {
      await migrationManager.migrate(1)
      const version = await migrationManager.getCurrentVersion()
      assertEqual(version, 1, '迁移后版本应为1')

      // 验证表可以正常操作
      const store = testDb.getStore()
      await store.insert('users', { id: 'u-001', name: '张三', email: 'z@test.com' })
      const pred = new relationalStore.RdbPredicates('users')
      const result = await store.query(pred)
      assertEqual(result.rowCount, 1)
      result.close()
    })

    it('V2迁移_添加age字段', 0, async () => {
      await migrationManager.migrate(2)
      const version = await migrationManager.getCurrentVersion()
      assertEqual(version, 2, '迁移后版本应为2')

      // 验证新字段可以正常使用
      const store = testDb.getStore()
      await store.insert('users', { id: 'u-002', name: '李四', email: 'l@test.com', age: 30 })
      const pred = new relationalStore.RdbPredicates('users')
      pred.equalTo('id', 'u-002')
      const result = await store.query(pred)
      result.goToFirstRow()
      assertEqual(result.getInt(result.getColumnIndex('age')), 30, 'age字段应可正常读写')
      result.close()
    })

    it('V3迁移_创建orders表', 0, async () => {
      await migrationManager.migrate(3)
      const version = await migrationManager.getCurrentVersion()
      assertEqual(version, 3, '迁移后版本应为3')

      // 验证orders表和外键约束
      const store = testDb.getStore()
      await store.insert('orders', { id: 'o-001', user_id: 'u-001', amount: 99.9, status: 'paid' })
      const pred = new relationalStore.RdbPredicates('orders')
      const result = await store.query(pred)
      assertEqual(result.rowCount, 1)
      result.close()
    })

    it('迁移幂等性_重复执行不出错', 0, async () => {
      // 已经迁移到V3了，再次迁移到V3应该无操作
      await migrationManager.migrate(3)
      const version = await migrationManager.getCurrentVersion()
      assertEqual(version, 3, '重复迁移版本不变')
    })

    it('数据兼容性_旧数据在新Schema下可正常访问', 0, async () => {
      // V1时插入的数据（没有age字段），在V2后应该可以正常访问
      const store = testDb.getStore()
      const pred = new relationalStore.RdbPredicates('users')
      pred.equalTo('id', 'u-001')
      const result = await store.query(pred)
      assertTrue(result.rowCount > 0, 'V1的旧数据应可正常访问')
      result.goToFirstRow()
      // age字段应有默认值
      const age = result.getInt(result.getColumnIndex('age'))
      assertEqual(age, 0, '旧数据的age应为默认值0')
      result.close()
    })
  })
}

四、踩坑与注意事项

坑点1：ResultSet不关闭导致内存泄漏

ResultSet用完必须close()。如果你在测试中查询了100次但没关闭ResultSet，内存会持续增长。更糟糕的是，未关闭的ResultSet会持有数据库锁，导致后续操作阻塞。

建议：用try-finally确保关闭，或者封装一个自动关闭的查询工具方法。

坑点2：异步操作中的事务嵌套

HarmonyOS的RDB事务不支持嵌套。如果你在一个事务中又调用了beginTransaction()，行为是未定义的——可能直接报错，也可能内层commit把外层的事务也提交了。

建议：事务管理集中在一个层级，不要在底层DAO方法中开事务，由上层Service统一管理事务边界。

坑点3：测试数据库和生产数据库不同步

你在测试数据库里建了3个表，生产数据库有5个表——测试全绿，但生产环境查第4个表就崩了。

解决方案：测试数据库的建表逻辑和生产环境完全一致——使用同一套迁移脚本。不要在测试中手写建表SQL，而是调用MigrationManager.migrate(latestVersion)。

坑点4：外键约束在测试中导致级联删除

你清空users表，但orders表还有关联数据——外键约束阻止删除，测试直接报错。

建议：清空数据时按依赖顺序删除——先删orders，再删users。或者在建表时不加外键约束（仅限测试环境），用应用层逻辑保证一致性。

坑点5：SQLite的ALTER TABLE限制

SQLite的ALTER TABLE只支持添加列和重命名表，不支持删除列、修改列类型。如果你的迁移脚本包含ALTER TABLE users DROP COLUMN age，直接报错。

解决方案：删除列需要重建表——创建新表、复制数据、删除旧表、重命名新表。这个过程在迁移脚本中要写完整，而且要测试。

坑点6：数据库文件锁导致并发测试失败

多个测试用例同时操作同一个数据库文件，SQLite的文件锁会导致"database is locked"错误。

建议：数据库测试串行执行，不要并发。如果必须并发，每个用例使用独立的数据库文件。

坑点7：日期时间类型处理不一致

SQLite没有原生的日期时间类型，日期存的是TEXT。datetime('now')返回的是UTC时间，不是本地时间。如果你在测试中断言"创建时间等于今天"，可能因为时区差异而失败。

建议：测试中不要断言精确的时间值，而是验证时间格式或时间范围。例如：assertTrue(createdAt.startsWith('2024-')) 或 assertTrue(new Date(createdAt).getTime() > Date.now() - 60000)。

五、HarmonyOS 6适配说明

API差异表

行为变更

1. Savepoint支持：HarmonyOS 6的事务支持保存点（Savepoint），可以在事务中设置回滚点，部分回滚而不是全部回滚。这对嵌套操作场景非常有用。

2. 官方迁移框架：@ohos.data.migration提供了标准化的迁移管理，支持版本记录、自动迁移、回滚等功能，不再需要自己实现MigrationManager。

3. 批量操作API：新增batchInsert、batchUpdate、batchDelete，一次调用处理多条记录，性能比循环单条操作高一个数量级。

适配代码

// HarmonyOS 6迁移测试
import { describe, it, assertEqual, assertTrue } from '@ohos/hypium'
import { DatabaseMigration, MigrationVersion } from '@ohos.data.migration'

export default function harmonyOS6MigrationTest() {
  describe('HarmonyOS 6迁移测试', () => {
    it('官方迁移框架_自动执行迁移', 0, async () => {
      const migration = new DatabaseMigration(store)

      // 注册迁移版本
      migration.addVersion(new MigrationVersion(1, [
        'CREATE TABLE users (id TEXT PRIMARY KEY, name TEXT)'
      ]))
      migration.addVersion(new MigrationVersion(2, [
        'ALTER TABLE users ADD COLUMN age INTEGER DEFAULT 0'
      ]))

      // 执行迁移
      await migration.migrateToLatest()
      assertEqual(await migration.getCurrentVersion(), 2)
    })

    it('Savepoint_部分回滚', 0, async () => {
      await store.beginTransaction()
      try {
        await store.insert('users', { id: 'u-001', name: '张三' })
        // 设置保存点
        await store.setSavepoint('sp1')

        await store.insert('users', { id: 'u-002', name: '李四' })
        // 回滚到保存点——只撤销u-002的插入
        await store.rollbackTo('sp1')

        await store.commit()
      } catch (e) {
        await store.rollBack()
      }

      // 验证只有u-001存在
      const result = await store.querySql('SELECT * FROM users')
      assertEqual(result.rowCount, 1, '部分回滚后只有1条记录')
      result.close()
    })
  })
}

六、总结

数据库测试的核心是三个字：对、稳、全。CRUD操作要验证对不对，事务要验证稳不稳，迁移要验证全不全。数据隔离是基础——测试用例之间互相污染，比没有测试还可怕。事务回滚是最推荐的隔离策略，又快又干净。迁移测试经常被忽视，但它是数据层最容易出现"悄悄出错"的地方——旧数据在新Schema下能不能正常访问？迁移脚本能不能重复执行？这些问题不测，生产环境就是你的测试场。数据库测试不是锦上添花，是数据安全的底线。