Angular组件中RxJS订阅式服务调用的单元测试问题

问题场景 #

组件代码 #

export class MyComp {
  destroy$ = new Subject<void>();
  submits$ = new Subject<MyFormState>(); // 在模板中调用.next()

  ngOnInit(){
    this.submits$.pipe(
      takeUntil(this.destroy$)
    ).subscribe(formState => this.service.doSomething(formState));
  }
}

错误测试写法1（假阳性） #

describe('MyComp', () => {
  let submits$ = new Subject<MyFormState>();
  let mock = jasmine.createSpyObj('MyService', [ 'doSomething' ]);

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [ MyComp ],
      providers: [
        { provide: MyService, useValue: mock },
      ],
    });
  });

  it('should be called when clicking the button', () => {
    const f: MyFormState = { name: 'my name' };
    component.submits$.next(formState);

    waitForAsync(() => {
      expect(mock.doSomething).toHaveBeenCalledWith(f); // 假阳性：任何断言都会通过
    });
  });
});

错误测试写法2（断言失败） #

it('should be called when clicking the button', fakeAsync(() => {
  // 调用component.submits$.next(...)

  tick();

  expect(mock.doSomething).toHaveBeenCalledWith(f);
}));

错误原因分析 #

1. waitForAsync假阳性问题：waitForAsync用于处理Angular框架级异步操作（如变更检测、HTTP请求），但Subject的订阅是同步执行的。如果没有在回调中触发fixture.whenStable()这类异步完成信号，Jasmine会直接跳过断言逻辑，导致任何断言都被判定为通过。
2. fakeAsync无效问题：核心是组件的ngOnInit未被正确触发，submits$的订阅关系还没建立。此外，Subject的next()是同步操作，不需要tick()来推进时间。

正确测试方式（不使用done()） #

核心：确保组件初始化完成，订阅关系建立 #

describe('MyComp', () => {
  let component: MyComp;
  let fixture: ComponentFixture<MyComp>;
  let mockService: jasmine.SpyObj<MyService>;

  beforeEach(() => {
    mockService = jasmine.createSpyObj('MyService', ['doSomething']);

    TestBed.configureTestingModule({
      declarations: [MyComp],
      providers: [
        { provide: MyService, useValue: mockService }
      ]
    });

    fixture = TestBed.createComponent(MyComp);
    component = fixture.componentInstance;
    // 触发ngOnInit，建立submits$的订阅关系
    fixture.detectChanges();
  });

  it('should call doSomething when submits$ emits', () => {
    const formState: MyFormState = { name: 'my name' };
    
    // 触发Subject发射值
    component.submits$.next(formState);
    
    // 直接断言：Subject的订阅回调同步执行
    expect(mockService.doSomething).toHaveBeenCalledWith(formState);
  });
});

用TestScheduler测试匿名订阅 #

对于复杂的RxJS流场景，可使用TestScheduler来精准测试订阅行为：

import { TestScheduler } from 'rxjs/testing';

describe('MyComp', () => {
  let component: MyComp;
  let mockService: jasmine.SpyObj<MyService>;
  let testScheduler: TestScheduler;

  beforeEach(() => {
    mockService = jasmine.createSpyObj('MyService', ['doSomething']);
    testScheduler = new TestScheduler((actual, expected) => {
      expect(actual).toEqual(expected);
    });

    TestBed.configureTestingModule({
      declarations: [MyComp],
      providers: [{ provide: MyService, useValue: mockService }]
    });

    component = TestBed.createComponent(MyComp).componentInstance;
    // 手动触发ngOnInit建立订阅
    component.ngOnInit();
  });

  it('should call service method when submits$ emits', () => {
    testScheduler.run((helpers) => {
      const { cold } = helpers;
      const testFormState: MyFormState = { name: 'my name' };

      // 模拟submits$发射值
      cold('a', { a: testFormState }).subscribe(val => component.submits$.next(val));

      // 断言服务方法被调用
      expect(mockService.doSomething).toHaveBeenCalledWith(testFormState);
    });
  });
});

注意 #

• 简单同步订阅场景，直接同步断言更高效；TestScheduler适合测试带时间依赖、操作符变换的复杂流。
• 无论哪种方式，都要确保ngOnInit已执行，订阅关系已建立。

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