@rxjs/test 完全指南：用虚拟时间与 Marble 测试验证 RxJS 9 Observable 契约

rxjs/test 完全指南用虚拟时间与 Marble 测试验证 RxJS 9 Observable 契约【免费下载链接】rxjsA reactive programming library for JavaScript项目地址: https://gitcode.com/gh_mirrors/rx/rxjsrxjs/test是 RxJS 9 附带的虚拟时间与 Marble 测试包为 Observable 契约提供与具体实现解耦implementation-neutral的确定性测试能力它把宿主环境真实的计时器、动画帧、空闲回调与时钟全部重定向到可推进的虚拟时间上让你能用cold(---a---b|)这样的字符串精确描述并验证异步流。本文以 packages/test/README.md 为核心骨架结合该包的源码实现src/与测试用例src/index.spec.ts完整讲解安装、三种测试源模型、Marble 语法、配置项、Context API、断言方式以及底层虚拟时间原理读完即可独立编写与迁移 RxJS 9 的确定性异步测试。安装与运行环境要求rxjs/test以独立包的形式发布通过 npm 安装即可注意两个依赖都需要next标签与 RxJS 9 预发布版本线对齐npm install --save-dev rxjs/testnext rxjsnext从 packages/test/package.json 可以看到该包的关键约束Node 版本engines.node声明为22.13.0仅 ESMtype: module构建时通过tshy的esmdialect 产出最终发布物只有dist/esm下的 ESM 代码精确 peer 依赖peerDependencies.rxjs为9.0.0-beta.0精确版本而非^范围确保与匹配的 RxJS 9 预发布版一起使用包体仅包含dist与README.md入口由 src/index.ts 定义。快速开始第一个 rxTest 测试README 给出了最简可运行的示例其中map采用 RxJS 9 的 Symbol 算子导入风格import rxjs; import { rxTest } from rxjs/test; import { map } from rxjs/map; await rxTest(({ cold, expectObservable }) { const source cold(-a-b-|, { a: 1, b: 2 }); expectObservable(sourcemap value * 10)).toBe(-a-b-|, { a: 10, b: 20, }); });要点拆解import rxjs负责把平台Observable安装到当前 realm见下文“平台 Observable 前提”随后才调用rxTestrxTest(callback, config?)返回Promisevoid回调是同步或异步的其中可以拿到cold、hot、observable、expectObservable等工具cold(-a-b-|, { a: 1, b: 2 })创建一个冷测试源-表示前进 1 帧虚拟 1 毫秒a/b是占位标记由第二个参数的值表value lookup替换为真实值|表示完成expectObservable(...).toBe(-a-b-|, {...})注册输出期望rxTest结束时统一求值toBe的第二个参数同样是把期望图里的a/b映射成具体值。平台 Observable 前提先初始化 realmREADME 特别强调rxjs/test永远不会替使用者安装平台 Observable。rxTest运行前当前 realm 必须已存在可用的平台Observable构造器通常通过以下两种方式之一提供import rxjs导入主包它会初始化平台 Observable或者import rxjs/observable-polyfill对应仓库内的 packages/observable-polyfill。源码层面的依据在 src/platform-observable.ts 的getObservableConstructor()它读取globalThis.Observable若不存在则直接抛出错误rxjs/test requires the active realm to initialize the platform Observable before rxTest is called.。也就是说rxTest只做“虚拟化”而不做“安装”这保证了被测代码跑在真实平台契约之上而不是某个被测试框架偷偷替换的假实现。仓库的导入测试 test/import/no-observable.mjs 专门验证了这一行为它先删除全局Observable/Subscriber再import(rxjs/test)随后断言导入本身会让globalThis.Observable变成函数——因为该测试的加载环境已由其他入口初始化过平台rxTest也就能正常运行。三种测试源cold、hot 与 observablerxTest提供三种来源分别对应不同的生命周期语义。README 给出的选择原则是只有测试刻意需要“每次直接订阅都得到独立生产者”时才用cold()当契约应该走活动平台的 Observable 构造器时用observable()。三者类型定义见 src/types.tsTestColdObservable/TestHotObservable/TestPlatformObservable实现见 src/test-sources.ts。cold()生产者随订阅独立创建cold(marbles, values?, error?)返回 RxJS 7 风格的冷源继承自rxjs/cold-observable的ColdObservable。每次subscribe都会为这一个观察者独立建立一条从订阅时刻开始计时的消息时间线await rxTest(({ cold, expectObservable, expectSubscriptions }) { const source cold(--a--b--|); expectObservable(source).toBe(--a--b--|); expectObservable(source, ---^).toBe(-----a--b--|); expectSubscriptions(source.subscriptions).toBe([^-------!, ---^-------!]); });第二次观察从第 3 帧---^的^开始于是只能看到从第 5 帧起的a、第 7 帧的b与完成订阅日志source.subscriptions记录两条^...!区间分别对应两次订阅^为订阅帧、!为退订帧单位为虚拟毫秒。hot()类 Subject 的绝对时间线hot(marbles, values?, error?)返回一个类 Subject 的源它的时间线是绝对测试时间在第一个观察者订阅之前就已经在“播出”因此晚订阅的观察者会错过前面的消息。它额外暴露active状态与next/error/complete方法可像 Subject 一样在测试中手动注入事件。hot 图支持用^将某帧声明为零点zeroFrame因此还能表达“零点之前的历史消息”await rxTest(({ hot, expectObservable }) { expectObservable(hot(a-^-b|)).toBe(--b|); });这里a发生在零点^之前即负时间的历史消息晚订阅的观察者只能收到-b|。注意^只能出现在 hot 图中且每图最多一个由 src/marble-parser.ts 的parseMarbles强制校验。observable()走平台共享/引用计数生命周期observable(marbles, values?, error?)用平台Observable构造器创建源new ObservableConstructor(...)因此它测试的是真实平台的共享生产者语义第一个观察者激活生产者并发观察者共享同一个生产者所有观察者退出ref-count 归零后关闭再次订阅会创建新生产者。测试用例 src/index.spec.ts 验证了“最终观察者离开后平台源会重启”await rxTest(({ observable, expectObservable, expectSubscriptions }) { const source observable(--a--|); expectObservable(source, ^--!).toBe(--a); expectObservable(source, -----^).toBe(-------a--|); expectSubscriptions(source.subscriptions).toBe([^--!, -----^----!]); });第一次观察在第 2 帧退订生产者随后关闭第二次从第 5 帧重新订阅触发全新的生产者时间线因此又能完整看到a与|。订阅日志里出现两段独立的^...!区间——这正是平台源与cold()的关键差异cold()的每个订阅都各自独立计时而observable()记录的是真实生产者窗口。订阅日志的结构所有测试源都带有subscriptions: readonly TestSubscriptionLog[]属性TestSubscriptionLog只含两个字段字段含义subscribedFrame订阅发生时的绝对虚拟时间毫秒unsubscribedFrame退订发生的绝对虚拟时间毫秒未退订为Infinity日志在 src/test-sources.ts 的openSubscriptionLog/closeSubscriptionLog中写入配合expectSubscriptions(logs).toBe(^-------!)使用。Marble 语法速查Marble 图的解析逻辑全部集中在 src/marble-parser.ts纯函数模块不依赖虚拟时钟与 Observable规则如下记号含义-前进 1 帧1 虚拟毫秒(ab)分组a、b落在同一帧不支持嵌套分组ab…普通值标记由values值表替换无值表时标记本身即字符串值\|完成notification{ kind: C }#错误notification{ kind: E }默认错误值为字符串error可用第三个参数error指定^仅 hot 图将该帧声明为零点之前为负时间历史!仅订阅图退订帧每图最多一个12ms/5s/1m时长 token等价于指定毫秒数数字前必须紧跟空白readDuration要求前一字符是空白或串首如12ms a 20ms (b|)几个由解析器保证的约束违反即抛错未闭合的分组(、多余的)普通非 hot图中出现^、任意图中出现!订阅图中出现^/!之外的字符hot 图中出现两个及以上^时长 token 必须是\d(\.\d)?加ms/s/m单位且必须非负有限validateDuration校验。帧与毫秒在这里是等价的一个-就是 1ms因此time(12ms ---|)的返回值是1512ms 3 帧。订阅图的解析走独立的parseSubscriptionMarbles^记录subscribedFrame!记录unsubscribedFrame未指定订阅图时默认订阅于第 0 帧、永不退订。expectObservable(source, ----!)这类带退订点的写法其退订帧是排他边界——第 4 帧的!意味着第 4 帧之前不含第 4 帧的值会被记录测试 src/index.spec.ts 中有--a--b--配--------!只得到--a--b--的验证。rxTest 配置项RxTestConfigrxTest(callback, config)的第二个参数支持以下配置类型定义见 src/types.ts默认值与校验逻辑见 src/rx-test.ts 的executeRxTest配置项类型默认值说明assertDeepEqual(actual, expected, info) void \| PromiseLikevoid内置深度严格相等自定义 Observable/订阅断言适配器默认实现直接抛RxTestAssertionError无需任何测试框架适配startTimenumber \| string \| Date0虚拟Date使用的纪元时间戳。注意now()与performance.now()仍从 0 开始只有Date.now()/new Date()返回startTime nowmaxVirtualTimeTestDuration数字毫秒或5s等Infinity允许进入的最大虚拟时间戳用于防止测试“跑飞”maxTaskExecutionsnumber100000回调执行上限用于诊断自调度死循环如setInterval(() {}, 0)必须为正整数否则抛TypeErroridleBudgetTestDuration50虚拟IdleDeadline.timeRemaining()的默认预算其中startTime支持Date实例与可解析的字符串/时间戳maxVirtualTime、idleBudget都会经过durationToMilliseconds归一化为毫秒。测试用例验证了maxVirtualTime与maxTaskExecutions的护栏行为// 超过 maxVirtualTime 会失败 await expect( rxTest(() setTimeout(() {}, 11), { maxVirtualTime: 10 }) ).rejects.toThrow(maxVirtualTime); // 无限周期任务会在执行 5 次后失败而不是挂起 await expect( rxTest(() setInterval(() {}, 0), { maxTaskExecutions: 5 }) ).rejects.toThrow(maxTaskExecutions);RxTestContext 全 API回调参数context是 src/context.ts 中RxTestContextImplementation的实例完整接口见 src/types.ts 的RxTestContextAPI签名作用signalAbortSignal测试完成或失败时中止的全局信号cold(marbles, values?, error?) TestColdObservable创建冷测试源hot(marbles, values?, error?) TestHotObservable创建热绝对时间线测试源observable(marbles, values?, error?) TestPlatformObservable创建走平台构造器的测试源time(marbles) number返回时序图中唯一\|的时间戳now() number返回自测试开始以来的虚拟毫秒数schedule(work, delay?, options?) ScheduledTestTask在虚拟任务队列上直接调度任务delay支持时长 tokenoptions.signal可外部取消expectObservable(actual, subscriptionMarbles?) ObservableExpectation注册 Observable 输出期望可选第二个参数声明订阅时机/退订点expectSubscriptions(logs) SubscriptionExpectation对测试源的实时订阅日志做断言animate(plan) void声明动画帧机会点虚拟requestAnimationFrame触发时刻只能在时间推进前调用一次idle(plan, options?) void声明空闲机会点options.budget覆盖本计划的空闲预算flush() Promisevoid排空有限虚拟工作并求值当前期望advanceBy(duration) Promisevoid相对推进并排空截止到该时刻的工作advanceTo(time) Promisevoid推进到绝对虚拟时间戳不允许回拨scheduled任务返回ScheduledTestTask只读的dueTime绝对虚拟毫秒、可监听取消的signal以及可重复调用但只生效一次的cancel(reason?)。animate/idle的plan类型TestTimingPlan是string | readonly number[]字符串形式中除-与保留标记\| # ^ ! ( )之外的任意字符都代表一个机会点如--------数组形式则必须是严格递增的绝对毫秒时间戳违反会抛错见parseTimingPlan。动画/空闲与手推进组合的典型用例来自 src/index.spec.tsawait rxTest(async ({ animate, idle, flush, now }) { animate(--------); idle([6, 12], { budget: 5 }); const events: string[] []; requestAnimationFrame(() events.push(paint:${now()})); setTimeout(() { requestAnimationFrame(() events.push(paint:${now()})); }, 5); requestIdleCallback((deadline) { events.push(idle:${now()}:${deadline.didTimeout}:${deadline.timeRemaining()}); }); await flush(); expect(events).toEqual([paint:4, idle:6:false:5, paint:9]); });断言与匹配器断言管理实现在 src/assertions.ts 的ExpectationManager中。ObservableExpectationexpectObservable(actual, subscriptionMarbles?)返回带三种匹配器的对象toBe(marbles, values?, error?)把实际记录的通知序列与 Marble 图解析出的期望消息做深度相等比较。这是最常用的形式toBe(messages)直接与一组精确的时间戳消息数组比较例如await rxTest(({ cold, expectObservable }) { expectObservable(cold(-(aaa)#)).toBe([ { frame: 1, notification: { kind: N, value: a } }, { frame: 1, notification: { kind: N, value: a } }, { frame: 1, notification: { kind: N, value: a } }, { frame: 6, notification: { kind: E, error: error } }, ]); });toEqual(expected: Observable)与另一个 Observable 在同一记录窗口内并行记录并比较例如expectObservable(cold(--a--|)).toEqual(cold(--a--|))。一个值得注意的细节toBe期望图里的值如果是测试源isTestSource会自动“物化”为该源的messages数组#materializeExpectedMessage这使得高阶 Observablemetastream测试非常自然const inner cold(-x|); expectObservable(cold(a|, { a: inner })).toBe(a|, { a: inner });SubscriptionExpectationexpectSubscriptions(logs).toBe(marbles | marbles[])用订阅图^订阅、!退订比较实时日志传入字符串数组时逐条对应多个日志expectSubscriptions(source.subscriptions).toBe([^-------!, ---^-------!]);默认深度相等与 RxTestAssertionError内置的defaultAssertDeepEqual使用一个完整的深度严格比较器deepEqual对以下类型都有专门处理Date比较时间戳、RegExp比较 sourceflags、Error比较 name/message/cause、TypedArray逐字节比较、Map、Set、普通对象枚举键递归比较以及循环引用通过WeakMap记录访问对。比较失败时抛出RxTestAssertionError该错误携带结构化断言信息export class RxTestAssertionError extends Error { readonly name RxTestAssertionError as const; constructor(readonly actual: unknown, readonly expected: unknown, readonly assertion: RxTestAssertionInfo) { ... } }其中assertionRxTestAssertionInfo包含kind: observable | subscriptions与可选的marbles便于测试框架适配器定位失败点。自定义断言适配器示例await rxTest( ({ cold, expectObservable }) { expectObservable(cold(--a|)).toBe(--a|); }, { assertDeepEqual(actual, expected, info) { // info.kind observable接入 jest 的 expect 或 node:assert 等 expect(actual).toEqual(expected); }, } );常见的失败模式rxTest结束时会依次执行finalizeExpectations、assertNoOpenObservations、assertNoPendingWork因此以下情况都会让测试以异常失败而不是静默通过或挂起注册了expectObservable/expectSubscriptions但没有调用任何 matcher“registered without a matcher”源没有完成且未提供退订图“open observation(s)”提示Complete the source or provide an unsubscription marble测试结束时仍有未取消的 interval、动画帧回调或空闲回调“pending virtual work”未处理的reportError或微任务中的异常会被收集并转为测试失败冷源错误无人处理时同样转为测试失败src/index.spec.ts 有对应用例。底层原理虚拟时间 RealmrxTest的核心是 src/virtual-time.ts 中的VirtualTimeController在回调运行的整个异步生命周期内把原生调度 API 替换为确定性虚拟时间实现结束后按逆序恢复所有被替换的属性描述符#patch/restore成对出现restore使用PatchRecord记录原始描述符恢复失败会以AggregateError报告。被虚拟化的宿主 APIinstall()依次安装以下补丁对应测试 src/index.spec.ts 中逐一验证恢复的属性清单类别被替换的 API计时器setTimeout/clearTimeout/setInterval/clearIntervalNode 下返回带ref/unref/hasRef/refresh与Symbol.dispose的VirtualNodeTimerHandle动画帧requestAnimationFrame/cancelAnimationFrame空闲回调requestIdleCallback/cancelIdleCallbackdeadline.timeRemaining()返回空闲预算超时用timeout选项驱动立即任务setImmediate/clearImmediate仅当宿主存在时取消AbortSignal.timeout虚拟化为可推进的调度任务时钟Date虚拟Date.now()/new Date()返回startTime now与performance.now()返回now微任务queueMicrotask计数包装失败转发到reportError记录错误上报reportError转为测试失败记录任务优先级同一虚拟时刻可能有多个任务控制器按“到期时间 → 任务类型优先级 → 注册顺序”排序执行#enqueue中的排序比较器与taskPriority表优先级任务类型-1observation-boundary退订边界0observation-start订阅起点、immediate1timer、scheduled2animation3idle这正是 hot 源“时间零点通知发生在同帧订阅之后”的保证测试attaches a time-zero observation before a hot time-zero notification订阅任务优先级 0 的 observation-start排在值通知之前。微任务则会在每个任务执行后通过#microtaskCheckpoint排空保证“同一帧的两个 timer 回调之间执行queueMicrotask的微任务”测试runs microtasks between same-time timer callbacks。Realm 锁串行化与嵌套防护多个rxTest在同一个 realm 并发运行是安全的rxTest通过Symbol.for(rxjs/test/realm-lock/v1)挂载的全局锁把测试串行化——每个测试都排在锁队列尾部执行lock.tail。同时嵌套调用被明确拒绝Nested rxTest calls are not supported.回调/配置非法非函数回调、非法startTime、非正maxTaskExecutions会直接以TypeError拒绝。测试serializes overlapping tests in the same realm验证了两个重叠rxTest的顺序执行。从 RxJS 7 的 TestScheduler.run 迁移packages/rxjs/MIGRATION.md 的 “Scheduling and testing” 一节说明了迁移时的关键决策RxJS 9 不再发布 RxJS 7 的 scheduler 系统与通用 scheduler 参数运行时能力走宿主计时器、动画帧与窄时钟提供方而rxjs/test正是用来虚拟化这些受支持的宿主 API、做确定性测试的官方路径。迁移 Marble 测试时必须显式选择源模型原文三条cold()每次订阅创建独立的生产者工作与时间线对应 RxJS 7 冷源语义hot()在观察者之前就存在的类 Subject 绝对时间线生产者observable()走平台的共享/引用计数生命周期。MIGRATION.md 还特别警告移除 scheduler 参数从来不是机械编辑除非该能力与计时声明已经被分类并测试过。迁移时还要注意生命周期与取消模型的差异——RxJS 9 使用AbortSignal表达取消不再有subscribe()返回Subscription的假设因此迁移测试前应先在 MIGRATION.md 与 docs/UNSUPPORTED_RXJS_7_SURFACES.md 中核对每个被测算子的生命周期目标。结语rxjs/test把“宿主时间”整体收进虚拟时钟让 RxJS 9 的异步契约可以用可读的 Marble 图精确表达与验证同时通过源模型的选择cold/hot/observable覆盖从 RxJS 7 冷源语义到平台共享生命周期的完整谱系。若要继续深入建议按以下顺序阅读仓库源码语法与类型/packages/test/src/marble-parser.ts、src/types.ts运行时与上下文src/rx-test.ts、src/context.ts、src/virtual-time.ts断言与测试源src/assertions.ts、src/test-sources.ts行为验证/packages/test/src/index.spec.ts几乎覆盖了上文所有行为点版本约束与脚本packages/test/package.json、packages/test/README.md。【免费下载链接】rxjsA reactive programming library for JavaScript项目地址: https://gitcode.com/gh_mirrors/rx/rxjs创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考