rootandClaude Opus 5.5
c9e8120fc6
- 新增 apps/server/test/support/{pg,engine}.ts:测试库地址、可用性探测、schema 咨询锁、
建 schema + 加锁迁移 + 删除、可选去重的假引擎;6 份 harness 改用它(acquisition-organize 与
file-manager 原先迁移不加锁,并行下有冲突隐患,一并修正)
- sse-pg 起子进程,并行下子进程启动超时,归入 serial
- 修正 b61335e 挪动 / 改名测试后遗留的 import 排序与格式(当时 lint 检查经管道被吞掉退出码)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
92 lines
3.2 KiB
TypeScript
92 lines
3.2 KiB
TypeScript
// server 测试共用的 PG 装配:测试库地址、可用性探测、schema 咨询锁、每文件独立 schema 的建 / 迁 / 删。
|
|
// 各模块 harness 只管自己的上下文与种子数据,建库这几步统一走这里。
|
|
import { type DB, migrateToLatest } from "@umt/db";
|
|
import { Kysely, PostgresDialect } from "kysely";
|
|
import pg from "pg";
|
|
|
|
export const PG_URL = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
|
|
|
|
/** 测试库连得上才跑 PG 用例;UMT_SKIP_DB_TESTS=1 显式跳过(deno task test:all 连不上库时直接失败) */
|
|
export async function pgAvailable(): Promise<boolean> {
|
|
if (process.env.UMT_SKIP_DB_TESTS === "1") return false;
|
|
const c = new pg.Client({ connectionString: PG_URL, connectionTimeoutMillis: 2000 });
|
|
try {
|
|
await c.connect();
|
|
await c.end();
|
|
return true;
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Kysely 迁移器会反射库里所有 schema 的表;别的测试文件同时 drop schema 时反射会报"schema 不存在"。
|
|
* 用 PG 咨询锁把"迁移 / 删 schema"串行化(跨进程有效),另加重试兜底。
|
|
*/
|
|
const SCHEMA_LOCK = 0x756d7431;
|
|
export async function withSchemaLock<T>(fn: () => Promise<T>): Promise<T> {
|
|
const c = new pg.Client({ connectionString: PG_URL });
|
|
await c.connect();
|
|
try {
|
|
await c.query("select pg_advisory_lock($1)", [SCHEMA_LOCK]);
|
|
for (let attempt = 1; ; attempt++) {
|
|
try {
|
|
return await fn();
|
|
} catch (err) {
|
|
if (attempt >= 4 || !/does not exist/.test(String((err as Error)?.message))) throw err;
|
|
await new Promise((r) => setTimeout(r, 200 * attempt));
|
|
}
|
|
}
|
|
} finally {
|
|
await c.query("select pg_advisory_unlock($1)", [SCHEMA_LOCK]).catch(() => {});
|
|
await c.end();
|
|
}
|
|
}
|
|
|
|
/** 用一次性连接执行一条管理语句(建 / 删 schema) */
|
|
export async function adminQuery(q: string): Promise<void> {
|
|
const c = new pg.Client({ connectionString: PG_URL });
|
|
await c.connect();
|
|
try {
|
|
await c.query(q);
|
|
} finally {
|
|
await c.end();
|
|
}
|
|
}
|
|
|
|
/** 进程 + 随机后缀,避免并行文件撞名 */
|
|
export const uniqueSchema = (prefix: string) =>
|
|
`${prefix}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
|
|
|
|
export async function dropSchemas(...schemas: string[]): Promise<void> {
|
|
await withSchemaLock(async () => {
|
|
for (const s of schemas) await adminQuery(`drop schema if exists ${s} cascade`);
|
|
});
|
|
}
|
|
|
|
export interface TestSchema {
|
|
schema: string;
|
|
db: Kysely<DB>;
|
|
/** 关连接池并删掉 schema(连同 extra,例如引擎 schema) */
|
|
drop(...extra: string[]): Promise<void>;
|
|
}
|
|
|
|
/** 建独立 schema → 连接池(search_path 指向它)→ 加锁迁移到最新 */
|
|
export async function createTestSchema(schema: string, opts: { max?: number } = {}): Promise<TestSchema> {
|
|
await adminQuery(`create schema ${schema}`);
|
|
const db = new Kysely<DB>({
|
|
dialect: new PostgresDialect({
|
|
pool: new pg.Pool({ connectionString: PG_URL, max: opts.max ?? 6, options: `-c search_path=${schema},public` }),
|
|
}),
|
|
});
|
|
await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: schema }));
|
|
return {
|
|
schema,
|
|
db,
|
|
async drop(...extra) {
|
|
await db.destroy();
|
|
await dropSchemas(schema, ...extra);
|
|
},
|
|
};
|
|
}
|