Angular单元测试中如何Mock RxJS Subject并调用next方法

问题描述 #

尝试Mock作为服务属性的RxJS Subject（该服务通过依赖注入注入到组件构造函数），组件的responseHandler()方法会订阅这个Subject。编写单元测试时，无法正确Mock该Subject并调用其next方法，测试未达到预期效果。

相关代码 #

app.component.ts #

import { Component } from '@angular/core';
import { UtilsService } from './services/utils.service';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.scss']
})

export class AppComponent {
message!: string;

constructor(private utilsService: UtilsService) { }

storeProduct() {
  this.utilsService.storeProductDetails();
  this.responseHandler();
}

responseHandler() {
  this.utilsService._httpResponse$.subscribe(({ message }) => {
  this.message = message;
  });
 }
}

utils.service.ts #

import { Injectable } from '@angular/core';
import { Subject } from 'rxjs';
import { CoreService } from './core.service';

@Injectable({
   providedIn: 'root'
})

export class UtilsService {
  _httpResponse$ = new Subject<any>()
  constructor(private coreService: CoreService) { }

  storeProductDetails() {
    this.coreService.storeProductDetails().subscribe(res => {
    this.httpResponse(res.content, null);
  })
 }

  httpResponse(content: any, errors: any) {
  this._httpResponse$.next({
    message: content,
    errors: errors
  });
 }
}

core.service.ts #

import { Injectable } from '@angular/core';
import { Observable, of } from 'rxjs';

@Injectable({
  providedIn: 'root'
})

export class CoreService {
  storeProductDetails(): Observable<any> {
    return of({ content: 'Product was stored succesffuly' });
  }
}

原测试代码 app.component.spec.ts #

import { TestBed, ComponentFixture } from '@angular/core/testing';
import { RouterTestingModule } from '@angular/router/testing';
import { HttpClientTestingModule } from '@angular/common/http/testing';
import { AppComponent } from './app.component';
import { CoreService } from './services/core.service';
import { UtilsService } from './services/utils.service';
import { of } from 'rxjs';

describe('AppComponent', () => {
  let app: AppComponent;
  let fixture: ComponentFixture<AppComponent>;
  let coreServiceSpy: jasmine.SpyObj<CoreService>;
  let utilsService: UtilsService;

  beforeEach(async () => {
    coreServiceSpy = jasmine.createSpyObj('CoreService', ['storeProductDetails']);

    await TestBed.configureTestingModule({
      imports: [
        RouterTestingModule,
        HttpClientTestingModule
      ],
      declarations: [
        AppComponent
      ],
      providers: [
        {
          provide: CoreService,
          useValue: coreServiceSpy
        }
      ]
    }).compileComponents();
  });

  beforeEach(() => {
    utilsService = TestBed.inject(UtilsService);
   fixture = TestBed.createComponent(AppComponent);
   app = fixture.componentInstance;
 })

  it('should store product details succesful and show message', () => {
    coreServiceSpy.storeProductDetails.and.returnValue(of({ content: 'The product details was stored succesful.' }))
    utilsService._httpResponse$.next({ message: 'The product details was stored succesful.', errors: null });
    fixture.detectChanges();
    app.storeProduct();
    expect(app.message).toEqual('The product details was stored succesful.');
  });
});

问题分析 #

• 未Mock UtilsService：测试中直接使用真实的UtilsService实例，而非Mock对象，无法完全控制_httpResponse$ Subject的行为，且真实服务依赖CoreService的调用逻辑，干扰测试隔离性。
• 测试逻辑顺序错误：先调用utilsService._httpResponse$.next()，再执行app.storeProduct()，此时responseHandler()还未订阅Subject，导致发出的值无法被捕获。
• 重复订阅风险：每次调用storeProduct()都会执行responseHandler()，新增一个订阅，未清理的订阅会导致内存泄漏，且多次测试会相互干扰。

修正后的测试代码 #

import { TestBed, ComponentFixture, fakeAsync, tick } from '@angular/core/testing';
import { RouterTestingModule } from '@angular/router/testing';
import { AppComponent } from './app.component';
import { CoreService } from './services/core.service';
import { UtilsService } from './services/utils.service';
import { of, Subject } from 'rxjs';

describe('AppComponent', () => {
  let app: AppComponent;
  let fixture: ComponentFixture<AppComponent>;
  let coreServiceSpy: jasmine.SpyObj<CoreService>;
  let utilsServiceSpy: jasmine.SpyObj<UtilsService>;
  let httpResponseSubject: Subject<any>;

  beforeEach(async () => {
    // 创建CoreService的Spy
    coreServiceSpy = jasmine.createSpyObj('CoreService', ['storeProductDetails']);
    // 创建Subject用于模拟_httpResponse$
    httpResponseSubject = new Subject<any>();
    // 创建UtilsService的Spy，模拟方法和属性
    utilsServiceSpy = jasmine.createSpyObj('UtilsService', ['storeProductDetails'], {
      _httpResponse$: httpResponseSubject
    });

    await TestBed.configureTestingModule({
      imports: [RouterTestingModule],
      declarations: [AppComponent],
      providers: [
        { provide: CoreService, useValue: coreServiceSpy },
        { provide: UtilsService, useValue: utilsServiceSpy }
      ]
    }).compileComponents();
  });

  beforeEach(() => {
    fixture = TestBed.createComponent(AppComponent);
    app = fixture.componentInstance;
  });

  // 清理订阅，避免内存泄漏
  afterEach(() => {
    httpResponseSubject.complete();
  });

  it('should store product details successfully and show message', fakeAsync(() => {
    // 模拟CoreService的返回值
    const mockResponse = { content: 'The product details was stored successful.' };
    coreServiceSpy.storeProductDetails.and.returnValue(of(mockResponse));

    // 执行组件方法，触发订阅
    app.storeProduct();
    tick(); // 触发异步操作

    // 模拟UtilsService发出响应
    httpResponseSubject.next({ message: mockResponse.content, errors: null });
    tick(); // 让订阅回调执行

    expect(app.message).toEqual(mockResponse.content);
    // 验证storeProductDetails方法被调用
    expect(utilsServiceSpy.storeProductDetails).toHaveBeenCalled();
  }));
});

关键修改说明 #

• Mock UtilsService：通过jasmine.createSpyObj创建UtilsService的Mock对象，将_httpResponse$替换为可控的Subject实例，完全隔离真实服务的逻辑。
• 调整测试顺序：先执行app.storeProduct()让组件完成订阅，再调用Subject的next()方法，确保值能被订阅捕获。
• 使用fakeAsync/tick：处理RxJS的异步操作，确保订阅回调有足够时间执行，避免测试因异步时序问题失败。
• 清理订阅：在afterEach中调用complete()关闭Subject，避免内存泄漏和后续测试干扰。
• 增加验证：添加对storeProductDetails方法调用的验证，确保组件逻辑符合预期。

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