101 lines
3.6 KiB
JavaScript
101 lines
3.6 KiB
JavaScript
'use strict';
|
|
|
|
const assert = require('assert');
|
|
const fs = require('fs');
|
|
const path = require('path');
|
|
const {
|
|
runImportBrowserBatch,
|
|
normalizeConcurrency,
|
|
DEFAULT_MAX_CONCURRENCY,
|
|
} = require('../src/services/import_browser_batch_service');
|
|
const { _private: { RateLimiter } } = require('../src/pipeline/browser_lifecycle');
|
|
|
|
function sleep(ms) {
|
|
return new Promise(resolve => setTimeout(resolve, ms));
|
|
}
|
|
|
|
async function waitUntil(predicate, timeoutMs = 1000) {
|
|
const startedAt = Date.now();
|
|
while (!predicate()) {
|
|
if (Date.now() - startedAt > timeoutMs) throw new Error('等待条件超时');
|
|
await sleep(5);
|
|
}
|
|
}
|
|
|
|
(async () => {
|
|
assert.strictEqual(normalizeConcurrency(30, 200), 30);
|
|
assert.strictEqual(normalizeConcurrency(999, 200), DEFAULT_MAX_CONCURRENCY);
|
|
assert.strictEqual(normalizeConcurrency(0, 200), 1);
|
|
assert.strictEqual(normalizeConcurrency(30, 4), 4);
|
|
|
|
const accounts = Array.from({ length: 8 }, (_, i) => ({ email: `user${i}@example.com` }));
|
|
let releaseGate;
|
|
const gate = new Promise(resolve => { releaseGate = resolve; });
|
|
let active = 0;
|
|
let maxActive = 0;
|
|
let started = 0;
|
|
const progress = [];
|
|
|
|
const batchPromise = runImportBrowserBatch(accounts, {
|
|
concurrency: 3,
|
|
acquireRequestSlot: async () => {},
|
|
processAccount: async () => {
|
|
started += 1;
|
|
active += 1;
|
|
maxActive = Math.max(maxActive, active);
|
|
await gate;
|
|
active -= 1;
|
|
return { action: 'created' };
|
|
},
|
|
onProgress: ({ completed }) => progress.push(completed),
|
|
});
|
|
|
|
await waitUntil(() => started === 3);
|
|
assert.strictEqual(maxActive, 3, '应同时启动三个账号，而不是逐个串行');
|
|
releaseGate();
|
|
const batchResult = await batchPromise;
|
|
assert.strictEqual(batchResult.completed, accounts.length);
|
|
assert.deepStrictEqual(progress, [1, 2, 3, 4, 5, 6, 7, 8]);
|
|
|
|
const processed = [];
|
|
const errors = [];
|
|
let acquired = 0;
|
|
const isolated = await runImportBrowserBatch(accounts, {
|
|
concurrency: 4,
|
|
needsRequestSlot: (_account, index) => index % 2 === 0,
|
|
acquireRequestSlot: async () => { acquired += 1; },
|
|
processAccount: async (account, index) => {
|
|
processed.push(account.email);
|
|
if (index === 2) throw new Error('单号失败');
|
|
return { action: 'created' };
|
|
},
|
|
onAccountError: (error, account) => errors.push(`${account.email}:${error.message}`),
|
|
});
|
|
assert.strictEqual(processed.length, accounts.length, '单号异常后其余账号仍应继续');
|
|
assert.strictEqual(isolated.results.filter(item => item && item.status === 'rejected').length, 1);
|
|
assert.strictEqual(errors.length, 1);
|
|
assert.strictEqual(acquired, 4);
|
|
|
|
const limiter = new RateLimiter(50);
|
|
const releasedAt = [];
|
|
await Promise.all(Array.from({ length: 5 }, () => limiter.acquire().then(() => releasedAt.push(Date.now()))));
|
|
for (let i = 1; i < releasedAt.length; i += 1) {
|
|
assert.ok(releasedAt[i] - releasedAt[i - 1] >= 12, '限速器不应突发放行整批请求');
|
|
}
|
|
|
|
const taskRunnerSource = fs.readFileSync(
|
|
path.join(__dirname, '../src/workers/task_runner.js'),
|
|
'utf8'
|
|
);
|
|
const importStart = taskRunnerSource.indexOf("case 'import_setup'");
|
|
const importEnd = taskRunnerSource.indexOf('\n default:', importStart);
|
|
const importSection = taskRunnerSource.slice(importStart, importEnd);
|
|
assert.ok(importSection.includes('runImportBrowserBatch(readyAccounts'), '导入流程应接入并发批处理器');
|
|
assert.ok(!importSection.includes('setTimeout(r, 600)'), '导入流程不应再为每个账号固定等待 600ms');
|
|
|
|
console.log('import_browser_batch_service.test.js PASS');
|
|
})().catch(error => {
|
|
console.error(error);
|
|
process.exitCode = 1;
|
|
});
|