使用Jest对流式文件下载功能进行单元测试

我有一个用TypeScript编写的函数，它接收图片URL数组，循环遍历并将图片下载到本地目录。尝试为该函数编写Jest单元测试，但在模拟部分组件时遇到问题。核心代码如下：

const finished = promisify(stream.finished);
const writer = fs.createWriteStream(filepath);

axios({
    method: 'get',
    url: url,
    responseType: 'stream'  // 让响应以流形式返回
}).then(async (response) => {
    response.data.pipe(writer);  // 将响应流导入文件可写流
    await finished(writer);  // 监听流完成/错误事件
}).catch(error => {
    console.error('Image Download');
    console.error(error);
    process.exit(1);
});

我已在其他非流式API请求的单元测试中使用axios-mock-adapter来模拟axios，请问如何正确模拟axios流、fs.createWriteStream和stream.finished函数？

解决方案 #

1. 模拟fs.createWriteStream

用Jest直接mockfs模块，让createWriteStream返回一个模拟的可写流对象，需要包含pipe方法，并且支持监听finish事件（stream.finished会等待这个事件）：

jest.mock('fs');
const mockCreateWriteStream = jest.fn(() => {
  const mockStream = {
    pipe: jest.fn().mockReturnThis(),
    on: jest.fn((event, callback) => {
      // 测试成功场景下，自动触发finish事件
      if (event === 'finish') {
        process.nextTick(callback);
      }
      return mockStream;
    }),
    emit: jest.fn((event) => {
      // 暴露emit方法，方便测试错误场景手动触发error事件
      return mockStream;
    })
  };
  return mockStream;
});

// 替换原方法
(fs.createWriteStream as jest.Mock) = mockCreateWriteStream;

2. 模拟stream.finished

mockstream模块的finished方法，根据测试场景返回resolved或rejected的Promise：

jest.mock('stream', () => ({
  ...jest.requireActual('stream'), // 保留其他原生方法
  finished: jest.fn(() => Promise.resolve()) // 默认模拟成功
}));

// 若要测试错误场景，可在单个测试用例中重写：
// (stream.finished as jest.Mock).mockImplementation(() => Promise.reject(new Error('Stream failed')));

3. 用axios-mock-adapter模拟流式响应

设置mock的响应类型为stream，并返回一个带有pipe方法的模拟流对象，在pipe调用时触发文件流的finish事件：

import axios from 'axios';
import MockAdapter from 'axios-mock-adapter';

const mockAxios = new MockAdapter(axios);

// 模拟成功的流式请求
mockAxios.onGet('https://test-image.com/1.jpg')
  .reply(200, {
    pipe: jest.fn((writer) => {
      // 触发文件流的finish事件，模拟数据传输完成
      writer.emit('finish');
      return this;
    })
  }, { responseType: 'stream' });

// 模拟请求失败场景
mockAxios.onGet('https://test-image.com/fail.jpg')
  .reply(500);

4. 完整测试用例示例

import axios from 'axios';
import MockAdapter from 'axios-mock-adapter';
import fs from 'fs';
import stream from 'stream';
import { downloadImages } from './your-download-function'; // 导入你的下载函数

jest.mock('fs');
jest.mock('stream');

describe('downloadImages', () => {
  let mockAxios: MockAdapter;

  beforeEach(() => {
    mockAxios = new MockAdapter(axios);
    // 重置所有mock
    jest.clearAllMocks();
    // 默认模拟stream.finished成功
    (stream.finished as jest.Mock).mockImplementation(() => Promise.resolve());
  });

  it('应该成功下载图片', async () => {
    const testUrls = ['https://test-image.com/1.jpg'];
    const testFilepath = './downloads/1.jpg';

    // 模拟axios响应
    mockAxios.onGet(testUrls[0])
      .reply(200, {
        pipe: jest.fn((writer) => {
          writer.emit('finish');
          return this;
        })
      }, { responseType: 'stream' });

    await downloadImages(testUrls);

    // 验证createWriteStream被调用
    expect(fs.createWriteStream).toHaveBeenCalledWith(testFilepath);
    // 验证axios请求被触发
    expect(mockAxios.history.get.length).toBe(1);
    // 验证stream.finished被调用
    expect(stream.finished).toHaveBeenCalled();
  });

  it('应该处理下载错误', async () => {
    const testUrls = ['https://test-image.com/fail.jpg'];
    const mockExit = jest.spyOn(process, 'exit').mockImplementation(() => {
      throw new Error('process.exit called');
    });

    // 模拟请求失败
    mockAxios.onGet(testUrls[0]).reply(500);

    await expect(downloadImages(testUrls)).rejects.toThrow('process.exit called');

    // 验证process.exit被调用
    expect(mockExit).toHaveBeenCalledWith(1);
    mockExit.mockRestore();
  });
});

内容的提问来源于stack exchange，提问作者Noldona