From c9e8120fc68602f7c603a2eba48451d5e7dc56c5 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 2 Oct 2026 17:40:43 +0800 Subject: [PATCH] =?UTF-8?q?test:=20server=20=E6=B5=8B=E8=AF=95=E5=85=B1?= =?UTF-8?q?=E7=94=A8=E5=BB=BA=E5=BA=93=E4=B8=8E=E5=81=87=E5=BC=95=E6=93=8E?= =?UTF-8?q?,sse-pg=20=E5=BD=92=E5=85=A5=E4=B8=B2=E8=A1=8C=E7=BB=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 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 --- .../background-loop.test.ts | 2 +- .../classification.test.ts | 2 +- .../enqueue-recovery.test.ts | 6 +- .../test/acquisition-organize/harness.ts | 82 +++-------------- .../acquisition-organize/organize-fs.test.ts | 6 +- .../organize-move.test.ts | 6 +- .../organize-sources.test.ts | 2 +- .../test/acquisition-organize/scrape.test.ts | 7 +- .../acquisition-organize/subscription.test.ts | 6 +- .../acquisition-organize/takeover.test.ts | 8 +- .../acquisition-organize/vertical.test.ts | 6 +- apps/server/test/auth-settings/harness.ts | 77 ++-------------- .../test/browser/browser-identity.test.ts | 2 +- .../browser/browser-provider-config.test.ts | 2 +- apps/server/test/browser/harness.ts | 32 ++----- apps/server/test/e2e/harness.ts | 44 ++------- apps/server/test/file-manager/harness.ts | 42 ++------- .../test/inventory/media-servers.test.ts | 7 +- apps/server/test/rules/rule-groups.test.ts | 2 +- apps/server/test/sources-identity/harness.ts | 79 ++-------------- .../subscriptions/subscriptions-c1a.test.ts | 18 ++-- .../subscriptions/subscriptions-v2.test.ts | 2 +- apps/server/test/support/engine.ts | 48 ++++++++++ apps/server/test/support/pg.ts | 91 +++++++++++++++++++ apps/server/vitest.config.ts | 3 +- 25 files changed, 254 insertions(+), 328 deletions(-) create mode 100644 apps/server/test/support/engine.ts create mode 100644 apps/server/test/support/pg.ts diff --git a/apps/server/test/acquisition-organize/background-loop.test.ts b/apps/server/test/acquisition-organize/background-loop.test.ts index c274daae..101cbdd0 100644 --- a/apps/server/test/acquisition-organize/background-loop.test.ts +++ b/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, diff --git a/apps/server/test/acquisition-organize/classification.test.ts b/apps/server/test/acquisition-organize/classification.test.ts index dc5c3054..aff9c8ae 100644 --- a/apps/server/test/acquisition-organize/classification.test.ts +++ b/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(); diff --git a/apps/server/test/acquisition-organize/enqueue-recovery.test.ts b/apps/server/test/acquisition-organize/enqueue-recovery.test.ts index 2759fce7..656a2c67 100644 --- a/apps/server/test/acquisition-organize/enqueue-recovery.test.ts +++ b/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(); diff --git a/apps/server/test/acquisition-organize/harness.ts b/apps/server/test/acquisition-organize/harness.ts index f95eb87f..2fd9a74d 100644 --- a/apps/server/test/acquisition-organize/harness.ts +++ b/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 { - 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({ - 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(); }, }; } diff --git a/apps/server/test/acquisition-organize/organize-fs.test.ts b/apps/server/test/acquisition-organize/organize-fs.test.ts index cced2c61..948c8304 100644 --- a/apps/server/test/acquisition-organize/organize-fs.test.ts +++ b/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、 diff --git a/apps/server/test/acquisition-organize/organize-move.test.ts b/apps/server/test/acquisition-organize/organize-move.test.ts index 3193037d..7f87944a 100644 --- a/apps/server/test/acquisition-organize/organize-move.test.ts +++ b/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, diff --git a/apps/server/test/acquisition-organize/organize-sources.test.ts b/apps/server/test/acquisition-organize/organize-sources.test.ts index d5d7e2fc..9562cac5 100644 --- a/apps/server/test/acquisition-organize/organize-sources.test.ts +++ b/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(); diff --git a/apps/server/test/acquisition-organize/scrape.test.ts b/apps/server/test/acquisition-organize/scrape.test.ts index c1ad01de..3d7e7c6a 100644 --- a/apps/server/test/acquisition-organize/scrape.test.ts +++ b/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 与图片(临时库根);只补缺失 / 全部覆盖; diff --git a/apps/server/test/acquisition-organize/subscription.test.ts b/apps/server/test/acquisition-organize/subscription.test.ts index 9a2630ce..b316c7dd 100644 --- a/apps/server/test/acquisition-organize/subscription.test.ts +++ b/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); diff --git a/apps/server/test/acquisition-organize/takeover.test.ts b/apps/server/test/acquisition-organize/takeover.test.ts index f0ad8a66..d4ff8c36 100644 --- a/apps/server/test/acquisition-organize/takeover.test.ts +++ b/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、贴管理标签,没有站点资源)完成后按 种子名 + 主视频文件名 识别 → 直接整理; diff --git a/apps/server/test/acquisition-organize/vertical.test.ts b/apps/server/test/acquisition-organize/vertical.test.ts index e737b461..13975da5 100644 --- a/apps/server/test/acquisition-organize/vertical.test.ts +++ b/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 { diff --git a/apps/server/test/auth-settings/harness.ts b/apps/server/test/auth-settings/harness.ts index 699303e3..268a2ac5 100644 --- a/apps/server/test/auth-settings/harness.ts +++ b/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(fn: () => Promise): Promise { - 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 = ts.db; const engineSchema = `${schema}_ow`; - await adminQuery(`create schema ${schema}`); - const db = new Kysely({ - 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); }, }; } diff --git a/apps/server/test/browser/browser-identity.test.ts b/apps/server/test/browser/browser-identity.test.ts index e5d1551b..a0a8cb62 100644 --- a/apps/server/test/browser/browser-identity.test.ts +++ b/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, diff --git a/apps/server/test/browser/browser-provider-config.test.ts b/apps/server/test/browser/browser-provider-config.test.ts index beeb5ab5..f395b215 100644 --- a/apps/server/test/browser/browser-provider-config.test.ts +++ b/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) diff --git a/apps/server/test/browser/harness.ts b/apps/server/test/browser/harness.ts index 891d8cfd..433b2656 100644 --- a/apps/server/test/browser/harness.ts +++ b/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({ - 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({ - dialect: new PostgresDialect({ - pool: new pg.Pool({ connectionString: PG_URL, max: 8, options: `-c search_path=${schema},public` }), - }), - }) as unknown as Kysely; - await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: schema })); + const ts = await createTestSchema(uniqueSchema(`c1e_${name}`), { max: 8 }); + const db = ts.db as unknown as Kysely; 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 }); }, diff --git a/apps/server/test/e2e/harness.ts b/apps/server/test/e2e/harness.ts index 1bacec14..8be67c50 100644 --- a/apps/server/test/e2e/harness.ts +++ b/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 { - 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( } export async function createE2E(name: string, opts: E2EOptions = {}): Promise { - 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 = 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({ - 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 { - 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、结算快一点 diff --git a/apps/server/test/file-manager/harness.ts b/apps/server/test/file-manager/harness.ts index 5d97281d..54d1cfb8 100644 --- a/apps/server/test/file-manager/harness.ts +++ b/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 { - 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({ - 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 }); }, }; diff --git a/apps/server/test/inventory/media-servers.test.ts b/apps/server/test/inventory/media-servers.test.ts index 49adc540..69ca7b22 100644 --- a/apps/server/test/inventory/media-servers.test.ts +++ b/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。 diff --git a/apps/server/test/rules/rule-groups.test.ts b/apps/server/test/rules/rule-groups.test.ts index 924c8058..7210ce26 100644 --- a/apps/server/test/rules/rule-groups.test.ts +++ b/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 + 修订、抑制记录、试算、默认链、评估接入、旧端点别名。 diff --git a/apps/server/test/sources-identity/harness.ts b/apps/server/test/sources-identity/harness.ts index a3699558..839c2753 100644 --- a/apps/server/test/sources-identity/harness.ts +++ b/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 { - 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): SecretStore & { plain: Map } { @@ -75,41 +63,8 @@ export function plainSettings(db: () => Kysely): SettingsReader { return { all, get: async (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(); - 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; @@ -130,20 +85,8 @@ export interface HarnessOptions extends RuntimeOverrides { } export async function createHarness(name: string, opts: HarnessOptions = {}): Promise { - const schema = `w2b_${name}_${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`; - const admin = new K({ - 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({ - dialect: new PostgresDialect({ - pool: new pg.Pool({ connectionString: PG_URL, max: 8, options: `-c search_path=${schema},public` }), - }), - }) as unknown as Kysely; - // 与其他测试进程共用 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; 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(); }, }; } diff --git a/apps/server/test/subscriptions/subscriptions-c1a.test.ts b/apps/server/test/subscriptions/subscriptions-c1a.test.ts index 375b237b..75e49a0e 100644 --- a/apps/server/test/subscriptions/subscriptions-c1a.test.ts +++ b/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、库根用于首次整理、下载器实例、重置进度、日历。 diff --git a/apps/server/test/subscriptions/subscriptions-v2.test.ts b/apps/server/test/subscriptions/subscriptions-v2.test.ts index f74a398c..25214608 100644 --- a/apps/server/test/subscriptions/subscriptions-v2.test.ts +++ b/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/旧字段兼容、暂停/恢复/重新订阅/批量/排序。 diff --git a/apps/server/test/support/engine.ts b/apps/server/test/support/engine.ts new file mode 100644 index 00000000..6bb8c614 --- /dev/null +++ b/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(); + 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() {}, + }; +} diff --git a/apps/server/test/support/pg.ts b/apps/server/test/support/pg.ts new file mode 100644 index 00000000..9e9f9b45 --- /dev/null +++ b/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 { + 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(fn: () => Promise): Promise { + 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 { + 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 { + await withSchemaLock(async () => { + for (const s of schemas) await adminQuery(`drop schema if exists ${s} cascade`); + }); +} + +export interface TestSchema { + schema: string; + db: Kysely; + /** 关连接池并删掉 schema(连同 extra,例如引擎 schema) */ + drop(...extra: string[]): Promise; +} + +/** 建独立 schema → 连接池(search_path 指向它)→ 加锁迁移到最新 */ +export async function createTestSchema(schema: string, opts: { max?: number } = {}): Promise { + await adminQuery(`create schema ${schema}`); + const db = new Kysely({ + 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); + }, + }; +} diff --git a/apps/server/vitest.config.ts b/apps/server/vitest.config.ts index cfc08ea7..2d0b2852 100644 --- a/apps/server/vitest.config.ts +++ b/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",