From 683d37a35f2054cb85d0ebbaa15b2f8a3a687746 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 13 Aug 2026 00:58:17 -0700 Subject: [PATCH] test(telegram): align ingress fixtures with Bot API (#123059) --- .../telegram/src/polling-session.test.ts | 98 +++++++++++-------- extensions/telegram/src/webhook.test.ts | 86 +++++++++------- 2 files changed, 108 insertions(+), 76 deletions(-) diff --git a/extensions/telegram/src/polling-session.test.ts b/extensions/telegram/src/polling-session.test.ts index 9106506a5cb8..6591c5c34bcf 100644 --- a/extensions/telegram/src/polling-session.test.ts +++ b/extensions/telegram/src/polling-session.test.ts @@ -5,6 +5,7 @@ import path from "node:path"; import { Worker } from "node:worker_threads"; import { expectDefined } from "@openclaw/normalization-core"; import { Bot } from "grammy"; +import type { Update } from "grammy/types"; import type { ChannelAccountSnapshot } from "openclaw/plugin-sdk/channel-contract"; import { DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS as TELEGRAM_SPOOLED_RETRY_MAX_ATTEMPTS } from "openclaw/plugin-sdk/channel-outbound"; import { toErrorObject as toLintErrorObject } from "openclaw/plugin-sdk/error-runtime"; @@ -582,34 +583,52 @@ async function waitForApiMiddleware( throw new Error("Telegram API middleware was not installed"); } -type TestTelegramUpdate = { - update_id: number; - message: { - text: string; - chat: { id: number; type: "private" | "supergroup"; is_forum?: boolean }; - message_thread_id?: number; - is_topic_message?: boolean; - }; +type TestTelegramUpdate = Update & { + message: NonNullable & { text: string }; +}; + +const testTelegramSender = { + id: 111, + is_bot: false as const, + first_name: "Ada", }; function topicUpdate(updateId: number, threadId: number, text: string): TestTelegramUpdate { return { update_id: updateId, message: { + message_id: updateId, + date: 1_736_380_800, + from: testTelegramSender, text, message_thread_id: threadId, is_topic_message: true, - chat: { id: -100, type: "supergroup" }, + chat: { id: -100, type: "supergroup", title: "Test group" }, }, }; } function directUpdate(updateId: number, chatId: number, text: string): TestTelegramUpdate { + const message = { + message_id: updateId, + date: 1_736_380_800, + from: testTelegramSender, + text, + }; + if (chatId < 0) { + return { + update_id: updateId, + message: { + ...message, + chat: { id: chatId, type: "supergroup", title: "Test group" }, + }, + }; + } return { update_id: updateId, message: { - text, - chat: { id: chatId, type: chatId < 0 ? "supergroup" : "private" }, + ...message, + chat: { id: chatId, type: "private", first_name: "Ada" }, }, }; } @@ -1293,9 +1312,10 @@ describe("TelegramPollingSession", () => { const abort = new AbortController(); const handleUpdate = vi.fn(async () => undefined); const init = vi.fn(async () => undefined); + const update = directUpdate(42, 123, "hello"); await writeTelegramSpooledUpdate({ spoolDir: tempDir, - update: { update_id: 42, message: { text: "hello" } }, + update, }); const { createWorker, runPromise } = startIsolatedIngressSession({ @@ -1333,7 +1353,7 @@ describe("TelegramPollingSession", () => { persistenceFloorUpdateId: null, }); expect(init).toHaveBeenCalledBefore(handleUpdate); - expect(handleUpdate).toHaveBeenCalledWith({ update_id: 42, message: { text: "hello" } }); + expect(handleUpdate).toHaveBeenCalledWith(update); }); }); @@ -1342,6 +1362,7 @@ describe("TelegramPollingSession", () => { const abort = new AbortController(); const handleUpdate = vi.fn(async () => undefined); const worker = createListeningIngressWorker(); + const update = directUpdate(42, 123, "hello"); const { runPromise } = startIsolatedIngressSession({ abort, spoolDir: tempDir, @@ -1353,7 +1374,7 @@ describe("TelegramPollingSession", () => { worker.emit({ type: "update", requestId: "write-1", - update: { update_id: 42, message: { text: "hello" } }, + update, queued: 1, }); await waitForTelegramTestState(() => @@ -1362,9 +1383,7 @@ describe("TelegramPollingSession", () => { updateId: 42, }), ); - await waitForTelegramTestState(() => - expect(handleUpdate).toHaveBeenCalledWith({ update_id: 42, message: { text: "hello" } }), - ); + await waitForTelegramTestState(() => expect(handleUpdate).toHaveBeenCalledWith(update)); await waitForTelegramTestState(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]), ); @@ -1487,10 +1506,11 @@ describe("TelegramPollingSession", () => { }); try { await waitForTelegramTestState(() => expect(worker.hasListener()).toBe(true)); + const update = directUpdate(42, 123, "hello"); worker.emit({ type: "update", requestId: "offset-gap", - update: { update_id: 42, message: { text: "hello" } }, + update, queued: 1, }); await waitForTelegramTestState(() => @@ -1528,10 +1548,11 @@ describe("TelegramPollingSession", () => { }); try { await waitForTelegramTestState(() => expect(worker.hasListener()).toBe(true)); + const update = directUpdate(43, 123, "hello"); worker.emit({ type: "update", requestId: "offset-failure", - update: { update_id: 43, message: { text: "hello" } }, + update, queued: 1, }); await waitForTelegramTestState(() => @@ -1566,10 +1587,11 @@ describe("TelegramPollingSession", () => { }); try { await waitForTelegramTestState(() => expect(worker.hasListener()).toBe(true)); + const update = directUpdate(44, 123, "hello"); worker.emit({ type: "update", requestId: "offset-catching-up", - update: { update_id: 44, message: { text: "hello" } }, + update, queued: 1, }); await waitForTelegramTestState(() => @@ -1612,12 +1634,13 @@ describe("TelegramPollingSession", () => { getCommittedUpdateId: firstOffsetPersistence.getCommittedUpdateId, persistUpdateId: firstOffsetPersistence.persistUpdateId, }); + const update = directUpdate(42, 123, "hello"); try { await waitForTelegramTestState(() => expect(firstWorker.hasListener()).toBe(true)); firstWorker.emit({ type: "update", requestId: "first-delivery", - update: { update_id: 42, message: { text: "hello" } }, + update, queued: 1, }); await waitForTelegramTestState(() => @@ -1664,7 +1687,7 @@ describe("TelegramPollingSession", () => { restartWorker.emit({ type: "update", requestId: "restart-replay", - update: { update_id: 42, message: { text: "hello" } }, + update, queued: 1, }); await waitForTelegramTestState(() => @@ -1722,6 +1745,7 @@ describe("TelegramPollingSession", () => { const abort = new AbortController(); const handleUpdate = vi.fn(async () => abort.abort()); const worker = createListeningIngressWorker(); + const update = directUpdate(42, 123, "hello"); const { runPromise } = startIsolatedIngressSession({ abort, spoolDir: tempDir, @@ -1734,7 +1758,7 @@ describe("TelegramPollingSession", () => { worker.emit({ type: "update", requestId: "write-1", - update: { update_id: 42, message: { text: "hello" } }, + update, queued: 1, }); await waitForTelegramTestState(() => @@ -1744,9 +1768,7 @@ describe("TelegramPollingSession", () => { }), ); worker.emit({ type: "spooled", updateId: 42, queued: 1 }); - await waitForTelegramTestState(() => - expect(handleUpdate).toHaveBeenCalledWith({ update_id: 42, message: { text: "hello" } }), - ); + await waitForTelegramTestState(() => expect(handleUpdate).toHaveBeenCalledWith(update)); await waitForTelegramTestState(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]), ); @@ -1791,9 +1813,11 @@ describe("TelegramPollingSession", () => { }, } as TelegramRuntime); + const firstUpdate = directUpdate(1, 123, "pre-seeded"); + const secondUpdate = directUpdate(2, 123, "during-drain"); await writeTelegramSpooledUpdate({ spoolDir: tempDir, - update: { update_id: 1, message: { text: "pre-seeded" } }, + update: firstUpdate, }); const handleUpdate = vi.fn(async () => undefined); const worker = createListeningIngressWorker(); @@ -1812,7 +1836,7 @@ describe("TelegramPollingSession", () => { worker.emit({ type: "update", requestId: "write-2", - update: { update_id: 2, message: { text: "during-drain" } }, + update: secondUpdate, queued: 1, }); expect(worker.ackSpooledUpdate).not.toHaveBeenCalledWith("write-2", expect.anything()); @@ -1827,16 +1851,10 @@ describe("TelegramPollingSession", () => { worker.emit({ type: "spooled", updateId: 2, queued: 1 }); await waitForTelegramTestState(() => - expect(handleUpdate).toHaveBeenCalledWith({ - update_id: 1, - message: { text: "pre-seeded" }, - }), + expect(handleUpdate).toHaveBeenCalledWith(firstUpdate), ); await waitForTelegramTestState(() => - expect(handleUpdate).toHaveBeenCalledWith({ - update_id: 2, - message: { text: "during-drain" }, - }), + expect(handleUpdate).toHaveBeenCalledWith(secondUpdate), ); await waitForTelegramTestState(async () => expect(await pendingUpdateIds(tempDir, "all")).toEqual([]), @@ -1853,9 +1871,10 @@ describe("TelegramPollingSession", () => { await withTempSpool(async (tempDir) => { const abort = new AbortController(); const handleUpdate = vi.fn(async () => undefined); + const update = directUpdate(42, 123, "pre-upgrade pending"); await writeTelegramSpooledUpdate({ spoolDir: tempDir, - update: { update_id: 42, message: { text: "pre-upgrade pending" } }, + update, }); const { createWorker, runPromise } = startIsolatedIngressSession({ @@ -1886,10 +1905,7 @@ describe("TelegramPollingSession", () => { lastUpdateId: null, persistenceFloorUpdateId: 42, }); - expect(handleUpdate).toHaveBeenCalledWith({ - update_id: 42, - message: { text: "pre-upgrade pending" }, - }); + expect(handleUpdate).toHaveBeenCalledWith(update); }); }); diff --git a/extensions/telegram/src/webhook.test.ts b/extensions/telegram/src/webhook.test.ts index 9312576044ab..2d8f54c060ef 100644 --- a/extensions/telegram/src/webhook.test.ts +++ b/extensions/telegram/src/webhook.test.ts @@ -5,6 +5,7 @@ import { createServer, request, type IncomingMessage } from "node:http"; import os from "node:os"; import nodePath from "node:path"; import { setTimeout as sleep } from "node:timers/promises"; +import type { Update } from "grammy/types"; import { DEFAULT_INGRESS_ADOPTION_STALL_MS } from "openclaw/plugin-sdk/channel-outbound"; import { closeOpenClawStateDatabaseForTest, @@ -74,6 +75,23 @@ const TELEGRAM_WEBHOOK_PATH = "/hook"; const WEBHOOK_DRAIN_GUARD_MS = 5; const TELEGRAM_WEBHOOK_RATE_LIMIT_BURST = WEBHOOK_RATE_LIMIT_DEFAULTS.maxRequests + 10; +type TestTelegramMessageUpdate = Update & { + message: NonNullable & { text: string }; +}; + +function telegramMessageUpdate(updateId: number, text: string): TestTelegramMessageUpdate { + return { + update_id: updateId, + message: { + message_id: updateId, + date: 1_736_380_800, + from: { id: 111, is_bot: false, first_name: "Ada" }, + chat: { id: 111, type: "private", first_name: "Ada" }, + text, + }, + }; +} + async function waitForWebhookState( assertion: () => T | Promise, options: { timeout?: number; interval?: number } = {}, @@ -486,16 +504,13 @@ async function postWebhookPayloadWithChunkPlan(params: { function createNearLimitTelegramPayload(): { payload: string; sizeBytes: number } { const maxBytes = 1_024 * 1_024; const targetBytes = maxBytes - 4_096; - const shell = { update_id: 77_777, message: { text: "" } }; + const shell = telegramMessageUpdate(77_777, ""); const shellSize = Buffer.byteLength(JSON.stringify(shell), "utf-8"); const textLength = Math.max(1, targetBytes - shellSize); const pattern = "the quick brown fox jumps over the lazy dog "; const repeats = Math.ceil(textLength / pattern.length); const text = pattern.repeat(repeats).slice(0, textLength); - const payload = JSON.stringify({ - update_id: 77_777, - message: { text }, - }); + const payload = JSON.stringify(telegramMessageUpdate(77_777, text)); return { payload, sizeBytes: Buffer.byteLength(payload, "utf-8") }; } @@ -543,8 +558,8 @@ async function withStartedWebhook( } function expectSingleNearLimitUpdate(params: { - seenUpdates: Array<{ update_id: number; message: { text: string } }>; - expected: { update_id: number; message: { text: string } }; + seenUpdates: TestTelegramMessageUpdate[]; + expected: TestTelegramMessageUpdate; }) { expect(params.seenUpdates).toHaveLength(1); expect(params.seenUpdates[0]?.update_id).toBe(params.expected.update_id); @@ -557,15 +572,15 @@ function expectSingleNearLimitUpdate(params: { async function runNearLimitPayloadTestAndExpectUpdate( mode: "single" | "random-chunked", ): Promise { - const seenUpdates: Array<{ update_id: number; message: { text: string } }> = []; + const seenUpdates: TestTelegramMessageUpdate[] = []; handleUpdateSpy.mockImplementationOnce((update: unknown) => { - seenUpdates.push(update as { update_id: number; message: { text: string } }); + seenUpdates.push(update as TestTelegramMessageUpdate); }); const { payload, sizeBytes } = createNearLimitTelegramPayload(); expect(sizeBytes).toBeLessThan(1_024 * 1_024); expect(sizeBytes).toBeGreaterThan(256 * 1_024); - const expected = JSON.parse(payload) as { update_id: number; message: { text: string } }; + const expected = JSON.parse(payload) as TestTelegramMessageUpdate; await withStartedWebhook( { @@ -1046,7 +1061,7 @@ describe("startTelegramWebhook", () => { ); expect(botParams.accountId).toBe("opie"); expect(requireRecord(botParams.config, "telegram config").bindings).toEqual([]); - const payload = JSON.stringify({ update_id: 1, message: { text: "hello" } }); + const payload = JSON.stringify(telegramMessageUpdate(1, "hello")); const response = await postWebhookJson({ url: webhookUrl(port, TELEGRAM_WEBHOOK_PATH), payload, @@ -1060,11 +1075,12 @@ describe("startTelegramWebhook", () => { }); it("acks before webhook update processing finishes", async () => { + const slowUpdate = telegramMessageUpdate(2, "slow"); let finishWork: (() => void) | undefined; let workStarted = false; let workFinished = false; handleUpdateSpy.mockImplementationOnce(async (update: unknown) => { - expect(update).toEqual({ update_id: 2, message: { text: "slow" } }); + expect(update).toEqual(telegramMessageUpdate(2, "slow")); workStarted = true; await new Promise((resolve) => { finishWork = resolve; @@ -1080,7 +1096,7 @@ describe("startTelegramWebhook", () => { async ({ port }) => { const response = await postWebhookJson({ url: webhookUrl(port, TELEGRAM_WEBHOOK_PATH), - payload: JSON.stringify({ update_id: 2, message: { text: "slow" } }), + payload: JSON.stringify(slowUpdate), secret: TELEGRAM_SECRET, timeoutMs: 1_000, }); @@ -1117,7 +1133,7 @@ describe("startTelegramWebhook", () => { try { const response = await postWebhookJson({ url: webhookUrl(getServerPort(started.server), TELEGRAM_WEBHOOK_PATH), - payload: JSON.stringify({ update_id: 3, message: { text: "stuck" } }), + payload: JSON.stringify(telegramMessageUpdate(3, "stuck")), secret: TELEGRAM_SECRET, }); expect(response.status).toBe(200); @@ -1197,7 +1213,7 @@ describe("startTelegramWebhook", () => { let responseSettled = false; const responseTask = postWebhookJson({ url: webhookUrl(port, TELEGRAM_WEBHOOK_PATH), - payload: JSON.stringify({ update_id: 4, message: { text: "commit gate" } }), + payload: JSON.stringify(telegramMessageUpdate(4, "commit gate")), secret: TELEGRAM_SECRET, }).then((response) => { responseSettled = true; @@ -1231,7 +1247,7 @@ describe("startTelegramWebhook", () => { throw new Error("agent turn failed"); } }); - const payload = JSON.stringify({ update_id: 3, message: { text: "boom" } }); + const payload = JSON.stringify(telegramMessageUpdate(3, "boom")); try { await withStartedWebhook( @@ -1297,7 +1313,7 @@ describe("startTelegramWebhook", () => { } = {}; await writeTelegramSpooledUpdate({ spoolDir: requireWebhookSpoolDir(), - update: { update_id: 39, message: { chat: { id: 123 }, text: "stalled" } }, + update: telegramMessageUpdate(39, "stalled"), }); handleUpdateSpy.mockImplementationOnce(async () => { active.dispatchStartedAt = Date.now(); @@ -1339,8 +1355,8 @@ describe("startTelegramWebhook", () => { try { let finishFirstUpdate: (() => void) | undefined; const seenUpdateIds: number[] = []; - const firstUpdate = { update_id: 40, message: { chat: { id: 123 }, text: "slow" } }; - const secondUpdate = { update_id: 41, message: { chat: { id: 123 }, text: "blocked" } }; + const firstUpdate = telegramMessageUpdate(40, "slow"); + const secondUpdate = telegramMessageUpdate(41, "blocked"); await writeTelegramSpooledUpdate({ spoolDir: requireWebhookSpoolDir(), update: firstUpdate, @@ -1386,7 +1402,7 @@ describe("startTelegramWebhook", () => { it("holds buffered timeout settlement behind durable webhook adoption", async () => { vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); try { - const update = { update_id: 42, message: { chat: { id: 123 }, text: "held adoption" } }; + const update = telegramMessageUpdate(42, "held adoption"); await writeTelegramSpooledUpdate({ spoolDir: requireWebhookSpoolDir(), update, @@ -1437,7 +1453,7 @@ describe("startTelegramWebhook", () => { }); it("drains spooled webhook updates left by a previous process on startup", async () => { - const update = { update_id: 30, message: { text: "leftover" } }; + const update = telegramMessageUpdate(30, "leftover"); await writeTelegramSpooledUpdate({ spoolDir: requireWebhookSpoolDir(), update, @@ -2187,8 +2203,8 @@ describe("startTelegramWebhook", () => { }, }, } as TelegramRuntime); - const firstUpdate = { update_id: 50, message: { chat: { id: 123 }, text: "first" } }; - const secondUpdate = { update_id: 51, message: { chat: { id: 123 }, text: "second" } }; + const firstUpdate = telegramMessageUpdate(50, "first"); + const secondUpdate = telegramMessageUpdate(51, "second"); await writeTelegramSpooledUpdate({ spoolDir: requireWebhookSpoolDir(), update: firstUpdate, @@ -2266,7 +2282,7 @@ describe("startTelegramWebhook", () => { } as unknown as TelegramRuntime); await writeTelegramSpooledUpdate({ spoolDir: requireWebhookSpoolDir(), - update: { update_id: 52, message: { chat: { id: 123 }, text: "stop retry" } }, + update: telegramMessageUpdate(52, "stop retry"), }); const runtimeLog = vi.fn(); const started = await startTelegramWebhook({ @@ -2296,7 +2312,7 @@ describe("startTelegramWebhook", () => { try { vi.setSystemTime(10_000_000); const runtimeLog = vi.fn(); - const update = { update_id: 31, message: { text: "young poison" } }; + const update = telegramMessageUpdate(31, "young poison"); await writeTelegramSpooledUpdate({ spoolDir: requireWebhookSpoolDir(), update, @@ -2336,7 +2352,7 @@ describe("startTelegramWebhook", () => { try { vi.setSystemTime(10_000_000); const runtimeLog = vi.fn(); - const update = { update_id: 32, message: { text: "old poison" } }; + const update = telegramMessageUpdate(32, "old poison"); await writeTelegramSpooledUpdate({ spoolDir: requireWebhookSpoolDir(), update, @@ -2448,7 +2464,7 @@ describe("startTelegramWebhook", () => { const validResponse = await postWebhookJson({ url: webhookUrl(port, TELEGRAM_WEBHOOK_PATH), - payload: JSON.stringify({ update_id: 999, message: { text: "hello" } }), + payload: JSON.stringify(telegramMessageUpdate(999, "hello")), secret: TELEGRAM_SECRET, }); expect(validResponse.status).toBe(200); @@ -2470,7 +2486,7 @@ describe("startTelegramWebhook", () => { for (let i = 0; i < TELEGRAM_WEBHOOK_RATE_LIMIT_BURST; i += 1) { const response = await postWebhookJson({ url: webhookUrl(port, TELEGRAM_WEBHOOK_PATH), - payload: JSON.stringify({ update_id: 10_000 + i, message: { text: `valid ${i}` } }), + payload: JSON.stringify(telegramMessageUpdate(10_000 + i, `valid ${i}`)), secret: TELEGRAM_SECRET, }); expect(response.status).toBe(200); @@ -2522,7 +2538,7 @@ describe("startTelegramWebhook", () => { "x-forwarded-for": "203.0.113.20", "x-telegram-bot-api-secret-token": TELEGRAM_SECRET, }, - body: JSON.stringify({ update_id: 201, message: { text: "hello" } }), + body: JSON.stringify(telegramMessageUpdate(201, "hello")), }, 5_000, ); @@ -2571,7 +2587,7 @@ describe("startTelegramWebhook", () => { const secondResponse = await postWebhookJson({ url: webhookUrl(secondPort, TELEGRAM_WEBHOOK_PATH), - payload: JSON.stringify({ update_id: 301, message: { text: "hello" } }), + payload: JSON.stringify(telegramMessageUpdate(301, "hello")), secret: TELEGRAM_SECRET, }); @@ -2630,7 +2646,7 @@ describe("startTelegramWebhook", () => { path: TELEGRAM_WEBHOOK_PATH, }, async ({ port }) => { - const payload = JSON.stringify({ update_id: 1, message: { text: "hello" } }); + const payload = JSON.stringify(telegramMessageUpdate(1, "hello")); const res = await postWebhookJson({ url: webhookUrl(port, TELEGRAM_WEBHOOK_PATH), payload, @@ -2735,8 +2751,8 @@ describe("startTelegramWebhook", () => { }, async ({ port }) => { const payloads = [ - JSON.stringify({ update_id: 1, message: { text: "first" } }), - JSON.stringify({ update_id: 2, message: { text: "second" } }), + JSON.stringify(telegramMessageUpdate(1, "first")), + JSON.stringify(telegramMessageUpdate(2, "second")), ]; for (const payload of payloads) { @@ -2770,8 +2786,8 @@ describe("startTelegramWebhook", () => { path: TELEGRAM_WEBHOOK_PATH, }, async ({ port }) => { - const firstPayload = JSON.stringify({ update_id: 100, message: { text: "first" } }); - const secondPayload = JSON.stringify({ update_id: 101, message: { text: "second" } }); + const firstPayload = JSON.stringify(telegramMessageUpdate(100, "first")); + const secondPayload = JSON.stringify(telegramMessageUpdate(101, "second")); const firstResponse = await postWebhookPayloadWithChunkPlan({ port, path: TELEGRAM_WEBHOOK_PATH,