import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'; import type { Server } from 'node:http'; import type { AddressInfo } from 'node:net'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { createAssessorHttpServer } from '../../server/assessor-http-server'; import { RegulatoryCorpusUploadService } from '../../server/regulatory/regulatory-corpus-upload-service'; import type { RegulatoryIngestionService } from '../../server/regulatory/regulatory-rag-service'; import type { RegulatoryMappingService } from '../../server/regulatory/regulatory-mapping-service'; import type { RegulatoryCorpusV2 } from '../../src/domain/regulatory-rag'; import { LocalAssessorRepository } from '../../src/persistence/local-assessor-repository'; const token = 'synthetic-operator-token-at-least-32-characters'; const pdf = Buffer.from('%PDF-synthetic-only'); const path = '/api/v1/reference-data/iso-corpus'; describe('operator ISO corpus upload from Data Hub', () => { let directory: string; let server: Server; let baseUrl: string; let service: RegulatoryCorpusUploadService; let corpusVersion: string; let failEmbedding: boolean; let indexedCount: number; let prepareGate: Promise | undefined; const pruneIndexed = vi.fn(async () => undefined); const activateCorpus = vi.fn(async () => undefined); const ingest = vi.fn(async () => { if (failEmbedding) throw new Error('Synthetic embedding failure.'); indexedCount = 2; return { documentCount: 1, chunkCount: 2, embeddingModel: 'synthetic-model' }; }); beforeEach(async () => { directory = await mkdtemp(join(tmpdir(), 'assessor-iso-upload-')); await writeFile(join(directory, 'index.html'), '
synthetic app
'); corpusVersion = 'synthetic-v1'; failEmbedding = false; indexedCount = 0; prepareGate = undefined; pruneIndexed.mockClear(); activateCorpus.mockClear(); ingest.mockClear(); service = makeService(); await service.initialize(); server = createAssessorHttpServer({ repository: new LocalAssessorRepository(), staticRoot: directory, regulatoryCorpusUploadService: service, isoUploadToken: token }); await new Promise((resolve, reject) => { server.once('error', reject); server.listen(0, '127.0.0.1', resolve); }); baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; }); afterEach(async () => { await service.close(); await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); await rm(directory, { recursive: true, force: true }); }); function makeService(): RegulatoryCorpusUploadService { return new RegulatoryCorpusUploadService({ root: join(directory, 'artifacts', 'regulatory-corpora'), embeddingModel: 'synthetic-model', countIndexed: async () => indexedCount, ingestion: { ingest } as unknown as RegulatoryIngestionService, mapping: { activateCorpus } as unknown as RegulatoryMappingService, pruneIndexed, prepare: async () => { await prepareGate; return ({ schemaVersion: 2, corpusId: 'synthetic-iso', corpusVersion, language: 'de', parserVersion: 'synthetic', generatedAt: '2026-09-23T00:00:00.000Z', sources: [ { standard: 'ISO/IEC 27001', sourceEdition: '2022', nationalEdition: 'synthetic', sha256: 'a'.repeat(64), pageCount: 2 }, { standard: 'ISO/IEC 27002', sourceEdition: '2022', nationalEdition: 'synthetic', sha256: 'b'.repeat(64), pageCount: 3 } ], documents: [{ schemaVersion: 2, corpusId: 'synthetic-iso', corpusVersion, documentId: 'synthetic-control', framework: 'Synthetic', frameworkVersion: '1', controlId: '5.1', language: 'de', title: 'Synthetic control', text: 'Private synthetic control text.' }] } satisfies RegulatoryCorpusV2); } }); } function upload(uploadToken?: string): Promise { const file = { sizeBytes: pdf.length, bytesBase64: pdf.toString('base64') }; return fetch(`${baseUrl}${path}`, { method: 'POST', headers: { 'Content-Type': 'application/json', 'X-Assessor-Request': '1', Origin: baseUrl, ...(uploadToken ? { 'X-Assessor-Iso-Upload-Token': uploadToken } : {}) }, body: JSON.stringify({ iso27001: file, iso27002: file }) }); } it('requires the operator token and never returns the licensed corpus text', async () => { expect((await upload()).status).toBe(403); expect((await upload('wrong-token')).status).toBe(403); expect(ingest).not.toHaveBeenCalled(); const accepted = await upload(token); expect(accepted.status).toBe(202); expect((await accepted.json()).job.status).toBe('running'); await service.close(); const status = await fetch(`${baseUrl}${path}`); expect(status.status).toBe(200); const body = await status.json(); expect(body.active).toMatchObject({ corpusVersion: 'synthetic-v1', documentCount: 1, chunkCount: 2 }); expect(body.indexStatus).toBe('ready'); expect(body.job.status).toBe('succeeded'); expect(JSON.stringify(body)).not.toContain('Private synthetic control text'); expect(JSON.stringify(body)).not.toContain(token); expect((await readFile(join(directory, 'artifacts', 'regulatory-corpora', 'synthetic-v1.json'), 'utf8'))) .toContain('Private synthetic control text'); expect(activateCorpus).toHaveBeenCalledWith('synthetic-iso', 'synthetic-v1'); }); it('keeps the previous version active after a failed replacement and reloads it after restart', async () => { await upload(token); await service.close(); corpusVersion = 'synthetic-v2'; failEmbedding = true; expect((await upload(token)).status).toBe(202); await service.close(); expect(service.status().active?.corpusVersion).toBe('synthetic-v1'); expect(service.status().job?.status).toBe('failed'); expect(activateCorpus).toHaveBeenCalledTimes(1); const restarted = makeService(); await restarted.initialize(); expect(restarted.status().active?.corpusVersion).toBe('synthetic-v1'); expect(activateCorpus).toHaveBeenCalledTimes(2); }); it('reuses a complete active index and rebuilds it from the same upload when Qdrant points are missing', async () => { await upload(token); await service.close(); expect(ingest).toHaveBeenCalledTimes(1); indexedCount = 2; await upload(token); await service.close(); expect(ingest).toHaveBeenCalledTimes(1); indexedCount = 0; expect((await (await fetch(`${baseUrl}${path}`)).json()).indexStatus).toBe('incomplete'); await upload(token); await service.close(); expect(ingest).toHaveBeenCalledTimes(2); }); it('rejects an overlapping second upload with 409 instead of starting a parallel job', async () => { const [first, second] = await Promise.all([upload(token), upload(token)]); expect([first.status, second.status].sort()).toEqual([202, 409]); await service.close(); expect(ingest).toHaveBeenCalledTimes(1); expect(service.status().job?.status).toBe('succeeded'); }); it('prunes superseded corpus versions only after a successful activation', async () => { await upload(token); await service.close(); expect(pruneIndexed).toHaveBeenCalledWith('synthetic-iso', 'synthetic-v1'); corpusVersion = 'synthetic-v2'; failEmbedding = true; await upload(token); await service.close(); expect(pruneIndexed).toHaveBeenCalledTimes(1); }); it('stops a running upload on server shutdown without embedding or activating it', async () => { let release!: () => void; prepareGate = new Promise(resolve => { release = resolve; }); expect((await upload(token)).status).toBe(202); const stopped = service.stop(); release(); await stopped; expect(ingest).not.toHaveBeenCalled(); expect(activateCorpus).not.toHaveBeenCalled(); expect(service.status().job).toMatchObject({ status: 'failed', error: expect.stringContaining('server shutdown') }); }); it('removes original PDFs left by an interrupted process before restoring corpus status', async () => { const temporary = join(directory, 'artifacts', 'regulatory-corpora', '.upload-12345678-1234-1234-1234-123456789abc'); await mkdir(temporary); await writeFile(join(temporary, '27001.pdf'), pdf); const restarted = makeService(); await restarted.initialize(); await expect(stat(temporary)).rejects.toMatchObject({ code: 'ENOENT' }); }); });