状态流与时间测试：把竞态变成断言

状态流的关键不是某个 Subject，而是谁能写、谁只能读，以及状态在失败和退订后如何恢复。内部可用 BehaviorSubject 保存快照，但对外只暴露 asObservable()；写入入口集中在 reducer 或 service。

从事件到只读状态

import { BehaviorSubject } from 'rxjs';
import { scan, distinctUntilChanged } from 'rxjs/operators';
const events$ = new BehaviorSubject({type:'init'});
const state$ = events$.pipe(
  scan((state, event) => {
    if (event.type === 'loading') return {...state, loading:true, error:null};
    if (event.type === 'success') return {loading:false, data:event.data, error:null};
    if (event.type === 'failure') return {loading:false, data:state.data, error:event.error};
    return state;
  }, {loading:false, data:null, error:null}),
  distinctUntilChanged((a,b) => a.loading===b.loading && a.data===b.data && a.error===b.error)
);
// service 对外只暴露 state$，不暴露 events$

测试顺序至少覆盖初始 → loading → failure → loading → success，以及卸载时订阅清理；只断言最后数据会漏掉 loading 闪烁、旧错误残留和重复订阅。

不依赖 sleep 的时间线测试

RxJS 7 的 TestScheduler 使用虚拟时间：- 是时间帧，a 是值，| 是完成，# 是错误。

import { TestScheduler } from 'rxjs/testing';
import { switchMap } from 'rxjs/operators';
const scheduler = new TestScheduler((actual, expected) => {
  if (JSON.stringify(actual) !== JSON.stringify(expected)) throw new Error('时间线不符');
});
scheduler.run(({hot, cold, expectObservable}) => {
  const query$ = hot('-a--b----|');
  const response = cold('---x|');
  expectObservable(query$.pipe(switchMap(() => response))).toBe('-------x-|');
});

改变 b 的时间后，旧响应若尚未完成应被取消；这比等待真实计时器更能证明协议。故意删除 teardown 或把 catchError 移到外层，让测试先失败，再恢复实现并记录回归。

练习

为 loading/success/failure 写三条事件和一个非法事件测试，非法事件保留旧状态；将 switchMap 改成 concatMap，更新 marble 期望并解释输出为何变晚。

查看本节答案