浅念
77687e2330
Task 6 质量评审四项修复： - 失败报告部分冲刷（Fix 1）：orchestratedPipeline 分派/finish 包 try/catch—— 中途抛错时把可安全取得的部分产出冲入 ctx.report 后重抛：term/content 的 finish 是纯返回（已算成复用、未算成就地补算 trySyncFinish，自身抛错放弃 该部分不掩盖原错）、已收集的 adapter 问题/清洗统计、进度计数 eventsDispatched（orchestrator 键，仅失败路径出现——成功路径不占 18 键面）。 media/route finish 依赖库与先序产物，失败路径不调用（其键缺席如实反映）。 - 列表路径校验（Fix 3）：ImportErrorCode 增 INVALID_LIST_PATH；解析期防线 移至阶段 2.5（建 job 行前——建行/置 running 后再抛会留悬挂行）；复用 route-writer 导出的 isSafePath；normalizeListPath 自带安全拒绝（写入器 自调用位双覆盖）。CLI 无需新码（既有 ImportError 通道按 code 呈现）。 - JSDoc 求和措辞（Fix 2）：content-writer/media-writer 的 counts 键契约 注记由"返回值覆盖同名键 CLOBBER"改为"同名键求和（Task 6 钉死——防覆盖 安全网）"，与 ImportPipelineOutcome.counts 契约对齐。 - reason 归档（Fix 4）：分派循环对 unsupported/error 事件把 reason/issue 推入 adapter 问题组（此前被静默丢弃）。 测试：import-full 增 2 例（手构流 3 事件后爆炸 → failed + eventsDispatched=3 + 部分 content/term 计数 + unsupported/error reason 双归档；--products-path /../x/ 等 → INVALID_LIST_PATH 零 job 行）。全量 325 通过；typecheck/lint 干净。 Co-Authored-By: Claude Code <noreply@anthropic.com>
605 lines
31 KiB
TypeScript
605 lines
31 KiB
TypeScript
// Plan 3 Task 6 失败测试（先于实现）：编排集成 + 导入报告——四家写入器汇流。
|
||
// 覆盖契约（对照计划 Task 6 与前置事实 8/11/13、Task 3/5 审查移交清单）：
|
||
// 1. 全链（register wxr+media+media-map → 默认编排管道·真事件流）：content_entries=3 /
|
||
// source_records=5 / terms=2+自建 0 / content_terms=1 / media_assets=2 / routes=12；
|
||
// published_revision_id 全库 NULL（导入不发布——ADR-010）；mapConfig.locale 接线
|
||
// （site.default_locale='zh-CN' → 全部 source_records.locale='zh-CN'——locale 只可能
|
||
// 来自编排接线，mini 无自带 locale）；报告 = 四家 counts 18 键全量 + 清洗统计 +
|
||
// 版本块 + issues 分组 + 路径块 + 适配器分段计时；
|
||
// 2. 二导幂等（换 key 重跑）：全部 counts 归零语义（termsProcessed/mediaDanglingRefs 等
|
||
// 按事件计键如实再现）、数据库行数不变、job 仍 succeeded_with_issues
|
||
// （attachment 降级汇总 issue 必在——Plan 3 钉死语义：降级是处置事实非写入计数）；
|
||
// 3. 无 media 快照失败路径：media 域全 missing 报告不失败（空 manifest 语义已备）；
|
||
// 4. NULL-entry 占位行收编（Task 3 审查移交①）：六列命中 content_entry_id NULL 行 →
|
||
// UPDATE 挂 entry（不撞 23505、sourceRecords 不虚增）；
|
||
// 5. stale 回收（义务④）：同 site 超 24h 的 queued/running 置 stale，新近 job 不受影响；
|
||
// 6. REPORT_WRITE_FAILED（Task 1 遗留）：报告写盘失败 → failed 终态、report_path 不落。
|
||
// 7. 失败报告部分冲刷（Task 6 审查 Fix 1）：事件流中途抛错 → job failed 且报告携带
|
||
// eventsDispatched 进度计数 + 部分 content/term 计数（media/route finish 依赖库不调用）；
|
||
// unsupported/error 事件的 reason/issue 归档 adapter 组（Fix 4——此前被丢弃）；
|
||
// 8. 主列表路径解析期防线（Fix 3）：--products-path/--news-path 含穿越段 →
|
||
// INVALID_LIST_PATH 且零 job 行（建行前抛出——建行后再抛会留悬挂 running 行）。
|
||
// 运行于 globalSetup 临时库（ADR-015）；storage/报告根 os.tmpdir，finally 清理。
|
||
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 type { AdapterEvent } from '@sf/adapter-wp'
|
||
import { createImportJob, ImportError } 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')
|
||
|
||
const PNG_SIG = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a])
|
||
|
||
/** 伪 PNG 体（魔数校验只看签名——载荷唯一避免跨测试撞哈希） */
|
||
function pngBytes(seed: string): Buffer {
|
||
return Buffer.concat([PNG_SIG, Buffer.from(`png-body:${seed}`, 'latin1')])
|
||
}
|
||
|
||
/** 本文件写入的行清场：FK 反序（与 import-routes 同清单；两张不可变表走触发器开关） */
|
||
async function cleanupSiteRows(db: Db, siteId: string): Promise<void> {
|
||
await db.execute(`delete from route_registry where site_id = '${siteId}'`)
|
||
await db.execute(`delete from media_usages where site_id = '${siteId}'`)
|
||
await db.execute(`delete from media_assets where site_id = '${siteId}'`)
|
||
await db.execute(`delete from content_terms where site_id = '${siteId}'`)
|
||
await db.execute(`delete from terms where site_id = '${siteId}'`)
|
||
await db.execute(`delete from taxonomies where site_id = '${siteId}'`)
|
||
await db.execute(`delete from source_records where site_id = '${siteId}'`)
|
||
await db.execute(`delete from entry_locales where site_id = '${siteId}'`)
|
||
await withImmutableCleanup(db, 'content_revisions', 'content_revisions_immutable', `site_id = '${siteId}'`)
|
||
await db.execute(`delete from content_entries where site_id = '${siteId}'`)
|
||
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}'`)
|
||
}
|
||
|
||
interface Work {
|
||
work: string
|
||
storageRoot: string
|
||
reportRoot: string
|
||
allowedRoots: string[]
|
||
}
|
||
|
||
async function makeWork(prefix: string): Promise<Work> {
|
||
const work = await mkdtemp(join(tmpdir(), `${prefix}-`))
|
||
return {
|
||
work,
|
||
storageRoot: join(work, 'storage'),
|
||
reportRoot: join(work, 'artifacts', 'import'),
|
||
allowedRoots: [FIXTURES_ROOT, work],
|
||
}
|
||
}
|
||
|
||
/** 全量登记（wxr + 媒体目录 + 侧车）——形态同 import-routes 全链用例；locale 可选注入 */
|
||
async function registerFull(
|
||
db: Db,
|
||
w: Work,
|
||
spec: { locale?: string } = {},
|
||
): Promise<{ key: string; siteId: string; wxrId: string }> {
|
||
const mediaDir = join(w.work, 'media')
|
||
await mkdir(join(mediaDir, '2026/10'), { recursive: true })
|
||
await writeFile(join(mediaDir, '2026/10/fixture-a.png'), pngBytes(`full-a-${rand()}`))
|
||
const guidB = `b-${rand()}.png`
|
||
await writeFile(join(mediaDir, guidB), pngBytes(`full-b-${rand()}`))
|
||
const sidecarPath = join(w.work, 'media-map.json')
|
||
await writeFile(
|
||
sidecarPath,
|
||
JSON.stringify([
|
||
{ sourceUrl: 'https://fixture.example/wp-content/uploads/2026/10/fixture-b.png', filePath: guidB },
|
||
]),
|
||
)
|
||
const key = `sffull-${rand()}`
|
||
const reg = await registerSource(db, {
|
||
siteKey: key,
|
||
wxrPath: WXR_FIXTURE,
|
||
mediaDir,
|
||
mediaMapPath: sidecarPath,
|
||
allowedRoots: w.allowedRoots,
|
||
storageRoot: w.storageRoot,
|
||
reportRoot: join(w.work, 'artifacts', 'register'),
|
||
...(spec.locale === undefined ? {} : { siteDefaults: { defaultLocale: spec.locale } }),
|
||
})
|
||
return { key, siteId: reg.siteId, wxrId: reg.snapshots.find((s) => s.kind === 'wxr')!.id }
|
||
}
|
||
|
||
/** 仅 wxr 登记（失败路径/收编场景——media_bundle/manual_url_list 缺席） */
|
||
async function registerWxrOnly(db: Db, w: Work): Promise<{ key: string; siteId: string; wxrId: string }> {
|
||
const key = `sffull-${rand()}`
|
||
const reg = await registerSource(db, {
|
||
siteKey: key,
|
||
wxrPath: WXR_FIXTURE,
|
||
allowedRoots: w.allowedRoots,
|
||
storageRoot: w.storageRoot,
|
||
reportRoot: join(w.work, 'artifacts', 'register'),
|
||
})
|
||
return { key, siteId: reg.siteId, wxrId: reg.snapshots.find((s) => s.kind === 'wxr')!.id }
|
||
}
|
||
|
||
/** 导入报告读取（默认编排管道的产物断言面） */
|
||
async function readReport(w: Work, jobId: string): Promise<any> {
|
||
return JSON.parse(await readFile(join(w.reportRoot, jobId, 'reports', 'import.json'), 'utf8'))
|
||
}
|
||
|
||
/** 单值计数便捷 */
|
||
async function countOf(db: Db, sql: string): Promise<number> {
|
||
const rows = (await db.execute(sql)) as any
|
||
return Number(rows[0].n)
|
||
}
|
||
|
||
describe('编排集成 + 导入报告（四家写入器汇流）', () => {
|
||
it('全链（真事件流 + 默认编排管道）：18 键 counts 全量、published 全 NULL、locale 接线、清洗统计、issues 分组、报告路径/版本/计时块', async () => {
|
||
await withTestDb(async (db) => {
|
||
const w = await makeWork('sf-full-chain')
|
||
try {
|
||
// locale 接线断言的实证设计（前置事实 8）：site 默认 locale 取 'zh-CN'——mini 无
|
||
// 自带 locale，行 locale 只可能来自编排接线 mapConfig.locale=site.default_locale
|
||
const { key, siteId, wxrId } = await registerFull(db, w, { locale: 'zh-CN' })
|
||
try {
|
||
const result = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxrId,
|
||
reportRoot: w.reportRoot,
|
||
})
|
||
expect(result.reused).toBe(false)
|
||
// 悬空引用（media 203/204）×2 + attachment 降级汇总 ×1 → issues 非空
|
||
expect(result.status).toBe('succeeded_with_issues')
|
||
|
||
// 数据库行数对拍（计划 Task 6 钉死清单）
|
||
expect(await countOf(db, `select count(*)::int as n from content_entries where site_id = '${siteId}'`)).toBe(3)
|
||
expect(await countOf(db, `select count(*)::int as n from source_records where site_id = '${siteId}'`)).toBe(5)
|
||
expect(await countOf(db, `select count(*)::int as n from terms where site_id = '${siteId}'`)).toBe(2)
|
||
expect(await countOf(db, `select count(*)::int as n from content_terms where site_id = '${siteId}'`)).toBe(1)
|
||
expect(await countOf(db, `select count(*)::int as n from media_assets where site_id = '${siteId}'`)).toBe(2)
|
||
expect(await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}'`)).toBe(12)
|
||
|
||
// published 全库 NULL（导入不发布——ADR-010）+ current 全设（导入产出工作修订）
|
||
expect(
|
||
await countOf(db, `select count(*)::int as n from entry_locales where site_id = '${siteId}' and published_revision_id is not null`),
|
||
).toBe(0)
|
||
expect(
|
||
await countOf(db, `select count(*)::int as n from entry_locales where site_id = '${siteId}' and current_revision_id is not null`),
|
||
).toBe(3)
|
||
|
||
// locale 接线（前置事实 8）：全部 source_records 行 locale = site 默认（非空串）
|
||
const locales = (await db.execute(`select distinct locale from source_records where site_id = '${siteId}'`)) as any
|
||
expect(locales.map((r: any) => r.locale).sort()).toStrictEqual(['zh-CN'])
|
||
|
||
// attachment 分派规则（前置事实 11）：2 条 attachment 行是 record-only 足迹（无 entry）
|
||
expect(
|
||
await countOf(db, `select count(*)::int as n from source_records where site_id = '${siteId}' and source_type = 'wp:attachment' and content_entry_id is null`),
|
||
).toBe(2)
|
||
|
||
// content_terms：post 101 的 category 引用（is_primary=true——WP 单主分类近似）
|
||
// （投影转 Boolean——postgres.js 原型差异会让 toStrictEqual 报"无视觉差异"失败）
|
||
const ct = (await db.execute(`select ct.is_primary, t.slug from content_terms ct
|
||
join terms t on t.id = ct.term_id where ct.site_id = '${siteId}'`)) as any
|
||
expect(ct.map((r: any) => ({ isPrimary: Boolean(r.is_primary), slug: r.slug }))).toStrictEqual([
|
||
{ isPrimary: true, slug: 'company' },
|
||
])
|
||
|
||
// 报告：四家 counts 18 键全量（全局命名空间求和聚合——键前缀互斥）
|
||
const report = await readReport(w, result.jobId)
|
||
expect(report.counts).toStrictEqual({
|
||
contentEntries: 3,
|
||
revisionsCreated: 3,
|
||
sourceRecords: 5,
|
||
skipped: 0,
|
||
termsProcessed: 2,
|
||
autoCreatedTerms: 0,
|
||
contentTermsCreated: 1,
|
||
mediaRegistered: 2,
|
||
mediaFromSidecar: 1,
|
||
mediaMissing: 0,
|
||
mediaQuarantined: 0,
|
||
mediaDanglingRefs: 2,
|
||
mediaAmbiguous: 0,
|
||
routesCreated: 12,
|
||
routesUpdated: 0,
|
||
routesRedirects: 2,
|
||
routesCollisions: 0,
|
||
routesExcludedQueries: 0,
|
||
})
|
||
|
||
// 清洗统计（分派循环收集 row 事件 sanitizeDiffs/changed）：mini 仅 post 101 正文
|
||
// img 自闭合规范化零 diff 变形（changedRows=1、无 diff 条目——sparse 形态）
|
||
expect(report.sanitize).toStrictEqual({ changedRows: 1, byDiffType: {} })
|
||
|
||
// issues 分组（按写入器前缀）：flat issues 保持全量（3 条），分组归位
|
||
expect(report.issues).toHaveLength(3)
|
||
expect(report.issueGroups).toStrictEqual({
|
||
content: [],
|
||
term: [],
|
||
media: report.issues.filter((i: string) => i.includes('悬空')),
|
||
route: report.issues.filter((i: string) => i.includes('attachment 页降级')),
|
||
adapter: [],
|
||
orchestrator: [],
|
||
})
|
||
expect(report.issueGroups.media).toHaveLength(2)
|
||
expect(report.issueGroups.route).toHaveLength(1)
|
||
|
||
// 路径块（前置事实 9：CLI 参数来源记入报告——缺省值）+ 版本块 + 分段计时（真流）
|
||
expect(report.paths).toStrictEqual({ productsPath: '/products/', newsPath: '/news/' })
|
||
expect(report.versions).toStrictEqual({
|
||
adapter: report.versions.adapter,
|
||
mapping: report.versions.mapping,
|
||
sanitizer: report.versions.sanitizer,
|
||
parser: report.versions.parser,
|
||
})
|
||
expect(Object.keys(report.versions).sort()).toStrictEqual(['adapter', 'mapping', 'parser', 'sanitizer'])
|
||
expect(typeof report.adapterTiming.parseMs).toBe('number')
|
||
expect(typeof report.adapterTiming.sanitizeMs).toBe('number')
|
||
} finally {
|
||
await cleanupSiteRows(db, siteId)
|
||
}
|
||
} finally {
|
||
await rm(w.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('二导幂等（换 key）：全部写入计数归零、数据库行数不变、job 仍 succeeded_with_issues（降级汇总 issue 必在）', async () => {
|
||
await withTestDb(async (db) => {
|
||
const w = await makeWork('sf-full-idem')
|
||
try {
|
||
const { key, siteId, wxrId } = await registerFull(db, w)
|
||
try {
|
||
const base = { siteKey: key, snapshotId: wxrId, reportRoot: w.reportRoot } as const
|
||
const first = await createImportJob(db, { ...base })
|
||
expect(first.status).toBe('succeeded_with_issues')
|
||
|
||
const again = await createImportJob(db, { ...base, idempotencyKey: `sffull-rerun-${rand()}` })
|
||
expect(again.reused).toBe(false)
|
||
expect(again.status).toBe('succeeded_with_issues')
|
||
|
||
// 写入计数全零语义：按事件计的键（termsProcessed/mediaDanglingRefs）如实再现
|
||
const report = await readReport(w, again.jobId)
|
||
expect(report.counts).toStrictEqual({
|
||
contentEntries: 0,
|
||
revisionsCreated: 0,
|
||
sourceRecords: 0,
|
||
skipped: 3,
|
||
termsProcessed: 2,
|
||
autoCreatedTerms: 0,
|
||
contentTermsCreated: 0,
|
||
mediaRegistered: 0,
|
||
mediaFromSidecar: 0,
|
||
mediaMissing: 0,
|
||
mediaQuarantined: 0,
|
||
mediaDanglingRefs: 2,
|
||
mediaAmbiguous: 0,
|
||
routesCreated: 0,
|
||
routesUpdated: 0,
|
||
routesRedirects: 0,
|
||
routesCollisions: 0,
|
||
routesExcludedQueries: 0,
|
||
})
|
||
|
||
// 数据库行数不变（幂等：零新行零修订零更新）
|
||
expect(await countOf(db, `select count(*)::int as n from content_entries where site_id = '${siteId}'`)).toBe(3)
|
||
expect(await countOf(db, `select count(*)::int as n from content_revisions where site_id = '${siteId}'`)).toBe(3)
|
||
expect(await countOf(db, `select count(*)::int as n from source_records where site_id = '${siteId}'`)).toBe(5)
|
||
expect(await countOf(db, `select count(*)::int as n from terms where site_id = '${siteId}'`)).toBe(2)
|
||
expect(await countOf(db, `select count(*)::int as n from content_terms where site_id = '${siteId}'`)).toBe(1)
|
||
expect(await countOf(db, `select count(*)::int as n from media_assets where site_id = '${siteId}'`)).toBe(2)
|
||
expect(await countOf(db, `select count(*)::int as n from media_usages where site_id = '${siteId}'`)).toBe(3)
|
||
expect(await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}'`)).toBe(12)
|
||
|
||
// 降级汇总 issue 必在（Plan 3 钉死：降级是处置事实，重跑如实再现）+ 悬空 2 条
|
||
const downgrade = report.issues.filter((i: string) => i.includes('attachment 页降级'))
|
||
expect(downgrade).toHaveLength(1)
|
||
expect(downgrade[0]).toContain('2 条')
|
||
expect(report.issues.filter((i: string) => i.includes('悬空'))).toHaveLength(2)
|
||
expect(report.issues).toHaveLength(3)
|
||
} finally {
|
||
await cleanupSiteRows(db, siteId)
|
||
}
|
||
} finally {
|
||
await rm(w.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('无 media 快照（只 wxr）：media 域全 missing 报告不失败（空 manifest 语义）+ 主列表路径参数入报告', async () => {
|
||
await withTestDb(async (db) => {
|
||
const w = await makeWork('sf-full-nomedia')
|
||
try {
|
||
const { key, siteId, wxrId } = await registerWxrOnly(db, w)
|
||
try {
|
||
const result = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxrId,
|
||
reportRoot: w.reportRoot,
|
||
productsPath: '/shop/',
|
||
})
|
||
expect(result.status).toBe('succeeded_with_issues')
|
||
|
||
// 空清单语义：2 attachment 全 missing（不插行）、5 引用全悬空、路由 8 行
|
||
// （3 站点页 + 3 内容 + 2 分类——301/asset 无资产可指）
|
||
expect(await countOf(db, `select count(*)::int as n from media_assets where site_id = '${siteId}'`)).toBe(0)
|
||
expect(await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}'`)).toBe(8)
|
||
expect(
|
||
await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}' and route_path = '/shop/'`),
|
||
).toBe(1)
|
||
expect(
|
||
await countOf(db, `select count(*)::int as n from route_registry where site_id = '${siteId}' and route_path = '/products/'`),
|
||
).toBe(0)
|
||
|
||
const report = await readReport(w, result.jobId)
|
||
expect(report.counts.mediaMissing).toBe(2)
|
||
expect(report.counts.mediaRegistered).toBe(0)
|
||
expect(report.counts.mediaDanglingRefs).toBe(5)
|
||
expect(report.counts.routesCreated).toBe(8)
|
||
expect(report.paths).toStrictEqual({ productsPath: '/shop/', newsPath: '/news/' })
|
||
// attachment 301 目标缺席逐条报告（§7.6）+ missing/悬空——全部在 issues
|
||
expect(report.issues.filter((i: string) => i.includes('301 目标缺席'))).toHaveLength(2)
|
||
expect(report.issues.filter((i: string) => i.includes('missing'))).toHaveLength(2)
|
||
expect(report.issues.filter((i: string) => i.includes('悬空'))).toHaveLength(5)
|
||
} finally {
|
||
await cleanupSiteRows(db, siteId)
|
||
}
|
||
} finally {
|
||
await rm(w.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('NULL-entry 占位行收编（Task 3 审查移交①）：六列命中 content_entry_id NULL 行 → UPDATE 挂 entry、不撞 23505、sourceRecords 不虚增', async () => {
|
||
await withTestDb(async (db) => {
|
||
const w = await makeWork('sf-full-adopt')
|
||
try {
|
||
const { key, siteId, wxrId } = await registerWxrOnly(db, w)
|
||
try {
|
||
// 外源占位行（seed 模拟：六列身份 = wp:post/101 本快照、entry NULL）
|
||
const seedJob = (await db.execute(`insert into import_jobs (id, site_id, source_snapshot_id, status,
|
||
mapping_version, idempotency_key, attempt) values (gen_random_uuid(), '${siteId}', '${wxrId}',
|
||
'succeeded', 'seed', 'seed-${rand()}', 1) returning id`)) as any
|
||
const connRow = (await db.execute(`select source_connection_id from source_snapshots where id = '${wxrId}'`)) as any
|
||
const seeded = (await db.execute(`insert into source_records (id, site_id, source_connection_id,
|
||
source_snapshot_id, import_job_id, source_type, source_id, locale, raw_location, parse_status,
|
||
content_entry_id) values (gen_random_uuid(), '${siteId}', '${connRow[0].source_connection_id}',
|
||
'${wxrId}', '${seedJob[0].id}', 'wp:post', '101', 'en', 'wxr://item/101', 'imported', NULL)
|
||
returning id`)) as any
|
||
|
||
const result = await createImportJob(db, { siteKey: key, snapshotId: wxrId, reportRoot: w.reportRoot })
|
||
expect(result.status).toBe('succeeded_with_issues')
|
||
|
||
// 占位行被收编：挂 entry + checksum 前移（非新建行——sourceRecords=4 不是 5）
|
||
const adopted = (await db.execute(`select content_entry_id, normalized_checksum from source_records
|
||
where id = '${seeded[0].id}'`)) as any
|
||
expect(adopted[0].content_entry_id).not.toBeNull()
|
||
expect(adopted[0].normalized_checksum).not.toBeNull()
|
||
const el = (await db.execute(`select current_revision_id, published_revision_id from entry_locales
|
||
where site_id = '${siteId}' and content_entry_id = '${adopted[0].content_entry_id}'`)) as any
|
||
expect(el[0].current_revision_id).not.toBeNull()
|
||
expect(el[0].published_revision_id).toBeNull()
|
||
|
||
const report = await readReport(w, result.jobId)
|
||
expect(report.counts.contentEntries).toBe(3)
|
||
expect(report.counts.revisionsCreated).toBe(3)
|
||
expect(report.counts.sourceRecords).toBe(4) // 101 收编非新建（2 attachment + 111/121）
|
||
expect(report.counts.skipped).toBe(0)
|
||
// 收编行照常参与 term 关联与路由（101 的 entry 锚点就位）
|
||
expect(report.counts.contentTermsCreated).toBe(1)
|
||
expect(report.counts.routesCreated).toBe(8)
|
||
} finally {
|
||
await cleanupSiteRows(db, siteId)
|
||
}
|
||
} finally {
|
||
await rm(w.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('stale 回收（义务④）：同 site 超 24h 的 queued/running 置 stale；新近 job 不受影响；新 job 正常推进', async () => {
|
||
await withTestDb(async (db) => {
|
||
const w = await makeWork('sf-full-stale')
|
||
try {
|
||
const { key, siteId, wxrId } = await registerWxrOnly(db, w)
|
||
try {
|
||
const oldH = 25
|
||
const freshH = 1
|
||
const seed = async (status: string, hoursAgo: number, idem: string): Promise<string> => {
|
||
const rows = (await db.execute(`insert into import_jobs (id, site_id, source_snapshot_id, status,
|
||
mapping_version, idempotency_key, attempt, started_at, created_at) values (gen_random_uuid(),
|
||
'${siteId}', '${wxrId}', '${status}', 'seed', '${idem}', 1,
|
||
now() - interval '${hoursAgo} hours', now() - interval '${hoursAgo} hours') returning id`)) as any
|
||
return rows[0].id
|
||
}
|
||
const oldRunning = await seed('running', oldH, `stale-run-${rand()}`)
|
||
const oldQueued = await seed('queued', oldH + 2, `stale-q-${rand()}`)
|
||
const freshRunning = await seed('running', freshH, `stale-fresh-${rand()}`)
|
||
|
||
const result = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxrId,
|
||
reportRoot: w.reportRoot,
|
||
pipeline: async () => ({ counts: {}, issues: [] }),
|
||
})
|
||
expect(result.status).toBe('succeeded')
|
||
|
||
const st = (rows: string[]) =>
|
||
(db.execute(`select status, finished_at from import_jobs where id in (${rows.map((r) => `'${r}'`).join(',')})`) as any) as Promise<any>
|
||
const oldRows = await st([oldRunning, oldQueued])
|
||
for (const r of oldRows) {
|
||
expect(r.status).toBe('stale')
|
||
expect(r.finished_at).not.toBeNull()
|
||
}
|
||
const freshRows = await st([freshRunning])
|
||
expect(freshRows[0].status).toBe('running') // 未超时不动（心跳列决策留 ADR-017）
|
||
expect(freshRows[0].finished_at).toBeNull()
|
||
} finally {
|
||
await cleanupSiteRows(db, siteId)
|
||
}
|
||
} finally {
|
||
await rm(w.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('REPORT_WRITE_FAILED（Task 1 遗留）：报告目录不可建 → failed 终态、error_code 落列、report_path 不落', async () => {
|
||
await withTestDb(async (db) => {
|
||
const w = await makeWork('sf-full-rwfail')
|
||
try {
|
||
const { key, siteId, wxrId } = await registerWxrOnly(db, w)
|
||
try {
|
||
// 用普通文件占住报告根：mkdir <file>/<jobId>/reports 必败（ENOTDIR/EEXIST）
|
||
const blocker = join(w.work, 'blocker')
|
||
await writeFile(blocker, 'not a directory')
|
||
const result = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxrId,
|
||
reportRoot: blocker,
|
||
pipeline: async () => ({ counts: { ok: 1 }, issues: [] }),
|
||
})
|
||
expect(result.status).toBe('failed')
|
||
expect(result.reportPath).toBeNull()
|
||
const job = (await db.execute(`select status, error_code, error_summary, report_path, finished_at
|
||
from import_jobs where id = '${result.jobId}'`)) as any
|
||
expect(job[0].status).toBe('failed')
|
||
expect(job[0].error_code).toBe('REPORT_WRITE_FAILED')
|
||
expect(String(job[0].error_summary)).toMatch(/^\[report\] /)
|
||
expect(job[0].report_path).toBeNull() // 文件不存在不悬挂路径
|
||
expect(job[0].finished_at).not.toBeNull()
|
||
} finally {
|
||
await cleanupSiteRows(db, siteId)
|
||
}
|
||
} finally {
|
||
await rm(w.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('失败报告部分冲刷（Fix 1）：事件流中途抛错 → failed + 报告含 eventsDispatched 与部分 content/term 计数；unsupported/error reason 归档 adapter 组（Fix 4）', async () => {
|
||
await withTestDb(async (db) => {
|
||
const w = await makeWork('sf-full-flush')
|
||
try {
|
||
const { key, siteId } = await registerWxrOnly(db, w)
|
||
try {
|
||
// 手构流 + 真编排（注入 events 而不注入 pipeline）：3 个事件成功分派后，
|
||
// 第 4 次拉取抛错——分派循环内炸（非写入器抛），编排器须冲刷部分产出后重抛
|
||
async function* bomb(): AsyncGenerator<AdapterEvent, void, void> {
|
||
yield {
|
||
kind: 'row',
|
||
row: {
|
||
sourceType: 'wp:post',
|
||
sourceId: '301',
|
||
status: 'publish',
|
||
locale: 'en',
|
||
taxonomyRefs: [{ taxonomy: 'category', sourceTermId: 'company' }],
|
||
attachmentRefs: [],
|
||
seo: {},
|
||
knownFields: {},
|
||
extras: {},
|
||
rawLocation: 'wxr://item/301',
|
||
},
|
||
issues: [],
|
||
sanitizeDiffs: [],
|
||
changed: false,
|
||
}
|
||
yield {
|
||
kind: 'unsupported',
|
||
sourceType: 'wp:nav_menu_item',
|
||
reason: 'post_type 未配置映射',
|
||
rawLocation: 'wxr://item/7',
|
||
postId: '68',
|
||
}
|
||
yield { kind: 'error', sourceType: 'wp:post', issue: 'post_id 缺失或非法', rawLocation: 'wxr://item/9' }
|
||
throw new Error('BOOM: 事件流在第 4 次拉取时爆炸')
|
||
}
|
||
const result = await createImportJob(db, { siteKey: key, reportRoot: w.reportRoot, events: bomb() })
|
||
expect(result.status).toBe('failed')
|
||
expect(result.reportPath).not.toBeNull()
|
||
|
||
const job = (await db.execute(`select error_code, error_summary, report_path, finished_at
|
||
from import_jobs where id = '${result.jobId}'`)) as any
|
||
expect(job[0].error_code).toBe('PIPELINE_FAILED')
|
||
expect(String(job[0].error_summary)).toMatch(/^\[pipeline\] /)
|
||
expect(job[0].report_path).not.toBeNull() // 失败也落报告（部分产出可审计）
|
||
expect(job[0].finished_at).not.toBeNull()
|
||
|
||
const report = await readReport(w, result.jobId)
|
||
expect(report.status).toBe('failed')
|
||
expect(report.error.code).toBe('PIPELINE_FAILED')
|
||
// 部分冲刷面：成功分派的 3 事件的 content/term 计数 + 进度计数 eventsDispatched
|
||
//（media/route finish 依赖库与先序产物——失败路径不调用，其键缺席）
|
||
expect(report.counts).toStrictEqual({
|
||
eventsDispatched: 3,
|
||
contentEntries: 1,
|
||
revisionsCreated: 1,
|
||
sourceRecords: 3,
|
||
skipped: 0,
|
||
unsupportedRecords: 1,
|
||
failedRecords: 1,
|
||
termsProcessed: 0,
|
||
autoCreatedTerms: 1,
|
||
contentTermsCreated: 1,
|
||
})
|
||
// unsupported/error 的 reason/issue 归档 adapter 组（Fix 4——此前被静默丢弃）
|
||
expect(report.issueGroups.adapter).toHaveLength(2)
|
||
expect(report.issueGroups.adapter[0]).toContain('不支持对象')
|
||
expect(report.issueGroups.adapter[0]).toContain('post_type 未配置映射')
|
||
expect(report.issueGroups.adapter[1]).toContain('适配器错误')
|
||
expect(report.issueGroups.adapter[1]).toContain('post_id 缺失或非法')
|
||
// 冲刷的 issue 同时在 flat 清单；content/term/media/route 组空（本流无其问题）
|
||
expect(report.issues).toHaveLength(2)
|
||
expect(report.issueGroups).toStrictEqual({
|
||
content: [],
|
||
term: [],
|
||
media: [],
|
||
route: [],
|
||
adapter: report.issues,
|
||
orchestrator: [],
|
||
})
|
||
// 落库审计：已分派事件的部分写入保留（各写入器单语句已提交，失败不回滚）
|
||
expect(await countOf(db, `select count(*)::int as n from content_entries where site_id = '${siteId}'`)).toBe(1)
|
||
} finally {
|
||
await cleanupSiteRows(db, siteId)
|
||
}
|
||
} finally {
|
||
await rm(w.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('主列表路径解析期防线（Fix 3）：--products-path/--news-path 含穿越段 → INVALID_LIST_PATH、零 job 行（建行前抛出）', async () => {
|
||
await withTestDb(async (db) => {
|
||
const w = await makeWork('sf-full-path')
|
||
try {
|
||
const { key, siteId } = await registerWxrOnly(db, w)
|
||
try {
|
||
const cases: Array<[string, { productsPath?: string; newsPath?: string }]> = [
|
||
['productsPath 穿越段', { productsPath: '/../x/' }],
|
||
['newsPath 穿越段', { newsPath: '/news/../../etc/' }],
|
||
]
|
||
for (const [label, paths] of cases) {
|
||
const err: ImportError = await createImportJob(db, {
|
||
siteKey: key,
|
||
reportRoot: w.reportRoot,
|
||
...paths,
|
||
}).then(
|
||
() => {
|
||
throw new Error(`用例「${label}」不应成功`)
|
||
},
|
||
(e: ImportError) => e,
|
||
)
|
||
expect(err, label).toBeInstanceOf(ImportError)
|
||
expect(err.code, label).toBe('INVALID_LIST_PATH')
|
||
expect(err.message, label).toContain('..') // 消息携带原始入参（操作者可定位）
|
||
}
|
||
// 零 job 行：解析期失败（建行前）——建行后再抛会留悬挂 running 行
|
||
expect(await countOf(db, `select count(*)::int as n from import_jobs where site_id = '${siteId}'`)).toBe(0)
|
||
} finally {
|
||
await cleanupSiteRows(db, siteId)
|
||
}
|
||
} finally {
|
||
await rm(w.work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
})
|