happy-export/backend/test/data-transfer.test.js

/**
 * 业务数据导入导出的回归测试。
 *
 * 全程在临时 DATA_DIR 里跑，不碰 backend/data/trade.db。
 * 覆盖：
 *   ① 系统表绝不进包（users / settings / audit_log …）
 *   ② 导出 → 导入到「另一个空部署」后行数与内容一致，BLOB 完整
 *   ③ 重复导入同一包：全部跳过，不产生重复数据（冲突不覆盖）
 *   ④ 只导订单：客户自动补壳、产品不跟着进来
 *   ⑤ 包被改坏：校验失败并拒绝导入
 */
const test = require('node:test')
const assert = require('node:assert/strict')
const fs = require('node:fs')
const os = require('node:os')
const path = require('node:path')
const zlib = require('node:zlib')

const DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), 'happy-transfer-'))
process.env.DATA_DIR = DATA_DIR

const db = require('../db/db')
const transfer = require('../lib/dataTransfer')
const { readBundle, writeBundleFile, foreignKeys } = require('../lib/dataTransfer/bundle')
const { TABLES } = require('../lib/dataTransfer/catalog')

const workDir = path.join(DATA_DIR, 'work')
fs.mkdirSync(workDir, { recursive: true })

const IMAGE_BYTES = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0xff])

function seed() {
  const now = '2026-01-01T00:00:00.000Z'
  const admin = db.prepare('SELECT * FROM users ORDER BY id LIMIT 1').get()
  db.prepare('INSERT INTO suppliers (name, created_at) VALUES (?, ?)').run('宁波腾达', now)
  db.prepare('INSERT INTO product_groups (name, group_key, created_at) VALUES (?, ?, ?)').run('扭矩扳手', 'torque', now)
  db.prepare(`INSERT INTO products (sku, name, price, currency, supplier_id, group_id, created_at)
              VALUES (?, ?, ?, ?, ?, ?, ?)`).run('TD-100', '扭矩扳手 100N', 88, 'CNY', 1, 1, now)
  db.prepare(`INSERT INTO products (sku, name, price, currency, supplier_id, created_at)
              VALUES (?, ?, ?, ?, ?, ?)`).run('TD-200', '扭矩扳手 200N', 128, 'CNY', 1, now)
  db.prepare('INSERT INTO product_packaging (sku, carton_qty, created_at) VALUES (?, ?, ?)').run('TD-100', 10, now)
  db.prepare('INSERT INTO product_images (sku, filename, created_at) VALUES (?, ?, ?)').run('TD-100', 'a.png', now)
  db.prepare('INSERT INTO product_image_files (image_id, bytes, size, stored_at) VALUES (?, ?, ?, ?)')
    .run(1, IMAGE_BYTES, IMAGE_BYTES.length, now)
  db.prepare('INSERT INTO price_history (sku, old_price, new_price, changed_at) VALUES (?, ?, ?, ?)')
    .run('TD-100', 80, 88, now)
  db.prepare(`INSERT INTO customers (company, country, owner_id, created_by, org_id, pool_status, status, created_at)
              VALUES (?, ?, ?, ?, ?, ?, ?, ?)`).run('ACME GmbH', 'DE', admin.id, admin.id, admin.org_id || 1, 'owned', 'active', now)
  db.prepare('INSERT INTO contacts (customer_id, name, email, created_at) VALUES (?, ?, ?, ?)')
    .run(1, 'Hans', '[email protected]', now)
  db.prepare(`INSERT INTO orders (order_no, customer_id, order_date, total_amount, currency, status, owner_id, created_at)
              VALUES (?, ?, ?, ?, ?, ?, ?, ?)`).run('PO-2026-001', 1, '2026-01-02', 880, 'USD', 'confirmed', admin.id, now)
  db.prepare('INSERT INTO order_items (order_id, sku, product_name, qty, unit_price, total_price, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)')
    .run(1, 'TD-100', '扭矩扳手 100N', 10, 88, 880, now)
}

