mirror of
https://github.com/hansjone/dsh-im-ops.git
synced 2026-10-09 01:53:21 +08:00
330 lines
11 KiB
JavaScript
330 lines
11 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import { execFile } from 'node:child_process';
|
|
import { createServer } from 'node:http';
|
|
import { promisify } from 'node:util';
|
|
import { resolve } from 'node:path';
|
|
import test from 'node:test';
|
|
|
|
import { TelegramApi } from '../../../src/channels/telegram/telegram-api.mjs';
|
|
import { createTelegramHttpTransport } from '../../../src/channels/telegram/telegram-http.mjs';
|
|
import { TelegramRuntime } from '../../../src/channels/telegram/telegram-runtime.mjs';
|
|
|
|
const TOKEN = '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZabcdef123456';
|
|
const execFileAsync = promisify(execFile);
|
|
const packageRoot = resolve(import.meta.dirname, '../../..');
|
|
|
|
function deferred() {
|
|
let resolvePromise;
|
|
let rejectPromise;
|
|
const promise = new Promise((resolveValue, rejectValue) => {
|
|
resolvePromise = resolveValue;
|
|
rejectPromise = rejectValue;
|
|
});
|
|
return { promise, resolve: resolvePromise, reject: rejectPromise };
|
|
}
|
|
|
|
async function bounded(promise, message, timeoutMs = 1_000) {
|
|
let timer;
|
|
try {
|
|
return await Promise.race([
|
|
promise,
|
|
new Promise((_, reject) => {
|
|
timer = setTimeout(() => reject(new Error(message)), timeoutMs);
|
|
}),
|
|
]);
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
}
|
|
|
|
async function listen(server) {
|
|
await new Promise((resolvePromise, rejectPromise) => {
|
|
server.once('error', rejectPromise);
|
|
server.listen(0, '127.0.0.1', resolvePromise);
|
|
});
|
|
const address = server.address();
|
|
return `http://127.0.0.1:${address.port}/`;
|
|
}
|
|
|
|
async function close(server) {
|
|
await new Promise((resolvePromise, rejectPromise) => {
|
|
server.close((error) => (error ? rejectPromise(error) : resolvePromise()));
|
|
});
|
|
}
|
|
|
|
function withLocalhostProxyBypass() {
|
|
const names = ['http_proxy', 'HTTP_PROXY', 'https_proxy', 'HTTPS_PROXY', 'no_proxy', 'NO_PROXY'];
|
|
const previous = Object.fromEntries(names.map((name) => [name, process.env[name]]));
|
|
for (const name of names) delete process.env[name];
|
|
process.env.no_proxy = '127.0.0.1,localhost';
|
|
process.env.NO_PROXY = '127.0.0.1,localhost';
|
|
return () => {
|
|
for (const name of names) {
|
|
if (previous[name] === undefined) delete process.env[name];
|
|
else process.env[name] = previous[name];
|
|
}
|
|
};
|
|
}
|
|
|
|
test('Telegram private transport sends while a long poll remains open', async () => {
|
|
const pollStarted = deferred();
|
|
const releasePoll = deferred();
|
|
const events = [];
|
|
const server = createServer((request, response) => {
|
|
request.resume();
|
|
if (request.url.endsWith('/getUpdates')) {
|
|
events.push('poll-started');
|
|
pollStarted.resolve();
|
|
void releasePoll.promise.then(() => {
|
|
events.push('poll-released');
|
|
response.setHeader('content-type', 'application/json');
|
|
response.end(JSON.stringify({ ok: true, result: [] }));
|
|
});
|
|
return;
|
|
}
|
|
events.push('send-completed');
|
|
response.setHeader('content-type', 'application/json');
|
|
response.end(JSON.stringify({ ok: true, result: true }));
|
|
});
|
|
const baseUrl = await listen(server);
|
|
const restoreProxy = withLocalhostProxyBypass();
|
|
const transport = createTelegramHttpTransport();
|
|
const api = new TelegramApi({
|
|
token: TOKEN,
|
|
baseUrl,
|
|
fetchImpl: transport.fetchImpl,
|
|
FormDataImpl: transport.FormDataImpl,
|
|
});
|
|
let pollSettled = false;
|
|
const poll = api.getUpdates().finally(() => { pollSettled = true; });
|
|
|
|
try {
|
|
await bounded(pollStarted.promise, 'Telegram long poll did not start');
|
|
assert.equal(await bounded(
|
|
api.sendChatAction({ chatId: 42 }),
|
|
'Telegram send was blocked behind its long poll',
|
|
), true);
|
|
assert.equal(pollSettled, false);
|
|
assert.deepEqual(events, ['poll-started', 'send-completed']);
|
|
releasePoll.resolve();
|
|
assert.deepEqual(await poll, []);
|
|
assert.deepEqual(events, ['poll-started', 'send-completed', 'poll-released']);
|
|
} finally {
|
|
releasePoll.resolve();
|
|
await poll.catch(() => undefined);
|
|
await transport.destroy();
|
|
restoreProxy();
|
|
await close(server);
|
|
}
|
|
});
|
|
|
|
test('Telegram private transport is independent from a saturated global dispatcher', async () => {
|
|
const source = String.raw`
|
|
import { createServer } from 'node:http';
|
|
import { Agent, fetch, setGlobalDispatcher } from 'undici';
|
|
import { createTelegramHttpTransport } from './src/channels/telegram/telegram-http.mjs';
|
|
|
|
let releaseHold;
|
|
let markHoldStarted;
|
|
const holdStarted = new Promise((resolve) => { markHoldStarted = resolve; });
|
|
const holdReleased = new Promise((resolve) => { releaseHold = resolve; });
|
|
const server = createServer((request, response) => {
|
|
request.resume();
|
|
if (request.url === '/hold') {
|
|
markHoldStarted();
|
|
void holdReleased.then(() => response.end('held'));
|
|
return;
|
|
}
|
|
response.end('private');
|
|
});
|
|
await new Promise((resolve, reject) => {
|
|
server.once('error', reject);
|
|
server.listen(0, '127.0.0.1', resolve);
|
|
});
|
|
const origin = 'http://127.0.0.1:' + server.address().port;
|
|
const globalAgent = new Agent({ connections: 1 });
|
|
setGlobalDispatcher(globalAgent);
|
|
const occupied = fetch(origin + '/hold');
|
|
await holdStarted;
|
|
const transport = createTelegramHttpTransport();
|
|
let timeout;
|
|
try {
|
|
const response = await Promise.race([
|
|
transport.fetchImpl(origin + '/send'),
|
|
new Promise((_, reject) => { timeout = setTimeout(
|
|
() => reject(new Error('private transport used the saturated global dispatcher')),
|
|
1_000,
|
|
); }),
|
|
]);
|
|
if (await response.text() !== 'private') throw new Error('unexpected private response');
|
|
process.stdout.write('independent');
|
|
} finally {
|
|
clearTimeout(timeout);
|
|
releaseHold();
|
|
await occupied;
|
|
await transport.destroy();
|
|
await globalAgent.destroy();
|
|
await new Promise((resolve, reject) => server.close(
|
|
(error) => error ? reject(error) : resolve(),
|
|
));
|
|
}
|
|
`;
|
|
const childEnv = { ...process.env };
|
|
for (const name of ['http_proxy', 'HTTP_PROXY', 'https_proxy', 'HTTPS_PROXY']) {
|
|
delete childEnv[name];
|
|
}
|
|
childEnv.no_proxy = '127.0.0.1,localhost';
|
|
childEnv.NO_PROXY = '127.0.0.1,localhost';
|
|
const { stdout, stderr } = await execFileAsync(process.execPath, [
|
|
'--input-type=module',
|
|
'--eval',
|
|
source,
|
|
], {
|
|
cwd: packageRoot,
|
|
env: childEnv,
|
|
timeout: 5_000,
|
|
});
|
|
assert.equal(stderr, '');
|
|
assert.equal(stdout, 'independent');
|
|
});
|
|
|
|
test('Telegram API sends real undici multipart data without losing metadata or bytes', async () => {
|
|
const received = deferred();
|
|
const server = createServer((request, response) => {
|
|
const chunks = [];
|
|
request.on('data', (chunk) => chunks.push(chunk));
|
|
request.on('end', () => {
|
|
received.resolve({
|
|
contentType: request.headers['content-type'],
|
|
body: Buffer.concat(chunks),
|
|
});
|
|
response.setHeader('content-type', 'application/json');
|
|
response.end(JSON.stringify({ ok: true, result: { message_id: 901 } }));
|
|
});
|
|
});
|
|
const baseUrl = await listen(server);
|
|
const restoreProxy = withLocalhostProxyBypass();
|
|
const transport = createTelegramHttpTransport();
|
|
const bytes = Buffer.from([0, 1, 2, 3, 0xff, 0x80, 0x41]);
|
|
const api = new TelegramApi({
|
|
token: TOKEN,
|
|
baseUrl,
|
|
fetchImpl: transport.fetchImpl,
|
|
FormDataImpl: transport.FormDataImpl,
|
|
});
|
|
|
|
try {
|
|
assert.deepEqual(await api.sendDocument({
|
|
chatId: -100123,
|
|
replyToMessageId: 44,
|
|
messageThreadId: 55,
|
|
file: {
|
|
fileName: 'result.bin',
|
|
mediaType: 'application/octet-stream',
|
|
bytes,
|
|
},
|
|
}), { message_id: 901 });
|
|
const request = await bounded(received.promise, 'Telegram multipart request was not received');
|
|
assert.match(request.contentType, /^multipart\/form-data; boundary=/);
|
|
assert.ok(request.body.includes(Buffer.from('name="chat_id"\r\n\r\n-100123')));
|
|
assert.ok(request.body.includes(Buffer.from('name="message_thread_id"\r\n\r\n55')));
|
|
assert.ok(request.body.includes(Buffer.from('name="reply_parameters"')));
|
|
assert.ok(request.body.includes(Buffer.from('"message_id":44')));
|
|
assert.ok(request.body.includes(Buffer.from(
|
|
'name="document"; filename="result.bin"\r\nContent-Type: application/octet-stream',
|
|
)));
|
|
assert.ok(request.body.includes(bytes));
|
|
} finally {
|
|
await transport.destroy();
|
|
restoreProxy();
|
|
await close(server);
|
|
}
|
|
});
|
|
|
|
function runtimeState(cursor = 0) {
|
|
return {
|
|
cursor: () => cursor,
|
|
setCursor: async (value) => { cursor = value; },
|
|
hasSeen: () => false,
|
|
markSeen: async () => undefined,
|
|
sessionFor: () => null,
|
|
};
|
|
}
|
|
|
|
function runtimeConfig() {
|
|
return {
|
|
botId: 'telegram_transport_test',
|
|
platformId: '123456789',
|
|
username: 'HarnessBot',
|
|
};
|
|
}
|
|
|
|
function healthyApi() {
|
|
return {
|
|
getMe: async () => ({ id: 123456789, is_bot: true }),
|
|
getWebhookInfo: async () => ({ url: '' }),
|
|
setMyCommands: async () => true,
|
|
setChatMenuButton: async () => true,
|
|
getUpdates: async ({ signal }) => new Promise((_, reject) => {
|
|
if (signal.aborted) reject(signal.reason);
|
|
else signal.addEventListener('abort', () => reject(signal.reason), { once: true });
|
|
}),
|
|
};
|
|
}
|
|
|
|
function fakeTransport() {
|
|
let destroyCalls = 0;
|
|
class FakeFormData {}
|
|
const fetchImpl = async () => undefined;
|
|
return {
|
|
fetchImpl,
|
|
FormDataImpl: FakeFormData,
|
|
destroy: async () => { destroyCalls += 1; },
|
|
get destroyCalls() { return destroyCalls; },
|
|
};
|
|
}
|
|
|
|
test('Telegram Runtime injects and destroys one private transport exactly once', async () => {
|
|
const transport = fakeTransport();
|
|
let apiOptions;
|
|
const runtime = new TelegramRuntime({
|
|
config: runtimeConfig(),
|
|
token: TOKEN,
|
|
harness: { ensureRunning: async () => true },
|
|
state: runtimeState(),
|
|
createHttpTransport: () => transport,
|
|
createApi: (options) => {
|
|
apiOptions = options;
|
|
return healthyApi();
|
|
},
|
|
});
|
|
|
|
await runtime.start();
|
|
assert.equal(apiOptions.fetchImpl, transport.fetchImpl);
|
|
assert.equal(apiOptions.FormDataImpl, transport.FormDataImpl);
|
|
await runtime.stop();
|
|
await runtime.stop();
|
|
assert.equal(transport.destroyCalls, 1);
|
|
});
|
|
|
|
for (const failurePoint of ['getMe', 'createApi']) {
|
|
test(`Telegram Runtime destroys its private transport when ${failurePoint} fails`, async () => {
|
|
const transport = fakeTransport();
|
|
const failure = new Error(`${failurePoint} failed`);
|
|
const runtime = new TelegramRuntime({
|
|
config: runtimeConfig(),
|
|
token: TOKEN,
|
|
harness: { ensureRunning: async () => true },
|
|
state: runtimeState(),
|
|
createHttpTransport: () => transport,
|
|
createApi: failurePoint === 'createApi'
|
|
? () => { throw failure; }
|
|
: () => ({ ...healthyApi(), getMe: async () => { throw failure; } }),
|
|
});
|
|
|
|
await assert.rejects(runtime.start(), (error) => error === failure);
|
|
assert.equal(transport.destroyCalls, 1);
|
|
await runtime.stop();
|
|
assert.equal(transport.destroyCalls, 1);
|
|
});
|
|
}
|