mirror of
https://github.com/hansjone/dsh-im-ops.git
synced 2026-10-09 00:33:20 +08:00
405 lines
13 KiB
JavaScript
405 lines
13 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import test from 'node:test';
|
|
|
|
import { DingtalkRuntime } from '../../../src/channels/dingtalk/dingtalk-runtime.mjs';
|
|
import { rememberConnectionTestTarget } from '../../../src/channels/shared/connection-test.mjs';
|
|
|
|
async function eventually(predicate, timeoutMs = 1_000) {
|
|
const deadline = Date.now() + timeoutMs;
|
|
while (Date.now() < deadline) {
|
|
if (predicate()) return;
|
|
await new Promise((resolve) => setTimeout(resolve, 5));
|
|
}
|
|
assert.fail('condition did not become true');
|
|
}
|
|
|
|
function stateFixture() {
|
|
const sessions = new Map();
|
|
const seen = new Set();
|
|
const pending = new Map();
|
|
return {
|
|
hasSeen: (id) => seen.has(id),
|
|
markSeen: async (id) => seen.add(id),
|
|
sessionFor: (key) => sessions.get(key) ?? null,
|
|
setSession: async (key, value) => sessions.set(key, value),
|
|
clearSession: async (key) => sessions.delete(key),
|
|
pendingSenders: () => [...pending.values()],
|
|
pendingSender: (requestId) => pending.get(requestId) ?? null,
|
|
recordPendingSender: async ({ staffId, displayName, lastSeenAt }) => {
|
|
const entry = {
|
|
requestId: `request-${staffId}`,
|
|
staffId,
|
|
displayName,
|
|
requestedAt: lastSeenAt,
|
|
lastSeenAt,
|
|
};
|
|
pending.set(entry.requestId, entry);
|
|
return entry;
|
|
},
|
|
removePendingSenderByStaffId: async (staffId) => {
|
|
const entry = [...pending.values()].find((value) => value.staffId === staffId);
|
|
if (!entry) return false;
|
|
pending.delete(entry.requestId);
|
|
return true;
|
|
},
|
|
};
|
|
}
|
|
|
|
test('runtime sends a DingTalk connection test only through the remembered private webhook', async () => {
|
|
const state = stateFixture();
|
|
const sends = [];
|
|
const client = {
|
|
connected: true,
|
|
socket: { readyState: 1 },
|
|
registerCallbackListener() {},
|
|
async connect() {},
|
|
socketCallBackResponse() {},
|
|
disconnect() {},
|
|
};
|
|
const runtime = new DingtalkRuntime({
|
|
config: { clientId: 'ding-client', approvedSenders: [] },
|
|
clientSecret: 'host-secret',
|
|
harness: { ensureRunning: async () => true },
|
|
state,
|
|
api: { sendText: async (request) => sends.push(request) },
|
|
streamFactory: async () => ({ client, topic: 'robot-topic' }),
|
|
});
|
|
|
|
await runtime.start();
|
|
await assert.rejects(() => runtime.sendConnectionTest('连接测试'), {
|
|
code: 'test-target-unavailable',
|
|
});
|
|
rememberConnectionTestTarget(state, {
|
|
sessionWebhook: 'https://oapi.dingtalk.com/robot/reply?ticket=inbound-private',
|
|
});
|
|
assert.deepEqual(await runtime.sendConnectionTest('连接测试'), { sent: true });
|
|
assert.equal(sends.length, 1);
|
|
assert.equal(sends[0].clientId, 'ding-client');
|
|
assert.equal(sends[0].clientSecret, 'host-secret');
|
|
assert.equal(sends[0].sessionWebhook, 'https://oapi.dingtalk.com/robot/reply?ticket=inbound-private');
|
|
assert.equal(sends[0].text, '连接测试');
|
|
await runtime.stop();
|
|
});
|
|
|
|
test('runtime owns one DWClient, waits for socket OPEN, acknowledges first, and disconnects on stop', async () => {
|
|
const order = [];
|
|
const state = stateFixture();
|
|
await state.recordPendingSender({
|
|
staffId: 'staff-approved',
|
|
displayName: '已批准用户',
|
|
lastSeenAt: '2026-08-15T01:00:00.000Z',
|
|
});
|
|
let callback;
|
|
const client = {
|
|
connected: false,
|
|
socket: { readyState: 0 },
|
|
registerCallbackListener(topic, listener) {
|
|
order.push(['register', topic]);
|
|
callback = listener;
|
|
},
|
|
async connect() {
|
|
order.push(['connect']);
|
|
setTimeout(() => {
|
|
this.connected = true;
|
|
this.socket.readyState = 1;
|
|
}, 10);
|
|
},
|
|
socketCallBackResponse(messageId, body) {
|
|
order.push(['ack', messageId, body]);
|
|
},
|
|
disconnect() {
|
|
order.push(['disconnect']);
|
|
this.connected = false;
|
|
this.socket.readyState = 3;
|
|
},
|
|
};
|
|
const runtime = new DingtalkRuntime({
|
|
config: {
|
|
clientId: 'ding-client',
|
|
approvedSenders: [{ staffId: 'staff-approved' }],
|
|
},
|
|
clientSecret: 'host-secret',
|
|
harness: {
|
|
ensureRunning: async () => order.push(['harness-ready']),
|
|
sessionExists: async () => true,
|
|
createSession: async () => 'session-one',
|
|
ask: async () => {
|
|
order.push(['ask']);
|
|
return 'Harness 回答';
|
|
},
|
|
},
|
|
state,
|
|
api: {
|
|
sendText: async ({ text }) => order.push(['send', text]),
|
|
},
|
|
streamFactory: async ({ clientId, clientSecret }) => {
|
|
assert.equal(clientId, 'ding-client');
|
|
assert.equal(clientSecret, 'host-secret');
|
|
return { client, topic: 'robot-topic' };
|
|
},
|
|
connectPollIntervalMs: 2,
|
|
});
|
|
|
|
const started = await runtime.start();
|
|
assert.equal(started.ready, true);
|
|
assert.equal(started.dingtalkStreamState, 'connected');
|
|
assert.deepEqual(started.pendingSenders, []);
|
|
assert.deepEqual(order.slice(0, 3), [
|
|
['harness-ready'],
|
|
['register', 'robot-topic'],
|
|
['connect'],
|
|
]);
|
|
|
|
callback({
|
|
headers: { messageId: 'callback-one' },
|
|
data: JSON.stringify({
|
|
msgId: 'business-one',
|
|
msgtype: 'text',
|
|
text: { content: '问题' },
|
|
conversationType: '1',
|
|
senderStaffId: 'staff-approved',
|
|
senderNick: '用户',
|
|
sessionWebhook: 'https://oapi.dingtalk.com/robot/reply?ticket=one',
|
|
}),
|
|
});
|
|
assert.deepEqual(order.at(-1), ['ack', 'callback-one', { success: true }]);
|
|
await eventually(() => runtime.status.messagesReplied === 1);
|
|
assert.ok(order.findIndex(([action]) => action === 'ack') < order.findIndex(([action]) => action === 'ask'));
|
|
assert.ok(order.findIndex(([action]) => action === 'ack') < order.findIndex(([action]) => action === 'send'));
|
|
|
|
await runtime.stop();
|
|
assert.deepEqual(order.at(-1), ['disconnect']);
|
|
assert.equal(runtime.status.ready, false);
|
|
assert.equal(runtime.status.dingtalkStreamState, 'idle');
|
|
});
|
|
|
|
test('runtime sends visible-scope messages to Harness without local sender approval', async () => {
|
|
const state = stateFixture();
|
|
const asked = [];
|
|
const sent = [];
|
|
let callback;
|
|
const client = {
|
|
connected: true,
|
|
socket: { readyState: 1 },
|
|
registerCallbackListener(_topic, listener) { callback = listener; },
|
|
async connect() {},
|
|
socketCallBackResponse() {},
|
|
disconnect() {},
|
|
};
|
|
const runtime = new DingtalkRuntime({
|
|
config: { clientId: 'ding-client', approvedSenders: [] },
|
|
clientSecret: 'host-secret',
|
|
harness: {
|
|
ensureRunning: async () => true,
|
|
sessionExists: async () => false,
|
|
createSession: async () => 'session-visible-sender',
|
|
ask: async (sessionId, text) => {
|
|
asked.push({ sessionId, text });
|
|
return '直接回答';
|
|
},
|
|
},
|
|
state,
|
|
api: { sendText: async ({ text }) => sent.push(text) },
|
|
streamFactory: async () => ({ client, topic: 'robot-topic' }),
|
|
});
|
|
await runtime.start();
|
|
callback({
|
|
headers: { messageId: 'callback-pending' },
|
|
data: JSON.stringify({
|
|
msgId: 'business-pending',
|
|
msgtype: 'text',
|
|
text: { content: '请求使用' },
|
|
conversationType: '1',
|
|
senderStaffId: 'raw-staff-id',
|
|
senderNick: '可见范围用户',
|
|
sessionWebhook: 'https://oapi.dingtalk.com/robot/reply?ticket=pending',
|
|
}),
|
|
});
|
|
|
|
await eventually(() => runtime.status.messagesReplied === 1);
|
|
assert.deepEqual(asked, [{
|
|
sessionId: 'session-visible-sender',
|
|
text: '请求使用',
|
|
}]);
|
|
assert.deepEqual(sent, ['直接回答']);
|
|
assert.deepEqual(runtime.pendingSenders(), []);
|
|
await runtime.stop();
|
|
});
|
|
|
|
test('runtime accepts an OPEN DingTalk socket when the SDK registered flag remains false', async () => {
|
|
const client = {
|
|
connected: true,
|
|
registered: false,
|
|
socket: { readyState: 1 },
|
|
registerCallbackListener() {},
|
|
async connect() {},
|
|
socketCallBackResponse() {},
|
|
disconnect() {},
|
|
};
|
|
const runtime = new DingtalkRuntime({
|
|
config: { clientId: 'ding-client', approvedSenders: [] },
|
|
clientSecret: 'host-secret',
|
|
harness: { ensureRunning: async () => true },
|
|
state: stateFixture(),
|
|
api: { sendText: async () => true },
|
|
streamFactory: async () => ({ client, topic: 'robot-topic' }),
|
|
});
|
|
|
|
assert.equal((await runtime.start()).ready, true);
|
|
assert.equal(client.registered, false);
|
|
await runtime.stop();
|
|
});
|
|
|
|
test('runtime never reports ready when connect resolves before the socket opens permanently', async () => {
|
|
let disconnects = 0;
|
|
const client = {
|
|
connected: false,
|
|
socket: { readyState: 0 },
|
|
registerCallbackListener() {},
|
|
async connect() {},
|
|
socketCallBackResponse() {},
|
|
disconnect() { disconnects += 1; },
|
|
};
|
|
const runtime = new DingtalkRuntime({
|
|
config: { clientId: 'ding-client', approvedSenders: [] },
|
|
clientSecret: 'host-secret',
|
|
harness: { ensureRunning: async () => true },
|
|
state: stateFixture(),
|
|
api: { sendText: async () => true },
|
|
streamFactory: async () => ({ client, topic: 'robot-topic' }),
|
|
connectTimeoutMs: 15,
|
|
connectPollIntervalMs: 2,
|
|
logger: { warn() {}, error() {} },
|
|
});
|
|
|
|
await assert.rejects(runtime.start(), /handshake timed out/);
|
|
assert.equal(disconnects, 1);
|
|
assert.equal(runtime.status.ready, false);
|
|
assert.equal(runtime.status.dingtalkStreamState, 'failed');
|
|
});
|
|
|
|
test('runtime bounds a stalled SDK gateway lookup and disconnects a late connection', async () => {
|
|
let finishConnect;
|
|
let disconnects = 0;
|
|
const client = {
|
|
connected: false,
|
|
socket: { readyState: 0 },
|
|
registerCallbackListener() {},
|
|
connect: async () => new Promise((resolve) => { finishConnect = resolve; }),
|
|
socketCallBackResponse() {},
|
|
disconnect() { disconnects += 1; },
|
|
};
|
|
const runtime = new DingtalkRuntime({
|
|
config: { clientId: 'ding-client', approvedSenders: [] },
|
|
clientSecret: 'host-secret',
|
|
harness: { ensureRunning: async () => true },
|
|
state: stateFixture(),
|
|
api: { sendText: async () => true },
|
|
streamFactory: async () => ({ client, topic: 'robot-topic' }),
|
|
connectTimeoutMs: 10,
|
|
connectPollIntervalMs: 2,
|
|
logger: { warn() {}, error() {} },
|
|
});
|
|
|
|
await assert.rejects(runtime.start(), /handshake timed out after 10ms/);
|
|
assert.equal(disconnects, 1);
|
|
finishConnect();
|
|
await eventually(() => disconnects === 2);
|
|
assert.equal(runtime.status.ready, false);
|
|
assert.equal(runtime.status.dingtalkStreamState, 'failed');
|
|
});
|
|
|
|
test('a callback from a stopped stream is not acknowledged or processed', async () => {
|
|
const events = [];
|
|
let callback;
|
|
const client = {
|
|
connected: true,
|
|
socket: { readyState: 1 },
|
|
registerCallbackListener(_topic, listener) { callback = listener; },
|
|
async connect() {},
|
|
socketCallBackResponse() { events.push('ack'); },
|
|
disconnect() { this.connected = false; this.socket.readyState = 3; },
|
|
};
|
|
const runtime = new DingtalkRuntime({
|
|
config: { clientId: 'ding-client', approvedSenders: [{ staffId: 'approved' }] },
|
|
clientSecret: 'host-secret',
|
|
harness: { ensureRunning: async () => true, ask: async () => events.push('ask') },
|
|
state: stateFixture(),
|
|
api: { sendText: async () => events.push('send') },
|
|
streamFactory: async () => ({ client, topic: 'robot-topic' }),
|
|
});
|
|
await runtime.start();
|
|
await runtime.stop();
|
|
|
|
callback({
|
|
headers: { messageId: 'late-callback' },
|
|
data: JSON.stringify({
|
|
msgId: 'late-message',
|
|
msgtype: 'text',
|
|
text: { content: '不应处理' },
|
|
conversationType: '1',
|
|
senderStaffId: 'approved',
|
|
sessionWebhook: 'https://oapi.dingtalk.com/robot/reply?ticket=late',
|
|
}),
|
|
});
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
|
|
assert.deepEqual(events, []);
|
|
});
|
|
|
|
test('stop aborts in-flight Harness work without waiting for the reply timeout', async () => {
|
|
let callback;
|
|
let askStarted;
|
|
const askStartedPromise = new Promise((resolve) => { askStarted = resolve; });
|
|
const client = {
|
|
connected: true,
|
|
socket: { readyState: 1 },
|
|
registerCallbackListener(_topic, listener) { callback = listener; },
|
|
async connect() {},
|
|
socketCallBackResponse() {},
|
|
disconnect() { this.connected = false; this.socket.readyState = 3; },
|
|
};
|
|
const runtime = new DingtalkRuntime({
|
|
config: { clientId: 'ding-client', approvedSenders: [{ staffId: 'approved' }] },
|
|
clientSecret: 'host-secret',
|
|
harness: {
|
|
ensureRunning: async () => true,
|
|
sessionExists: async () => true,
|
|
createSession: async () => 'session-one',
|
|
ask: async (_sessionId, _text, { signal }) => {
|
|
askStarted();
|
|
await new Promise((resolve, reject) => {
|
|
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
|
|
});
|
|
},
|
|
},
|
|
state: stateFixture(),
|
|
api: { sendText: async () => true },
|
|
streamFactory: async () => ({ client, topic: 'robot-topic' }),
|
|
logger: { warn() {}, error() {} },
|
|
});
|
|
await runtime.start();
|
|
callback({
|
|
headers: { messageId: 'callback-hanging' },
|
|
data: JSON.stringify({
|
|
msgId: 'business-hanging',
|
|
msgtype: 'text',
|
|
text: { content: '长时间问题' },
|
|
conversationType: '1',
|
|
senderStaffId: 'approved',
|
|
sessionWebhook: 'https://oapi.dingtalk.com/robot/reply?ticket=hanging',
|
|
}),
|
|
});
|
|
await askStartedPromise;
|
|
|
|
await Promise.race([
|
|
runtime.stop(),
|
|
new Promise((_, reject) => setTimeout(
|
|
() => reject(new Error('runtime stop did not abort Harness work')),
|
|
100,
|
|
)),
|
|
]);
|
|
|
|
assert.equal(runtime.status.ready, false);
|
|
assert.equal(runtime.status.dingtalkStreamState, 'idle');
|
|
});
|