/** * Integration Test Harness * 启动真实 API + Worker 进程,通过真实 HTTP 和 BullMQ 执行测试。 */ import { spawn, ChildProcess } from 'child_process'; import * as http from 'http'; import { Queue } from 'bullmq'; import * as jwt from 'jsonwebtoken'; // ══════════════════════════════════════════════════════════════════ // JWT helper — matches TokenService.generateAccessToken() // ══════════════════════════════════════════════════════════════════ export function generateJWT(userId: string, email: string, role: string = 'USER', secret?: string): string { const jwtSecret = secret || process.env.JWT_SECRET || 'test-integration-secret'; return jwt.sign({ sub: userId, email, role, type: 'user' }, jwtSecret, { expiresIn: '1h' }); } let redisConn: { host: string; port: number; password?: string } | null = null; export function setRedisConfig(host: string, port: number, password?: string) { redisConn = { host, port, password: password || undefined }; } /** * Enqueue a job directly to BullMQ Redis queue. * Worker process consumes from this queue. */ export async function enqueueJob(queueName: string, data: Record, jobId?: string): Promise { if (!redisConn) throw new Error('Redis config not set — call setRedisConfig() first'); const queue = new Queue(queueName, { connection: redisConn }); const job = await queue.add(queueName, data, { attempts: 3, backoff: { type: 'exponential', delay: 1000 }, ...(jobId ? { jobId } : {}), }); await queue.close(); return job.id as string; } let apiProc: ChildProcess | null = null; let workerProc: ChildProcess | null = null; let mockAIUrl = ''; const API_PORT = 3000; // ══════════════════════════════════════════════════════════════════ // HTTP helpers // ══════════════════════════════════════════════════════════════════ export function httpGet(path: string): Promise<{ status: number; data: any }> { return new Promise((resolve, reject) => { http.get(`http://127.0.0.1:${API_PORT}${path}`, (res) => { let body = ''; res.on('data', (c) => (body += c)); res.on('end', () => { try { resolve({ status: res.statusCode || 0, data: JSON.parse(body) }); } catch { resolve({ status: res.statusCode || 0, data: body }); } }); }).on('error', reject); }); } export function httpPost(path: string, body: unknown, token?: string): Promise<{ status: number; data: any }> { return new Promise((resolve, reject) => { const payload = JSON.stringify(body); const headers: Record = { 'Content-Type': 'application/json', 'Content-Length': String(Buffer.byteLength(payload)), }; if (token) headers['Authorization'] = `Bearer ${token}`; const req = http.request({ hostname: '127.0.0.1', port: API_PORT, path, method: 'POST', headers, }, (res) => { let b = ''; res.on('data', (c) => (b += c)); res.on('end', () => { try { resolve({ status: res.statusCode || 0, data: JSON.parse(b) }); } catch { resolve({ status: res.statusCode || 0, data: b }); } }); }); req.on('error', reject); req.write(payload); req.end(); }); } // ══════════════════════════════════════════════════════════════════ // Process management // ══════════════════════════════════════════════════════════════════ function envForProcess(baseEnv: NodeJS.ProcessEnv): NodeJS.ProcessEnv { return { ...baseEnv, NODE_ENV: 'test', PORT: String(API_PORT), DEEPSEEK_BASE_URL: mockAIUrl, AI_PROVIDER: 'deepseek', DEEPSEEK_API_KEY: 'sk-test', STORAGE_DRIVER: 'local', ENABLE_SWAGGER: 'false', // No Bull Board in test }; } export async function startAPI(mockUrl: string, dbUrl: string, redisHost: string, redisPort: string, redisPw: string): Promise { mockAIUrl = mockUrl; return new Promise((resolve, reject) => { const env = envForProcess(process.env); Object.assign(env, { DATABASE_URL: dbUrl, REDIS_HOST: redisHost, REDIS_PORT: redisPort, REDIS_PASSWORD: redisPw, REDIS_DB: '0', }); apiProc = spawn('node', ['dist/src/main.js'], { env, stdio: ['ignore', 'pipe', 'pipe'] }); let done = false; const onData = (chunk: Buffer) => { if (done) return; const msg = chunk.toString(); // API ready signal if (msg.includes('Server running') || msg.includes('successfully started')) { done = true; resolve(apiProc!.pid!); } }; apiProc.stdout!.on('data', onData); apiProc.stderr!.on('data', onData); setTimeout(() => { if (!done) { done = true; reject(new Error('API timeout')); } }, 30_000); }); } export async function startWorker(dbUrl: string, redisHost: string, redisPort: string, redisPw: string, instanceId: string): Promise { return new Promise((resolve, reject) => { const env = envForProcess(process.env); Object.assign(env, { DATABASE_URL: dbUrl, REDIS_HOST: redisHost, REDIS_PORT: redisPort, REDIS_PASSWORD: redisPw, REDIS_DB: '0', WORKER_INSTANCE_ID: instanceId, }); workerProc = spawn('node', ['dist/src/worker.main.js'], { env, stdio: ['ignore', 'pipe', 'pipe'] }); let done = false; const onData = (chunk: Buffer) => { if (done) return; const msg = chunk.toString(); if (msg.includes('started — waiting for jobs')) { done = true; resolve(workerProc!.pid!); } }; workerProc!.stdout!.on('data', onData); workerProc!.stderr!.on('data', onData); setTimeout(() => { if (!done) { done = true; reject(new Error('Worker timeout')); } }, 30_000); }); } export function getApiPid(): number | undefined { return apiProc?.pid; } export function getWorkerPid(): number | undefined { return workerProc?.pid; } export async function stopWorker(signal: NodeJS.Signals = 'SIGTERM'): Promise { if (!workerProc) return; return new Promise((resolve) => { workerProc!.on('exit', () => resolve()); workerProc!.kill(signal); setTimeout(() => resolve(), 10_000); }); } export async function stopAPI(): Promise { if (!apiProc) return; return new Promise((resolve) => { apiProc!.on('exit', () => resolve()); apiProc!.kill('SIGTERM'); setTimeout(() => resolve(), 10_000); }); } // ══════════════════════════════════════════════════════════════════ // DB helpers (direct PrismaClient, no NestJS DI, no mocks) // ══════════════════════════════════════════════════════════════════ import { PrismaClient } from '@prisma/client'; let prisma: PrismaClient | null = null; export function getPrisma(dbUrl: string): PrismaClient { if (!prisma) { prisma = new PrismaClient({ datasources: { db: { url: dbUrl } } }); } return prisma; } export async function dbExec(dbUrl: string, sql: string, params: any[] = []): Promise { const p = getPrisma(dbUrl); await p.$executeRawUnsafe(sql, ...params); } export async function dbQuery(dbUrl: string, sql: string, params: any[] = []): Promise { const p = getPrisma(dbUrl); return (await p.$queryRawUnsafe(sql, ...params)) as T[]; } export async function dbClose(): Promise { if (prisma) { await prisma.$disconnect(); prisma = null; } } export async function setupFixtures(dbUrl: string, userId: string, kbId: string, kiId: string, sessionId: string): Promise { await dbExec(dbUrl, `INSERT IGNORE INTO User (id, email, nickname, role, status, updatedAt) VALUES (?, ?, ?, ?, ?, NOW())`, [userId, `test-${Date.now()}@zhixi.app`, 'Integration Test', 'USER', 'active']); await dbExec(dbUrl, `INSERT IGNORE INTO KnowledgeBase (id, userId, title, updatedAt) VALUES (?, ?, ?, NOW())`, [kbId, userId, 'Test KB']); await dbExec(dbUrl, `INSERT IGNORE INTO KnowledgeItem (id, userId, knowledgeBaseId, itemType, title, content, status, updatedAt) VALUES (?, ?, ?, ?, ?, ?, ?, NOW())`, [kiId, userId, kbId, 'concept', 'Test Item', 'Test content.', 'active']); await dbExec(dbUrl, `INSERT IGNORE INTO LearningSession (id, userId, knowledgeBaseId, knowledgeItemId, mode, status, startedAt, updatedAt) VALUES (?, ?, ?, ?, ?, ?, NOW(), NOW())`, [sessionId, userId, kbId, kiId, 'active_recall', 'active']); } export async function cleanupFixtures(dbUrl: string, userId: string): Promise { const tables = ['ReviewCard', 'FocusItem', 'AiAnalysisResult', 'AiAnalysisJob', 'LearningSession', 'KnowledgeItem', 'KnowledgeBase']; for (const t of tables) { await dbExec(dbUrl, `DELETE FROM \`${t}\` WHERE userId = ?`, [userId]).catch(() => {}); } await dbExec(dbUrl, `DELETE FROM User WHERE id = ?`, [userId]).catch(() => {}); }