import type { IncomingMessage, ServerResponse } from "node:http"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-runtime"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; import { WEBHOOK_IN_FLIGHT_DEFAULTS } from "openclaw/plugin-sdk/webhook-request-guards"; import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; type LineNodeWebhookHandler = (req: IncomingMessage, res: ServerResponse) => Promise; const { createLineBotMock, createLineNodeWebhookHandlerMock, registerPluginHttpRouteMock, unregisterHttpMock, } = vi.hoisted(() => ({ createLineBotMock: vi.fn(() => ({ account: { accountId: "default" }, handleWebhook: vi.fn(), })), createLineNodeWebhookHandlerMock: vi.fn<() => LineNodeWebhookHandler>(() => vi.fn(async () => {}), ), registerPluginHttpRouteMock: vi.fn(), unregisterHttpMock: vi.fn(), })); let monitorLineProvider: typeof import("./monitor.js").monitorLineProvider; let getLineRuntimeState: typeof import("./monitor.js").getLineRuntimeState; let clearLineRuntimeStateForTests: typeof import("./monitor.js").clearLineRuntimeStateForTests; let innerLineWebhookHandlerMock: ReturnType>; vi.mock("./bot.js", () => ({ createLineBot: createLineBotMock, })); vi.mock("openclaw/plugin-sdk/reply-runtime", () => ({ chunkMarkdownText: vi.fn(), dispatchReplyWithBufferedBlockDispatcher: vi.fn(), })); vi.mock("openclaw/plugin-sdk/runtime-env", async () => { const actual = await vi.importActual( "openclaw/plugin-sdk/runtime-env", ); return { ...actual, danger: (value: unknown) => String(value), logVerbose: vi.fn(), waitForAbortSignal: vi.fn(), }; }); vi.mock("openclaw/plugin-sdk/channel-reply-pipeline", () => ({ createChannelReplyPipeline: vi.fn(() => ({})), })); vi.mock("openclaw/plugin-sdk/webhook-ingress", () => ({ normalizePluginHttpPath: (_path: string | undefined, fallback: string) => fallback, registerPluginHttpRoute: registerPluginHttpRouteMock, })); vi.mock("./webhook-node.js", () => ({ createLineNodeWebhookHandler: createLineNodeWebhookHandlerMock, })); vi.mock("./auto-reply-delivery.js", () => ({ deliverLineAutoReply: vi.fn(), })); vi.mock("./markdown-to-line.js", () => ({ processLineMessage: vi.fn(), })); vi.mock("./reply-chunks.js", () => ({ sendLineReplyChunks: vi.fn(), })); vi.mock("./send.js", () => ({ createFlexMessage: vi.fn(), createImageMessage: vi.fn(), createLocationMessage: vi.fn(), createQuickReplyItems: vi.fn(), createTextMessageWithQuickReplies: vi.fn(), getUserDisplayName: vi.fn(), pushMessageLine: vi.fn(), pushMessagesLine: vi.fn(), pushTextMessageWithQuickReplies: vi.fn(), replyMessageLine: vi.fn(), showLoadingAnimation: vi.fn(), })); vi.mock("./template-messages.js", () => ({ buildTemplateMessageFromPayload: vi.fn(), })); describe("monitorLineProvider lifecycle", () => { beforeAll(async () => { ({ monitorLineProvider, getLineRuntimeState, clearLineRuntimeStateForTests } = await import("./monitor.js")); }); beforeEach(() => { clearLineRuntimeStateForTests(); createLineBotMock.mockReset(); createLineBotMock.mockReturnValue({ account: { accountId: "default" }, handleWebhook: vi.fn(), }); innerLineWebhookHandlerMock = vi.fn(async () => {}); createLineNodeWebhookHandlerMock .mockReset() .mockImplementation(() => innerLineWebhookHandlerMock); unregisterHttpMock.mockReset(); registerPluginHttpRouteMock.mockReset().mockReturnValue(unregisterHttpMock); }); const createRouteResponse = () => { const resObj = { statusCode: 0, headersSent: false, setHeader: vi.fn(), end: vi.fn(() => { resObj.headersSent = true; }), }; return resObj as unknown as ServerResponse & { end: ReturnType }; }; it("waits for abort before resolving", async () => { const abort = new AbortController(); let resolved = false; const task = monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, abortSignal: abort.signal, }).then((monitor) => { resolved = true; return monitor; }); expect(registerPluginHttpRouteMock).toHaveBeenCalledTimes(1); expect(registerPluginHttpRouteMock).toHaveBeenCalledWith( expect.objectContaining({ auth: "plugin" }), ); expect(resolved).toBe(false); abort.abort(); await task; expect(unregisterHttpMock).toHaveBeenCalledTimes(1); }); it("stops immediately when signal is already aborted", async () => { const abort = new AbortController(); abort.abort(); await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, abortSignal: abort.signal, }); expect(unregisterHttpMock).toHaveBeenCalledTimes(1); }); it("returns immediately without abort signal and stop is idempotent", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); expect(unregisterHttpMock).not.toHaveBeenCalled(); monitor.stop(); monitor.stop(); expect(unregisterHttpMock).toHaveBeenCalledTimes(1); }); it("records startup state under configured defaultAccount when accountId is omitted", async () => { const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: { channels: { line: { defaultAccount: "work", accounts: { work: { channelAccessToken: "work-token", channelSecret: "work-secret", }, }, }, }, } as OpenClawConfig, runtime: {} as RuntimeEnv, }); expect(getLineRuntimeState("work")).toEqual( expect.objectContaining({ running: true, }), ); expect(getLineRuntimeState("default")).toBeUndefined(); monitor.stop(); }); it("rejects webhook requests above the shared in-flight limit before body handling", async () => { const limit = WEBHOOK_IN_FLIGHT_DEFAULTS.maxInFlightPerKey; const releaseRequests: Array<() => void> = []; let reachLimit!: () => void; const reachedLimit = new Promise((resolve) => { reachLimit = resolve; }); innerLineWebhookHandlerMock.mockImplementation( async (_req: IncomingMessage, res: ServerResponse) => { if (releaseRequests.length === limit - 1) { reachLimit(); } await new Promise((resolve) => { releaseRequests.push(resolve); }); res.statusCode = 200; res.end(); }, ); const monitor = await monitorLineProvider({ channelAccessToken: "token", channelSecret: "secret", // pragma: allowlist secret config: {} as OpenClawConfig, runtime: {} as RuntimeEnv, }); const route = registerPluginHttpRouteMock.mock.calls[0]?.[0] as | { handler: (req: IncomingMessage, res: ServerResponse) => Promise } | undefined; expect(route).toBeDefined(); const createPostRequest = () => ({ method: "POST", headers: {}, }) as IncomingMessage; const firstRequests = Array.from({ length: limit }, () => route!.handler(createPostRequest(), createRouteResponse()), ); await reachedLimit; const overflowResponse = createRouteResponse(); await route!.handler(createPostRequest(), overflowResponse); expect(innerLineWebhookHandlerMock).toHaveBeenCalledTimes(limit); expect(overflowResponse.statusCode).toBe(429); expect(overflowResponse.end).toHaveBeenCalledWith("Too Many Requests"); releaseRequests.splice(0).forEach((release) => release()); await Promise.all(firstRequests); monitor.stop(); }); });