bot-28/tests/Regression/admin_realtime_probe.mjs

import assert from 'node:assert/strict';
import { readFileSync } from 'node:fs';
import { test } from 'node:test';

const source = readFileSync(new URL('../../public/static/js/realtime-probe.js', import.meta.url));
const { RealtimeProbe, ProbeError, apiAddress, apiEndpoint, redact } = await import(`data:text/javascript;base64,${source.toString('base64')}`);
const A = 'admin.orders.withdraw';
const B = 'admin.users.online';
const flush = () => new Promise((resolve) => setImmediate(resolve));
const event = (seq, channel = A) => ({ seq, id: seq, channel, event: channel === A ? 'order.updated' : 'user.online',
  action: channel === A ? 'updated' : 'online', key: { id: 88 }, data: { status: 1 } });
const config = { code: 0, data: { key: 'fixture-key', ws_host: 'localhost', ws_port: 8080, force_tls: false,
  channels: [A, B, 'admin.stats.platform_count', 'admin.stats.unprocessed'], last_seq: 0, session_expires_at: 1900000000, auth_endpoint: '/admin/broadcasting/auth' } };

class Subscription {
  constructor() { this.listeners = new Map(); }
  listen(name, callback) { this.listeners.set(name, callback); return this; }
  subscribed(callback) { this.ready = callback; return this; }
  error(callback) { this.failed = callback; return this; }
  emit(message) { this.listeners.get(`.${message.event}`)?.(message); }
}
class Echo {
  constructor(options) {
    this.options = options; this.channels = new Map(); this.id = '1.1';
    this.connector = { pusher: { connection: { bind: (name, callback) => { this[name] = callback; } } } };
  }
  private(name) { if (!this.channels.has(name)) this.channels.set(name, new Subscription()); return this.channels.get(name); }
  leave(name) { this.channels.delete(name); }
  socketId() { return this.id; }
  disconnect() { this.state_change?.({ current: 'disconnected' }); }
}
const deferred = () => { let resolve; const promise = new Promise((done) => { resolve = done; }); return { resolve, promise }; };

async function setup(request) {
  const logs = [], events = [], requests = [];
  const probe = new RealtimeProbe({ Echo, Pusher: function Pusher() {}, onLog: (entry) => logs.push(entry), onEvent: (entry) => events.push(entry),
    request: async (url, options) => {
      requests.push({ url, options });
      if (request) return request(url, options);
      if (url.includes('/config')) return structuredClone(config);
      if (url.includes('/auth')) return { auth: 'fixture-signature' };
      return { code: 0, data: { last_seq: 0, resync: false, events: [] } };
    } });
  await probe.configure('https://api.example.test/', 'Bearer fixture-token');
  return { probe, logs, events, requests };
}

test('configuration and historical replay never count as real-time delivery', async () => {
  const { probe } = await setup();
  probe.connect([A]);
  probe.echo.state_change({ current: 'connected' });
  probe.echo.private(A).ready(); await flush();
  probe.receive(event(1), 'replay');
  assert.equal(probe.state.liveCount, 0);
  assert.equal(probe.state.replayCount, 1);
  probe.echo.private(A).emit(event(2));
  assert.equal(probe.state.liveCount, 1);
});

test('manual reconnect retains each channel resume point and ignores a stale HTTP completion', async () => {
  const old = deferred(); let replayCalls = 0;
  const { probe, events, requests } = await setup((url) => {
    if (url.includes('/config')) return structuredClone(config);
    if (++replayCalls === 1) return old.promise;
    return { code: 0, data: { resync: false, last_seq: 2, events: [event(2)] } };
  });
  probe.connect([A]); const first = probe.echo;
  first.private(A).ready();
  first.private(A).emit(event(1));
  probe.disconnect();
  old.resolve({ code: 0, data: { resync: false, last_seq: 10, events: [event(10)] } });
  await flush();
  probe.connect([A]); probe.echo.private(A).ready(); await flush();
  assert.equal(new URL(requests.at(-1).url).searchParams.get('after_seq'), '0');
  assert.deepEqual(events.map((e) => e.event.seq), [1, 2]);
});

