浅念
fe72df3afa
审查 cheap riders 四项：handleUnmappable 分项 unsupportedRecords/ failedRecords（按事件计——拆分只在分派点可知，Task 6 无从重derive，测试 对拍两键）；ImportPipelineOutcome.counts 键契约注记（跨写入器扁平全局 命名空间、同名 CLOBBER 不求和、各键归属声明、Task 4 勿复用 'skipped'）； 六列 NULL-entry 已知死角缝注释（23505 → Task 6 收编）；态 1 checksum-diff 的 import_job_id lineage 语义注记（source_records 留建行 job、新 job 记 content_revisions）。296 全绿；typecheck/lint 通过。 Co-Authored-By: Claude Code <noreply@anthropic.com>
677 lines
32 KiB
TypeScript
677 lines
32 KiB
TypeScript
// Plan 3 Task 3 失败测试（先于实现）：内容写入器（稳定身份 + checksum skip）。
|
||
// 覆盖契约（对照计划 Task 3 与前置事实 6/8/11/12、ADR-006/010）：
|
||
// 1. mini.xml 全链（3 内容行 + 2 attachment）：3 content_entries（typeKey 映射前置事实 11）
|
||
// + 3 entry_locales（slug/locale/current_revision_id 设、**published NULL**）+
|
||
// 3 content_revisions（source_kind='import'、checksum 按前置事实 6 公式**独立重算对拍**、
|
||
// data={title?,slug?,status,bodyHtml,seo,knownFields,extras}）+ 5 source_records
|
||
// （attachment 2 行 content_entry_id NULL、parse_status='imported'）；
|
||
// 2. 二导同快照（换幂等键重跑）：0 新修订 0 新 entry（checksum skip）、counts.skipped=3；
|
||
// 跨快照（变体文件新 snapshot、业务内容不变）：同样 0 新修订 0 新 entry——五列稳定
|
||
// 身份复用既有 entry（ADR-006），本快照 source_records 补行（六列身份含 snapshot_id）；
|
||
// 3. 内容变化（改 knownFields.title）：新修订 + current 前移 + published 仍 NULL
|
||
// （ADR-010 导入不发布）+ 旧修订原样（不可变触发器——写入路径纯 INSERT 纪律）；
|
||
// 4. permalink 变化（改 sourceUrl）：**不建新 entry**（身份键不含 URL）+ 新修订
|
||
// （sourceUrl 在 route_source_values 哈希域内）+ source_records.source_url 更新；
|
||
// 5. unsupported/error 也写 source_records（前置事实 12）：parse_status='unsupported'/
|
||
// 'failed'、locale=站点默认、content_entry_id NULL；error 无 postId → source_id 以
|
||
// rawLocation 回退（不硬造）；
|
||
// 6. locale 校验（前置事实 8）：row.locale ≠ site.default_locale → issue + 跳过（对照行
|
||
// 证明拒收不断流）。
|
||
// 期望值全部从适配器真实产物行派生（collect 后独立组装 checksum 输入对拍，不与写入器
|
||
// 共享组装代码），不猜清洗细节。运行于 globalSetup 临时库（ADR-015）；报告根用
|
||
// os.tmpdir 临时目录，finally 清理。
|
||
import { createHash } from 'node:crypto'
|
||
import { createReadStream } from 'node:fs'
|
||
import { 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,
|
||
SANITIZER_VERSION,
|
||
runWxrAdapter,
|
||
type AdapterEvent,
|
||
} from '@sf/adapter-wp'
|
||
import { UNIVERSAL_ROW_VERSION, type UniversalRow } from '@sf/contracts'
|
||
import { canonicalJson } from '../../apps/api/src/common/canonical.js'
|
||
import { createImportJob } from '../../apps/api/src/modules/import/import.service.js'
|
||
import { createContentWriter } from '../../apps/api/src/modules/import/content-writer.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')
|
||
|
||
/** 前置事实 11 的 typeKey 映射（断言侧独立持有——与实现常量对拍而非引用） */
|
||
const EXPECTED_TYPE_KEYS: Readonly<Record<string, string>> = Object.freeze({
|
||
'wp:post': 'news_article',
|
||
'wp:page': 'basic_page',
|
||
'wp:product': 'product',
|
||
})
|
||
|
||
/** 手构事件流的 AsyncIterable 包装（每次调用产新迭代器——重跑各流独立） */
|
||
async function* eventsOf(events: AdapterEvent[]): AsyncGenerator<AdapterEvent, void, void> {
|
||
yield* events
|
||
}
|
||
|
||
/** db.execute 行对象规整：postgres.js 行对象的原型差异会让 toStrictEqual 报"无视觉差异"失败——浅拷贝到普通对象 */
|
||
function plain(rows: any[]): any[] {
|
||
return rows.map((r) => ({ ...r }))
|
||
}
|
||
|
||
/** 事件流收集（mini 全链一饮而尽；断言期望从产物行派生，不猜清洗细节） */
|
||
async function collectMiniEvents(): Promise<AdapterEvent[]> {
|
||
const out: AdapterEvent[] = []
|
||
for await (const ev of runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } })) {
|
||
out.push(ev)
|
||
}
|
||
return out
|
||
}
|
||
|
||
/** 事件流中的 row 事件提取（UniversalRow 列表） */
|
||
function rowsOf(events: AdapterEvent[]): UniversalRow[] {
|
||
return events.flatMap((ev) => (ev.kind === 'row' ? [ev.row] : []))
|
||
}
|
||
|
||
/** 深拷贝事件数组并对指定 sourceId 的 row 施加变异（整组克隆——两导的事件互不串扰） */
|
||
function withRowMutated(
|
||
events: AdapterEvent[],
|
||
sourceId: string,
|
||
mutate: (row: UniversalRow) => void,
|
||
): AdapterEvent[] {
|
||
return events.map((ev) => {
|
||
const copy = structuredClone(ev)
|
||
if (copy.kind === 'row' && copy.row.sourceId === sourceId) mutate(copy.row)
|
||
return copy
|
||
})
|
||
}
|
||
|
||
/**
|
||
* 独立重算 normalized checksum（前置事实 6 公式对拍）：测试自行组装输入对象（mapped_content
|
||
* 结构钉死 {title?,slug?,status,bodyHtml,seo,knownFields,extras}、rawLocation 显式排除、
|
||
* 版本三常量必入），与写入器零共享代码；序列化基形用被 tests/unit/canonical.test.ts 独立
|
||
* 钉死的 canonicalJson，哈希与 register 同法（sha256 hex）。
|
||
*/
|
||
function expectedChecksum(row: UniversalRow): string {
|
||
const mappedContent: Record<string, unknown> = {
|
||
status: row.status,
|
||
seo: row.seo,
|
||
knownFields: row.knownFields,
|
||
extras: row.extras,
|
||
}
|
||
if (row.title !== undefined) mappedContent.title = row.title
|
||
if (row.slug !== undefined) mappedContent.slug = row.slug
|
||
if (row.bodyHtml !== undefined) mappedContent.bodyHtml = row.bodyHtml
|
||
const routeSourceValues: Record<string, unknown> = {}
|
||
if (row.sourceUrl !== undefined) routeSourceValues.sourceUrl = row.sourceUrl
|
||
return createHash('sha256')
|
||
.update(
|
||
canonicalJson({
|
||
universal_row_version: UNIVERSAL_ROW_VERSION,
|
||
adapter_version: ADAPTER_VERSION,
|
||
mapping_version: MAPPING_VERSION,
|
||
sanitizer_version: SANITIZER_VERSION,
|
||
locale: row.locale,
|
||
mapped_content: mappedContent,
|
||
taxonomy_relations: row.taxonomyRefs,
|
||
media_reference_identities: row.attachmentRefs,
|
||
route_source_values: routeSourceValues,
|
||
}),
|
||
)
|
||
.digest('hex')
|
||
}
|
||
|
||
/** revision.data 期望结构（前置事实 6 钉死的顶层形状——与 checksum 的 mapped_content 同构） */
|
||
function expectedRevisionData(row: UniversalRow): Record<string, unknown> {
|
||
const data: Record<string, unknown> = {
|
||
status: row.status,
|
||
seo: row.seo,
|
||
knownFields: row.knownFields,
|
||
extras: row.extras,
|
||
}
|
||
if (row.title !== undefined) data.title = row.title
|
||
if (row.slug !== undefined) data.slug = row.slug
|
||
if (row.bodyHtml !== undefined) data.bodyHtml = row.bodyHtml
|
||
return data
|
||
}
|
||
|
||
/** jsonb 往返规整（期望侧：去 null 原型/undefined 键——与 PG 取回的普通对象同形） */
|
||
function jsonbNorm(value: unknown): Record<string, unknown> {
|
||
return JSON.parse(JSON.stringify(value)) as Record<string, unknown>
|
||
}
|
||
|
||
/** 手构 row 事件（全必填字段的最小 UniversalRow；可选字段按 spec 传入） */
|
||
function rowEvent(spec: {
|
||
sourceType?: string
|
||
sourceId: string
|
||
locale?: string
|
||
slug?: string
|
||
title?: string
|
||
bodyHtml?: string
|
||
sourceUrl?: string
|
||
}): AdapterEvent {
|
||
const row: UniversalRow = {
|
||
sourceType: spec.sourceType ?? 'wp:post',
|
||
sourceId: spec.sourceId,
|
||
status: 'publish',
|
||
locale: spec.locale ?? 'en',
|
||
taxonomyRefs: [],
|
||
attachmentRefs: [],
|
||
seo: {},
|
||
knownFields: {},
|
||
extras: {},
|
||
rawLocation: `wxr://item/${spec.sourceId}`,
|
||
}
|
||
if (spec.slug !== undefined) row.slug = spec.slug
|
||
if (spec.title !== undefined) row.title = spec.title
|
||
if (spec.bodyHtml !== undefined) row.bodyHtml = spec.bodyHtml
|
||
if (spec.sourceUrl !== undefined) row.sourceUrl = spec.sourceUrl
|
||
return { kind: 'row', row, issues: [], sanitizeDiffs: [], changed: false }
|
||
}
|
||
|
||
/** 本文件写入的行清场：FK 反序（两张不可变表走触发器开关封装） */
|
||
async function cleanupSiteRows(db: Db, siteId: string): Promise<void> {
|
||
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}'`)
|
||
}
|
||
|
||
/** mini 登记（各用例公共前置）：register → {siteId, wxr 快照} */
|
||
async function registerMini(db: Db, work: string) {
|
||
const key = `sfcontent-${rand()}`
|
||
const reg = await registerSource(db, {
|
||
siteKey: key,
|
||
wxrPath: WXR_FIXTURE,
|
||
allowedRoots: [FIXTURES_ROOT, work],
|
||
storageRoot: join(work, 'storage'),
|
||
reportRoot: join(work, 'artifacts', 'register'),
|
||
})
|
||
return { key, reg, wxr: reg.snapshots.find((s) => s.kind === 'wxr')! }
|
||
}
|
||
|
||
/** 导入报告读取（import.service 的报告落盘纪律：终态先行落文件） */
|
||
async function readReport(work: string, jobId: string): Promise<{
|
||
status: string
|
||
counts: Record<string, number>
|
||
issues: string[]
|
||
}> {
|
||
return JSON.parse(await readFile(join(work, 'artifacts', 'import', jobId, 'reports', 'import.json'), 'utf8'))
|
||
}
|
||
|
||
describe('内容写入器（稳定身份 + checksum skip）', () => {
|
||
it('mini 首导全链：typeKey 映射 + entry_locales 双指针（published NULL）+ revision（checksum 独立对拍/data 钉死结构）+ 5 source_records + counts', async () => {
|
||
await withTestDb(async (db) => {
|
||
const work = await mkdtemp(join(tmpdir(), 'sf-content-'))
|
||
try {
|
||
const events = await collectMiniEvents()
|
||
// 流形状 sanity：5 row（3 内容 + 2 attachment）+ 2 term + 0 unsupported/error
|
||
expect(rowsOf(events)).toHaveLength(5)
|
||
expect(events.filter((e) => e.kind === 'term')).toHaveLength(2)
|
||
expect(events.filter((e) => e.kind !== 'row' && e.kind !== 'term')).toStrictEqual([])
|
||
const contentRows = rowsOf(events).filter((r) => r.sourceType !== 'wp:attachment')
|
||
const attachmentRows = rowsOf(events).filter((r) => r.sourceType === 'wp:attachment')
|
||
expect(contentRows).toHaveLength(3)
|
||
expect(attachmentRows).toHaveLength(2)
|
||
|
||
const { key, reg, wxr } = await registerMini(db, work)
|
||
try {
|
||
const result = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: join(work, 'artifacts', 'import'),
|
||
events: eventsOf(events),
|
||
pipeline: createContentWriter(),
|
||
})
|
||
expect(result.reused).toBe(false)
|
||
expect(result.status).toBe('succeeded')
|
||
|
||
// content_entries：typeKey 映射（前置事实 11）+ lifecycle active
|
||
const entries = (await db.execute(`select type_key, lifecycle_status from content_entries
|
||
where site_id = '${reg.siteId}' order by type_key`)) as any
|
||
expect(plain(entries)).toStrictEqual([
|
||
{ type_key: 'basic_page', lifecycle_status: 'active' },
|
||
{ type_key: 'news_article', lifecycle_status: 'active' },
|
||
{ type_key: 'product', lifecycle_status: 'active' },
|
||
])
|
||
|
||
// source_records：5 行；内容行挂 entry、checksum=独立重算；attachment 行
|
||
// content_entry_id NULL、parse_status='imported'（前置事实 11：不建 entry）
|
||
const recs = (await db.execute(`select source_type, source_id, locale, parse_status,
|
||
content_entry_id is null as no_entry, normalized_checksum, source_url, raw_location
|
||
from source_records where site_id = '${reg.siteId}' and source_snapshot_id = '${wxr.id}'
|
||
order by source_id`)) as any
|
||
expect(recs).toHaveLength(5)
|
||
for (const row of contentRows) {
|
||
const rec = recs.find((r: any) => r.source_id === row.sourceId)!
|
||
expect(rec.no_entry).toBe(false)
|
||
expect(rec.parse_status).toBe('imported')
|
||
expect(rec.locale).toBe('en')
|
||
expect(rec.normalized_checksum).toBe(expectedChecksum(row))
|
||
expect(rec.source_url).toBe(row.sourceUrl ?? null)
|
||
}
|
||
for (const row of attachmentRows) {
|
||
const rec = recs.find((r: any) => r.source_id === row.sourceId)!
|
||
expect(rec.no_entry).toBe(true)
|
||
expect(rec.parse_status).toBe('imported')
|
||
expect(rec.locale).toBe('en')
|
||
}
|
||
|
||
// revision + 双指针（经 current 指针 join）：source='import'、schema_version=
|
||
// UNIVERSAL_ROW_VERSION、data 钉死结构、published NULL（导入不发布）
|
||
const cur = (await db.execute(`select r.source_id, e.type_key, el.slug, el.locale as locale_of_row,
|
||
el.current_revision_id, el.published_revision_id, cr.checksum, cr.data, cr.source_kind,
|
||
cr.schema_version, cr.locale as locale_of_rev
|
||
from source_records r
|
||
join content_entries e on e.id = r.content_entry_id
|
||
join entry_locales el on el.content_entry_id = e.id and el.locale = r.locale
|
||
join content_revisions cr on cr.id = el.current_revision_id
|
||
where r.site_id = '${reg.siteId}' and r.content_entry_id is not null
|
||
order by r.source_id`)) as any
|
||
expect(cur).toHaveLength(3)
|
||
for (const row of contentRows) {
|
||
const c = cur.find((x: any) => x.source_id === row.sourceId)!
|
||
expect(c.type_key).toBe(EXPECTED_TYPE_KEYS[row.sourceType])
|
||
expect(c.slug).toBe(row.slug)
|
||
expect(c.locale_of_row).toBe('en')
|
||
expect(c.locale_of_rev).toBe('en')
|
||
expect(c.current_revision_id).toBeTruthy()
|
||
expect(c.published_revision_id).toBe(null)
|
||
expect(c.source_kind).toBe('import')
|
||
expect(c.schema_version).toBe(UNIVERSAL_ROW_VERSION)
|
||
expect(c.checksum).toBe(expectedChecksum(row))
|
||
expect(jsonbNorm(c.data)).toStrictEqual(jsonbNorm(expectedRevisionData(row)))
|
||
}
|
||
|
||
// 报告：counts 四键 + issues 空（正常路径）
|
||
const report = await readReport(work, result.jobId)
|
||
expect(report.status).toBe('succeeded')
|
||
expect(report.counts).toStrictEqual({
|
||
contentEntries: 3,
|
||
revisionsCreated: 3,
|
||
sourceRecords: 5,
|
||
skipped: 0,
|
||
})
|
||
expect(report.issues).toStrictEqual([])
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('二导同快照（换幂等键）：0 新修订 0 新 entry（checksum skip）、counts.skipped=3', async () => {
|
||
await withTestDb(async (db) => {
|
||
const work = await mkdtemp(join(tmpdir(), 'sf-content-'))
|
||
try {
|
||
const events = await collectMiniEvents()
|
||
const { key, reg, wxr } = await registerMini(db, work)
|
||
try {
|
||
const base = {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: join(work, 'artifacts', 'import'),
|
||
pipeline: createContentWriter(),
|
||
}
|
||
const first = await createImportJob(db, { ...base, events: eventsOf(events) })
|
||
expect(first.status).toBe('succeeded')
|
||
|
||
// 同流重跑：显式幂等键（默认键会复用既有 job 不重跑——Task 1 语义，此处钉写入器幂等）
|
||
const again = await createImportJob(db, {
|
||
...base,
|
||
idempotencyKey: `sfcontent-rerun-${rand()}`,
|
||
events: eventsOf(events),
|
||
})
|
||
expect(again.reused).toBe(false)
|
||
expect(again.status).toBe('succeeded')
|
||
|
||
// 全表不增行
|
||
const after = (await db.execute(`select
|
||
(select count(*)::int from content_entries where site_id = '${reg.siteId}') as entries,
|
||
(select count(*)::int from entry_locales where site_id = '${reg.siteId}') as locales,
|
||
(select count(*)::int from content_revisions where site_id = '${reg.siteId}') as revisions,
|
||
(select count(*)::int from source_records where site_id = '${reg.siteId}') as records`)) as any
|
||
expect(plain(after)[0]).toStrictEqual({ entries: 3, locales: 3, revisions: 3, records: 5 })
|
||
|
||
// skipped=3：三内容行 checksum 命中（attachment 命中不计——媒体域是 Task 4 主场）
|
||
const report = await readReport(work, again.jobId)
|
||
expect(report.counts).toStrictEqual({
|
||
contentEntries: 0,
|
||
revisionsCreated: 0,
|
||
sourceRecords: 0,
|
||
skipped: 3,
|
||
})
|
||
expect(report.issues).toStrictEqual([])
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('跨快照（变体文件、业务内容不变）：0 新修订 0 新 entry（五列稳定身份）+ 本快照 source_records 补行', async () => {
|
||
await withTestDb(async (db) => {
|
||
const work = await mkdtemp(join(tmpdir(), 'sf-content-xsnap-'))
|
||
try {
|
||
const { key, reg, wxr } = await registerMini(db, work)
|
||
try {
|
||
const first = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: join(work, 'artifacts', 'import'),
|
||
events: runWxrAdapter(createReadStream(WXR_FIXTURE), { mapConfig: { locale: 'en' } }),
|
||
pipeline: createContentWriter(),
|
||
})
|
||
expect(first.status).toBe('succeeded')
|
||
|
||
// 变体文件：mini 复制 + 通道内注释（字节不同 → 新 snapshot；item 业务内容不变）
|
||
const variantPath = join(work, 'mini-v2.xml')
|
||
const xml = await readFile(WXR_FIXTURE, 'utf8')
|
||
await writeFile(variantPath, xml.replace('<channel>', '<channel>\n <!-- v2: byte-different, business-identical -->'))
|
||
const reg2 = await registerSource(db, {
|
||
siteKey: key,
|
||
wxrPath: variantPath,
|
||
allowedRoots: [FIXTURES_ROOT, work],
|
||
storageRoot: join(work, 'storage'),
|
||
reportRoot: join(work, 'artifacts', 'register'),
|
||
})
|
||
const wxr2 = reg2.snapshots.find((s) => s.kind === 'wxr')!
|
||
expect(wxr2.id).not.toBe(wxr.id)
|
||
expect(reg2.connectionId).toBe(reg.connectionId) // 同连接——五列稳定身份的前提
|
||
|
||
const second = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxr2.id,
|
||
reportRoot: join(work, 'artifacts', 'import'),
|
||
events: runWxrAdapter(createReadStream(variantPath), { mapConfig: { locale: 'en' } }),
|
||
pipeline: createContentWriter(),
|
||
})
|
||
expect(second.status).toBe('succeeded')
|
||
|
||
// 0 新 entry、0 新修订；source_records 每快照各行（六列身份含 snapshot_id）：5+5
|
||
const after = (await db.execute(`select
|
||
(select count(*)::int from content_entries where site_id = '${reg.siteId}') as entries,
|
||
(select count(*)::int from content_revisions where site_id = '${reg.siteId}') as revisions,
|
||
(select count(*)::int from source_records where site_id = '${reg.siteId}') as records`)) as any
|
||
expect(plain(after)[0]).toStrictEqual({ entries: 3, revisions: 3, records: 10 })
|
||
|
||
// 新快照的内容行挂**同一批 entry**（稳定身份跨快照复用）
|
||
const linkage = (await db.execute(`select r.source_id, r.source_snapshot_id, r.content_entry_id
|
||
from source_records r
|
||
where r.site_id = '${reg.siteId}' and r.source_snapshot_id in ('${wxr.id}', '${wxr2.id}')
|
||
and r.content_entry_id is not null
|
||
order by r.source_id, r.source_snapshot_id`)) as any
|
||
expect(linkage).toHaveLength(6)
|
||
for (const sourceId of ['101', '111', '121']) {
|
||
const pair = linkage.filter((l: any) => l.source_id === sourceId)
|
||
expect(pair).toHaveLength(2)
|
||
expect(pair[0].content_entry_id).toBe(pair[1].content_entry_id)
|
||
}
|
||
|
||
const report = await readReport(work, second.jobId)
|
||
expect(report.counts).toStrictEqual({
|
||
contentEntries: 0,
|
||
revisionsCreated: 0,
|
||
sourceRecords: 5,
|
||
skipped: 3,
|
||
})
|
||
expect(report.issues).toStrictEqual([])
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('内容变化（改 knownFields.title）：新修订 + current 前移 + published 仍 NULL（ADR-010）+ 旧修订原样', async () => {
|
||
await withTestDb(async (db) => {
|
||
const work = await mkdtemp(join(tmpdir(), 'sf-content-chg-'))
|
||
try {
|
||
const events = await collectMiniEvents()
|
||
const row101 = rowsOf(events).find((r) => r.sourceId === '101')!
|
||
const { key, reg, wxr } = await registerMini(db, work)
|
||
try {
|
||
const base = {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: join(work, 'artifacts', 'import'),
|
||
pipeline: createContentWriter(),
|
||
}
|
||
const first = await createImportJob(db, { ...base, events: eventsOf(events) })
|
||
expect(first.status).toBe('succeeded')
|
||
const before = (await db.execute(`select el.content_entry_id as entry, el.current_revision_id as cur,
|
||
el.published_revision_id as pub
|
||
from entry_locales el join source_records r on r.content_entry_id = el.content_entry_id
|
||
where r.site_id = '${reg.siteId}' and r.source_id = '101'`)) as any
|
||
expect(before).toHaveLength(1)
|
||
expect(before[0].pub).toBe(null)
|
||
|
||
// 手构内容变化：WXR 标题变更在适配器产物上是 knownFields.title 变化（mapItem 提升
|
||
// 纪律：title 留在 knownFields）
|
||
const changed = withRowMutated(events, '101', (row) => {
|
||
row.knownFields.title = 'Fixture News A (edited)'
|
||
})
|
||
const changed101 = rowsOf(changed).find((r) => r.sourceId === '101')!
|
||
expect(changed101.knownFields.title).not.toBe(row101.knownFields.title)
|
||
const second = await createImportJob(db, {
|
||
...base,
|
||
idempotencyKey: `sfcontent-chg-${rand()}`,
|
||
events: eventsOf(changed),
|
||
})
|
||
expect(second.status).toBe('succeeded')
|
||
|
||
// current 前移、published 仍 NULL（导入不发布）、同 entry（不建新）
|
||
const after = (await db.execute(`select el.content_entry_id as entry, el.current_revision_id as cur,
|
||
el.published_revision_id as pub
|
||
from entry_locales el join source_records r on r.content_entry_id = el.content_entry_id
|
||
where r.site_id = '${reg.siteId}' and r.source_id = '101'`)) as any
|
||
expect(after[0].entry).toBe(before[0].entry)
|
||
expect(after[0].cur).not.toBe(before[0].cur)
|
||
expect(after[0].pub).toBe(null)
|
||
|
||
// 该 entry 两修订（旧原样——不可变触发器下写入路径纯 INSERT；新旧 checksum 各归位）
|
||
const revs = (await db.execute(`select checksum from content_revisions
|
||
where site_id = '${reg.siteId}' and content_entry_id = '${before[0].entry}'
|
||
order by created_at`)) as any
|
||
expect(plain(revs)).toStrictEqual([
|
||
{ checksum: expectedChecksum(row101) },
|
||
{ checksum: expectedChecksum(changed101) },
|
||
])
|
||
expect((await db.execute(`select count(*)::int as n from content_entries
|
||
where site_id = '${reg.siteId}'`)) as any).toMatchObject([{ n: 3 }])
|
||
|
||
// 其余两行未变 → skipped=2；一行新修订
|
||
const report = await readReport(work, second.jobId)
|
||
expect(report.counts).toStrictEqual({
|
||
contentEntries: 0,
|
||
revisionsCreated: 1,
|
||
sourceRecords: 0,
|
||
skipped: 2,
|
||
})
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('permalink 变化（改 sourceUrl）：不建新 entry（身份键不含 URL）+ 新修订（sourceUrl 在哈希内）+ source_url 更新', async () => {
|
||
await withTestDb(async (db) => {
|
||
const work = await mkdtemp(join(tmpdir(), 'sf-content-perma-'))
|
||
try {
|
||
const events = await collectMiniEvents()
|
||
const { key, reg, wxr } = await registerMini(db, work)
|
||
try {
|
||
const base = {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: join(work, 'artifacts', 'import'),
|
||
pipeline: createContentWriter(),
|
||
}
|
||
const first = await createImportJob(db, { ...base, events: eventsOf(events) })
|
||
expect(first.status).toBe('succeeded')
|
||
const before = (await db.execute(`select r.content_entry_id as entry, r.source_url
|
||
from source_records r
|
||
where r.site_id = '${reg.siteId}' and r.source_id = '101'`)) as any
|
||
|
||
// 手构 permalink 变化：sourceUrl 进 route_source_values 哈希域（前置事实 6）
|
||
const newUrl = 'https://fixture.example/news/fixture-news-a-renamed.html'
|
||
const changed = withRowMutated(events, '101', (row) => {
|
||
row.sourceUrl = newUrl
|
||
})
|
||
const changed101 = rowsOf(changed).find((r) => r.sourceId === '101')!
|
||
const second = await createImportJob(db, {
|
||
...base,
|
||
idempotencyKey: `sfcontent-perma-${rand()}`,
|
||
events: eventsOf(changed),
|
||
})
|
||
expect(second.status).toBe('succeeded')
|
||
|
||
// 不建新 entry（count 仍 3）；source_records 同行 source_url 前移 + checksum 前移
|
||
const entries = (await db.execute(`select count(*)::int as n from content_entries
|
||
where site_id = '${reg.siteId}'`)) as any
|
||
expect(entries[0].n).toBe(3)
|
||
const rec101 = (await db.execute(`select content_entry_id, source_url, normalized_checksum
|
||
from source_records where site_id = '${reg.siteId}' and source_id = '101'`)) as any
|
||
expect(rec101[0].content_entry_id).toBe(before[0].entry) // 同 entry——稳定身份
|
||
expect(rec101[0].source_url).toBe(newUrl)
|
||
expect(rec101[0].normalized_checksum).toBe(expectedChecksum(changed101))
|
||
|
||
// 新修订（permalink 产新修订不建新 entry）；其余两行 skipped
|
||
const revs = (await db.execute(`select count(*)::int as n from content_revisions
|
||
where site_id = '${reg.siteId}' and content_entry_id = '${before[0].entry}'`)) as any
|
||
expect(revs[0].n).toBe(2)
|
||
const report = await readReport(work, second.jobId)
|
||
expect(report.counts).toStrictEqual({
|
||
contentEntries: 0,
|
||
revisionsCreated: 1,
|
||
sourceRecords: 0,
|
||
skipped: 2,
|
||
})
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('unsupported/error 也写 source_records（前置事实 12）：parse_status/locale/无 entry；error 无 postId 以 rawLocation 回退', async () => {
|
||
await withTestDb(async (db) => {
|
||
const work = await mkdtemp(join(tmpdir(), 'sf-content-unsup-'))
|
||
try {
|
||
const { key, reg, wxr } = await registerMini(db, work)
|
||
try {
|
||
const events: AdapterEvent[] = [
|
||
{ kind: 'unsupported', sourceType: 'wp:nav_menu_item', reason: 'post_type 未配置映射', rawLocation: 'wxr://item/7', postId: '68' },
|
||
{ kind: 'error', sourceType: 'wp:post', issue: 'post_id 缺失或非法', rawLocation: 'wxr://item/8' },
|
||
]
|
||
const result = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: join(work, 'artifacts', 'import'),
|
||
events: eventsOf(events),
|
||
pipeline: createContentWriter(),
|
||
})
|
||
expect(result.status).toBe('succeeded')
|
||
|
||
// 两行：locale=站点默认、content_entry_id NULL；error 无 postId → source_id 用
|
||
// rawLocation 回退（不硬造）
|
||
const recs = (await db.execute(`select source_type, source_id, locale, parse_status,
|
||
content_entry_id is null as no_entry
|
||
from source_records where site_id = '${reg.siteId}' order by source_id`)) as any
|
||
expect(plain(recs)).toStrictEqual([
|
||
{ source_type: 'wp:nav_menu_item', source_id: '68', locale: 'en', parse_status: 'unsupported', no_entry: true },
|
||
{ source_type: 'wp:post', source_id: 'wxr://item/8', locale: 'en', parse_status: 'failed', no_entry: true },
|
||
])
|
||
// 零 entry 零修订
|
||
expect(((await db.execute(`select count(*)::int as n from content_entries
|
||
where site_id = '${reg.siteId}'`)) as any)[0].n).toBe(0)
|
||
expect(((await db.execute(`select count(*)::int as n from content_revisions
|
||
where site_id = '${reg.siteId}'`)) as any)[0].n).toBe(0)
|
||
|
||
const report = await readReport(work, result.jobId)
|
||
expect(report.counts).toStrictEqual({
|
||
contentEntries: 0,
|
||
revisionsCreated: 0,
|
||
sourceRecords: 2,
|
||
skipped: 0,
|
||
unsupportedRecords: 1,
|
||
failedRecords: 1,
|
||
})
|
||
// unsupported/error 是"有明确结果"的预期输入（source_records 行即审计），非 issue
|
||
expect(report.issues).toStrictEqual([])
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
|
||
it('locale 校验（前置事实 8）：row.locale ≠ site.default_locale → issue + 跳过（对照行不断流）', async () => {
|
||
await withTestDb(async (db) => {
|
||
const work = await mkdtemp(join(tmpdir(), 'sf-content-locale-'))
|
||
try {
|
||
const { key, reg, wxr } = await registerMini(db, work)
|
||
try {
|
||
const events: AdapterEvent[] = [
|
||
rowEvent({ sourceId: '301', locale: 'fr', slug: 'post-301', bodyHtml: '<p>bonjour</p>' }),
|
||
rowEvent({ sourceId: '302', slug: 'post-302', bodyHtml: '<p>ok</p>' }),
|
||
]
|
||
const result = await createImportJob(db, {
|
||
siteKey: key,
|
||
snapshotId: wxr.id,
|
||
reportRoot: join(work, 'artifacts', 'import'),
|
||
events: eventsOf(events),
|
||
pipeline: createContentWriter(),
|
||
})
|
||
expect(result.status).toBe('succeeded_with_issues')
|
||
|
||
// 只落对照行 302（en）；301 无任何行
|
||
const recs = (await db.execute(`select source_id from source_records
|
||
where site_id = '${reg.siteId}' order by source_id`)) as any
|
||
expect(plain(recs)).toStrictEqual([{ source_id: '302' }])
|
||
expect(((await db.execute(`select count(*)::int as n from content_entries
|
||
where site_id = '${reg.siteId}'`)) as any)[0].n).toBe(1)
|
||
|
||
const report = await readReport(work, result.jobId)
|
||
expect(report.counts).toStrictEqual({
|
||
contentEntries: 1,
|
||
revisionsCreated: 1,
|
||
sourceRecords: 1,
|
||
skipped: 1,
|
||
})
|
||
expect(report.issues).toHaveLength(1)
|
||
expect(report.issues[0]).toContain('301')
|
||
expect(report.issues[0]).toContain('fr')
|
||
expect(report.issues[0]).toContain('en')
|
||
} finally {
|
||
await cleanupSiteRows(db, reg.siteId)
|
||
}
|
||
} finally {
|
||
await rm(work, { recursive: true, force: true })
|
||
}
|
||
})
|
||
})
|
||
})
|