/** * 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 apiPid = await startAPI(mockAI.url, DB_URL, REDIS_HOST, String(REDIS_PORT), REDIS_PW); 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()}`); 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); }); }); });