Files
cms-website/tests/integration/import-job.test.ts
T

565 lines
25 KiB
TypeScript
Raw Normal View History

2026-10-21 15:30:42 +08:00
// Plan 3 Task 1 失败测试(先于实现):import 命令骨架 + import_jobs 生命周期。
// 覆盖契约(对照计划 Task 1 与前置事实 6/12/13):
// 1. register(wxr + media + media-map 三快照)→ createImportJob:job 行 created(queued)→
// running→succeeded 全程可观察(管道钩子内读 running 中间态)、mapping_version 落
// MAPPING_VERSION、报告记录全四版本块 + 三快照绑定(wxr 主输入 / media_bundle /
// manual_url_list 按 (connection,kind) 解析最新记入报告,job 行不扩列——前置事实 13);
// 2. 无 --snapshot:按 (connection, kind='wxr') 选最新快照;
// 3. idempotency_key 唯一:重复提交同默认 key(site+snapshot+版本)→ 复用既有 job 不重跑
// (reused 标记 + 管道零调用 + 恰一行);显式 key 覆盖默认;并发同 key 收敛同 job;
// 4. 无 wxr 快照(站点仅有非 wxr 输入 / 显式 id 指向非 wxr 快照)→ NO_WXR_SNAPSHOT;
// 显式 id 不存在或属他站 → SNAPSHOT_NOT_FOUND;站点不存在 → SITE_NOT_FOUND;均零 job 行;
// 5. failed 路径:管道抛错 → running→failed(error_code/error_summary/finished_at,
// '[pipeline]' 前缀),重提交同 key 复用失败 job 不重跑;管道抛 ImportError 保留原 code;
2026-10-21 20:08:06 +08:00
// 6. Outcome 契约:ctx.report 累加 ⊕ 返回值(counts 同名键**求和**——Task 6 编排聚合)→ succeeded_with_issues;
// 7. 复用口径忠实化:reuse 返回只含 job 行锚定字段(快照/版本字段缺席);
// 8. 23505 收敛确定性探针:未提交孪生 job 持有幂等键(真实 PG 锁语义,零代理)。
2026-10-21 15:30:42 +08:00
// 运行于 globalSetup 临时库(ADR-015);storage/报告根用 os.tmpdir 临时目录,finally 清理。
import { existsSync } from 'node:fs'
import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join, resolve } from 'node:path'
import { fileURLToPath } from 'node:url'
import { describe, expect, it } from 'vitest'
import { ADAPTER_VERSION, MAPPING_VERSION, PARSER_VERSION, SANITIZER_VERSION } from '@sf/adapter-wp'
import {
ImportError,
createImportJob,
type CreateImportJobInput,
} from '../../apps/api/src/modules/import/import.service.js'
import { registerSource } from '../../apps/api/src/modules/source/register.service.js'
import { rand, withImmutableCleanup, withTestDb, type Db } from './helpers.js'
const FIXTURES_ROOT = fileURLToPath(new URL('../fixtures', import.meta.url))
const WXR_FIXTURE = resolve(FIXTURES_ROOT, 'wxr', 'mini.xml')
interface Ctx {
work: string
storageRoot: string
registerReportRoot: string
importReportRoot: string
mediaDir: string
mediaMapFile: string
allowedRoots: string[]
}
/** 独立媒体树 + 侧车 JSON(形态同 source-register.test.ts) */
async function makeCtx(): Promise<Ctx> {
const work = await mkdtemp(join(tmpdir(), 'sf-import-'))
const mediaDir = join(work, 'media')
await mkdir(join(mediaDir, 'sub'), { recursive: true })
await writeFile(join(mediaDir, 'a.png'), 'sf-import-media-a')
await writeFile(join(mediaDir, 'sub', 'b.png'), 'sf-import-media-b')
const mediaMapFile = join(work, 'media-map.json')
await writeFile(
mediaMapFile,
`${JSON.stringify([
{ sourceUrl: 'https://fixture.example/wp-content/uploads/2026/10/fixture-a.png', filePath: 'a.png' },
])}\n`,
'utf8',
)
return {
work,
storageRoot: join(work, 'storage'),
registerReportRoot: join(work, 'artifacts', 'register'),
importReportRoot: join(work, 'artifacts', 'import'),
mediaDir,
mediaMapFile,
allowedRoots: [FIXTURES_ROOT, work],
}
}
/** 服务写入的 job 行清场:import_jobs 可变直删;快照不可变走触发器开关;FK 反序 */
async function cleanupSiteRows(db: Db, siteId: string): Promise<void> {
await db.execute(`delete from import_jobs where site_id = '${siteId}'`)
await withImmutableCleanup(db, 'source_snapshots', 'source_snapshots_immutable', `site_id = '${siteId}'`)
await db.execute(`delete from source_connections where site_id = '${siteId}'`)
await db.execute(`delete from sites where id = '${siteId}'`)
}
/** 登记三快照(wxr + media_bundle + manual_url_list)的便捷封装 */
async function registerAll(
db: Db,
ctx: Ctx,
siteKey: string,
): Promise<Awaited<ReturnType<typeof registerSource>>> {
return registerSource(db, {
siteKey,
wxrPath: WXR_FIXTURE,
mediaDir: ctx.mediaDir,
mediaMapPath: ctx.mediaMapFile,
allowedRoots: ctx.allowedRoots,
storageRoot: ctx.storageRoot,
reportRoot: ctx.registerReportRoot,
})
}
describe('import_jobs 生命周期(Task 1 骨架)', () => {
it('三快照 register → createImportJob:queued→running→succeeded、mapping_version、报告含版本块与三快照绑定', async () => {
await withTestDb(async (db) => {
const ctx = await makeCtx()
try {
const key = `sfimp-${rand()}`
const reg = await registerAll(db, ctx, key)
try {
const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
const media = reg.snapshots.find((s) => s.kind === 'media_bundle')!
const map = reg.snapshots.find((s) => s.kind === 'manual_url_list')!
// 管道钩子内观察中间态:此刻 job 已 created(queued)→running(attempt 记首次执行)
let midStatus: string | undefined
let midAttempt: string | number | undefined
let midStartedAt: Date | null | undefined
let midFinishedAt: Date | null | undefined
const result = await createImportJob(db, {
siteKey: key,
snapshotId: wxr.id,
reportRoot: ctx.importReportRoot,
pipeline: async (jobCtx) => {
expect(jobCtx.jobId).toBeTruthy()
const rows = (await db.execute(
`select status, attempt, started_at, finished_at from import_jobs where id = '${jobCtx.jobId}'`,
)) as any
midStatus = rows[0].status
midAttempt = rows[0].attempt
midStartedAt = rows[0].started_at
midFinishedAt = rows[0].finished_at
return { counts: {}, issues: [] }
2026-10-21 15:30:42 +08:00
},
})
expect(result.reused).toBe(false)
expect(result.status).toBe('succeeded')
expect(midStatus).toBe('running')
expect(Number(midAttempt)).toBe(1)
expect(midStartedAt).not.toBeNull()
expect(midFinishedAt).toBeNull()
// 终态行:身份/版本/时间戳/report_path
const job = (await db.execute(
`select status, mapping_version, source_snapshot_id, started_at, finished_at, report_path, attempt, error_code
from import_jobs where id = '${result.jobId}'`,
)) as any
expect(job[0].status).toBe('succeeded')
expect(job[0].mapping_version).toBe(MAPPING_VERSION)
expect(job[0].source_snapshot_id).toBe(wxr.id)
expect(job[0].started_at).not.toBeNull()
expect(job[0].finished_at).not.toBeNull()
expect(job[0].error_code).toBeNull()
expect(Number(job[0].attempt)).toBe(1)
// 报告:全四版本块 + 三快照绑定(前置事实 13:侧快照只入报告,job 行不扩列)
const reportPath = join(ctx.importReportRoot, result.jobId, 'reports', 'import.json')
expect(job[0].report_path).toBe(reportPath)
expect(existsSync(reportPath)).toBe(true)
const report = JSON.parse(await readFile(reportPath, 'utf8')) as {
jobId: string
status: string
idempotencyKey: string
attempt: number
counts: Record<string, number>
issues: string[]
2026-10-21 15:30:42 +08:00
versions: Record<string, string>
snapshots: {
wxr: { id: string }
media_bundle: { id: string } | null
manual_url_list: { id: string } | null
}
}
expect(report.jobId).toBe(result.jobId)
expect(report.status).toBe('succeeded')
// 报告携带幂等键与 attempt(审计锚点——Fix 7)
expect(report.idempotencyKey).toBe(
`import:${key}:${wxr.id}:${ADAPTER_VERSION}/${MAPPING_VERSION}/${SANITIZER_VERSION}`,
)
expect(report.attempt).toBe(1)
expect(report.counts).toStrictEqual({})
expect(report.issues).toStrictEqual([])
2026-10-21 15:30:42 +08:00
expect(report.versions).toStrictEqual({
adapter: ADAPTER_VERSION,
mapping: MAPPING_VERSION,
sanitizer: SANITIZER_VERSION,
parser: PARSER_VERSION,
})
expect(report.snapshots.wxr.id).toBe(wxr.id)
expect(report.snapshots.media_bundle?.id).toBe(media.id)
expect(report.snapshots.manual_url_list?.id).toBe(map.id)
} finally {
await cleanupSiteRows(db, reg.siteId)
}
} finally {
await rm(ctx.work, { recursive: true, force: true })
}
})
})
it('无 snapshotId:按 (connection, kind=wxr) 选最新;同默认 key 重提交复用不重跑', async () => {
await withTestDb(async (db) => {
const ctx = await makeCtx()
try {
const key = `sfimp-${rand()}`
const r1 = await registerSource(db, {
siteKey: key,
wxrPath: WXR_FIXTURE,
allowedRoots: ctx.allowedRoots,
storageRoot: ctx.storageRoot,
reportRoot: ctx.registerReportRoot,
})
try {
// 第二个 wxr 快照:内容加尾随换行(sha 不同、仍合法 XML);稍歇保证 capturedAt 严格更晚
const v2 = join(ctx.work, 'mini-v2.xml')
await writeFile(v2, `${await readFile(WXR_FIXTURE, 'utf8')}\n`, 'utf8')
await new Promise((r) => setTimeout(r, 25))
const r2 = await registerSource(db, {
siteKey: key,
wxrPath: v2,
allowedRoots: ctx.allowedRoots,
storageRoot: ctx.storageRoot,
reportRoot: ctx.registerReportRoot,
})
const wxr1 = r1.snapshots.find((s) => s.kind === 'wxr')!
const wxr2 = r2.snapshots.find((s) => s.kind === 'wxr')!
expect(wxr2.id).not.toBe(wxr1.id)
let pipelineCalls = 0
const input: CreateImportJobInput = {
siteKey: key,
reportRoot: ctx.importReportRoot,
pipeline: async () => {
pipelineCalls++
return { counts: {}, issues: [] }
2026-10-21 15:30:42 +08:00
},
}
const a = await createImportJob(db, input)
expect(a.reused).toBe(false)
expect(a.status).toBe('succeeded')
// 最新快照 = 第二个(v2)
const job = (await db.execute(
`select source_snapshot_id, report_path from import_jobs where id = '${a.jobId}'`,
2026-10-21 15:30:42 +08:00
)) as any
expect(job[0].source_snapshot_id).toBe(wxr2.id)
// 同默认 key 重提交:复用既有 job、管道零追加调用
const b = await createImportJob(db, input)
expect(b.reused).toBe(true)
expect(b.jobId).toBe(a.jobId)
expect(b.status).toBe('succeeded')
// 复用口径忠实化(Fix 1):只返回 job 行锚定字段——快照/版本解析字段缺席
expect('wxrSnapshot' in b).toBe(false)
expect('mediaBundleSnapshot' in b).toBe(false)
expect('versions' in b).toBe(false)
expect('connectionId' in b).toBe(false)
expect(b.mappingVersion).toBe(MAPPING_VERSION)
expect(b.reportPath).toBe(job[0].report_path)
2026-10-21 15:30:42 +08:00
expect(pipelineCalls).toBe(1)
const count = (await db.execute(
`select count(*)::int as n from import_jobs where site_id = '${r1.siteId}'`,
)) as any
expect(count[0].n).toBe(1)
// 显式 idempotencyKey 覆盖默认 key:同 key 复用、异 key(回默认)不新建重复
const c = await createImportJob(db, { ...input, idempotencyKey: 'manual-key-1' })
expect(c.reused).toBe(false)
const d = await createImportJob(db, { ...input, idempotencyKey: 'manual-key-1' })
expect(d.reused).toBe(true)
expect(d.jobId).toBe(c.jobId)
expect(pipelineCalls).toBe(2)
} finally {
await cleanupSiteRows(db, r1.siteId)
}
} finally {
await rm(ctx.work, { recursive: true, force: true })
}
})
})
it('并发同默认 key:恰一 job 行、双方同 jobId、管道恰执行一次(唯一约束收敛)', async () => {
await withTestDb(async (db) => {
const ctx = await makeCtx()
try {
const key = `sfimp-${rand()}`
const reg = await registerAll(db, ctx, key)
try {
const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
let pipelineCalls = 0
const input: CreateImportJobInput = {
siteKey: key,
snapshotId: wxr.id,
reportRoot: ctx.importReportRoot,
pipeline: async () => {
pipelineCalls++
return { counts: {}, issues: [] }
2026-10-21 15:30:42 +08:00
},
}
const [a, b] = await Promise.all([createImportJob(db, input), createImportJob(db, input)])
expect(a.jobId).toBe(b.jobId)
expect(Number(a.reused) + Number(b.reused)).toBe(1)
expect(pipelineCalls).toBe(1)
const count = (await db.execute(
`select count(*)::int as n from import_jobs where site_id = '${reg.siteId}'`,
)) as any
expect(count[0].n).toBe(1)
} finally {
await cleanupSiteRows(db, reg.siteId)
}
} finally {
await rm(ctx.work, { recursive: true, force: true })
}
})
})
it('可预期失败均零 job 行:SITE_NOT_FOUND / SNAPSHOT_NOT_FOUND(不存在或属他站)/ NO_WXR_SNAPSHOT(无非 wxr 主输入或显式指向非 wxr)', async () => {
await withTestDb(async (db) => {
const ctx = await makeCtx()
try {
// 站点 A:仅 media-map 输入(有 connection、无非 wxr 主输入)
const keyA = `sfimp-${rand()}`
const regA = await registerSource(db, {
siteKey: keyA,
mediaMapPath: ctx.mediaMapFile,
allowedRoots: ctx.allowedRoots,
storageRoot: ctx.storageRoot,
reportRoot: ctx.registerReportRoot,
})
// 站点 B:三快照齐全
const keyB = `sfimp-${rand()}`
const regB = await registerAll(db, ctx, keyB)
try {
const mapSnapA = regA.snapshots[0]!
const mediaSnapB = regB.snapshots.find((s) => s.kind === 'media_bundle')!
const cases: { label: string; input: CreateImportJobInput; code: string }[] = [
{
label: '站点不存在',
input: { siteKey: `sfimp-no-such-${rand()}` },
code: 'SITE_NOT_FOUND',
},
{
label: '仅有 manual_url_list 输入的站点(无任何 wxr 快照)',
input: { siteKey: keyA, reportRoot: ctx.importReportRoot },
code: 'NO_WXR_SNAPSHOT',
},
{
label: '显式 snapshotId 指向非 wxr 快照(media_bundle)',
input: { siteKey: keyB, snapshotId: mediaSnapB.id, reportRoot: ctx.importReportRoot },
code: 'NO_WXR_SNAPSHOT',
},
{
label: '显式 snapshotId 不存在',
input: {
siteKey: keyB,
snapshotId: '00000000-0000-4000-8000-000000000000',
reportRoot: ctx.importReportRoot,
},
code: 'SNAPSHOT_NOT_FOUND',
},
{
label: '显式 snapshotId 属于他站',
input: { siteKey: keyB, snapshotId: mapSnapA.id, reportRoot: ctx.importReportRoot },
code: 'SNAPSHOT_NOT_FOUND',
},
]
for (const c of cases) {
const err: ImportError = await createImportJob(db, c.input).then(
() => {
throw new Error(`用例「${c.label}」不应成功`)
},
(e: ImportError) => e,
)
expect(err, c.label).toBeInstanceOf(ImportError)
expect(err.code, c.label).toBe(c.code)
expect(err.message, c.label).not.toBe('')
}
// 零 job 行(两站点均无)
const count = (await db.execute(
`select count(*)::int as n from import_jobs where site_id in ('${regA.siteId}', '${regB.siteId}')`,
)) as any
expect(count[0].n).toBe(0)
} finally {
await cleanupSiteRows(db, regB.siteId)
await cleanupSiteRows(db, regA.siteId)
}
} finally {
await rm(ctx.work, { recursive: true, force: true })
}
})
})
it('failed 路径:管道抛错 → running→failed(error_code/error_summary/finished_at);重提交同 key 复用失败 job 不重跑', async () => {
await withTestDb(async (db) => {
const ctx = await makeCtx()
try {
const key = `sfimp-${rand()}`
const reg = await registerAll(db, ctx, key)
try {
const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
let pipelineCalls = 0
const input: CreateImportJobInput = {
siteKey: key,
snapshotId: wxr.id,
reportRoot: ctx.importReportRoot,
pipeline: async () => {
pipelineCalls++
throw new Error('模拟管道失败:断链 attachment')
},
}
const a = await createImportJob(db, input)
expect(a.reused).toBe(false)
expect(a.status).toBe('failed')
const job = (await db.execute(
`select status, error_code, error_summary, finished_at, started_at from import_jobs where id = '${a.jobId}'`,
)) as any
expect(job[0].status).toBe('failed')
expect(job[0].error_code).toBe('PIPELINE_FAILED')
// '[pipeline]' 前缀:管道阶段错误与解析阶段错误在 error_summary 上可区分(Fix 5)
expect(String(job[0].error_summary)).toMatch(/^\[pipeline\] Error: 模拟管道失败/)
2026-10-21 15:30:42 +08:00
expect(job[0].started_at).not.toBeNull()
expect(job[0].finished_at).not.toBeNull()
// 同 key 重提交:复用失败 job(不重跑、不改行)
const b = await createImportJob(db, input)
expect(b.reused).toBe(true)
expect(b.jobId).toBe(a.jobId)
expect(b.status).toBe('failed')
expect(pipelineCalls).toBe(1)
const count = (await db.execute(
`select count(*)::int as n from import_jobs where site_id = '${reg.siteId}'`,
)) as any
expect(count[0].n).toBe(1)
} finally {
await cleanupSiteRows(db, reg.siteId)
}
} finally {
await rm(ctx.work, { recursive: true, force: true })
}
})
})
2026-10-21 20:08:06 +08:00
it('Outcome 契约:ctx.report 累加 ⊕ 返回值按**键求和**(Task 6 编排聚合),issues 非空 → succeeded_with_issues,计数与问题入报告', async () => {
await withTestDb(async (db) => {
const ctx = await makeCtx()
try {
const key = `sfimp-${rand()}`
const reg = await registerAll(db, ctx, key)
try {
const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
const result = await createImportJob(db, {
siteKey: key,
snapshotId: wxr.id,
reportRoot: ctx.importReportRoot,
pipeline: async (jobCtx) => {
// 写入器式累加(Task 6 写入器的使用形态)
jobCtx.report.counts['terms'] = 2
2026-10-21 20:08:06 +08:00
jobCtx.report.counts['contentEntries'] = 100 // 与返回值同名键**求和**(Task 6 钉死:防覆盖安全网)
jobCtx.report.issues.push('自建缺失 term:1(category/uncategorized)')
return {
counts: { contentEntries: 3, routes: 8 },
issues: ['悬空 attachment 引用:2 条(mini 203/204)'],
}
},
})
expect(result.reused).toBe(false)
expect(result.status).toBe('succeeded_with_issues')
const job = (await db.execute(
`select status, error_code from import_jobs where id = '${result.jobId}'`,
)) as any
expect(job[0].status).toBe('succeeded_with_issues')
expect(job[0].error_code).toBeNull()
const report = JSON.parse(
await readFile(join(ctx.importReportRoot, result.jobId, 'reports', 'import.json'), 'utf8'),
) as { status: string; counts: Record<string, number>; issues: string[] }
expect(report.status).toBe('succeeded_with_issues')
2026-10-21 20:08:06 +08:00
// 汇合规则:counts 按键**求和**(Task 6 钉死——CLOBBER 废弃)、issues 累加器在前返回值在后
expect(report.counts).toStrictEqual({ terms: 2, contentEntries: 103, routes: 8 })
expect(report.issues).toStrictEqual([
'自建缺失 term:1(category/uncategorized)',
'悬空 attachment 引用:2 条(mini 203/204)',
])
} finally {
await cleanupSiteRows(db, reg.siteId)
}
} finally {
await rm(ctx.work, { recursive: true, force: true })
}
})
})
it('管道抛 ImportError:error_code 保留原 code、error_summary 带 [pipeline] 前缀(Fix 5)', async () => {
await withTestDb(async (db) => {
const ctx = await makeCtx()
try {
const key = `sfimp-${rand()}`
const reg = await registerAll(db, ctx, key)
try {
const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
const result = await createImportJob(db, {
siteKey: key,
snapshotId: wxr.id,
reportRoot: ctx.importReportRoot,
pipeline: async () => {
// 管道内的可预期失败复用同一套错误码面(如侧车快照在管道期校验不过)
throw new ImportError('NO_WXR_SNAPSHOT', '管道内示例:主快照副本读取失败')
},
})
expect(result.status).toBe('failed')
const job = (await db.execute(
`select error_code, error_summary from import_jobs where id = '${result.jobId}'`,
)) as any
expect(job[0].error_code).toBe('NO_WXR_SNAPSHOT')
expect(String(job[0].error_summary)).toMatch(/^\[pipeline\] ImportError: 管道内示例/)
} finally {
await cleanupSiteRows(db, reg.siteId)
}
} finally {
await rm(ctx.work, { recursive: true, force: true })
}
})
})
it('23505 收敛确定性探针:未提交孪生 job 持有幂等键 → 落败方插入撞唯一约束后重查复用(Fix 6)', async () => {
await withTestDb(async (db) => {
const ctx = await makeCtx()
try {
const key = `sfimp-${rand()}`
const reg = await registerAll(db, ctx, key)
try {
const wxr = reg.snapshots.find((s) => s.kind === 'wxr')!
const idemKey = `import:${key}:${wxr.id}:${ADAPTER_VERSION}/${MAPPING_VERSION}/${SANITIZER_VERSION}`
let pipelineCalls = 0
// 事务内插孪生 job(status=running、不提交):业务连接的预查看不到它(read
// committed),插入撞其未提交唯一索引项——事务提交后 23505 确定性触发,
// 服务重查命中已提交行 → 收敛复用。真实并发窗口的库级语义复现,零代理。
let loser: ReturnType<typeof createImportJob> | undefined
await db.transaction(async (tx) => {
await tx.execute(
`insert into import_jobs (id, site_id, source_snapshot_id, status, mapping_version, idempotency_key, attempt, started_at)
values (gen_random_uuid(), '${reg.siteId}', '${wxr.id}', 'running', '${MAPPING_VERSION}', '${idemKey}', 1, now())`,
)
loser = createImportJob(db, {
siteKey: key,
reportRoot: ctx.importReportRoot,
pipeline: async () => {
pipelineCalls++
return { counts: {}, issues: [] }
},
})
// 让落败方先跑完预查、抵达插入(阻塞在孪生的未提交唯一项上)
await new Promise((r) => setTimeout(r, 150))
})
const result = await loser!
expect(result.reused).toBe(true)
expect(result.status).toBe('running') // 孪生行的原状态——复用不改行
expect(pipelineCalls).toBe(0) // 落败方从未执行管道
const count = (await db.execute(
`select count(*)::int as n from import_jobs where site_id = '${reg.siteId}'`,
)) as any
expect(count[0].n).toBe(1) // 只剩孪生行(孪生 status=running 由清理删除)
} finally {
await cleanupSiteRows(db, reg.siteId)
}
} finally {
await rm(ctx.work, { recursive: true, force: true })
}
})
})
2026-10-21 15:30:42 +08:00
})