Files
2026-07-13 12:47:58 +08:00

277 lines
8.4 KiB
TypeScript

import { afterEach, expect, test, vi } from 'vitest'
afterEach(() => {
vi.restoreAllMocks()
vi.resetModules()
vi.doUnmock('undici')
vi.doUnmock('../../src/utils')
delete process.env.E2B_ENVD_RPC_CONNECTIONS
delete process.env.E2B_ENVD_INFLIGHT_REQUESTS
delete process.env.E2B_ENVD_RPC_INFLIGHT_REQUESTS
})
test('uses undici with HTTP/2 enabled in Node', async () => {
const agents: Array<{ allowH2?: boolean; connections?: number }> = []
const requests: Array<{ init?: RequestInit & { dispatcher?: unknown } }> = []
class Agent {
constructor(options: { allowH2?: boolean; connections?: number }) {
agents.push(options)
}
}
const undiciFetch = vi.fn((input, init) => {
requests.push({ init })
return Promise.resolve(new Response('ok'))
})
const { createEnvdFetchForRuntime } = await import('../../src/envd/http2')
const fetcher = createEnvdFetchForRuntime('node', {
connectionLimit: 1,
loadUndici: () => Promise.resolve({ Agent, fetch: undiciFetch }),
})
const res = await fetcher('https://example.com/status')
expect(await res.text()).toBe('ok')
expect(agents).toEqual([{ allowH2: true, connections: 1 }])
expect(requests[0].init?.dispatcher).toBeInstanceOf(Agent)
})
test('uses a ProxyAgent dispatcher when a proxy is configured', async () => {
const proxyAgents: Array<{
uri?: string
allowH2?: boolean
connections?: number
}> = []
const agents: Array<unknown> = []
const requests: Array<{ init?: RequestInit & { dispatcher?: unknown } }> = []
class Agent {
constructor() {
agents.push(this)
}
}
class ProxyAgent {
constructor(options: {
uri?: string
allowH2?: boolean
connections?: number
}) {
proxyAgents.push(options)
}
}
const undiciFetch = vi.fn((input, init) => {
requests.push({ init })
return Promise.resolve(new Response('ok'))
})
const { createEnvdFetchForRuntime } = await import('../../src/envd/http2')
const fetcher = createEnvdFetchForRuntime('node', {
connectionLimit: 1,
proxy: 'http://127.0.0.1:8080',
loadUndici: () =>
Promise.resolve({ Agent, ProxyAgent, fetch: undiciFetch }),
})
await fetcher('https://example.com/status')
expect(agents).toHaveLength(0)
expect(proxyAgents).toEqual([
{ uri: 'http://127.0.0.1:8080', allowH2: true, connections: 1 },
])
expect(requests[0].init?.dispatcher).toBeInstanceOf(ProxyAgent)
})
test('caches envd fetchers per proxy', async () => {
const { createEnvdFetch, createEnvdRpcFetch } = await import(
'../../src/envd/http2'
)
const noProxy = createEnvdFetch()
const proxyA = createEnvdFetch('http://127.0.0.1:8080')
expect(createEnvdFetch()).toBe(noProxy)
expect(createEnvdFetch('http://127.0.0.1:8080')).toBe(proxyA)
expect(proxyA).not.toBe(noProxy)
const rpcNoProxy = createEnvdRpcFetch()
const rpcProxyA = createEnvdRpcFetch('http://127.0.0.1:8080')
expect(createEnvdRpcFetch()).toBe(rpcNoProxy)
expect(createEnvdRpcFetch('http://127.0.0.1:8080')).toBe(rpcProxyA)
expect(rpcProxyA).not.toBe(rpcNoProxy)
})
test('passes Request objects to undici as URL plus init', async () => {
const requests: Array<{ input: RequestInfo | URL; init?: RequestInit }> = []
class Agent {}
const undiciFetch = vi.fn((input, init) => {
requests.push({ input, init })
return Promise.resolve(new Response('ok'))
})
const { createEnvdFetchForRuntime } = await import('../../src/envd/http2')
const fetcher = createEnvdFetchForRuntime('node', {
connectionLimit: 1,
loadUndici: () => Promise.resolve({ Agent, fetch: undiciFetch }),
})
const body = JSON.stringify({ ok: true })
await fetcher(
new Request('https://example.com/rpc', {
body,
headers: { 'content-type': 'application/json' },
method: 'POST',
})
)
expect(requests[0].input).toBe('https://example.com/rpc')
expect(requests[0].init?.method).toBe('POST')
expect(requests[0].init?.headers).toBeInstanceOf(Headers)
expect(requests[0].init?.body).toBeInstanceOf(ReadableStream)
})
test('can create a bounded dispatcher for RPC streams', async () => {
const agents: Array<{ allowH2?: boolean; connections?: number }> = []
class Agent {
constructor(options: { allowH2?: boolean; connections?: number }) {
agents.push(options)
}
}
const undiciFetch = vi.fn(() => Promise.resolve(new Response('ok')))
const { createEnvdFetchForRuntime } = await import('../../src/envd/http2')
const fetcher = createEnvdFetchForRuntime('node', {
connectionLimit: 100,
loadUndici: () => Promise.resolve({ Agent, fetch: undiciFetch }),
})
await fetcher('https://example.com/rpc')
expect(agents).toEqual([{ allowH2: true, connections: 100 }])
})
test('reads RPC stream dispatcher connection limit from env', async () => {
process.env.E2B_ENVD_RPC_CONNECTIONS = '200'
const { getEnvdRpcConnectionLimit } = await import('../../src/envd/http2')
expect(getEnvdRpcConnectionLimit()).toBe(200)
})
test('getEnvdRpcConnectionLimit throws on malformed env value', async () => {
process.env.E2B_ENVD_RPC_CONNECTIONS = 'bogus'
const { getEnvdRpcConnectionLimit } = await import('../../src/envd/http2')
expect(() => getEnvdRpcConnectionLimit()).toThrow(/E2B_ENVD_RPC_CONNECTIONS/)
})
test('getEnvdInflightLimit throws on malformed env value', async () => {
process.env.E2B_ENVD_INFLIGHT_REQUESTS = 'bogus'
const { getEnvdInflightLimit } = await import('../../src/envd/http2')
expect(() => getEnvdInflightLimit()).toThrow(/E2B_ENVD_INFLIGHT_REQUESTS/)
})
test('getEnvdRpcInflightLimit throws on malformed env value', async () => {
process.env.E2B_ENVD_RPC_INFLIGHT_REQUESTS = 'bogus'
const { getEnvdRpcInflightLimit } = await import('../../src/envd/http2')
expect(() => getEnvdRpcInflightLimit()).toThrow(
/E2B_ENVD_RPC_INFLIGHT_REQUESTS/
)
})
test('inflight limit env vars return 0 when explicitly disabled', async () => {
process.env.E2B_ENVD_INFLIGHT_REQUESTS = '0'
process.env.E2B_ENVD_RPC_INFLIGHT_REQUESTS = '0'
const { getEnvdInflightLimit, getEnvdRpcInflightLimit } = await import(
'../../src/envd/http2'
)
expect(getEnvdInflightLimit()).toBe(0)
expect(getEnvdRpcInflightLimit()).toBe(0)
})
test('getEnvdInflightLimit throws on negative env value', async () => {
process.env.E2B_ENVD_INFLIGHT_REQUESTS = '-1'
const { getEnvdInflightLimit } = await import('../../src/envd/http2')
expect(() => getEnvdInflightLimit()).toThrow(/E2B_ENVD_INFLIGHT_REQUESTS=-1/)
})
test('getEnvdRpcInflightLimit throws on negative env value', async () => {
process.env.E2B_ENVD_RPC_INFLIGHT_REQUESTS = '-5'
const { getEnvdRpcInflightLimit } = await import('../../src/envd/http2')
expect(() => getEnvdRpcInflightLimit()).toThrow(
/E2B_ENVD_RPC_INFLIGHT_REQUESTS=-5/
)
})
test('defers loading undici until the first Node request', async () => {
const Agent = vi.fn()
const undiciFetch = vi.fn(() => Promise.resolve(new Response('ok')))
const loadUndici = vi.fn(() => Promise.resolve({ Agent, fetch: undiciFetch }))
const { createEnvdFetchForRuntime } = await import('../../src/envd/http2')
const fetcher = createEnvdFetchForRuntime('node', {
connectionLimit: 1,
loadUndici,
})
expect(loadUndici).not.toHaveBeenCalled()
expect(Agent).not.toHaveBeenCalled()
expect(undiciFetch).not.toHaveBeenCalled()
await fetcher('https://example.com/status')
expect(loadUndici).toHaveBeenCalledOnce()
expect(Agent).toHaveBeenCalledOnce()
expect(undiciFetch).toHaveBeenCalledOnce()
})
test('falls back to global fetch when undici cannot be loaded', async () => {
const fallbackFetch = vi.fn(() =>
Promise.resolve(new Response('fallback ok'))
) as unknown as typeof fetch
vi.stubGlobal('fetch', fallbackFetch)
const { createEnvdFetchForRuntime } = await import('../../src/envd/http2')
const fetcher = createEnvdFetchForRuntime('node', {
loadUndici: () => Promise.resolve(undefined),
})
const res = await fetcher('https://example.com/status')
expect(await res.text()).toBe('fallback ok')
expect(fallbackFetch).toHaveBeenCalledWith(
'https://example.com/status',
undefined
)
})
test('uses global fetch outside Node', async () => {
const fallbackFetch = vi.fn() as unknown as typeof fetch
vi.stubGlobal('fetch', fallbackFetch)
const { createEnvdFetchForRuntime } = await import('../../src/envd/http2')
expect(createEnvdFetchForRuntime('browser')).toBe(fallbackFetch)
expect(createEnvdFetchForRuntime('vercel-edge')).toBe(fallbackFetch)
})