forked from ChaiHao/BPOS-MVP
端到端集成测试:整条流在真库上测死 —— 八条分支矩阵全绿
真容器 + 真迁移 + 真种子;推进不靠定时器,靠手动泵 (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) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m
This commit is contained in:
Claude Opus 5
@@ -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<void> } | 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<void> {
|
||||||
|
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<string> {
|
||||||
|
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<TaskId> {
|
||||||
|
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<string, string> = {},
|
||||||
|
selectedPorts: Record<string, string> = {},
|
||||||
|
): Promise<void> {
|
||||||
|
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<void> {
|
||||||
|
await container.process.onApprovalVerdict(workflowId, 'A5', verdict, { by: chaihao });
|
||||||
|
await pump();
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 节点实例摘要:`A1#1=released` —— 断言拓扑走向的通用形状 */
|
||||||
|
async function seq(workflowId: string): Promise<string[]> {
|
||||||
|
const instances = await container.process.instancesOf(workflowId);
|
||||||
|
return instances.map((i) => `${i.nodeId}#${i.instance}=${i.terminal ?? '…'}`);
|
||||||
|
}
|
||||||
|
|
||||||
|
async function flowState(workflowId: string): Promise<string> {
|
||||||
|
return (await container.process.getWorkflow(workflowId)).state;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 走完 A1(按旗选路)→ … → 停在 A6 的验收单前 */
|
||||||
|
async function walkToA6(ref: string): Promise<string> {
|
||||||
|
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);
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in New Issue
Block a user