如何测试Angular中Dispatcher订阅内的私有字段以提升覆盖率？

组件代码 #

@Component({
  selector: 'page',
  templateUrl: './page.html',
  styleUrls: ['./page.scss']
})
export class PageComponent implements OnInit, OnDestroy {
  private lts: Igst;
  constructor(
    private router: Router,
    private route: ActivatedRoute,
    private dispatcher: Dispatcher
    ) {
    this.dispatcher
      .ofType(ProductsPageActions.TILE_UPDATE_STATE)
      .pipe(
        map((action: ProductsPageActions.St) => action.payload.st),
        takeUntil(this.componentDestroyed$)
      )
      .subscribe((st: Igst) => {
        this.lts = st;
      }
  }

  handleTs(st: Igst): void {
    this.dispatcher.dispatch(new ProductsPageActions.St({st}));
  }
}

测试代码（原尝试） #

describe('Testing pageComponent', () => {
 let cmp: PageComponent;
 let fixture: ComponentFixture<PageComponent>;
 let dispatcher: Dispatcher;

 beforeEach(() => {
    TestBed.configureTestingModule({
      imports: [HttpClientTestingModule, RouterTestingModule],
      schemas: [NO_ERRORS_SCHEMA],
      providers: [
        {provide: ActivatedRoute, useValue: {params: of({section: 'test'})}},
        Dispatcher
      ]
    });


      it('should handleTileSelect', () => {
        spyOn(dispatcher, 'dispatch');
        const payload = {
            agrType: 'L',
        };
        cmp.handleTs(payload); // payload is st
    
        expect(dispatcher.dispatch).toHaveBeenCalledTimes(1); // ok
    
        expect(cmp['lts']).toEqual(payload); // lts is undefined
      });
}

依赖与补充定义 #

package.json核心依赖 #

"dependencies": {
  "@angular/core": "^13.3.0",
  "rxjs": "^6.5.5"
},
"devDependencies": {
  "@angular/cli": "~13.3.0",
  "jasmine": "^4.0.2",
  "karma": "^6.3.17"
}

Dispatcher类 #

import {IAction} from './action';
import {filter} from 'rxjs/operators';
import {Observable, Subject} from 'rxjs';
export class Dispatcher extends Subject<IAction> {
  dispatch(action: IAction): void {
    super.next(action);
  }

  ofType(type: string): Observable<IAction> {
    return this.pipe(filter(action => action.type === type));
  }
}

export const ofType = (type: string) => (source: Observable<IAction>) =>
  new Observable(observer => {
    source
      .pipe(
        filter((action: IAction) => action.type === type)
      ).subscribe(
      (data) => {
        observer.next(data);
      },
      (err) => {
        observer.error(err);
      },
      () => {
        observer.complete();
      }
    );
  });

IAction接口 #

export interface IAction {
  readonly type: string;
  readonly reducerId?: string;
  payload?: any;
}

问题原因分析 #

1. 组件缺失必要的销毁信号Subject：组件中使用了takeUntil(this.componentDestroyed$)但未定义该变量，导致订阅流立即完成，无法接收后续的action事件，this.lts = st永远不会执行。
2. 未处理异步逻辑：Dispatcher的订阅回调是异步执行的，原测试中调用handleTs后立即断言，此时回调还未触发，lts仍为undefined。
3. 测试实例初始化不完整：原测试中未正确获取Dispatcher实例和创建组件实例，导致测试逻辑无法关联到真实的组件订阅。

解决方案 #

第一步：修复组件代码，补充销毁信号 #

@Component({
  selector: 'page',
  templateUrl: './page.html',
  styleUrls: ['./page.scss']
})
export class PageComponent implements OnInit, OnDestroy {
  private lts: Igst;
  // 补充销毁信号Subject
  private componentDestroyed$ = new Subject<void>();

  constructor(
    private router: Router,
    private route: ActivatedRoute,
    private dispatcher: Dispatcher
  ) {
    this.dispatcher
      .ofType(ProductsPageActions.TILE_UPDATE_STATE)
      .pipe(
        map((action: ProductsPageActions.St) => action.payload.st),
        takeUntil(this.componentDestroyed$)
      )
      .subscribe((st: Igst) => {
        this.lts = st;
      }); // 补充分号
  }

  // 实现OnDestroy接口，发送销毁信号
  ngOnDestroy(): void {
    this.componentDestroyed$.next();
    this.componentDestroyed$.complete();
  }

  handleTs(st: Igst): void {
    this.dispatcher.dispatch(new ProductsPageActions.St({st}));
  }
}

第二步：调整测试代码，处理异步逻辑并完善实例初始化 #

describe('Testing pageComponent', () => {
  let cmp: PageComponent;
  let fixture: ComponentFixture<PageComponent>;
  let dispatcher: Dispatcher;

  beforeEach(() => {
    TestBed.configureTestingModule({
      imports: [HttpClientTestingModule, RouterTestingModule],
      schemas: [NO_ERRORS_SCHEMA],
      providers: [
        {provide: ActivatedRoute, useValue: {params: of({section: 'test'})}},
        Dispatcher
      ]
    });

    // 获取Dispatcher实例并创建组件
    dispatcher = TestBed.inject(Dispatcher);
    fixture = TestBed.createComponent(PageComponent);
    cmp = fixture.componentInstance;
    fixture.detectChanges(); // 触发组件初始化，建立订阅
  });

  it('should update lts when handleTs is called', (done) => {
    const payload = { agrType: 'L' };
    
    // 监听目标action，确保订阅回调执行后再断言
    dispatcher.ofType(ProductsPageActions.TILE_UPDATE_STATE).subscribe(() => {
      expect(cmp['lts']).toEqual(payload);
      done(); // 告知测试异步操作完成
    });

    cmp.handleTs(payload);
    expect(dispatcher.dispatch).toHaveBeenCalledTimes(1);
  });

  // 可选：直接测试订阅逻辑，不通过handleTs中转
  it('should update lts when TILE_UPDATE_STATE action is dispatched', (done) => {
    const payload = { agrType: 'L' };
    const action = new ProductsPageActions.St({ st: payload });

    dispatcher.ofType(ProductsPageActions.TILE_UPDATE_STATE).subscribe(() => {
      expect(cmp['lts']).toEqual(payload);
      done();
    });

    dispatcher.dispatch(action);
  });
});

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