232 lines
9.7 KiB
TypeScript
232 lines
9.7 KiB
TypeScript
/**
|
|
* 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<string, unknown>, jobId?: string): Promise<string> {
|
|
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<string, string> = {
|
|
'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',
|
|
JWT_SECRET: baseEnv.JWT_SECRET || 'test-integration-secret',
|
|
JWT_ACCESS_EXPIRES_IN: '1h',
|
|
CREDENTIAL_ENCRYPTION_KEY: baseEnv.CREDENTIAL_ENCRYPTION_KEY || '0123456789abcdef0123456789abcdef',
|
|
};
|
|
}
|
|
|
|
export async function startAPI(mockUrl: string, dbUrl: string, redisHost: string, redisPort: string, redisPw: string): Promise<number> {
|
|
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<number> {
|
|
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<void> {
|
|
if (!workerProc) return;
|
|
return new Promise((resolve) => {
|
|
workerProc!.on('exit', () => resolve());
|
|
workerProc!.kill(signal);
|
|
setTimeout(() => resolve(), 10_000);
|
|
});
|
|
}
|
|
|
|
export async function stopAPI(): Promise<void> {
|
|
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<void> {
|
|
const p = getPrisma(dbUrl);
|
|
await p.$executeRawUnsafe(sql, ...params);
|
|
}
|
|
|
|
export async function dbQuery<T = any>(dbUrl: string, sql: string, params: any[] = []): Promise<T[]> {
|
|
const p = getPrisma(dbUrl);
|
|
return (await p.$queryRawUnsafe(sql, ...params)) as T[];
|
|
}
|
|
|
|
export async function dbClose(): Promise<void> {
|
|
if (prisma) { await prisma.$disconnect(); prisma = null; }
|
|
}
|
|
|
|
export async function setupFixtures(dbUrl: string, userId: string, kbId: string, kiId: string, sessionId: string): Promise<void> {
|
|
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<void> {
|
|
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(() => {});
|
|
}
|