108 lines
5.1 KiB
JavaScript
108 lines
5.1 KiB
JavaScript
'use strict';
|
|
|
|
const test = require('node:test');
|
|
const assert = require('node:assert/strict');
|
|
const fs = require('node:fs');
|
|
const os = require('node:os');
|
|
const path = require('node:path');
|
|
const { createTransferService } = require('../../src/services/transfer-service');
|
|
|
|
test('server transfer queue reports progress, supports pause/cancel and caps concurrency', async () => {
|
|
const updates = [];
|
|
const service = createTransferService({
|
|
ftp: {},
|
|
broadcast: (channel, state) => { if (channel === 'transfer:update') updates.push(state); },
|
|
logger: { info() {}, warn() {}, error() {} },
|
|
getCredentials: () => ({}),
|
|
});
|
|
|
|
service.pause();
|
|
const protectedJob = service.enqueue('external', { label: 'Upstreamtaak', cancellable: false }, async () => ({ ok: true }));
|
|
assert.equal(service.cancel(protectedJob.jobId), false);
|
|
const cancelled = service.enqueue('upload', { label: 'Annuleren' }, async () => ({ ok: true }));
|
|
const cancelledResult = cancelled.promise.catch(error => error.message);
|
|
assert.equal(service.state().queuedCount, 2);
|
|
assert.equal(service.cancel(cancelled.jobId), true);
|
|
assert.match(await cancelledResult, /geannuleerd/i);
|
|
assert.equal(service.state().jobs.find(job => job.id === cancelled.jobId).status, 'cancelled');
|
|
|
|
service.resume();
|
|
await protectedJob.promise;
|
|
const completed = service.enqueue('upload', { label: 'Payload' }, async ({ progress }) => {
|
|
progress({ bytesOverall: 5, size: 10, name: 'payload.elf' });
|
|
progress({ bytesOverall: 10, size: 10, name: 'payload.elf' });
|
|
return { ok: true };
|
|
});
|
|
assert.deepEqual(await completed.promise, { ok: true });
|
|
const done = service.state().jobs.find(job => job.id === completed.jobId);
|
|
assert.equal(done.status, 'done');
|
|
assert.equal(done.progress.percent, 100);
|
|
assert.equal(done.progress.file, 'payload.elf');
|
|
assert.equal(service.state().throughput24h.jobs, 2);
|
|
assert.ok(service.state().throughput24h.bytes >= 10);
|
|
assert.ok(updates.some(state => state.activeCount === 1));
|
|
|
|
assert.equal(service.setConcurrent(99), 4);
|
|
assert.equal(service.state().maxConcurrent, 4);
|
|
service.clearDone();
|
|
assert.equal(service.state().jobs.length, 0);
|
|
});
|
|
|
|
test('durable transfer queue restores interrupted jobs and requires an explicit safe retry', async () => {
|
|
const root = fs.mkdtempSync(path.join(os.tmpdir(), 'chimera-transfer-'));
|
|
const storageFile = path.join(root, 'transfer-queue.json');
|
|
const logger = { info() {}, warn() {}, error() {} };
|
|
try {
|
|
const first = createTransferService({ ftp: {}, broadcast() {}, logger, getCredentials: () => ({}), storageFile });
|
|
first.registerExecutor('fixture', descriptor => async () => ({ value: descriptor.value }));
|
|
first.pause();
|
|
first.setPolicy({ windowStart: '23:00', windowEnd: '06:00' });
|
|
const queued = first.enqueue('download', {
|
|
label: 'Duurzame test',
|
|
descriptor: { kind: 'fixture', value: 42 },
|
|
idempotencyKey: 'fixture:42',
|
|
provenance: 'test',
|
|
});
|
|
const duplicate = first.enqueue('download', {
|
|
label: 'Dubbel', descriptor: { kind: 'fixture', value: 42 }, idempotencyKey: 'fixture:42',
|
|
});
|
|
assert.equal(duplicate.deduplicated, true);
|
|
assert.equal(duplicate.jobId, queued.jobId);
|
|
first.close();
|
|
assert.equal(fs.existsSync(storageFile), true);
|
|
|
|
let executions = 0;
|
|
const second = createTransferService({ ftp: {}, broadcast() {}, logger, getCredentials: () => ({}), storageFile });
|
|
second.registerExecutor('fixture', descriptor => async () => { executions += 1; return { value: descriptor.value }; });
|
|
const restored = second.state();
|
|
assert.equal(restored.jobs.length, 1);
|
|
assert.equal(restored.jobs[0].id, queued.jobId);
|
|
assert.equal(restored.jobs[0].status, 'interrupted');
|
|
assert.equal(restored.jobs[0].retryable, true);
|
|
assert.deepEqual(restored.policy, { windowStart: '23:00', windowEnd: '06:00', bandwidthLimitSupported: false });
|
|
second.setPolicy({ windowStart: null, windowEnd: null });
|
|
second.resume();
|
|
assert.equal(second.retry(queued.jobId), true);
|
|
for (let attempt = 0; attempt < 50 && second.state().jobs[0].status !== 'done'; attempt += 1) await new Promise(resolve => setTimeout(resolve, 5));
|
|
assert.equal(second.state().jobs[0].status, 'done');
|
|
assert.equal(executions, 1);
|
|
second.close();
|
|
} finally { fs.rmSync(root, { recursive: true, force: true }); }
|
|
});
|
|
|
|
test('durable transfer descriptors reject secrets and automatic retries are bounded', async () => {
|
|
let attempts = 0;
|
|
const service = createTransferService({ ftp: {}, broadcast() {}, logger: { info() {}, warn() {}, error() {} }, getCredentials: () => ({}) });
|
|
service.registerExecutor('flaky', () => async () => {
|
|
attempts += 1;
|
|
if (attempts < 2) throw new Error('temporary');
|
|
return { ok: true };
|
|
});
|
|
assert.throws(() => service.enqueue('upload', { descriptor: { kind: 'flaky', password: 'never-persist' } }), /Secrets/);
|
|
const queued = service.enqueue('upload', { descriptor: { kind: 'flaky' }, maxAttempts: 2 });
|
|
assert.deepEqual(await queued.promise, { ok: true });
|
|
assert.equal(attempts, 2);
|
|
assert.equal(service.state().jobs[0].status, 'done');
|
|
service.close();
|
|
});
|