/** 用「源库的副本 + 清空业务表」模拟另一台全新部署：schema 与线上完全一致，只有系统数据 */
function cloneEmptyDeployment(source, label) {
  const file = path.join(workDir, `${label}.db`)
  source.exec(`VACUUM INTO '${file.replace(/'/g, "''")}'`)
  const Database = require('better-sqlite3-multiple-ciphers')
  const copy = new Database(file)
  copy.pragma('foreign_keys = ON')
  for (const table of ['order_items', 'orders', 'contacts', 'customers', 'price_history', 'product_image_files',
    'product_images', 'product_packaging', 'products', 'product_groups', 'suppliers', 'contracts']) {
    copy.exec(`DELETE FROM "${table}"`)
  }
  return copy
}

test.after(() => {
  try { db.close() } catch { /* 已关闭 */ }
  try { fs.rmSync(DATA_DIR, { recursive: true, force: true }) } catch { /* Windows 偶发占用 */ }
})

test('导出包只含业务表，系统表一律不出现', async () => {
  seed()
  const file = path.join(workDir, 'all.hdata.gz')
  await writeBundleFile(db, file, { categories: ['all'], exporter: db.prepare('SELECT * FROM users LIMIT 1').get() })
  const bundle = await readBundle(file)

  for (const table of ['users', 'sessions', 'settings', 'audit_log', 'organizations', 'permission_groups']) {
    assert.equal(bundle.rows.has(table), false, `${table} 不应出现在数据包里`)
  }
  assert.equal(bundle.rows.get('products').length, 2)
  assert.equal(bundle.rows.get('orders').length, 1)
  assert.deepEqual(bundle.categories.sort(), ['catalog', 'customers', 'orders', 'prospects', 'todos'])
})

test('导入到另一台空部署：行数一致、归属给导入人、BLOB 完整', async () => {
  const file = path.join(workDir, 'all.hdata.gz')
  const target = cloneEmptyDeployment(db, 'target-all')
  const importer = target.prepare('SELECT * FROM users ORDER BY id LIMIT 1').get()

  const result = await transfer.applyImport(target, file, {
    user: importer,
    backupDir: path.join(workDir, 'backups')
  })

  assert.ok(result.backup && fs.existsSync(result.backup), '导入前必须留一份备份')
  assert.equal(result.plan.totals.failed, 0)
  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM products').get().n, 2)
  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM suppliers').get().n, 1)
  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM orders').get().n, 1)
  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM customers').get().n, 1)

  const customer = target.prepare('SELECT * FROM customers WHERE company = ?').get('ACME GmbH')
  assert.equal(customer.owner_id, importer.id, '客户归属应改写为导入人')

  const bytes = target.prepare('SELECT bytes FROM product_image_files').get().bytes
  assert.deepEqual(Buffer.from(bytes), IMAGE_BYTES, 'BLOB 必须逐字节一致')

  /* 系统数据没被覆盖：导入前后用户表还是目标库自己那份 */
  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM users').get().n, 1)

  target.close()
})

test('同一包再导一次：全部跳过，不产生重复', async () => {
  const file = path.join(workDir, 'all.hdata.gz')
  const target = cloneEmptyDeployment(db, 'target-repeat')
  const importer = target.prepare('SELECT * FROM users ORDER BY id LIMIT 1').get()

  const first = await transfer.applyImport(target, file, { user: importer, backupDir: path.join(workDir, 'backups') })
  assert.ok(first.plan.totals.insert > 0)

  const second = await transfer.applyImport(target, file, { user: importer, backupDir: path.join(workDir, 'backups') })
  assert.equal(second.plan.totals.insert, 0, '第二次导入不应新增任何行')
  assert.equal(second.plan.totals.skip, first.plan.totals.rows)
  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM products').get().n, 2)

  target.close()
})