test('changing credentials invalidates in-flight responses and clears old session data', async () => {
  const response = deferred();
  const { probe } = await setup();
  probe.request = () => response.promise;
  const loading = probe.configure('https://api.example.test', 'new-token');
  probe.invalidate();
  response.resolve(structuredClone(config));
  assert.equal(await loading, null);
  assert.equal(probe.state.config, null);
  assert.equal(probe.token, '');
  assert.equal(probe.channels.size, 0);
});

test('duplicate detection is scoped by seq and channel, not just seq', async () => {
  const { probe, events } = await setup();
  probe.receive(event(1), 'live'); probe.receive(event(1), 'replay'); probe.receive(event(1, B), 'live');
  assert.deepEqual(events.map((e) => e.duplicate), [false, true, false]);
  assert.equal(probe.state.duplicateCount, 1);
});

test('rejected authorization stays visible and 401 stops the expired session', async () => {
  const { probe, logs } = await setup();
  probe.connect([A]);
  probe.request = async () => { throw new ProbeError('expired', 401); };
  let callbackError;
  await probe.authorize(`private-${A}`, '1.1', (error) => { callbackError = error; }, probe.epoch);
  assert.equal(callbackError.status, 401);
  assert.equal(probe.state.expired, true);
  assert.equal(probe.echo, null);
  assert.throws(() => probe.connect([A]), /登录已过期/);
  assert.ok(logs.some((entry) => entry.message.includes('登录已过期')));
});

test('applying subscriptions twice does not add duplicate handlers; leaving stops that channel', async () => {
  const { probe, events } = await setup();
  probe.connect([A]); const sub = probe.echo.private(A);
  probe.subscribe([A]);
  sub.emit(event(1)); assert.equal(events.length, 1);
  probe.subscribe([]); sub.emit(event(2)); assert.equal(events.length, 1);
});

test('redaction removes credentials from log messages and nested event payloads', () => {
  const token = 'fixture-token';
  const safe = redact({ message: `Echo failed for Bearer ${token}`, data: { token, password: 'hidden', text: token } }, token);
  assert.equal(JSON.stringify(safe).includes(token), false);
  assert.equal(safe.data.password, '[已隐藏]');
});

test('base and auth URLs cannot carry credentials or change the request origin', () => {
  assert.throws(() => apiAddress('https://name:password@example.test'), /账号/);
  assert.throws(() => apiAddress('javascript:alert(1)'), /http/);
  assert.throws(() => apiEndpoint('https://api.example.test', '//other.example.test/auth'), /路径/);
  assert.equal(apiEndpoint('https://api.example.test/base', '/admin/broadcasting/auth'), 'https://api.example.test/base/admin/broadcasting/auth');
});

test('invalid reconfiguration does not leave a previous usable configuration', async () => {
  const { probe } = await setup();
  await assert.rejects(probe.configure('not-an-address', 'new-token'), /完整 API 地址/);
  assert.equal(probe.state.config, null);
  assert.throws(() => probe.connect([A]), /先获取/);
});

test('statistics channels receive stats.updated snapshots without treating them as list rows', async () => {
  const { probe, events } = await setup();
  const name = 'admin.stats.platform_count';
  probe.connect([name]);
  probe.echo.private(name).emit({ seq: 1, channel: name, event: 'stats.updated', action: 'updated', key: { scope: 'all' },
    data: { total: 2, apple: 1, android: 0, pc: 1, mobile: 0, wechat: 0, unknown: 0 } });
  assert.equal(probe.state.liveCount, 1);
  assert.equal(events[0].event.data.total, 2);
  assert.deepEqual(events[0].event.key, { scope: 'all' });
});