test: server 测试共用建库与假引擎,sse-pg 归入串行组

- 新增 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]>

apps/server/test/acquisition-organize/background-loop.test.ts

@@ -1,10 +1,10 @@
 import { sql } from "@umt/db";
 import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { createSchedulesModule } from "../../src/modules/schedules/index.ts";
 import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
 import { submitDownload, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
 import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
+import { createSchedulesModule } from "../../src/modules/schedules/index.ts";
 import {
   createHarness,
   fakeDisk,

apps/server/test/acquisition-organize/classification.test.ts

@@ -3,10 +3,10 @@ import type { TmdbClient } from "@umt/adapters";
 import { DEFAULT_CLASSIFICATION_YAML } from "@umt/domain";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import type { AuthEnv } from "../../src/http/auth.ts";
+import { classifyEntity } from "../../src/modules/acquisition-organize/services/classification.ts";
 import { createSettingsReader } from "../../src/modules/auth-settings/provided.ts";
 import { registerSettingsRoutes } from "../../src/modules/auth-settings/settings-routes.ts";
 import { SourcesRuntime } from "../../src/modules/sources-identity/runtime.ts";
-import { classifyEntity } from "../../src/modules/acquisition-organize/services/classification.ts";
 import { createHarness, type Harness, pgAvailable, seedEntity } from "./harness.ts";
 
 const hasPg = await pgAvailable();

apps/server/test/acquisition-organize/enqueue-recovery.test.ts

@@ -3,7 +3,11 @@ import { createTempTree, type TempTree } from "@umt/testkit";
 import { sql } from "kysely";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { enqueueSweep, enqueueTracked } from "../../src/modules/acquisition-organize/services/common.ts";
-import { type QueueScrapeInput, queueScrape, runScrapeRecord } from "../../src/modules/acquisition-organize/services/scrape.ts";
+import {
+  type QueueScrapeInput,
+  queueScrape,
+  runScrapeRecord,
+} from "../../src/modules/acquisition-organize/services/scrape.ts";
 import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
 
 const hasPg = await pgAvailable();

apps/server/test/acquisition-organize/harness.ts

@@ -3,25 +3,28 @@ import { join } from "node:path";
 import type { DiskUsage } from "@umt/adapters";
 import { createQbClient } from "@umt/adapters";
 import { DEFAULT_SETTINGS, GlobalSettings, type SettingsKey } from "@umt/contracts";
-import { type DB, migrateToLatest } from "@umt/db";
-import type { TaskEngine } from "@umt/tasks";
-import { Kysely, PostgresDialect, sql } from "kysely";
-import pg from "pg";
+import type { DB } from "@umt/db";
+import { type Kysely, sql } from "kysely";
 import { ConfigSchema } from "../../src/config.ts";
 import { createOutboxPort } from "../../src/context.ts";
 import { createApp } from "../../src/http/app.ts";
 import { hashApiKey } from "../../src/http/auth.ts";
 import { EventBus } from "../../src/http/sse.ts";
 import { createLogger } from "../../src/logger.ts";
+import { type AoOverrides, createAcquisitionOrganizeModule } from "../../src/modules/acquisition-organize/index.ts";
+import type { AoDeps } from "../../src/modules/acquisition-organize/services/common.ts";
 import { createCapabilityRegistry } from "../../src/modules/agent/capability-registry.ts";
 import { createSourcesIdentityModule } from "../../src/modules/sources-identity/index.ts";
 import type { ServerContext } from "../../src/modules/types.ts";
-import { type AoOverrides, createAcquisitionOrganizeModule } from "../../src/modules/acquisition-organize/index.ts";
-import type { AoDeps } from "../../src/modules/acquisition-organize/services/common.ts";
+import { fakeEngine } from "../support/engine.ts";
+import { createTestSchema, PG_URL as DB_URL, uniqueSchema } from "../support/pg.ts";
 
 // 真实 PG(独立 schema)+ mock qB + 临时目录 的测试装配;引擎用记录入队的假实现,测试直接驱动服务函数。
 
-export const DB_URL = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
+export type { Enqueued } from "../support/engine.ts";
+export { fakeEngine } from "../support/engine.ts";
+export { pgAvailable } from "../support/pg.ts";
+export { DB_URL };
 export const DEV_KEY = "dev-admin-ao-test-key";
 /** 只读整理记录 / 可发起整理 的 API Key */
 export const READ_KEY = "ao-organize-read-key";
@@ -31,52 +34,6 @@ const KEYS = [
   { key: EXEC_KEY, scopes: ["organize.execute"] },
 ];
 
-export async function pgAvailable() {
-  if (process.env.UMT_SKIP_DB_TESTS === "1") return false;
-  const c = new pg.Client({ connectionString: DB_URL, connectionTimeoutMillis: 2000 });
-  try {
-    await c.connect();
-    await c.end();
-    return true;
-  } catch {
-    return false;
-  }
-}
-
-export interface Enqueued {
-  task: string;
-  input: unknown;
-  key: string;
-}
-
-/** 记录入队、不执行的引擎(处理器由测试直接调服务函数) */
-export function fakeEngine(): TaskEngine & { enqueued: Enqueued[] } {
-  const enqueued: Enqueued[] = [];
-  let n = 0;
-  return {
-    enqueued,
-    async enqueue(task, input, opts) {
-      enqueued.push({ task, input, key: opts.idempotencyKey });
-      n++;
-      return { runId: `run-${n}`, deduplicated: false };
-    },
-    async signal() {
-      return { delivered: 0 };
-    },
-    async cancel() {
-      return false;
-    },
-    async get() {
-      return null;
-    },
-    async list() {
-      return [];
-    },
-    async startWorker() {},
-    async stop() {},
-  };
-}
-
 /** 可调的磁盘用量:used = base + 目录下文件总大小(清理删文件后自然下降) */
 export function fakeDisk(total: number) {
   const state = { total, base: 0, watch: [] as string[], fail: false };
@@ -136,17 +93,8 @@ export async function createHarness(
   /** withSources:同时装配 sources-identity 模块(待处理中心路由等) */
   opts: { withSources?: boolean } = {},
 ): Promise<Harness> {
-  const schema = `ao_test_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
-  const admin = new pg.Client({ connectionString: DB_URL });
-  await admin.connect();
-  await admin.query(`create schema ${schema}`);
-  await admin.end();
-  const db = new Kysely<DB>({
-    dialect: new PostgresDialect({
-      pool: new pg.Pool({ connectionString: DB_URL, max: 6, options: `-c search_path=${schema},public` }),
-    }),
-  });
-  await migrateToLatest(db, { migrationTableSchema: schema });
+  const ts = await createTestSchema(uniqueSchema(`ao_test_${name}`));
+  const { db } = ts;
   const logger = createLogger({ level: "silent" });
   const engine = fakeEngine();
   const config = ConfigSchema.parse({
@@ -254,11 +202,7 @@ export async function createHarness(
       return r.rows;
     },
     async close() {
-      await db.destroy();
-      const c = new pg.Client({ connectionString: DB_URL });
-      await c.connect();
-      await c.query(`drop schema if exists ${schema} cascade`);
-      await c.end();
+      await ts.drop();
     },
   };
 }

apps/server/test/acquisition-organize/organize-fs.test.ts

@@ -4,7 +4,11 @@ import { readdir, readFile, rm, utimes, writeFile } from "node:fs/promises";
 import { dirname, join } from "node:path";
 import { createTempTree, type TempTree } from "@umt/testkit";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { executeRecord, type FrozenPlan, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
+import {
+  executeRecord,
+  type FrozenPlan,
+  planImport,
+} from "../../src/modules/acquisition-organize/services/organize.ts";
 import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
 
 // W4-E4 执行侧:replace_with_verify 接线(写临时→校验→原子替换,旧文件进回收目录)、写前清 partial、

apps/server/test/acquisition-organize/organize-move.test.ts

@@ -14,7 +14,11 @@ import {
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
 import { submitDownload, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
-import { executeRecord, type FrozenPlan, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
+import {
+  executeRecord,
+  type FrozenPlan,
+  planImport,
+} from "../../src/modules/acquisition-organize/services/organize.ts";
 import {
   createHarness,
   EXEC_KEY,

apps/server/test/acquisition-organize/organize-sources.test.ts

@@ -4,9 +4,9 @@ import type { TmdbClient } from "@umt/adapters";
 import { DEFAULT_SETTINGS, type OrganizeSourceConfig } from "@umt/contracts";
 import { createMockQb, createTempTree, type MockQb, type TempTree } from "@umt/testkit";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
-import { SourcesRuntime } from "../../src/modules/sources-identity/runtime.ts";
 import { syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
 import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
+import { SourcesRuntime } from "../../src/modules/sources-identity/runtime.ts";
 import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
 
 const hasPg = await pgAvailable();

apps/server/test/acquisition-organize/scrape.test.ts

@@ -3,7 +3,12 @@ import { join, relative } from "node:path";
 import { createTempTree, type TempTree } from "@umt/testkit";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
-import { entityScrapeLocations, runScrapeRecord, SCRAPE_TASK, scrapeEntity } from "../../src/modules/acquisition-organize/services/scrape.ts";
+import {
+  entityScrapeLocations,
+  runScrapeRecord,
+  SCRAPE_TASK,
+  scrapeEntity,
+} from "../../src/modules/acquisition-organize/services/scrape.ts";
 import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
 
 // 刮削(SCR-1 / SCR-2):入库成功后自动对作品目录排 scrapeMedia → 写 NFO 与图片(临时库根);只补缺失 / 全部覆盖;

apps/server/test/acquisition-organize/subscription.test.ts

@@ -4,7 +4,11 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
 import { submitDownload, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
 import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
-import { applyEpisodeTotal, subscriptionProgress, sweepDueSlots } from "../../src/modules/acquisition-organize/services/subscriptions.ts";
+import {
+  applyEpisodeTotal,
+  subscriptionProgress,
+  sweepDueSlots,
+} from "../../src/modules/acquisition-organize/services/subscriptions.ts";
 import { createHarness, fakeDisk, type Harness, pgAvailable, seedBase, seedEntity, seedRelease } from "./harness.ts";
 
 // W5-C 订阅增强集成:Slot×单元多在途 + 忙时延后重评 + 移除释放;季包按缺集选文件(mock qB files/filePrio);

apps/server/test/acquisition-organize/takeover.test.ts

@@ -2,10 +2,14 @@ import { existsSync, readdirSync, statSync } from "node:fs";
 import { join } from "node:path";
 import { createMockQb, createTempTree, type MockQb, type TempTree } from "@umt/testkit";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { addAliasIfAbsent } from "../../src/modules/sources-identity/service/entities.ts";
 import { syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
 import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
-import { organizeDecisionHook, pickMainVideo, takeoverParsed } from "../../src/modules/acquisition-organize/services/takeover.ts";
+import {
+  organizeDecisionHook,
+  pickMainVideo,
+  takeoverParsed,
+} from "../../src/modules/acquisition-organize/services/takeover.ts";
+import { addAliasIfAbsent } from "../../src/modules/sources-identity/service/entities.ts";
 import { createHarness, type Harness, pgAvailable, seedBase, seedEntity } from "./harness.ts";
 
 // QB-1:接管任务(用户手动加进 qB、贴管理标签,没有站点资源)完成后按 种子名 + 主视频文件名 识别 → 直接整理;

apps/server/test/acquisition-organize/vertical.test.ts

@@ -5,7 +5,11 @@ import { sql } from "@umt/db";
 import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
-import { cleanupDownloadCache, submitDownload, syncDownloads } from "../../src/modules/acquisition-organize/services/downloads.ts";
+import {
+  cleanupDownloadCache,
+  submitDownload,
+  syncDownloads,
+} from "../../src/modules/acquisition-organize/services/downloads.ts";
 import { executeRecord, planImport } from "../../src/modules/acquisition-organize/services/organize.ts";
 import { reconcileLibrary } from "../../src/modules/acquisition-organize/services/reconcile.ts";
 import {

apps/server/test/auth-settings/harness.ts

@@ -1,7 +1,6 @@
-import { type DB, migrateToLatest } from "@umt/db";
+import type { DB } from "@umt/db";
 import { migrateEngineSchema } from "@umt/tasks";
-import { Kysely, PostgresDialect } from "kysely";
-import pg from "pg";
+import type { Kysely } from "kysely";
 import { vi } from "vitest";
 import { ConfigSchema } from "../../src/config.ts";
 import { createOutboxPort } from "../../src/context.ts";
@@ -15,69 +14,17 @@ import { DEV_KEY, memoryLogger } from "../helpers.ts";
 
 // 真实 PG 的 W2-A 测试夹具:独立业务 schema + 独立引擎 schema;任务引擎用假对象(只测 cancel 透传)。
 
-export const PG_URL = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
+import { createTestSchema, PG_URL, uniqueSchema, withSchemaLock } from "../support/pg.ts";
+
+export { PG_URL, pgAvailable, withSchemaLock } from "../support/pg.ts";
 export const MASTER_KEY = "test-master-key-0123456789";
 
-export async function pgAvailable() {
-  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();
-  }
-}
-
-async function adminQuery(q: string) {
-  const c = new pg.Client({ connectionString: PG_URL });
-  await c.connect();
-  try {
-    await c.query(q);
-  } finally {
-    await c.end();
-  }
-}
-
 export async function createHarness(opts: { engine?: boolean; extraModules?: ServerModule[] } = {}) {
-  const uniq = `${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
-  const schema = `w2a_test_${uniq}`;
+  const ts = await createTestSchema(uniqueSchema("w2a_test"), { max: 5 });
+  const { schema } = ts;
+  const db: Kysely<DB> = ts.db;
   const engineSchema = `${schema}_ow`;
-  await adminQuery(`create schema ${schema}`);
-  const db = new Kysely<DB>({
-    dialect: new PostgresDialect({
-      pool: new pg.Pool({ connectionString: PG_URL, max: 5, options: `-c search_path=${schema},public` }),
-    }),
-  });
-  await withSchemaLock(async () => {
-    await migrateToLatest(db, { migrationTableSchema: schema });
-    if (opts.engine) await migrateEngineSchema(PG_URL, engineSchema);
-  });
+  if (opts.engine) await withSchemaLock(() => migrateEngineSchema(PG_URL, engineSchema));
   const { logger, lines } = memoryLogger();
   const bus = new EventBus(100);
   const config = ConfigSchema.parse({
@@ -120,11 +67,7 @@ export async function createHarness(opts: { engine?: boolean; extraModules?: Ser
     schema,
     engineSchema,
     async close() {
-      await db.destroy();
-      await withSchemaLock(async () => {
-        await adminQuery(`drop schema if exists ${schema} cascade`);
-        await adminQuery(`drop schema if exists ${engineSchema} cascade`);
-      });
+      await ts.drop(engineSchema);
     },
   };
 }

apps/server/test/browser/browser-identity.test.ts

@@ -1,8 +1,8 @@
 import { MOCK_CAMOFOX_IDENTITY } from "@umt/testkit";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { egressProxy } from "../../src/modules/browser/service.ts";
 import { sourcesIdentityModule } from "../../src/modules/sources-identity/index.ts";
 import { readBrowserUserAgent } from "../../src/modules/sources-identity/service/user-agent.ts";
-import { egressProxy } from "../../src/modules/browser/service.ts";
 import { type BrowserHarness, createBrowserHarness, type Loose, ORIGIN, pgAvailable } from "./harness.ts";
 
 // BR-4:打开会话前按站点下发出口(站点开代理 = 与 HTTP 同一代理,否则直连);浏览器固定身份写进 browser_provider.identity,

apps/server/test/browser/browser-provider-config.test.ts

@@ -1,6 +1,6 @@
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import type { ServerContext } from "../../src/modules/types.ts";
 import { BrowserService, BrowserServiceError, ENV_PROVIDER_NAME } from "../../src/modules/browser/service.ts";
+import type { ServerContext } from "../../src/modules/types.ts";
 import { ACCESS_KEY, type BrowserHarness, createBrowserHarness, type Loose, pgAvailable } from "./harness.ts";
 
 // BR-1:浏览器服务登记 —— 环境变量(BROWSER_PROVIDER_URL / BROWSER_ACCESS_KEY / BROWSER_ENABLED → camofox-env)

apps/server/test/browser/harness.ts

@@ -2,7 +2,7 @@ import { mkdtemp, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import type { CapabilityRegistry, InvokeContext } from "@umt/agent";
-import { type DB, type Kysely, migrateToLatest, sql } from "@umt/db";
+import type { DB, Kysely } from "@umt/db";
 import {
   buildFakeTorrent,
   createMockCamofox,
@@ -10,22 +10,20 @@ import {
   type MockCamofox,
   type MockCamofoxSite,
 } from "@umt/testkit";
-import { Kysely as K, PostgresDialect } from "kysely";
-import pg from "pg";
-import { withSchemaLock } from "../auth-settings/harness.ts";
-import { DEV_KEY, memoryLogger, memoryStore } from "../helpers.ts";
-import { fakeEngine, fakeSecrets, PG_URL, plainSettings } from "../sources-identity/harness.ts";
 import { createOutboxPort } from "../../src/context.ts";
 import { createApp } from "../../src/http/app.ts";
 import { EventBus } from "../../src/http/sse.ts";
 import { createAcquisitionOrganizeModule } from "../../src/modules/acquisition-organize/index.ts";
 import { createCapabilityRegistry } from "../../src/modules/agent/capability-registry.ts";
 import { createAgentModule } from "../../src/modules/agent/index.ts";
-import { createSourcesIdentityModule } from "../../src/modules/sources-identity/index.ts";
-import type { ServerContext } from "../../src/modules/types.ts";
 import { registerBrowserCapabilities } from "../../src/modules/browser/capabilities.ts";
 import { createBrowserModule } from "../../src/modules/browser/index.ts";
 import type { BrowserModuleOptions } from "../../src/modules/browser/service.ts";
+import { createSourcesIdentityModule } from "../../src/modules/sources-identity/index.ts";
+import type { ServerContext } from "../../src/modules/types.ts";
+import { DEV_KEY, memoryLogger, memoryStore } from "../helpers.ts";
+import { fakeEngine, fakeSecrets, plainSettings } from "../sources-identity/harness.ts";
+import { createTestSchema, uniqueSchema } from "../support/pg.ts";
 
 // C1-E 浏览器模块测试装配:独立 PG schema + 真实迁移 + 交互式 mock camofox(观众风格站点)+ 真实能力注册中心。
 
@@ -94,18 +92,8 @@ export async function createBrowserHarness(
     },
   });
   const stagingDir = await mkdtemp(join(tmpdir(), "umt-browser-"));
-  const schema = `c1e_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
-  const admin = new K<unknown>({
-    dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: PG_URL, max: 1 }) }),
-  });
-  await sql.raw(`create schema ${schema}`).execute(admin);
-  await import("@umt/db");
-  const db = new K<DB>({
-    dialect: new PostgresDialect({
-      pool: new pg.Pool({ connectionString: PG_URL, max: 8, options: `-c search_path=${schema},public` }),
-    }),
-  }) as unknown as Kysely<DB>;
-  await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: schema }));
+  const ts = await createTestSchema(uniqueSchema(`c1e_${name}`), { max: 8 });
+  const db = ts.db as unknown as Kysely<DB>;
 
   const { logger } = memoryLogger();
   const secrets = fakeSecrets(() => db);
@@ -230,9 +218,7 @@ export async function createBrowserHarness(
         await db.insertInto("automation_grant").values({ action, granted_by: "test" }).execute();
     },
     async close() {
-      await db.destroy();
-      await withSchemaLock(() => sql.raw(`drop schema if exists ${schema} cascade`).execute(admin));
-      await admin.destroy();
+      await ts.drop();
       await mock.close();
       await rm(stagingDir, { recursive: true, force: true });
     },

apps/server/test/e2e/harness.ts

@@ -2,7 +2,7 @@ import { mkdtemp, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import type { FakeScript } from "@umt/agent";
-import { type DB, migrateToLatest } from "@umt/db";
+import type { DB } from "@umt/db";
 import {
   createMockQb,
   createMockTmdb,
@@ -13,8 +13,7 @@ import {
   TMDB_FIXTURES,
   type TmdbFixture,
 } from "@umt/testkit";
-import { Kysely, PostgresDialect, sql } from "kysely";
-import pg from "pg";
+import { type Kysely, sql } from "kysely";
 import { ConfigSchema } from "../../src/config.ts";
 import { type Bootstrapped, bootstrap } from "../../src/context.ts";
 import { createApp } from "../../src/http/app.ts";
@@ -26,30 +25,18 @@ import { setDefaultUpstream } from "../../src/modules/pi/store.ts";
 import type { ServerModule } from "../../src/modules/types.ts";
 import type { WebSearchModuleOptions } from "../../src/modules/websearch/index.ts";
 import { type RunningWorker, startWorker } from "../../src/worker/index.ts";
-import { withSchemaLock } from "../auth-settings/harness.ts";
+import { createTestSchema, dropSchemas, PG_URL, uniqueSchema } from "../support/pg.ts";
 import { buildModules } from "./fixtures/modules.ts";
 
 // 端到端夹具:真实 PG(每个文件独立业务 schema + 独立引擎 schema)+ 真实四个模块 + OpenWorkflow 引擎/worker/outbox 中继
 // + testkit mock(qB / TMDB / RSS)+ 临时目录(缓存卷、媒体库、Agent 会话)。HTTP 走 createApp(DEV key = admin)。
 
-export const PG_URL = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
+export { PG_URL, pgAvailable } from "../support/pg.ts";
 export const DEV_KEY = "dev-admin-e2e-key";
 export const MASTER_KEY = "e2e-master-key-0123456789";
 /** mock TMDB 的 v3 key(32 位十六进制,经设置 metadata.tmdbApiKey 录入) */
 export const TMDB_KEY = "0123456789abcdef0123456789abcdef";
 
-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;
-  }
-}
-
 // biome-ignore lint/suspicious/noExplicitAny: 测试断言直接读 JSON
 type Loose = any;
 
@@ -110,19 +97,10 @@ export async function waitFor<T>(
 }
 
 export async function createE2E(name: string, opts: E2EOptions = {}): Promise<E2E> {
-  const uniq = `${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
-  const biz = `e2e_${name}_${uniq}`;
+  const ts = await createTestSchema(uniqueSchema(`e2e_${name}`), { max: 10 });
+  const { schema: biz } = ts;
+  const db: Kysely<DB> = ts.db;
   const ow = `${biz}_ow`;
-  const admin = new pg.Client({ connectionString: PG_URL });
-  await admin.connect();
-  await admin.query(`create schema ${biz}`);
-  await admin.end();
-  const db = new Kysely<DB>({
-    dialect: new PostgresDialect({
-      pool: new pg.Pool({ connectionString: PG_URL, max: 10, options: `-c search_path=${biz},public` }),
-    }),
-  });
-  await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: biz }));
 
   const qb = await createMockQb({ addVisibleAfterPolls: 0 });
   const tmdb = await createMockTmdb({ apiKey: TMDB_KEY, fixtures: opts.tmdbFixtures ?? TMDB_FIXTURES });
@@ -225,13 +203,7 @@ export async function createE2E(name: string, opts: E2EOptions = {}): Promise<E2
       await lib.cleanup();
       await cache.cleanup();
       await rm(sessionDir, { recursive: true, force: true });
-      await withSchemaLock(async () => {
-        const c = new pg.Client({ connectionString: PG_URL });
-        await c.connect();
-        await c.query(`drop schema if exists ${biz} cascade`);
-        await c.query(`drop schema if exists ${ow} cascade`);
-        await c.end();
-      });
+      await dropSchemas(biz, ow);
     },
   };
   // 通用设置:fake 上游(脚本化假模型,设为默认)、TMDB 走设置里的 secret、结算快一点

apps/server/test/file-manager/harness.ts

@@ -10,26 +10,27 @@ import {
   SESSION_COOKIE,
   type SettingsKey,
 } from "@umt/contracts";
-import { type DB, migrateToLatest } from "@umt/db";
+import type { DB } from "@umt/db";
 import { normalizeTitle } from "@umt/domain";
-import { Kysely, PostgresDialect, sql } from "kysely";
-import pg from "pg";
+import { type Kysely, sql } from "kysely";
 import { ConfigSchema } from "../../src/config.ts";
 import { createOutboxPort } from "../../src/context.ts";
 import { createApp } from "../../src/http/app.ts";
 import { deriveCsrfToken, hashApiKey, hashSessionToken } from "../../src/http/auth.ts";
 import { EventBus } from "../../src/http/sse.ts";
 import { createLogger } from "../../src/logger.ts";
-import { fakeEngine } from "../acquisition-organize/harness.ts";
 import { executeRecord, type FrozenPlan } from "../../src/modules/acquisition-organize/services/organize.ts";
 import { runScrapeRecord } from "../../src/modules/acquisition-organize/services/scrape.ts";
 import { createCapabilityRegistry } from "../../src/modules/agent/capability-registry.ts";
 import { createFileManagerModule, type FmOverrides } from "../../src/modules/file-manager/index.ts";
 import type { ServerContext } from "../../src/modules/types.ts";
+import { fakeEngine } from "../support/engine.ts";
+import { createTestSchema, PG_URL as DB_URL, uniqueSchema } from "../support/pg.ts";
 
 // 文件管理测试装配:真实 PG(独立 schema)+ 临时目录(库根 / 下载根 / 根外目录)+ 记录入队的假引擎。
 
-export const DB_URL = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
+export { pgAvailable } from "../support/pg.ts";
+export { DB_URL };
 export const DEV_KEY = "dev-admin-fm-test-key";
 export const READ_KEY = "fm-organize-read-key";
 export const EXEC_KEY = "fm-organize-exec-key";
@@ -43,18 +44,6 @@ const KEYS = [
   { key: ADMIN_KEY, scopes: ["admin"] },
 ];
 
-export async function pgAvailable() {
-  if (process.env.UMT_SKIP_DB_TESTS === "1") return false;
-  const c = new pg.Client({ connectionString: DB_URL, connectionTimeoutMillis: 2000 });
-  try {
-    await c.connect();
-    await c.end();
-    return true;
-  } catch {
-    return false;
-  }
-}
-
 // biome-ignore lint/suspicious/noExplicitAny: 测试断言直接读响应 JSON
 type Loose = any;
 
@@ -98,17 +87,8 @@ export async function createFmHarness(
     runTransfers?: boolean;
   } = {},
 ): Promise<FmHarness> {
-  const schema = `fm_test_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
-  const admin = new pg.Client({ connectionString: DB_URL });
-  await admin.connect();
-  await admin.query(`create schema ${schema}`);
-  await admin.end();
-  const db = new Kysely<DB>({
-    dialect: new PostgresDialect({
-      pool: new pg.Pool({ connectionString: DB_URL, max: 6, options: `-c search_path=${schema},public` }),
-    }),
-  });
-  await migrateToLatest(db, { migrationTableSchema: schema });
+  const ts = await createTestSchema(uniqueSchema(`fm_test_${name}`));
+  const { db } = ts;
 
   const base = await mkdtemp(join(tmpdir(), `fm-${name}-`));
   const lib = join(base, "lib");
@@ -285,11 +265,7 @@ export async function createFmHarness(
       return Promise.resolve(app.request(`/api/v1${path}`, { method, headers: authHeaders(key) }));
     },
     async close() {
-      await db.destroy();
-      const c = new pg.Client({ connectionString: DB_URL });
-      await c.connect();
-      await c.query(`drop schema if exists ${schema} cascade`);
-      await c.end();
+      await ts.drop();
       await rm(base, { recursive: true, force: true });
     },
   };

apps/server/test/inventory/media-servers.test.ts

@@ -1,11 +1,6 @@
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import {
-  createHarness,
-  type Harness,
-  pgAvailable,
-  seedEntity,
-} from "../acquisition-organize/harness.ts";
 import { MEDIA_SERVER_LIMITS } from "../../src/modules/acquisition-organize/routes/library.ts";
+import { createHarness, type Harness, pgAvailable, seedEntity } from "../acquisition-organize/harness.ts";
 
 // C1-D:GET /inventory/media-servers 按实例并行(allSettled)+ 实例内并发上限 + 整实例超时;部分失败 partial=true。
 

apps/server/test/rules/rule-groups.test.ts

@@ -1,5 +1,6 @@
 import { makeTorrent } from "@umt/testkit";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
 import {
   createHarness,
   fakeDisk,
@@ -10,7 +11,6 @@ import {
   seedRelease,
   seedSlot,
 } from "../acquisition-organize/harness.ts";
-import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
 
 // W9-D 规则组:关键字表、规则串解析/序列化、CRUD + 修订、抑制记录、试算、默认链、评估接入、旧端点别名。
 

apps/server/test/sources-identity/harness.ts

@@ -1,36 +1,24 @@
 import { randomUUID } from "node:crypto";
 import { createTmdbClient } from "@umt/adapters";
 import { DEFAULT_SETTINGS, type GlobalSettings, type SettingsKey } from "@umt/contracts";
-import { type DB, type Kysely, migrateToLatest, sql } from "@umt/db";
-import type { TaskEngine } from "@umt/tasks";
-import { Kysely as K, PostgresDialect } from "kysely";
-import pg from "pg";
+import type { DB, Kysely } from "@umt/db";
 import { createOutboxPort } from "../../src/context.ts";
 import { createApp } from "../../src/http/app.ts";
 import { EventBus } from "../../src/http/sse.ts";
 import { createSourcesIdentityModule } from "../../src/modules/sources-identity/index.ts";
 import type { RuntimeOverrides } from "../../src/modules/sources-identity/runtime.ts";
 import type { SecretStore, ServerContext, SettingsReader } from "../../src/modules/types.ts";
-import { withSchemaLock } from "../auth-settings/harness.ts";
 import { DEV_KEY, memoryLogger, memoryStore } from "../helpers.ts";
+import { fakeEngine as baseFakeEngine } from "../support/engine.ts";
+import { createTestSchema, uniqueSchema } from "../support/pg.ts";
 
 export const READ_KEY = "w2b-read-key";
 export const EXEC_KEY = "w2b-exec-key";
 
 // W2-B 集成测试夹具:独立 PG schema + 真实迁移 + fake 引擎/凭证 + 模块运行时。
 
-export const PG_URL = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
-
-export async function pgAvailable(): Promise<boolean> {
-  const c = new pg.Client({ connectionString: PG_URL, connectionTimeoutMillis: 2000 });
-  try {
-    await c.connect();
-    await c.end();
-    return true;
-  } catch {
-    return false;
-  }
-}
+export type { Enqueued } from "../support/engine.ts";
+export { PG_URL, pgAvailable } from "../support/pg.ts";
 
 /** 内存凭证库(W2-A 的 SecretStore 实现前);记录明文以便断言"明文不入业务表" */
 export function fakeSecrets(db: () => Kysely<DB>): SecretStore & { plain: Map<string, string> } {
@@ -75,41 +63,8 @@ export function plainSettings(db: () => Kysely<DB>): SettingsReader {
   return { all, get: async <K extends SettingsKey>(k: K) => (await all())[k] } as SettingsReader;
 }
 
-export interface Enqueued {
-  task: string;
-  input: unknown;
-  key: string;
-}
-
-export function fakeEngine(): TaskEngine & { enqueued: Enqueued[] } {
-  const enqueued: Enqueued[] = [];
-  const keys = new Set<string>();
-  return {
-    enqueued,
-    async enqueue(task, input, opts) {
-      const dedup = keys.has(opts.idempotencyKey);
-      if (!dedup) {
-        keys.add(opts.idempotencyKey);
-        enqueued.push({ task, input, key: opts.idempotencyKey });
-      }
-      return { runId: randomUUID(), deduplicated: dedup };
-    },
-    async signal() {
-      return { delivered: 0 };
-    },
-    async cancel() {
-      return false;
-    },
-    async get() {
-      return null;
-    },
-    async list() {
-      return [];
-    },
-    async startWorker() {},
-    async stop() {},
-  };
-}
+/** 按幂等键去重的假引擎(贴近真实引擎) */
+export const fakeEngine = () => baseFakeEngine({ dedupe: true });
 
 export interface Harness {
   db: Kysely<DB>;
@@ -130,20 +85,8 @@ export interface HarnessOptions extends RuntimeOverrides {
 }
 
 export async function createHarness(name: string, opts: HarnessOptions = {}): Promise<Harness> {
-  const schema = `w2b_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
-  const admin = new K<unknown>({
-    dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: PG_URL, max: 1 }) }),
-  });
-  await sql.raw(`create schema ${schema}`).execute(admin);
-  // 与 @umt/db createDb 相同的 int8 解析(createDb 导入即注册全局解析器)
-  await import("@umt/db");
-  const db = new K<DB>({
-    dialect: new PostgresDialect({
-      pool: new pg.Pool({ connectionString: PG_URL, max: 8, options: `-c search_path=${schema},public` }),
-    }),
-  }) as unknown as Kysely<DB>;
-  // 与其他测试进程共用 PG:迁移内省会看到别处正在删除的 schema,用同一把咨询锁排队并重试
-  await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: schema }));
+  const ts = await createTestSchema(uniqueSchema(`w2b_${name}`), { max: 8 });
+  const db = ts.db as unknown as Kysely<DB>;
 
   const { logger } = memoryLogger();
   const engine = fakeEngine();