test('只导订单：客户自动补壳，产品数据不跟进来', async () => {
  const file = path.join(workDir, 'orders-only.hdata.gz')
  await writeBundleFile(db, file, { categories: ['orders'] })
  const bundle = await readBundle(file)
  assert.deepEqual([...bundle.rows.keys()].sort(), ['order_items', 'orders'], '只选订单类别时不应带出产品/客户')

  const target = cloneEmptyDeployment(db, 'target-orders')
  const importer = target.prepare('SELECT * FROM users ORDER BY id LIMIT 1').get()
  const result = await transfer.applyImport(target, file, { user: importer, backupDir: path.join(workDir, 'backups') })

  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM orders').get().n, 1)
  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM order_items').get().n, 1)
  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM products').get().n, 0, '未选产品类别就不该有产品')

  const stub = target.prepare('SELECT * FROM customers').get()
  assert.equal(stub.company, 'ACME GmbH', '应按 __refs 快照补出客户公司名')
  assert.equal(stub.owner_id, importer.id)
  assert.deepEqual(result.plan.stubbed.map(item => item.table), ['customers'])

  target.close()
})

test('流水表：包内逐列相同的记录全部保留，重导仍全部跳过', async () => {
  /* 源库确实会这样：一次批量改价会在同一毫秒写两条一模一样的 price_history。
     拿 (sku, changed_at) 当去重键会把其中一条当重复丢掉（实测 643 行丢 186 行），
     所以流水表改成「整行内容比对」，且允许同一个包里出现相同的两行。 */
  const now = '2026-02-02T00:00:00.000Z'
  const insert = db.prepare('INSERT INTO price_history (sku, old_price, new_price, changed_at) VALUES (?, ?, ?, ?)')
  insert.run('TD-200', 100, 110, now)
  insert.run('TD-200', 100, 110, now)

  const file = path.join(workDir, 'history.hdata.gz')
  await writeBundleFile(db, file, { categories: ['catalog'] })
  const bundle = await readBundle(file)
  assert.equal(bundle.rows.get('price_history').length, 3, '源库 1 条 + 新增 2 条都应进包')

  const target = cloneEmptyDeployment(db, 'target-history')
  const importer = target.prepare('SELECT * FROM users ORDER BY id LIMIT 1').get()
  const first = await transfer.applyImport(target, file, { user: importer, backupDir: path.join(workDir, 'backups') })
  assert.equal(first.plan.totals.failed, 0)
  assert.equal(target.prepare('SELECT COUNT(*) AS n FROM price_history').get().n, 3, '逐列相同的历史记录也要全部带过去')

  const second = await transfer.applyImport(target, file, { user: importer, backupDir: path.join(workDir, 'backups') })
  assert.equal(second.plan.totals.insert, 0, '第二次导入不应新增')

  target.close()
})

test('目录顺序满足外键依赖：父表一定排在子表前面', () => {
  /* 导入是逐表逐行插入的，顺序错了就会撞外键。这条测试保证以后新增业务表时
     有人把表插到错误位置会立刻红。 */
  const known = new Set(TABLES.map(item => item.table))
  const seen = new Set()
  for (const spec of TABLES) {
    for (const ref of foreignKeys(db, spec.table)) {
      if (ref.table === spec.table) continue
      if (!known.has(ref.table)) continue // 指向 users / organizations 等系统表的列不参与
      assert.ok(seen.has(ref.table), `${spec.table}.${ref.from} 引用的 ${ref.table} 必须排在它前面`)
    }
    seen.add(spec.table)
  }
})

test('包被改坏：sha256 校验失败并拒绝导入', async () => {
  const file = path.join(workDir, 'all.hdata.gz')
  const lines = zlib.gunzipSync(fs.readFileSync(file)).toString('utf8').split('\n').filter(Boolean)
  const productLine = lines.findIndex(line => line.includes('"t":"products"') && line.includes('TD-100'))
  assert.ok(productLine > 0)
  lines[productLine] = lines[productLine].replace('TD-100', 'TD-999')
  const broken = path.join(workDir, 'broken.hdata')
  fs.writeFileSync(broken, lines.join('\n') + '\n')

  await assert.rejects(() => readBundle(broken), /校验失败/)
})