From a57da2f5ca6009dcf09d140e8fc07f6fb11bf1a9 Mon Sep 17 00:00:00 2001 From: 18628386721 <18628386721@163.com> Date: Tue, 28 Jul 2026 16:45:11 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AB=AF=E5=88=B0=E7=AB=AF=E9=9B=86=E6=88=90?= =?UTF-8?q?=E6=B5=8B=E8=AF=95:=E6=95=B4=E6=9D=A1=E6=B5=81=E5=9C=A8?= =?UTF-8?q?=E7=9C=9F=E5=BA=93=E4=B8=8A=E6=B5=8B=E6=AD=BB=20=E2=80=94?= =?UTF-8?q?=E2=80=94=20=E5=85=AB=E6=9D=A1=E5=88=86=E6=94=AF=E7=9F=A9?= =?UTF-8?q?=E9=98=B5=E5=85=A8=E7=BB=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 真容器 + 真迁移 + 真种子;推进不靠定时器,靠手动泵 (dispatchOnce + advanceOnce 循环到静止)—— 每一拍都是生产同一份代码,零 mock。 **库绝不回落到开发库**:E2E 要清库播种,碰开发库就是清人数据。次序: JCAI_TEST_DATABASE_URL → 在开发实例上另建独立库 jcai_it_flow → Testcontainers; 都不可得则跳过并告警(skipIf,跳过可见,不变成沉默的通过)。 每次从零:四 schema drop-cascade 重建 → 迁移 → 种子 —— E2E 的前提是起点确定。 八条(本机 PG 实测 15s): 主路径 不选型不寻源 → A4 → A5 → G3 → A6 选成交 → end_won,completed G1 需选型 进 A2,A2 办结(不再寻源)→ 经 G2/G2J 进 A4 —— 异或汇合不等的回归 G2 需寻源 进 A3 → G2J → A4 G3 驳回 走图上的驳回边:A4#2 重开,A4#1 released 留痕不动 G4 有异议 A4#2 重开,流仍 running G4 拒绝 A7 归档 → end_arch,completed 恰选一口 A6 放行不选口 → 同步拒收、工单还活着;补选 won 后 completed 零死路 全程 route-dead-end 事件 0 条 这套矩阵是后续一切编排层改动(包括网关去节点化)的护栏: 改完跑一遍,九个网关分支的行为有没有变,15 秒内见分晓。 typecheck ✓ lint ✓ depcruise ✓ 506 测试 ✓ api:check ✓ Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m --- .../__tests__/unified-quote-flow.it.test.ts | 280 ++++++++++++++++++ 1 file changed, 280 insertions(+) create mode 100644 apps/api/src/app/__tests__/unified-quote-flow.it.test.ts diff --git a/apps/api/src/app/__tests__/unified-quote-flow.it.test.ts b/apps/api/src/app/__tests__/unified-quote-flow.it.test.ts new file mode 100644 index 0000000..e08f169 --- /dev/null +++ b/apps/api/src/app/__tests__/unified-quote-flow.it.test.ts @@ -0,0 +1,280 @@ +/** + * 「统一报价」端到端集成测试 —— 把整条流**在真库上**测死。 + * + * 起真容器、跑真迁移、播真种子;推进不靠定时器,靠手动泵 + * (`dispatchOnce` + `advanceOnce` 循环到静止)—— 每一拍都是生产同一份代码。 + * 系统内部机制零 mock(README);唯一的「假」是占位活动本身,而这正是被测物: + * 编排层要在活动只有契约的情况下把流转走对。 + * + * **库的选择绝不回落到开发库**:E2E 要 seed 与断言全库状态,碰开发库就是清人数据。 + * 次序:JCAI_TEST_DATABASE_URL → 在开发实例上**另建**独立库 jcai_it_flow → Testcontainers; + * 都不可得则跳过并告警 —— 跳过必须可见,不能变成沉默的通过。 + */ +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { Pool } from 'pg'; +import { drizzle } from 'drizzle-orm/node-postgres'; +import { migrate } from 'drizzle-orm/node-postgres/migrator'; +import { existsSync } from 'node:fs'; +import { dirname, join, resolve } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import type { TaskId, WorkOrderId } from '@contracts'; +import { loadConfig } from '../config.ts'; +import { createContainer } from '../container.ts'; +import type { Container } from '../container.ts'; +import { seedMvp } from '../seed-mvp.ts'; +import { advanceOnce } from '../workers/task-advancer.ts'; +import { dispatchOnce } from '../workers/dispatcher.ts'; +import { createConsumers } from '../workers/consumers.ts'; + +/* ── 库:独立于开发库 ─────────────────────────────────────────────────── */ + +const IT_DB = 'jcai_it_flow'; + +async function provisionDatabase(): Promise<{ url: string; stop(): Promise } | undefined> { + const explicit = process.env['JCAI_TEST_DATABASE_URL']; + if (explicit !== undefined) return { url: explicit, stop: async () => undefined }; + + // 在开发实例上另建一个库 —— 同一个 PG,不同 database,动不到开发数据 + const base = process.env['DATABASE_URL'] ?? 'postgres://jcai:jcai@localhost:54329/jcai'; + try { + const admin = new Pool({ connectionString: base, max: 1, connectionTimeoutMillis: 1500 }); + const exists = await admin.query(`select 1 from pg_database where datname = $1`, [IT_DB]); + if (exists.rowCount === 0) await admin.query(`create database ${IT_DB}`); + await admin.end(); + return { url: base.replace(/\/[^/]*$/, `/${IT_DB}`), stop: async () => undefined }; + } catch { + /* 开发实例连不上,试 Testcontainers */ + } + try { + const { PostgreSqlContainer } = await import('@testcontainers/postgresql'); + const started = await new PostgreSqlContainer('postgres:16-alpine').start(); + return { url: started.getConnectionUri(), stop: () => started.stop().then(() => undefined) }; + } catch (error) { + process.stderr.write( + `⚠️ unified-quote E2E 跳过:连不上开发实例也起不了 Testcontainers(${error instanceof Error ? error.message : String(error)})\n`, + ); + return undefined; + } +} + +function migrationsFolder(): string { + let dir = dirname(fileURLToPath(import.meta.url)); + for (let i = 0; i < 8; i += 1) { + const candidate = join(dir, 'db', 'migrations'); + if (existsSync(candidate)) return candidate; + const parent = resolve(dir, '..'); + if (parent === dir) break; + dir = parent; + } + throw new Error('找不到 db/migrations'); +} + +// 顶层 await:可用性必须在收集期定下来,否则 it.skip 的判定早于 beforeAll +const database = await provisionDatabase(); + +let pool: Pool | undefined; +let container: Container; +let chaihao: string; + +beforeAll(async () => { + if (database === undefined) return; + pool = new Pool({ connectionString: database.url, max: 4 }); + + // 每次从零:四 schema 全重建 → 迁移 → 种子。E2E 的前提是「起点确定」。 + for (const schema of ['sys', 'ledger', 'mastra', 'queue']) { + await pool.query(`drop schema if exists "${schema}" cascade`); + await pool.query(`create schema "${schema}"`); + } + await migrate(drizzle(pool), { migrationsFolder: migrationsFolder(), migrationsSchema: 'sys' }); + + // 配置指向测试库 —— 经 loadConfig 的 env 形参,不污染 process.env + container = await createContainer(loadConfig({ ...process.env, DATABASE_URL: database.url })); + await container.org.ensureSeeded(); + await seedMvp(container); + + const { staff } = await container.org.listStaff(); + chaihao = staff.find((s) => s.account === 'Chaihao')!.id; +}, 180_000); + +afterAll(async () => { + if (database === undefined) return; + await container?.close(); + await pool?.end(); + await database.stop(); +}); + +/* ── 推进与走单的工具 ─────────────────────────────────────────────────── */ + +const consumers = createConsumers(); + +/** 手动泵:投递 + 推进循环到静止 —— 测试不靠定时器,每拍都可断言 */ +async function pump(): Promise { + for (let i = 0; i < 60; i += 1) { + const delivered = await dispatchOnce({ container, consumers }); + const advanced = await advanceOnce({ container }); + if (delivered === 0 && advanced === 0) return; + } + throw new Error('泵 60 轮未静止 —— 流转在空转'); +} + +async function start(ref: string): Promise { + const record = await container.process.start({ + packageId: 'unified-quote', + subject: ref, + source: { kind: 'registration', ref: 'it', by: chaihao }, + data: { customerCode: 'C014', rfqRef: ref, subject: ref, body: 'it' }, + }); + await pump(); + return record.id; +} + +/** 该流当前在途任务(恰一个 —— 占位期无并行支) */ +async function currentTask(workflowId: string): Promise { + const rows = await pool!.query<{ id: string }>( + `select id from sys.task where workflow_id = $1 and terminal is null order by started_at desc limit 1`, + [workflowId], + ); + expect(rows.rowCount, `流 ${workflowId} 应有在途任务`).toBe(1); + return rows.rows[0]!.id as TaskId; +} + +/** 答掉当前任务的收口验收(占位活动恰一张) */ +async function acceptCurrent( + workflowId: string, + values: Record = {}, + selectedPorts: Record = {}, +): Promise { + const taskId = await currentTask(workflowId); + const open = await container.humanTool.openWorkOrders(taskId); + const order = open.find((o) => o.mode === 'acceptance'); + expect(order, '应有待答验收单').toBeDefined(); + await container.humanTool.replyWorkOrder( + { mode: 'acceptance', workOrderId: order!.id as WorkOrderId, verdict: 'pass', values, selectedPorts }, + chaihao as never, + ); + await pump(); +} + +async function decideApproval(workflowId: string, verdict: 'pass' | 'reject'): Promise { + await container.process.onApprovalVerdict(workflowId, 'A5', verdict, { by: chaihao }); + await pump(); +} + +/** 节点实例摘要:`A1#1=released` —— 断言拓扑走向的通用形状 */ +async function seq(workflowId: string): Promise { + const instances = await container.process.instancesOf(workflowId); + return instances.map((i) => `${i.nodeId}#${i.instance}=${i.terminal ?? '…'}`); +} + +async function flowState(workflowId: string): Promise { + return (await container.process.getWorkflow(workflowId)).state; +} + +/** 走完 A1(按旗选路)→ … → 停在 A6 的验收单前 */ +async function walkToA6(ref: string): Promise { + const wf = await start(ref); + await acceptCurrent(wf, { needAlternative: 'false', needSourcing: 'false' }); + await acceptCurrent(wf); // A4 + await decideApproval(wf, 'pass'); // A5 → G3 通过 → A6 + return wf; +} + +/* ── 用例 ─────────────────────────────────────────────────────────────── */ + +describe.skipIf(database === undefined)('统一报价 · 端到端(真库真容器)', () => { + it('主路径:不选型不寻源 → A4 → A5 → G3 → A6 选成交 → end_won,流 completed', async () => { + const wf = await walkToA6('E2E-WON'); + await acceptCurrent(wf, {}, { 议价结果: 'won' }); + + const s = await seq(wf); + expect(s).toEqual(['A1#1=released', 'A4#1=released', 'A5#1=released', 'A6#1=released', 'end_won#1=released']); + expect(await flowState(wf)).toBe('completed'); + }, 120_000); + + it('G1 需选型 → 进 A2;A2 办结(不再寻源)→ 过 G2/G2J 进 A4', async () => { + const wf = await start('E2E-ALT'); + await acceptCurrent(wf, { needAlternative: 'true', needSourcing: 'false' }); + + let s = await seq(wf); + expect(s).toContain('A2#1=…'); + expect(s.some((x) => x.startsWith('A3#'))).toBe(false); + + await acceptCurrent(wf, { needSourcing: 'false' }); // A2:仍不寻源 + s = await seq(wf); + expect(s).toContain('A2#1=released'); + expect(s).toContain('A4#1=…'); // 经 G2 → G2J(异或汇合不等)→ A4 + }, 120_000); + + it('G2 需寻源 → 进 A3;A3 办结 → 经 G2J 进 A4', async () => { + const wf = await start('E2E-SRC'); + await acceptCurrent(wf, { needAlternative: 'false', needSourcing: 'true' }); + + let s = await seq(wf); + expect(s).toContain('A3#1=…'); + expect(s.some((x) => x.startsWith('A2#'))).toBe(false); + + await acceptCurrent(wf); // A3 + s = await seq(wf); + expect(s).toContain('A3#1=released'); + expect(s).toContain('A4#1=…'); + }, 120_000); + + it('G3 审批驳回 → 走图上的驳回边,重开 A4 新实例(旧实例留痕不动)', async () => { + const wf = await start('E2E-REJ'); + await acceptCurrent(wf, { needAlternative: 'false', needSourcing: 'false' }); + await acceptCurrent(wf); // A4#1 + await decideApproval(wf, 'reject'); + + const s = await seq(wf); + expect(s).toContain('A4#1=released'); // 旧实例不复活、不改写 + expect(s).toContain('A4#2=…'); // 重开新实例 + }, 120_000); + + it('G4 有异议 → 回边重开 A4 新实例', async () => { + const wf = await walkToA6('E2E-REV'); + await acceptCurrent(wf, {}, { 议价结果: 'revise' }); + + const s = await seq(wf); + expect(s).toContain('A6#1=released'); + expect(s).toContain('A4#2=…'); + expect(await flowState(wf)).toBe('running'); + }, 120_000); + + it('G4 拒绝/超时 → A7 归档 → end_arch,流 completed', async () => { + const wf = await walkToA6('E2E-LOST'); + await acceptCurrent(wf, {}, { 议价结果: 'lost' }); + await acceptCurrent(wf); // A7 + + const s = await seq(wf); + expect(s).toContain('A7#1=released'); + expect(s).toContain('end_arch#1=released'); + expect(await flowState(wf)).toBe('completed'); + }, 120_000); + + it('恰选一口是收单规矩:A6 放行不选口 → 同步 400 拒收,工单还活着', async () => { + const wf = await walkToA6('E2E-PORT'); + const taskId = await currentTask(wf); + const order = (await container.humanTool.openWorkOrders(taskId)).find((o) => o.mode === 'acceptance')!; + + await expect( + container.humanTool.replyWorkOrder( + { mode: 'acceptance', workOrderId: order.id as WorkOrderId, verdict: 'pass', values: {}, selectedPorts: {} }, + chaihao as never, + ), + ).rejects.toThrow(/议价结果/); + + // 工单未被吃掉 —— 补选就能继续 + const stillOpen = await container.humanTool.openWorkOrders(taskId); + expect(stillOpen.some((o) => o.id === order.id)).toBe(true); + + await acceptCurrent(wf, {}, { 议价结果: 'won' }); + expect(await flowState(wf)).toBe('completed'); + }, 120_000); + + it('全程零死路:route-dead-end 事件一条都不该有', async () => { + const dead = await pool!.query( + `select count(*)::int as n from sys.event where correlation->'cause'->>'ref' = 'route-dead-end'`, + ); + expect(dead.rows[0].n).toBe(0); + }); +});