@@ -207,9 +150,7 @@ export async function createHarness(name: string, opts: HarnessOptions = {}): Pr
       return (await q.execute()) as never;
     },
     async close() {
-      await db.destroy();
-      await withSchemaLock(() => sql.raw(`drop schema if exists ${schema} cascade`).execute(admin));
-      await admin.destroy();
+      await ts.drop();
     },
   };
 }

apps/server/test/subscriptions/subscriptions-c1a.test.ts

@@ -2,15 +2,6 @@ import { type DB, sql } from "@umt/db";
 import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit";
 import type { Kysely } from "kysely";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import {
-  createHarness,
-  fakeDisk,
-  type Harness,
-  pgAvailable,
-  seedBase,
-  seedEntity,
-  seedRelease,
-} from "../acquisition-organize/harness.ts";
 import {
   evaluateAcquisition,
   subscriptionSavePath,
@@ -25,6 +16,15 @@ import {
   parseMpSource,
   presetConditions,
 } from "../../src/modules/acquisition-organize/services/subscriptions-params.ts";
+import {
+  createHarness,
+  fakeDisk,
+  type Harness,
+  pgAvailable,
+  seedBase,
+  seedEntity,
+  seedRelease,
+} from "../acquisition-organize/harness.ts";
 
 // C1-A 订阅遗留收尾:包含 / 排除 → 订阅私有规则组(评估链纳入、随订阅删除)、MP 分辨率 / 来源 / 特效 → 收录版本、
 // 保存路径经路径映射传给 qB、库根用于首次整理、下载器实例、重置进度、日历。

apps/server/test/subscriptions/subscriptions-v2.test.ts

@@ -2,6 +2,7 @@ import type { DB } from "@umt/db";
 import { makeTorrent } from "@umt/testkit";
 import type { Kysely } from "kysely";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
 import {
   createHarness,
   fakeDisk,
@@ -11,7 +12,6 @@ import {
   seedEntity,
   seedRelease,
 } from "../acquisition-organize/harness.ts";
-import { evaluateAcquisition } from "../../src/modules/acquisition-organize/services/acquisition.ts";
 
 // W9-E 订阅重做后端:总览分组/排序/计数、覆盖矩阵(单集/季包/多版本/在途/临时单元/电影版本)、
 // 多版本自动补全(开关、规则组继承、补全预算、手动补全)、表单字段与 MP/旧字段兼容、暂停/恢复/重新订阅/批量/排序。

apps/server/test/support/engine.ts

@@ -0,0 +1,48 @@
+// 只记录入队、不执行的假引擎(处理器由测试直接调服务函数)。
+import { randomUUID } from "node:crypto";
+import type { TaskEngine } from "@umt/tasks";
+
+export interface Enqueued {
+  task: string;
+  input: unknown;
+  key: string;
+}
+
+/**
+ * dedupe:同一幂等键只记一次并返回 deduplicated(贴近真实引擎);缺省每次都记,runId 为 run-1、run-2…
+ */
+export function fakeEngine(opts: { dedupe?: boolean } = {}): TaskEngine & { enqueued: Enqueued[] } {
+  const enqueued: Enqueued[] = [];
+  const keys = new Set<string>();
+  let n = 0;
+  return {
+    enqueued,
+    async enqueue(task, input, o) {
+      if (opts.dedupe) {
+        const dup = keys.has(o.idempotencyKey);
+        if (!dup) {
+          keys.add(o.idempotencyKey);
+          enqueued.push({ task, input, key: o.idempotencyKey });
+        }
+        return { runId: randomUUID(), deduplicated: dup };
+      }
+      enqueued.push({ task, input, key: o.idempotencyKey });
+      n++;
+      return { runId: `run-${n}`, deduplicated: false };
+    },
+    async signal() {
+      return { delivered: 0 };
+    },
+    async cancel() {
+      return false;
+    },
+    async get() {
+      return null;
+    },
+    async list() {
+      return [];
+    },
+    async startWorker() {},
+    async stop() {},
+  };
+}

apps/server/test/support/pg.ts

@@ -0,0 +1,91 @@
+// 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);
+    },
+  };
+}

