File size: 4,323 Bytes
f0634fb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | import { mkdtemp, readFile, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { PassThrough, Writable } from 'node:stream';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { downloadToFile, runCommand } from '#/app/capability/host';
import type { IHostProcess, IHostProcessService } from '#/os/interface/hostProcess';
describe('capability host runCommand', () => {
it('does not leak a rejected promise when a timed-out process fails while being killed', async () => {
const stdout = new PassThrough();
const stderr = new PassThrough();
let rejectWait: ((error: Error) => void) | undefined;
const wait = new Promise<number>((_resolve, reject) => {
rejectWait = reject;
});
const proc = {
_serviceBrand: undefined,
pid: 1234,
exitCode: null,
stdin: new Writable({
write: (_chunk, _encoding, callback) => {
callback();
},
}),
stdout,
stderr,
wait: () => wait,
kill: () => {
stdout.destroy(new Error('stream closed after timeout'));
stderr.end();
rejectWait?.(new Error('process killed'));
return Promise.resolve();
},
dispose: () => undefined,
} as IHostProcess;
const host = {
_serviceBrand: undefined,
spawn: () => Promise.resolve(proc),
} as IHostProcessService;
const unhandled: unknown[] = [];
const onUnhandled = (error: unknown): void => {
unhandled.push(error);
};
process.on('unhandledRejection', onUnhandled);
try {
await expect(runCommand(host, 'hang', [], { timeout: 5 })).rejects.toThrow(
'command timed out after 5ms: hang',
);
await new Promise<void>((resolve) => {
setImmediate(resolve);
});
expect(unhandled).toEqual([]);
} finally {
process.off('unhandledRejection', onUnhandled);
}
});
});
describe('capability host downloadToFile', () => {
let root: string;
beforeEach(async () => {
root = await mkdtemp(path.join(tmpdir(), 'capability-download-'));
});
afterEach(async () => {
await rm(root, { recursive: true, force: true });
});
function fakeFetchWith(body: ReadableStream): typeof fetch {
return (() =>
Promise.resolve(
new Response(body, {
status: 200,
headers: { 'content-length': '100' },
}),
)) as unknown as typeof fetch;
}
it('aborts a response whose byte stream goes quiet', async () => {
const body = new ReadableStream({
start(controller) {
controller.enqueue(new Uint8Array([1, 2, 3]));
},
});
await expect(
downloadToFile(
'https://cdn.example.test/blob',
path.join(root, 'blob'),
undefined,
fakeFetchWith(body) as never,
{ idleTimeoutMs: 5 },
),
).rejects.toThrow(/stalled/);
});
it('aborts when the response headers never arrive', async () => {
const hangingFetch = ((_url: string, init?: { signal?: AbortSignal }) =>
new Promise((_resolve, reject) => {
init?.signal?.addEventListener('abort', () => {
reject(new DOMException('This operation was aborted.', 'AbortError'));
});
})) as never;
await expect(
downloadToFile(
'https://cdn.example.test/headers',
path.join(root, 'headers'),
undefined,
hangingFetch,
{ idleTimeoutMs: 5 },
),
).rejects.toThrow(/no response within 5ms/);
});
it('lets a slow but flowing download finish intact', async () => {
const chunks = ['hel', 'lo ', 'wor', 'ld'];
const body = new ReadableStream({
async start(controller) {
for (const chunk of chunks) {
controller.enqueue(new TextEncoder().encode(chunk));
await new Promise((resolve) => {
setTimeout(resolve, 30);
});
}
controller.close();
},
});
const dest = path.join(root, 'hello.txt');
const received = await downloadToFile(
'https://cdn.example.test/hello',
dest,
undefined,
fakeFetchWith(body) as never,
{ idleTimeoutMs: 50 },
);
expect(received).toBe(11);
expect(await readFile(dest, 'utf-8')).toBe('hello world');
});
});
|