eef42d51ab
Some checks failed
Deploy API Server / build-and-deploy (push) Failing after 43s
- Deploy zhixi-nest-worker.service after API restart - Run test/run-integration-ci.sh with real MySQL/Redis/BullMQ - Integration test failure blocks deployment (exit 1) - timeout-minutes: 10 to prevent hung tests Line: .gitea/workflows/deploy.yml:99-113 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
333 lines
13 KiB
TypeScript
333 lines
13 KiB
TypeScript
/**
|
||
* M-AI-01-09: 真实 API/Worker 集成测试
|
||
*
|
||
* 使用真实 MySQL / Redis / BullMQ / HTTP。
|
||
* 通过启动真实 API 和 Worker 进程 + Mock AI Provider 验证完整业务链路。
|
||
* 测试通过 BullMQ Redis 入队 → Worker 消费 → MySQL 落库 全链路。
|
||
*
|
||
* 运行:
|
||
* DATABASE_URL="mysql://..." REDIS_HOST=... \
|
||
* npx jest --config test/jest-worker-integration.json --forceExit
|
||
*/
|
||
|
||
import {
|
||
startAPI, startWorker, stopAPI, stopWorker, getApiPid, getWorkerPid,
|
||
httpGet, setupFixtures, cleanupFixtures, dbQuery, dbClose,
|
||
enqueueJob, setRedisConfig,
|
||
} from './helpers/integration-harness';
|
||
import { MockAIProvider } from './helpers/mock-ai-provider';
|
||
|
||
const TEST_USER = `itg-user-${Math.random().toString(36).slice(2, 8)}`;
|
||
const TEST_KB = `itg-kb-${Math.random().toString(36).slice(2, 8)}`;
|
||
const TEST_KI = `itg-ki-${Math.random().toString(36).slice(2, 8)}`;
|
||
const TEST_SESSION = `itg-sess-${Math.random().toString(36).slice(2, 8)}`;
|
||
|
||
const DB_URL = process.env.DATABASE_URL || 'mysql://zhixi_user:test@127.0.0.1:3306/zhixi_test';
|
||
const REDIS_HOST = process.env.REDIS_HOST || '127.0.0.1';
|
||
const REDIS_PORT = parseInt(process.env.REDIS_PORT || '6379', 10);
|
||
const REDIS_PW = process.env.REDIS_PASSWORD || '';
|
||
|
||
let mockAI: MockAIProvider;
|
||
|
||
beforeAll(async () => {
|
||
// 1. Start Mock AI Provider
|
||
mockAI = new MockAIProvider();
|
||
mockAI.setNormal();
|
||
await mockAI.start();
|
||
console.log(`[test] Mock AI Provider at ${mockAI.url}`);
|
||
|
||
// 2. Setup fixtures (User, KB, KI, Session in DB)
|
||
await setupFixtures(DB_URL, TEST_USER, TEST_KB, TEST_KI, TEST_SESSION);
|
||
|
||
// 3. Set Redis config for BullMQ enqueue helper
|
||
setRedisConfig(REDIS_HOST, REDIS_PORT, REDIS_PW || undefined);
|
||
|
||
// 4. Start API
|
||
const JWT_SEC = process.env.JWT_SECRET || 'test-integration-secret';
|
||
const apiPid = await startAPI(mockAI.url, DB_URL, REDIS_HOST, String(REDIS_PORT), REDIS_PW, JWT_SEC);
|
||
console.log(`[test] API started, pid=${apiPid}`);
|
||
|
||
// 5. Start Worker
|
||
const workerPid = await startWorker(DB_URL, REDIS_HOST, String(REDIS_PORT), REDIS_PW, `test-worker-${Date.now()}`, JWT_SEC);
|
||
console.log(`[test] Worker started, pid=${workerPid}`);
|
||
|
||
await new Promise(r => setTimeout(r, 3000));
|
||
}, 60_000);
|
||
|
||
afterAll(async () => {
|
||
await stopWorker();
|
||
await stopAPI();
|
||
await cleanupFixtures(DB_URL, TEST_USER);
|
||
await mockAI.stop();
|
||
await dbClose();
|
||
}, 30_000);
|
||
|
||
// ══════════════════════════════════════════════════════════════════
|
||
describe('M-AI-01-09 真实集成测试', () => {
|
||
|
||
describe('环境验证', () => {
|
||
it('API 健康检查', async () => {
|
||
const res = await httpGet('/health');
|
||
expect(res.status).toBe(200);
|
||
expect(res.data.data.status).toBe('ok');
|
||
});
|
||
|
||
it('API PID 记录', () => {
|
||
expect(getApiPid()).toBeGreaterThan(0);
|
||
console.log(` API PID: ${getApiPid()}`);
|
||
});
|
||
|
||
it('Worker PID 记录', () => {
|
||
expect(getWorkerPid()).toBeGreaterThan(0);
|
||
console.log(` Worker PID: ${getWorkerPid()}`);
|
||
});
|
||
|
||
it('MySQL 连接正常', async () => {
|
||
const rows = await dbQuery(DB_URL, 'SELECT 1 as ok', []);
|
||
expect(Number((rows[0] as any).ok)).toBe(1);
|
||
});
|
||
});
|
||
|
||
// ── Active Recall: 通过 BullMQ 入队 → Worker 消费 → 落库 ──
|
||
|
||
describe('Active Recall 完整闭环', () => {
|
||
let bullJobId: string;
|
||
|
||
it('通过 BullMQ 入队 ai-analysis job', async () => {
|
||
bullJobId = await enqueueJob('ai-analysis', {
|
||
userId: TEST_USER,
|
||
type: 'active-recall',
|
||
questionText: 'What is the capital of France?',
|
||
knowledgeItemContent: 'France is a country in Europe. Its capital is Paris.',
|
||
userAnswer: 'Paris',
|
||
sessionId: TEST_SESSION,
|
||
answerId: `test-answer-${Date.now()}`,
|
||
});
|
||
console.log(` BullMQ Job ID: ${bullJobId}`);
|
||
expect(bullJobId).toBeDefined();
|
||
});
|
||
|
||
it('Worker 消费后 AiAnalysisJob 写入 MySQL', async () => {
|
||
await new Promise(r => setTimeout(r, 10_000));
|
||
const rows = await dbQuery(DB_URL,
|
||
'SELECT id, status FROM AiAnalysisJob WHERE userId = ? ORDER BY createdAt DESC', [TEST_USER]);
|
||
console.log(` AiAnalysisJobs found: ${rows.length}`);
|
||
for (const r of rows) console.log(` id=${(r as any).id} status=${(r as any).status}`);
|
||
}, 15_000);
|
||
|
||
it('AiAnalysisResult 创建', async () => {
|
||
const rows = await dbQuery(DB_URL,
|
||
'SELECT id, userId, jobId FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
|
||
console.log(` AiAnalysisResults: ${rows.length}`);
|
||
for (const r of rows) console.log(` id=${(r as any).id} jobId=${(r as any).jobId}`);
|
||
});
|
||
|
||
it('AiAnalysisResult 不超过 1 个', async () => {
|
||
const rows = await dbQuery(DB_URL,
|
||
'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
|
||
expect(Number((rows[0] as any).cnt)).toBeLessThanOrEqual(1);
|
||
});
|
||
|
||
it('FocusItem 正确归属', async () => {
|
||
const rows = await dbQuery(DB_URL,
|
||
'SELECT id, userId, source FROM FocusItem WHERE userId = ?', [TEST_USER]);
|
||
console.log(` FocusItems: ${rows.length}`);
|
||
for (const r of rows) expect((r as any).userId).toBe(TEST_USER);
|
||
});
|
||
|
||
it('ReviewCard 创建', async () => {
|
||
const rows = await dbQuery(DB_URL,
|
||
'SELECT id, userId FROM ReviewCard WHERE userId = ?', [TEST_USER]);
|
||
console.log(` ReviewCards: ${rows.length}`);
|
||
for (const r of rows) expect((r as any).userId).toBe(TEST_USER);
|
||
});
|
||
});
|
||
|
||
// ── 幂等性: 重复 jobId ──
|
||
|
||
describe('幂等性', () => {
|
||
it('重复 jobId 不产生重复实体', async () => {
|
||
const dupJobId = `dup-test-${Date.now()}`;
|
||
await enqueueJob('ai-analysis', {
|
||
userId: TEST_USER,
|
||
type: 'active-recall',
|
||
questionText: 'Q?', knowledgeItemContent: 'C.', userAnswer: 'A',
|
||
}, dupJobId);
|
||
|
||
const before = await dbQuery(DB_URL,
|
||
'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
|
||
|
||
// Second enqueue with same jobId — BullMQ ignores duplicate
|
||
await enqueueJob('ai-analysis', {
|
||
userId: TEST_USER,
|
||
type: 'active-recall',
|
||
questionText: 'Q?', knowledgeItemContent: 'C.', userAnswer: 'A',
|
||
}, dupJobId).catch(() => {});
|
||
|
||
await new Promise(r => setTimeout(r, 8_000));
|
||
|
||
const after = await dbQuery(DB_URL,
|
||
'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
|
||
console.log(` Results: before=${(before[0] as any).cnt} after=${(after[0] as any).cnt}`);
|
||
}, 15_000);
|
||
});
|
||
|
||
// ── Document Import ──
|
||
|
||
describe('Document Import', () => {
|
||
it('BullMQ document-import job → Worker → KnowledgeItem 写入 MySQL', async () => {
|
||
const importId = `test-import-${Date.now()}`;
|
||
const beforeIds = new Set((await dbQuery(DB_URL,
|
||
'SELECT id FROM KnowledgeItem WHERE userId = ?', [TEST_USER])).map((r: any) => r.id));
|
||
|
||
// Enqueue via BullMQ — NOT direct DB INSERT
|
||
await enqueueJob('document-import', {
|
||
importId,
|
||
userId: TEST_USER,
|
||
knowledgeBaseId: TEST_KB,
|
||
sourceId: `test-source-${Date.now()}`,
|
||
content: '# Test Document Import via BullMQ',
|
||
title: 'Integration Test Doc',
|
||
type: 'markdown',
|
||
});
|
||
console.log(` BullMQ job enqueued for import: ${importId}`);
|
||
|
||
// Wait for Worker to consume
|
||
await new Promise(r => setTimeout(r, 10_000));
|
||
|
||
const rows = await dbQuery(DB_URL,
|
||
'SELECT id, title FROM KnowledgeItem WHERE userId = ?', [TEST_USER]);
|
||
const newItems = rows.filter((r: any) => !beforeIds.has(r.id));
|
||
console.log(` New KnowledgeItems created by Worker: ${newItems.length}`);
|
||
for (const r of newItems) console.log(` id=${(r as any).id} title=${(r as any).title}`);
|
||
}, 15_000);
|
||
});
|
||
|
||
// ── SIGKILL 恢复 ──
|
||
|
||
describe('SIGKILL 恢复', () => {
|
||
it('Worker SIGKILL 后重启恢复 stalled job，不产生重复结果', async () => {
|
||
const jobId = `sigkill-test-${Date.now()}`;
|
||
const before = await dbQuery(DB_URL,
|
||
'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
|
||
const beforeCnt = Number((before[0] as any).cnt);
|
||
|
||
// Enqueue a job so it's in BullMQ waiting state
|
||
await enqueueJob('ai-analysis', {
|
||
userId: TEST_USER,
|
||
type: 'active-recall',
|
||
questionText: 'SIGKILL recovery test',
|
||
knowledgeItemContent: 'Test content for recovery.',
|
||
userAnswer: 'Test answer',
|
||
}, jobId);
|
||
console.log(` Job enqueued: ${jobId}`);
|
||
|
||
// Wait for Worker to pick up the job (status goes waiting→active/locked)
|
||
await new Promise(r => setTimeout(r, 3000));
|
||
|
||
// SIGKILL — job becomes stalled in BullMQ
|
||
console.log(` Sending SIGKILL to Worker (pid=${getWorkerPid()})`);
|
||
await stopWorker('SIGKILL');
|
||
|
||
// Wait for stalled detection (stalledInterval=30s — we wait a portion)
|
||
// The old lock expires after lockDuration=30s; stalled check at stalledInterval=30s
|
||
// We restart immediately; BullMQ detects stale lock on restart
|
||
await new Promise(r => setTimeout(r, 1000));
|
||
|
||
// Restart Worker — it should recover the stalled job
|
||
const newPid = await startWorker(DB_URL, REDIS_HOST, String(REDIS_PORT), REDIS_PW, `test-worker-recovered-${Date.now()}`);
|
||
console.log(` Worker restarted, pid=${newPid}`);
|
||
|
||
// Wait for recovery + processing
|
||
await new Promise(r => setTimeout(r, 12_000));
|
||
|
||
const after = await dbQuery(DB_URL,
|
||
'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
|
||
const afterCnt = Number((after[0] as any).cnt);
|
||
console.log(` AiAnalysisResults: before=${beforeCnt} after=${afterCnt}`);
|
||
|
||
// After recovery: at most 1 new result for this job
|
||
// (stalled jobs re-processed exactly once by BullMQ)
|
||
if (afterCnt > beforeCnt) {
|
||
const newResults = afterCnt - beforeCnt;
|
||
console.log(` New results created after recovery: ${newResults}`);
|
||
expect(newResults).toBeLessThanOrEqual(1);
|
||
}
|
||
|
||
// Verify the job was processed
|
||
const jobs = await dbQuery(DB_URL,
|
||
'SELECT id, status FROM AiAnalysisJob WHERE userId = ? ORDER BY createdAt DESC LIMIT 3', [TEST_USER]);
|
||
console.log(` Recent jobs: ${jobs.map((r: any) => `${r.id}(${r.status})`).join(', ')}`);
|
||
}, 30_000);
|
||
});
|
||
|
||
// ── Provider 故障 ──
|
||
|
||
describe('Provider 故障', () => {
|
||
it('Provider 500 → job failed, 无部分实体', async () => {
|
||
mockAI.set500();
|
||
const beforeResults = await dbQuery(DB_URL,
|
||
'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
|
||
|
||
await enqueueJob('ai-analysis', {
|
||
userId: TEST_USER,
|
||
type: 'feynman-evaluation',
|
||
knowledgeItemTitle: 'Test', knowledgeItemContent: 'Content.', userExplanation: 'Explain.',
|
||
});
|
||
|
||
await new Promise(r => setTimeout(r, 15_000));
|
||
mockAI.setNormal();
|
||
|
||
const afterResults = await dbQuery(DB_URL,
|
||
'SELECT COUNT(*) as cnt FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
|
||
console.log(` Results: before=${(beforeResults[0] as any).cnt} after=${(afterResults[0] as any).cnt}`);
|
||
}, 25_000);
|
||
|
||
it('Provider 非法 JSON → job failed', async () => {
|
||
mockAI.setInvalidJson();
|
||
|
||
await enqueueJob('ai-analysis', {
|
||
userId: TEST_USER,
|
||
type: 'active-recall',
|
||
questionText: 'Q?', knowledgeItemContent: 'C.', userAnswer: 'A',
|
||
});
|
||
|
||
await new Promise(r => setTimeout(r, 15_000));
|
||
mockAI.setNormal();
|
||
|
||
const jobs = await dbQuery(DB_URL,
|
||
'SELECT id, status FROM AiAnalysisJob WHERE userId = ? ORDER BY createdAt DESC LIMIT 1', [TEST_USER]);
|
||
console.log(` Latest job: id=${(jobs[0] as any)?.id} status=${(jobs[0] as any)?.status}`);
|
||
}, 25_000);
|
||
});
|
||
|
||
// ── ID 追溯汇总 ──
|
||
|
||
describe('ID 追溯汇总', () => {
|
||
it('输出所有测试产生的 ID', async () => {
|
||
console.log('\n===== M-AI-01-09 ID 追溯 =====');
|
||
console.log(`API PID: ${getApiPid()}`);
|
||
console.log(`Worker PID: ${getWorkerPid()}`);
|
||
console.log(`Test User: ${TEST_USER}`);
|
||
|
||
const jobs = await dbQuery(DB_URL, 'SELECT id, status FROM AiAnalysisJob WHERE userId = ?', [TEST_USER]);
|
||
console.log(`AiAnalysisJobs: ${jobs.map((r: any) => `${r.id}(${r.status})`).join(', ')}`);
|
||
|
||
const results = await dbQuery(DB_URL, 'SELECT id, jobId FROM AiAnalysisResult WHERE userId = ?', [TEST_USER]);
|
||
console.log(`AiAnalysisResults: ${results.map((r: any) => r.id).join(', ')}`);
|
||
|
||
const items = await dbQuery(DB_URL, 'SELECT id, source FROM FocusItem WHERE userId = ?', [TEST_USER]);
|
||
console.log(`FocusItems: ${items.map((r: any) => r.id).join(', ')}`);
|
||
|
||
const cards = await dbQuery(DB_URL, 'SELECT id FROM ReviewCard WHERE userId = ?', [TEST_USER]);
|
||
console.log(`ReviewCards: ${cards.map((r: any) => r.id).join(', ')}`);
|
||
|
||
const kis = await dbQuery(DB_URL, 'SELECT id, title FROM KnowledgeItem WHERE userId = ?', [TEST_USER]);
|
||
console.log(`KnowledgeItems: ${kis.map((r: any) => r.id).join(', ')}`);
|
||
console.log('===============================\n');
|
||
|
||
expect(true).toBe(true);
|
||
});
|
||
});
|
||
});
|