vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-03-15 20:10:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: share telegram network test helpers

Browse Source
This commit is contained in:
Peter Steinberger
2026-03-13 18:17:47 +00:00
parent 41c9e3ead0
commit 05a1b0c3ae
2 changed files with 89 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
src/telegram/fetch.test.ts
View File

@@ -88,6 +88,44 @@ function buildFetchFallbackError(code: string) {
});
}
const STICKY_IPV4_FALLBACK_NETWORK = {
network: {
autoSelectFamily: true,
dnsResultOrder: "ipv4first" as const,
},
};
async function runDefaultStickyIpv4FallbackProbe(code = "EHOSTUNREACH"): Promise<void> {
undiciFetch
.mockRejectedValueOnce(buildFetchFallbackError(code))
.mockResolvedValueOnce({ ok: true } as Response)
.mockResolvedValueOnce({ ok: true } as Response);
const resolved = resolveTelegramFetchOrThrow(undefined, STICKY_IPV4_FALLBACK_NETWORK);
await resolved("https://api.telegram.org/botx/sendMessage");
await resolved("https://api.telegram.org/botx/sendChatAction");
}
function expectPinnedIpv4ConnectDispatcher(args: {
pinnedCall: number;
firstCall?: number;
followupCall?: number;
}): void {
const pinnedDispatcher = getDispatcherFromUndiciCall(args.pinnedCall);
expect(pinnedDispatcher?.options?.connect).toEqual(
expect.objectContaining({
family: 4,
autoSelectFamily: false,
}),
);
if (args.firstCall) {
expect(getDispatcherFromUndiciCall(args.firstCall)).not.toBe(pinnedDispatcher);
}
if (args.followupCall) {
expect(getDispatcherFromUndiciCall(args.followupCall)).toBe(pinnedDispatcher);
}
}
afterEach(() => {
undiciFetch.mockReset();
setGlobalDispatcher.mockReset();
@@ -307,9 +345,8 @@ describe("resolveTelegramFetch", () => {
it("treats ALL_PROXY-only env as direct transport and arms sticky IPv4 fallback", async () => {
vi.stubEnv("ALL_PROXY", "socks5://127.0.0.1:1080");
const fetchError = buildFetchFallbackError("EHOSTUNREACH");
undiciFetch
.mockRejectedValueOnce(fetchError)
.mockRejectedValueOnce(buildFetchFallbackError("EHOSTUNREACH"))
.mockResolvedValueOnce({ ok: true } as Response)
.mockResolvedValueOnce({ ok: true } as Response);
@@ -327,18 +364,11 @@ describe("resolveTelegramFetch", () => {
expect(EnvHttpProxyAgentCtor).not.toHaveBeenCalled();
expect(AgentCtor).toHaveBeenCalledTimes(2);
const firstDispatcher = getDispatcherFromUndiciCall(1);
const secondDispatcher = getDispatcherFromUndiciCall(2);
const thirdDispatcher = getDispatcherFromUndiciCall(3);
expect(firstDispatcher).not.toBe(secondDispatcher);
expect(secondDispatcher).toBe(thirdDispatcher);
expect(secondDispatcher?.options?.connect).toEqual(
expect.objectContaining({
family: 4,
autoSelectFamily: false,
}),
);
expectPinnedIpv4ConnectDispatcher({
firstCall: 1,
pinnedCall: 2,
followupCall: 3,
});
expect(transport.pinnedDispatcherPolicy).toEqual(
expect.objectContaining({
mode: "direct",
@@ -351,134 +381,52 @@ describe("resolveTelegramFetch", () => {
EnvHttpProxyAgentCtor.mockImplementationOnce(function ThrowingEnvProxyAgent() {
throw new Error("invalid proxy config");
});
const fetchError = buildFetchFallbackError("EHOSTUNREACH");
undiciFetch
.mockRejectedValueOnce(fetchError)
.mockResolvedValueOnce({ ok: true } as Response)
.mockResolvedValueOnce({ ok: true } as Response);
const resolved = resolveTelegramFetchOrThrow(undefined, {
network: {
autoSelectFamily: true,
dnsResultOrder: "ipv4first",
},
});
await resolved("https://api.telegram.org/botx/sendMessage");
await resolved("https://api.telegram.org/botx/sendChatAction");
await runDefaultStickyIpv4FallbackProbe();
expect(undiciFetch).toHaveBeenCalledTimes(3);
expect(EnvHttpProxyAgentCtor).toHaveBeenCalledTimes(1);
expect(AgentCtor).toHaveBeenCalledTimes(2);
const firstDispatcher = getDispatcherFromUndiciCall(1);
const secondDispatcher = getDispatcherFromUndiciCall(2);
const thirdDispatcher = getDispatcherFromUndiciCall(3);
expect(firstDispatcher).not.toBe(secondDispatcher);
expect(secondDispatcher).toBe(thirdDispatcher);
expect(secondDispatcher?.options?.connect).toEqual(
expect.objectContaining({
family: 4,
autoSelectFamily: false,
}),
);
expectPinnedIpv4ConnectDispatcher({
firstCall: 1,
pinnedCall: 2,
followupCall: 3,
});
});
it("arms sticky IPv4 fallback when NO_PROXY bypasses telegram under env proxy", async () => {
vi.stubEnv("HTTPS_PROXY", "http://127.0.0.1:7890");
vi.stubEnv("NO_PROXY", "api.telegram.org");
const fetchError = buildFetchFallbackError("EHOSTUNREACH");
undiciFetch
.mockRejectedValueOnce(fetchError)
.mockResolvedValueOnce({ ok: true } as Response)
.mockResolvedValueOnce({ ok: true } as Response);
const resolved = resolveTelegramFetchOrThrow(undefined, {
network: {
autoSelectFamily: true,
dnsResultOrder: "ipv4first",
},
});
await resolved("https://api.telegram.org/botx/sendMessage");
await resolved("https://api.telegram.org/botx/sendChatAction");
await runDefaultStickyIpv4FallbackProbe();
expect(undiciFetch).toHaveBeenCalledTimes(3);
expect(EnvHttpProxyAgentCtor).toHaveBeenCalledTimes(2);
expect(AgentCtor).not.toHaveBeenCalled();
const firstDispatcher = getDispatcherFromUndiciCall(1);
const secondDispatcher = getDispatcherFromUndiciCall(2);
const thirdDispatcher = getDispatcherFromUndiciCall(3);
expect(firstDispatcher).not.toBe(secondDispatcher);
expect(secondDispatcher).toBe(thirdDispatcher);
expect(secondDispatcher?.options?.connect).toEqual(
expect.objectContaining({
family: 4,
autoSelectFamily: false,
}),
);
expectPinnedIpv4ConnectDispatcher({
firstCall: 1,
pinnedCall: 2,
followupCall: 3,
});
});
it("uses no_proxy over NO_PROXY when deciding env-proxy bypass", async () => {
vi.stubEnv("HTTPS_PROXY", "http://127.0.0.1:7890");
vi.stubEnv("NO_PROXY", "");
vi.stubEnv("no_proxy", "api.telegram.org");
const fetchError = buildFetchFallbackError("EHOSTUNREACH");
undiciFetch
.mockRejectedValueOnce(fetchError)
.mockResolvedValueOnce({ ok: true } as Response)
.mockResolvedValueOnce({ ok: true } as Response);
const resolved = resolveTelegramFetchOrThrow(undefined, {
network: {
autoSelectFamily: true,
dnsResultOrder: "ipv4first",
},
});
await resolved("https://api.telegram.org/botx/sendMessage");
await resolved("https://api.telegram.org/botx/sendChatAction");
await runDefaultStickyIpv4FallbackProbe();
expect(EnvHttpProxyAgentCtor).toHaveBeenCalledTimes(2);
const secondDispatcher = getDispatcherFromUndiciCall(2);
expect(secondDispatcher?.options?.connect).toEqual(
expect.objectContaining({
family: 4,
autoSelectFamily: false,
}),
);
expectPinnedIpv4ConnectDispatcher({ pinnedCall: 2 });
});
it("matches whitespace and wildcard no_proxy entries like EnvHttpProxyAgent", async () => {
vi.stubEnv("HTTPS_PROXY", "http://127.0.0.1:7890");
vi.stubEnv("no_proxy", "localhost *.telegram.org");
const fetchError = buildFetchFallbackError("EHOSTUNREACH");
undiciFetch
.mockRejectedValueOnce(fetchError)
.mockResolvedValueOnce({ ok: true } as Response)
.mockResolvedValueOnce({ ok: true } as Response);
const resolved = resolveTelegramFetchOrThrow(undefined, {
network: {
autoSelectFamily: true,
dnsResultOrder: "ipv4first",
},
});
await resolved("https://api.telegram.org/botx/sendMessage");
await resolved("https://api.telegram.org/botx/sendChatAction");
await runDefaultStickyIpv4FallbackProbe();
expect(EnvHttpProxyAgentCtor).toHaveBeenCalledTimes(2);
const secondDispatcher = getDispatcherFromUndiciCall(2);
expect(secondDispatcher?.options?.connect).toEqual(
expect.objectContaining({
family: 4,
autoSelectFamily: false,
}),
);
expectPinnedIpv4ConnectDispatcher({ pinnedCall: 2 });
});
it("fails closed when explicit proxy dispatcher initialization fails", async () => {

121
src/telegram/monitor.test.ts
View File

@@ -129,6 +129,28 @@ function mockRunOnceAndAbort(abort: AbortController) {
runSpy.mockImplementationOnce(() => makeAbortRunner(abort));
}
function mockRunOnceWithStalledPollingRunner(): {
stop: ReturnType<typeof vi.fn<() => void | Promise<void>>>;
} {
let running = true;
let releaseTask: (() => void) | undefined;
const stop = vi.fn(async () => {
running = false;
releaseTask?.();
});
runSpy.mockImplementationOnce(() =>
makeRunnerStub({
task: () =>
new Promise<void>((resolve) => {
releaseTask = resolve;
}),
stop,
isRunning: () => running,
}),
);
return { stop };
}
function expectRecoverableRetryState(expectedRunCalls: number) {
expect(computeBackoff).toHaveBeenCalled();
expect(sleepWithAbort).toHaveBeenCalled();
@@ -434,31 +456,8 @@ describe("monitorTelegramProvider (grammY)", () => {
it("force-restarts polling when unhandled network rejection stalls runner", async () => {
const abort = new AbortController();
let running = true;
let releaseTask: (() => void) | undefined;
const stop = vi.fn(async () => {
running = false;
releaseTask?.();
});
runSpy
.mockImplementationOnce(() =>
makeRunnerStub({
task: () =>
new Promise<void>((resolve) => {
releaseTask = resolve;
}),
stop,
isRunning: () => running,
}),
)
.mockImplementationOnce(() =>
makeRunnerStub({
task: async () => {
abort.abort();
},
}),
);
const { stop } = mockRunOnceWithStalledPollingRunner();
mockRunOnceAndAbort(abort);
const monitor = monitorTelegramProvider({ token: "tok", abortSignal: abort.signal });
await vi.waitFor(() => expect(runSpy).toHaveBeenCalledTimes(1));
@@ -474,31 +473,8 @@ describe("monitorTelegramProvider (grammY)", () => {
it("aborts the active Telegram fetch when unhandled network rejection forces restart", async () => {
const abort = new AbortController();
let running = true;
let releaseTask: (() => void) | undefined;
const stop = vi.fn(async () => {
running = false;
releaseTask?.();
});
runSpy
.mockImplementationOnce(() =>
makeRunnerStub({
task: () =>
new Promise<void>((resolve) => {
releaseTask = resolve;
}),
stop,
isRunning: () => running,
}),
)
.mockImplementationOnce(() =>
makeRunnerStub({
task: async () => {
abort.abort();
},
}),
);
const { stop } = mockRunOnceWithStalledPollingRunner();
mockRunOnceAndAbort(abort);
const monitor = monitorTelegramProvider({ token: "tok", abortSignal: abort.signal });
await vi.waitFor(() => expect(createTelegramBotCalls.length).toBeGreaterThanOrEqual(1));
@@ -515,23 +491,7 @@ describe("monitorTelegramProvider (grammY)", () => {
it("ignores unrelated process-level network errors while telegram polling is active", async () => {
const abort = new AbortController();
let running = true;
let releaseTask: (() => void) | undefined;
const stop = vi.fn(async () => {
running = false;
releaseTask?.();
});
runSpy.mockImplementationOnce(() =>
makeRunnerStub({
task: () =>
new Promise<void>((resolve) => {
releaseTask = resolve;
}),
stop,
isRunning: () => running,
}),
);
const { stop } = mockRunOnceWithStalledPollingRunner();
const monitor = monitorTelegramProvider({ token: "tok", abortSignal: abort.signal });
await vi.waitFor(() => expect(runSpy).toHaveBeenCalledTimes(1));
@@ -600,31 +560,8 @@ describe("monitorTelegramProvider (grammY)", () => {
it("force-restarts polling when getUpdates stalls (watchdog)", async () => {
vi.useFakeTimers({ shouldAdvanceTime: true });
const abort = new AbortController();
let running = true;
let releaseTask: (() => void) | undefined;
const stop = vi.fn(async () => {
running = false;
releaseTask?.();
});
runSpy
.mockImplementationOnce(() =>
makeRunnerStub({
task: () =>
new Promise<void>((resolve) => {
releaseTask = resolve;
}),
stop,
isRunning: () => running,
}),
)
.mockImplementationOnce(() =>
makeRunnerStub({
task: async () => {
abort.abort();
},
}),
);
const { stop } = mockRunOnceWithStalledPollingRunner();
mockRunOnceAndAbort(abort);
const monitor = monitorTelegramProvider({ token: "tok", abortSignal: abort.signal });
await vi.waitFor(() => expect(runSpy).toHaveBeenCalledTimes(1));