240 lines
13 KiB
JavaScript
240 lines
13 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import {randomUUID} from 'node:crypto';
|
|
import {EventEmitter} from 'node:events';
|
|
import {createRequire, isBuiltin} from 'node:module';
|
|
import {fileURLToPath} from 'node:url';
|
|
import test from 'node:test';
|
|
import {build} from 'esbuild';
|
|
|
|
const root = fileURLToPath(new URL('../', import.meta.url));
|
|
const require = createRequire(import.meta.url);
|
|
async function bundle(mock_catalog = false) {
|
|
const compiled = await build({
|
|
stdin: {contents: `export * from './bot/forward/stream.ts';
|
|
export * from './bot/forward/manager.ts';
|
|
export * from './utils/forward_protocol.ts';
|
|
export * from './bot/workflow/catalog.ts';
|
|
export * from './service/server/update/workflow_catalog.ts';
|
|
export {report_logs} from './utils/console_print.ts';`, resolveDir: root, loader: 'ts'},
|
|
bundle: true, platform: 'node', format: 'cjs', write: false, metafile: true,
|
|
plugins: [{name: 'isolated-node', setup(build) {
|
|
build.onResolve({filter: /(?:instance\/index\.ts|^socket\.io-client$|console_print\.ts|path_utils\.ts)$/}, args =>
|
|
({path: args.path.includes('instance/') ? 'registry' : args.path === 'socket.io-client' ? 'socket'
|
|
: args.path.endsWith('path_utils.ts') ? 'path' : 'console', namespace: 'stub'}));
|
|
if (mock_catalog) build.onResolve({filter: /(?:^|\/)workflow\/catalog\.ts$/}, () =>
|
|
({path: 'catalog', namespace: 'stub'}));
|
|
build.onLoad({filter: /.*/, namespace: 'stub'}, ({path}) => ({loader: 'js', contents: {
|
|
registry: 'export const get_bot_context = () => undefined; export const observe_bot_context = () => () => {};',
|
|
socket: 'export const io = () => {throw Error("Unexpected real socket")};',
|
|
console: 'export const report_logs = []; export const console_print = {success: x => x, log: x => report_logs.push(x)};',
|
|
path: 'export const get_project_root_path = () => {throw Error("Unexpected filesystem access")};',
|
|
catalog: 'let reader; export const set_catalog_reader = value => {reader = value}; export const get_workflow_catalog = (...args) => reader(...args);',
|
|
}[path]}));
|
|
}}],
|
|
});
|
|
assert.ok(Object.keys(compiled.metafile.inputs).every(path => !path.includes('forward_server')));
|
|
const module = {exports: {}};
|
|
new Function('require', 'module', 'exports', compiled.outputFiles[0].text)(specifier => {
|
|
assert.ok(isBuiltin(specifier), `Unexpected external import: ${specifier}`);
|
|
return require(specifier);
|
|
}, module, module.exports);
|
|
return module.exports;
|
|
}
|
|
|
|
const real = await bundle();
|
|
const controlled = await bundle(true);
|
|
const {BotStream, validate_rpc, validate_workflow_payload, error_ack} = real;
|
|
const fails = (action, code = 'INVALID_REQUEST') => assert.throws(action, error => error.code === code);
|
|
const request = (payload = {}, scope = randomUUID(), deadline_at = Date.now() + 1000) => ({
|
|
request_id: randomUUID(), correlation_id: randomUUID(), method: 'workflow.catalog', payload,
|
|
forward_session_id: scope, deadline_at,
|
|
});
|
|
function stream(t, api = real, context, now = () => 100) {
|
|
const value = new api.BotStream(1, {now, source: {get: () => context, observe: () => () => {}}});
|
|
t.after(() => value.dispose());
|
|
return value;
|
|
}
|
|
const sample = () => ({catalog_version: 1, nodes: [{name: 'sample', config: [{name: 'nested', default: {value: 1}}]}],
|
|
start_nodes: [], events: [], queries: [], common: {references: []}});
|
|
|
|
test('catalog RPC accepts only an exact empty payload, including direct stream validation', t => {
|
|
const value = stream(t);
|
|
assert.equal(validate_rpc(request()).method, 'workflow.catalog');
|
|
assert.deepEqual(validate_workflow_payload('workflow.catalog', Object.create(null)), Object.create(null));
|
|
let reads = 0;
|
|
const accessor = Object.defineProperty({}, 'run_id', {enumerable: true, get() { reads++; return randomUUID(); }});
|
|
for (const payload of [undefined, null, [], 1, 'text', new Date(), Object.create({inherited: true}),
|
|
{run_id: randomUUID()}, {run_id: undefined}, {allow_code: true}, {extra: true}, {[Symbol('extra')]: true},
|
|
Object.defineProperty({}, 'hidden', {value: true}), accessor]) {
|
|
const raw = {...request(), payload};
|
|
fails(() => validate_rpc(raw));
|
|
fails(() => validate_workflow_payload('workflow.catalog', payload));
|
|
fails(() => value.workflow_catalog(payload, 200));
|
|
}
|
|
assert.equal(reads, 0);
|
|
fails(() => validate_workflow_payload('workflow.unknown', {}));
|
|
assert.deepEqual(validate_workflow_payload('workflow.snapshot', {}), {});
|
|
const run_id = randomUUID();
|
|
assert.deepEqual(validate_workflow_payload('workflow.snapshot', {run_id}), {run_id});
|
|
fails(() => validate_workflow_payload('workflow.snapshot', {run_id, extra: true}));
|
|
});
|
|
|
|
test('real catalog bundles independently and works offline or without any context, returning isolated clones', t => {
|
|
const absent = stream(t);
|
|
const offline = stream(t, real, Object.assign(new EventEmitter(), {bot_status: 'offline', bot_instance: null}));
|
|
const expected = real.get_workflow_catalog();
|
|
const result = absent.workflow_catalog({}, 200);
|
|
assert.equal(result.catalog_version, 1);
|
|
assert.ok(result.nodes.length > 0);
|
|
assert.deepEqual(result.nodes.find(node => node.name === 'code').availability,
|
|
{available: false, reason: 'CODE_EXECUTION_DISABLED'});
|
|
assert.deepEqual(result, expected);
|
|
assert.deepEqual(offline.workflow_catalog({}, 200), expected);
|
|
assert.notEqual(result, expected);
|
|
result.nodes[0].name = 'mutated';
|
|
result.common.references.length = 0;
|
|
assert.deepEqual(absent.workflow_catalog({}, 200), expected);
|
|
});
|
|
|
|
test('catalog reads pass no trusted code option and clone nested catalog values', t => {
|
|
const original = sample();
|
|
controlled.set_catalog_reader((...args) => { assert.deepEqual(args, []); return original; });
|
|
const value = stream(t, controlled);
|
|
const result = value.workflow_catalog({}, 200);
|
|
assert.deepEqual(result, original);
|
|
result.nodes[0].config[0].default.value = 2;
|
|
assert.equal(original.nodes[0].config[0].default.value, 1);
|
|
});
|
|
|
|
test('catalog deadlines, transport and disposal are checked before and after the read', t => {
|
|
let now = 100;
|
|
let reads = 0;
|
|
controlled.set_catalog_reader(() => { reads++; return sample(); });
|
|
const value = stream(t, controlled, undefined, () => now);
|
|
for (const deadline of [100, 99, NaN, Infinity, 1.5, undefined])
|
|
fails(() => value.workflow_catalog({}, deadline), 'TIMEOUT');
|
|
const stale = () => { throw Object.assign(Error('SECRET transport'), {code: 'SESSION_STALE'}); };
|
|
fails(() => value.workflow_catalog({}, 200, stale), 'SESSION_STALE');
|
|
assert.equal(reads, 0);
|
|
let checks = 0;
|
|
controlled.set_catalog_reader(() => { assert.equal(checks, 1); now = 200; return sample(); });
|
|
fails(() => value.workflow_catalog({}, 200, () => checks++), 'TIMEOUT');
|
|
assert.equal(checks, 2);
|
|
now = 100;
|
|
controlled.set_catalog_reader(() => sample());
|
|
checks = 0;
|
|
fails(() => value.workflow_catalog({}, 200, () => { if (++checks === 2) stale(); }), 'SESSION_STALE');
|
|
controlled.set_catalog_reader(() => { value.dispose(); return sample(); });
|
|
fails(() => value.workflow_catalog({}, 200), 'CANCELLED');
|
|
fails(() => value.workflow_catalog({}, 200), 'CANCELLED');
|
|
});
|
|
|
|
async function relay(t, value) {
|
|
const command = {forward_id: 1, forward_session_id: randomUUID(), node_session_id: 'session', node_id: 1, bot_id: 1,
|
|
url: 'http://127.0.0.1:1234', path: '/socket.io', namespace: '/', token: 'opaque',
|
|
expires_at: Date.now() + 10_000, deadline_at: Date.now() + 2000};
|
|
const socket = new EventEmitter();
|
|
socket.auth = {};
|
|
socket.io = {engine: {writeBuffer: []}};
|
|
socket.volatile = {emit() {}};
|
|
socket.connect = () => { socket.connected = true; queueMicrotask(() => socket.emit('connect')); };
|
|
socket.disconnect = () => { socket.connected = false; socket.emit('disconnect'); };
|
|
socket.timeout = () => ({emit(event, _payload, ack) {
|
|
const {forward_id, forward_session_id, node_session_id, node_id, bot_id, expires_at} = command;
|
|
const data = event === 'forward_node_ready' ? {node_ready: true}
|
|
: {forward_id, forward_session_id, node_session_id, node_id, bot_id, expires_at, role: 'node', node_ready: false};
|
|
queueMicrotask(() => ack(null, {status: 'success', data}));
|
|
}});
|
|
const manager = new controlled.NodeForwardManager(1, 'session', {socket_factory: () => socket, stream_factory: () => value});
|
|
t.after(() => manager.stop());
|
|
assert.equal((await manager.ensure(command)).status, 'success');
|
|
const handler = socket.listeners('relay_request')[0];
|
|
return {socket, manager, rpc: (payload = {}, deadline = Date.now() + 1000) => new Promise(resolve =>
|
|
handler(request(payload, command.forward_session_id, deadline), resolve))};
|
|
}
|
|
|
|
test('manager dispatches context-free catalog RPC and sanitizes stale transport and read failures', async t => {
|
|
const catalog = sample();
|
|
let reads = 0;
|
|
controlled.set_catalog_reader(() => { reads++; return catalog; });
|
|
const value = stream(t, controlled, undefined, Date.now);
|
|
const channel = await relay(t, value);
|
|
assert.deepEqual(await channel.rpc(), {status: 'success', data: catalog});
|
|
assert.deepEqual(await channel.rpc({run_id: randomUUID()}), error_ack('INVALID_REQUEST'));
|
|
assert.deepEqual(await channel.rpc({}, Date.now() - 1), error_ack('TIMEOUT'));
|
|
assert.equal(reads, 1);
|
|
controlled.set_catalog_reader(() => { throw Error('SECRET catalog'); });
|
|
assert.deepEqual(await channel.rpc(), error_ack('INTERNAL_ERROR'));
|
|
controlled.set_catalog_reader(() => { channel.socket.disconnect(); return catalog; });
|
|
assert.deepEqual(await channel.rpc(), error_ack('SESSION_STALE'));
|
|
assert.deepEqual(await channel.rpc(), error_ack('SESSION_STALE'));
|
|
});
|
|
|
|
function reporter(t, response = {status: 'success', data: 0}) {
|
|
const reports = [];
|
|
const logs = controlled.report_logs;
|
|
logs.length = 0;
|
|
t.after(() => { logs.length = 0; });
|
|
const catalog = sample();
|
|
controlled.set_catalog_reader((...args) => { assert.deepEqual(args, []); return catalog; });
|
|
const socket = {connected: true, timeout(ms) { assert.equal(ms, 5000); return this; },
|
|
async emitWithAck(...args) { reports.push(args); return response; }};
|
|
const state = {current: true};
|
|
return {socket, state, catalog, reports, logs,
|
|
run: () => controlled.update_workflow_catalog(socket, 7, () => state.current)};
|
|
}
|
|
|
|
test('catalog reporter emits node metadata, accepts success ACK and supports the default session guard', async t => {
|
|
const app = reporter(t);
|
|
await app.run();
|
|
assert.deepEqual(app.reports, [['update_workflow_catalog', {id: 7, workflow_catalog: app.catalog}]]);
|
|
assert.equal(app.logs.length, 1);
|
|
assert.equal(app.logs[0], '工作流目录更新成功');
|
|
app.socket.connected = false;
|
|
await controlled.update_workflow_catalog(app.socket, 7);
|
|
assert.equal(app.reports.length, 1);
|
|
app.socket.connected = true;
|
|
await controlled.update_workflow_catalog(app.socket, 7);
|
|
assert.equal(app.reports.length, 2);
|
|
});
|
|
|
|
test('catalog reporter safely rejects malformed/error ACKs, read errors and transport failures', async t => {
|
|
for (const response of [null, {}, {status: 'unknown'}, {status: 'error', data: 'SECRET rejection'}]) {
|
|
const app = reporter(t, response);
|
|
await assert.rejects(app.run(), error => error.message === 'Workflow catalog report failed' && error.cause === undefined);
|
|
assert.equal(app.logs.length, 0);
|
|
}
|
|
const app = reporter(t);
|
|
app.socket.emitWithAck = async () => undefined;
|
|
await assert.rejects(app.run(), /^Error: Workflow catalog report failed$/);
|
|
app.socket.emitWithAck = async () => { throw Error('SECRET timeout'); };
|
|
await assert.rejects(app.run(), /^Error: Workflow catalog report failed$/);
|
|
controlled.set_catalog_reader(() => { throw Error('SECRET catalog path'); });
|
|
await assert.rejects(app.run(), /^Error: Workflow catalog report failed$/);
|
|
});
|
|
|
|
test('catalog reporter skips stale sessions before emission and ignores stale ACKs and rejections', async t => {
|
|
for (const change of ['disconnect', 'session']) {
|
|
const app = reporter(t);
|
|
const stale = () => { if (change === 'disconnect') app.socket.connected = false; else app.state.current = false; };
|
|
stale();
|
|
await app.run();
|
|
assert.equal(app.reports.length, 0);
|
|
app.socket.connected = app.state.current = true;
|
|
controlled.set_catalog_reader(() => { stale(); return app.catalog; });
|
|
await app.run();
|
|
assert.equal(app.reports.length, 0);
|
|
for (const reject of [false, true]) {
|
|
app.socket.connected = app.state.current = true;
|
|
controlled.set_catalog_reader(() => app.catalog);
|
|
let finish;
|
|
app.socket.emitWithAck = () => new Promise((resolve, fail) => { finish = reject ? fail : resolve; });
|
|
const task = app.run();
|
|
stale();
|
|
finish(reject ? Error('SECRET late ACK') : {status: 'success'});
|
|
await task;
|
|
assert.equal(app.logs.length, 0);
|
|
}
|
|
}
|
|
});
|