apps/server/vitest.config.ts

@@ -5,7 +5,7 @@ import { defineConfig } from "vitest/config";
 //   pg     —— 真实 PG、假引擎(每文件独立 schema),文件并行;
 //   serial —— 起真实引擎 / worker / 后台循环 / 子进程的集成与 e2e,文件级串行。并发跑时 worker、轮询、迁移
 //             互相抢 CPU 与连接,会把"入库→状态推进"这类时序拉长到超时,串行让每个文件拿到稳定的时序预算。
-// 新测试缺省进 pg;起真实 worker / 引擎的加进 SERIAL;确认不碰 PG 的加进 UNIT。
+// 新测试缺省进 pg;起真实 worker / 引擎 / 子进程的加进 SERIAL;确认不碰 PG 的加进 UNIT。
 // 并行文件数受开发库连接数约束(max_connections 100,每个 harness 连接池 5–10):MAX_WORKERS 别随手调大。
 const UNIT = [
   "test/api-registry.test.ts",
@@ -27,6 +27,7 @@ const SERIAL = [
   "test/e2e/**/*.test.ts",
   "test/auth-settings/admin-api.test.ts",
   "test/auth-settings/engine-retry.test.ts",
+  "test/auth-settings/sse-pg.test.ts",
   "test/system/**/*.test.ts",
   "test/search/search.test.ts",
   "test/search/two-stage-search.test.ts",