使用RxJS Marble Testing测试返回Observable的函数遇问题求助

其实你不用纠结“热源”的问题——冷 Observable 本身就是订阅时才启动，刚好适配 rxjs-marbles 的测试逻辑。我给你捋清楚具体怎么写，结合你的场景举个实际例子：

先明确你的目标函数（假设长这样） #

比如你说的“接收数组、延迟依次发射每个值”的函数，大概是这种实现：

import { from, of } from 'rxjs';
import { mergeMap, delay } from 'rxjs/operators';

// 接收数组，每个元素间隔 delayMs 发射
export function delayedArrayEmitter(arr: any[], delayMs = 100) {
  return from(arr).pipe(
    mergeMap(item => of(item).pipe(delay(delayMs)))
  );
}

测试用例写法（以 Jest 为例） #

首先确保你已经安装了 rxjs-marbles，然后直接用它的测试绑定来写：

import { marbles } from 'rxjs-marbles/jest';
import { delayedArrayEmitter } from './your-file-path';

describe('delayedArrayEmitter', () => {
  // 核心测试：按延迟依次发射元素
  it('emits each array item with specified delay between them', marbles(m => {
    // 1. 定义测试输入
    const testInput = ['foo', 'bar', 'baz'];
    // 2. 定义预期的 marble 序列
    // 这里每个 `-` 代表 10ms（rxjs-marbles 默认帧时长），10个 `-` 就是 100ms
    // `|` 代表 Observable 完成
    const expected = m.cold('----------foo----------bar----------baz|', {
      foo: 'foo',
      bar: 'bar',
      baz: 'baz'
    });

    // 3. 获取实际的 Observable
    const actual$ = delayedArrayEmitter(testInput, 100);

    // 4. 对比实际和预期序列
    m.expect(actual$).toBeObservable(expected);
  }));

  // 边界测试：空数组直接完成
  it('completes immediately with empty array', marbles(m => {
    const expected = m.cold('|');
    const actual$ = delayedArrayEmitter([]);

    m.expect(actual$).toBeObservable(expected);
  }));
});

关键注意点 #

1. 不用热源！ 你的函数返回冷 Observable，直接用 m.cold() 创建预期序列就行——rxjs-marbles 会自动帮你订阅两个 Observable，对比它们的发射时机和内容。
2. 时间帧对应：默认每个 - 是 10ms，如果你的延迟是 200ms，那就要写 20 个 -；嫌麻烦也可以用 m.time() 直接指定毫秒：

   const expected = m.cold(`${m.time(100)}foo${m.time(100)}bar${m.time(100)}baz|`, { ... });
   

3. 不要漏完成信号：预期序列末尾一定要加 |，否则测试会因为 Observable 没完成而失败。

常见坑点排查 #

如果你的测试一直失败，大概率是这两个原因：

• 时间不匹配：比如函数里的延迟是 100ms，但预期序列的间隔是 50ms，调整一致就行；
• 没有正确导入 rxjs-marbles 的测试绑定：不同测试框架导入路径不一样，Jest 用 rxjs-marbles/jest，Jasmine 用 rxjs-marbles/jasmine。

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