Fix remaining tests

src/room/GroupCallView.test.tsx

@@ -16,7 +16,6 @@ import {
 import { render, waitFor, screen } from "@testing-library/react";
 import { type MatrixClient, JoinRule, type RoomState } from "matrix-js-sdk";
 import { type MatrixRTCSession } from "matrix-js-sdk/lib/matrixrtc";
-import { of } from "rxjs";
 import { BrowserRouter } from "react-router-dom";
 import userEvent from "@testing-library/user-event";
 import { type RelationsContainer } from "matrix-js-sdk/lib/models/relations-container";
@@ -43,6 +42,7 @@ import { MatrixRTCFocusMissingError } from "../utils/errors";
 import { ProcessorProvider } from "../livekit/TrackProcessorContext";
 import { MediaDevicesContext } from "../MediaDevicesContext";
 import { HeaderStyle } from "../UrlParams";
+import { constant } from "../state/Behavior";
 
 vi.mock("../soundUtils");
 vi.mock("../useAudioContext");
@@ -141,7 +141,7 @@ function createGroupCallView(
     room,
     localRtcMember,
     [],
-  ).withMemberships(of([]));
+  ).withMemberships(constant([]));
   rtcSession.joined = joined;
   const muteState = {
     audio: { enabled: false },

src/state/Behavior.ts

@@ -44,8 +44,19 @@ Observable.prototype.behavior = function <T>(
   scope: ObservableScope,
 ): Behavior<T> {
   const subject$ = new BehaviorSubject<T | typeof nothing>(nothing);
-  // Push values from the Observable into the BehaviorSubject
+  // Push values from the Observable into the BehaviorSubject.
-  this.pipe(scope.bind(), distinctUntilChanged()).subscribe(subject$);
+  // BehaviorSubjects have an undesirable feature where if you call 'complete',
+  // they will no longer re-emit their current value upon subscription. We want
+  // to support Observables that complete (for example `of({})`), so we have to
+  // take care to not propagate the completion event.
+  this.pipe(scope.bind(), distinctUntilChanged()).subscribe({
+    next(value) {
+      subject$.next(value);
+    },
+    error(err) {
+      subject$.error(err);
+    },
+  });
   if (subject$.value === nothing)
     throw new Error("Behavior failed to synchronously emit an initial value");
   return subject$ as Behavior<T>;

src/state/CallViewModel.test.ts

@@ -12,9 +12,9 @@ import {
   debounceTime,
   distinctUntilChanged,
   map,
+  NEVER,
   type Observable,
   of,
-  skip,
   switchMap,
 } from "rxjs";
 import { type MatrixClient } from "matrix-js-sdk";
@@ -75,11 +75,18 @@ import {
 import { ObservableScope } from "./ObservableScope";
 import { MediaDevices } from "./MediaDevices";
 import { getValue } from "../utils/observable";
-import { constant } from "./Behavior";
+import { type Behavior, constant } from "./Behavior";
 
 const getUrlParams = vi.hoisted(() => vi.fn(() => ({})));
 vi.mock("../UrlParams", () => ({ getUrlParams }));
 
+vi.mock("rxjs", async (importOriginal) => ({
+  ...(await importOriginal()),
+  // Disable interval Observables for the following tests since the test
+  // scheduler will loop on them forever and never call the test 'done'
+  interval: (): Observable<number> => NEVER,
+}));
+
 vi.mock("@livekit/components-core");
 
 const daveRtcMember = mockRtcMembership("@dave:example.org", "DDDD");
@@ -215,8 +222,8 @@ function summarizeLayout$(l$: Observable<Layout>): Observable<LayoutSummary> {
 }
 
 function withCallViewModel(
-  remoteParticipants$: Observable<RemoteParticipant[]>,
+  remoteParticipants$: Behavior<RemoteParticipant[]>,
-  rtcMembers$: Observable<Partial<CallMembership>[]>,
+  rtcMembers$: Behavior<Partial<CallMembership>[]>,
   connectionState$: Observable<ECConnectionState>,
   speaking: Map<Participant, Observable<boolean>>,
   mediaDevices: MediaDevices,
@@ -294,7 +301,7 @@ function withCallViewModel(
 }
 
 test("participants are retained during a focus switch", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     // Participants disappear on frame 2 and come back on frame 3
     const participantInputMarbles = "a-ba";
     // Start switching focus on frame 1 and reconnect on frame 3
@@ -303,12 +310,12 @@ test("participants are retained during a focus switch", () => {
     const expectedLayoutMarbles = "  a";
 
     withCallViewModel(
-      hot(participantInputMarbles, {
+      behavior(participantInputMarbles, {
         a: [aliceParticipant, bobParticipant],
         b: [],
       }),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
-      hot(connectionInputMarbles, {
+      behavior(connectionInputMarbles, {
         c: ConnectionState.Connected,
         s: ECAddonConnectionState.ECSwitchingFocus,
       }),
@@ -331,7 +338,7 @@ test("participants are retained during a focus switch", () => {
 });
 
 test("screen sharing activates spotlight layout", () => {
-  withTestScheduler(({ hot, schedule, expectObservable }) => {
+  withTestScheduler(({ behavior, schedule, expectObservable }) => {
     // Start with no screen shares, then have Alice and Bob share their screens,
     // then return to no screen shares, then have just Alice share for a bit
     const participantInputMarbles = "    abcda-ba";
@@ -344,13 +351,13 @@ test("screen sharing activates spotlight layout", () => {
     const expectedLayoutMarbles = "      abcdaefeg";
     const expectedShowSpeakingMarbles = "y----nyny";
     withCallViewModel(
-      hot(participantInputMarbles, {
+      behavior(participantInputMarbles, {
         a: [aliceParticipant, bobParticipant],
         b: [aliceSharingScreen, bobParticipant],
         c: [aliceSharingScreen, bobSharingScreen],
         d: [aliceParticipant, bobSharingScreen],
       }),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
       of(ConnectionState.Connected),
       new Map(),
       mockMediaDevices({}),
@@ -416,7 +423,7 @@ test("screen sharing activates spotlight layout", () => {
 });
 
 test("participants stay in the same order unless to appear/disappear", () => {
-  withTestScheduler(({ hot, schedule, expectObservable }) => {
+  withTestScheduler(({ behavior, schedule, expectObservable }) => {
     const visibilityInputMarbles = "a";
     // First Bob speaks, then Dave, then Alice
     const aSpeakingInputMarbles = " n- 1998ms - 1999ms y";
@@ -429,13 +436,22 @@ test("participants stay in the same order unless to appear/disappear", () => {
     const expectedLayoutMarbles = " a  1999ms b 1999ms a 57999ms c 1999ms a";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant, daveParticipant]),
+      constant([aliceParticipant, bobParticipant, daveParticipant]),
-      of([aliceRtcMember, bobRtcMember, daveRtcMember]),
+      constant([aliceRtcMember, bobRtcMember, daveRtcMember]),
       of(ConnectionState.Connected),
       new Map([
-        [aliceParticipant, hot(aSpeakingInputMarbles, { y: true, n: false })],
+        [
-        [bobParticipant, hot(bSpeakingInputMarbles, { y: true, n: false })],
+          aliceParticipant,
-        [daveParticipant, hot(dSpeakingInputMarbles, { y: true, n: false })],
+          behavior(aSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          bobParticipant,
+          behavior(bSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          daveParticipant,
+          behavior(dSpeakingInputMarbles, { y: true, n: false }),
+        ],
       ]),
       mockMediaDevices({}),
       (vm) => {
@@ -475,7 +491,7 @@ test("participants stay in the same order unless to appear/disappear", () => {
 });
 
 test("participants adjust order when space becomes constrained", () => {
-  withTestScheduler(({ hot, schedule, expectObservable }) => {
+  withTestScheduler(({ behavior, schedule, expectObservable }) => {
     // Start with all tiles on screen then shrink to 3
     const visibilityInputMarbles = "a-b";
     // Bob and Dave speak
@@ -487,12 +503,18 @@ test("participants adjust order when space becomes constrained", () => {
     const expectedLayoutMarbles = " a-b";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant, daveParticipant]),
+      constant([aliceParticipant, bobParticipant, daveParticipant]),
-      of([aliceRtcMember, bobRtcMember, daveRtcMember]),
+      constant([aliceRtcMember, bobRtcMember, daveRtcMember]),
       of(ConnectionState.Connected),
       new Map([
-        [bobParticipant, hot(bSpeakingInputMarbles, { y: true, n: false })],
+        [
-        [daveParticipant, hot(dSpeakingInputMarbles, { y: true, n: false })],
+          bobParticipant,
+          behavior(bSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          daveParticipant,
+          behavior(dSpeakingInputMarbles, { y: true, n: false }),
+        ],
       ]),
       mockMediaDevices({}),
       (vm) => {
@@ -526,7 +548,7 @@ test("participants adjust order when space becomes constrained", () => {
 });
 
 test("spotlight speakers swap places", () => {
-  withTestScheduler(({ hot, schedule, expectObservable }) => {
+  withTestScheduler(({ behavior, schedule, expectObservable }) => {
     // Go immediately into spotlight mode for the test
     const modeInputMarbles = "     s";
     // First Bob speaks, then Dave, then Alice
@@ -540,13 +562,22 @@ test("spotlight speakers swap places", () => {
     const expectedLayoutMarbles = "abcd";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant, daveParticipant]),
+      constant([aliceParticipant, bobParticipant, daveParticipant]),
-      of([aliceRtcMember, bobRtcMember, daveRtcMember]),
+      constant([aliceRtcMember, bobRtcMember, daveRtcMember]),
       of(ConnectionState.Connected),
       new Map([
-        [aliceParticipant, hot(aSpeakingInputMarbles, { y: true, n: false })],
+        [
-        [bobParticipant, hot(bSpeakingInputMarbles, { y: true, n: false })],
+          aliceParticipant,
-        [daveParticipant, hot(dSpeakingInputMarbles, { y: true, n: false })],
+          behavior(aSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          bobParticipant,
+          behavior(bSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          daveParticipant,
+          behavior(dSpeakingInputMarbles, { y: true, n: false }),
+        ],
       ]),
       mockMediaDevices({}),
       (vm) => {
@@ -590,8 +621,8 @@ test("layout enters picture-in-picture mode when requested", () => {
     const expectedLayoutMarbles = " aba";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant]),
+      constant([aliceParticipant, bobParticipant]),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
       of(ConnectionState.Connected),
       new Map(),
       mockMediaDevices({}),
@@ -632,8 +663,8 @@ test("spotlight remembers whether it's expanded", () => {
     const expectedLayoutMarbles = "abcbada";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant]),
+      constant([aliceParticipant, bobParticipant]),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
       of(ConnectionState.Connected),
       new Map(),
       mockMediaDevices({}),
@@ -681,7 +712,7 @@ test("spotlight remembers whether it's expanded", () => {
 });
 
 test("participants must have a MatrixRTCSession to be visible", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     // iterate through a number of combinations of participants and MatrixRTC memberships
     // Bob never has an MatrixRTC membership
     const scenarioInputMarbles = " abcdec";
@@ -689,14 +720,14 @@ test("participants must have a MatrixRTCSession to be visible", () => {
     const expectedLayoutMarbles = "a-bc-b";
 
     withCallViewModel(
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [bobParticipant],
         c: [aliceParticipant, bobParticipant],
         d: [aliceParticipant, daveParticipant, bobParticipant],
         e: [aliceParticipant, daveParticipant, bobSharingScreen],
       }),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [],
         c: [aliceRtcMember],
@@ -737,17 +768,17 @@ test("shows participants without MatrixRTCSession when enabled in settings", ()
   try {
     // enable the setting:
     showNonMemberTiles.setValue(true);
-    withTestScheduler(({ hot, expectObservable }) => {
+    withTestScheduler(({ behavior, expectObservable }) => {
       const scenarioInputMarbles = " abc";
       const expectedLayoutMarbles = "abc";
 
       withCallViewModel(
-        hot(scenarioInputMarbles, {
+        behavior(scenarioInputMarbles, {
           a: [],
           b: [aliceParticipant],
           c: [aliceParticipant, bobParticipant],
         }),
-        of([]), // No one joins the MatrixRTC session
+        constant([]), // No one joins the MatrixRTC session
         of(ConnectionState.Connected),
         new Map(),
         mockMediaDevices({}),
@@ -782,15 +813,15 @@ test("shows participants without MatrixRTCSession when enabled in settings", ()
 });
 
 it("should show at least one tile per MatrixRTCSession", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     // iterate through some combinations of MatrixRTC memberships
     const scenarioInputMarbles = " abcd";
     // There should always be one tile for each MatrixRTCSession
     const expectedLayoutMarbles = "abcd";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [aliceRtcMember],
         c: [aliceRtcMember, daveRtcMember],
@@ -832,13 +863,13 @@ it("should show at least one tile per MatrixRTCSession", () => {
 });
 
 test("should disambiguate users with the same displayname", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     const scenarioInputMarbles = "abcde";
     const expectedLayoutMarbles = "abcde";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [aliceRtcMember],
         c: [aliceRtcMember, aliceDoppelgangerRtcMember],
@@ -849,50 +880,46 @@ test("should disambiguate users with the same displayname", () => {
       new Map(),
       mockMediaDevices({}),
       (vm) => {
-        // Skip the null state.
+        expectObservable(vm.memberDisplaynames$).toBe(expectedLayoutMarbles, {
-        expectObservable(vm.memberDisplaynames$.pipe(skip(1))).toBe(
+          // Carol has no displayname - So userId is used.
-          expectedLayoutMarbles,
+          a: new Map([[carolId, carol.userId]]),
-          {
+          b: new Map([
-            // Carol has no displayname - So userId is used.
+            [carolId, carol.userId],
-            a: new Map([[carolId, carol.userId]]),
+            [aliceId, alice.rawDisplayName],
-            b: new Map([
+          ]),
-              [carolId, carol.userId],
+          // The second alice joins.
-              [aliceId, alice.rawDisplayName],
+          c: new Map([
-            ]),
+            [carolId, carol.userId],
-            // The second alice joins.
+            [aliceId, "Alice (@alice:example.org)"],
-            c: new Map([
+            [aliceDoppelgangerId, "Alice (@alice2:example.org)"],
-              [carolId, carol.userId],
+          ]),
-              [aliceId, "Alice (@alice:example.org)"],
+          // Bob also joins
-              [aliceDoppelgangerId, "Alice (@alice2:example.org)"],
+          d: new Map([
-            ]),
+            [carolId, carol.userId],
-            // Bob also joins
+            [aliceId, "Alice (@alice:example.org)"],
-            d: new Map([
+            [aliceDoppelgangerId, "Alice (@alice2:example.org)"],
-              [carolId, carol.userId],
+            [bobId, bob.rawDisplayName],
-              [aliceId, "Alice (@alice:example.org)"],
+          ]),
-              [aliceDoppelgangerId, "Alice (@alice2:example.org)"],
+          // Alice leaves, and the displayname should reset.
-              [bobId, bob.rawDisplayName],
+          e: new Map([
-            ]),
+            [carolId, carol.userId],
-            // Alice leaves, and the displayname should reset.
+            [aliceDoppelgangerId, "Alice"],
-            e: new Map([
+            [bobId, bob.rawDisplayName],
-              [carolId, carol.userId],
+          ]),
-              [aliceDoppelgangerId, "Alice"],
+        });
-              [bobId, bob.rawDisplayName],
-            ]),
-          },
-        );
       },
     );
   });
 });
 
 test("should disambiguate users with invisible characters", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     const scenarioInputMarbles = "ab";
     const expectedLayoutMarbles = "ab";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [bobRtcMember, bobZeroWidthSpaceRtcMember],
       }),
@@ -900,36 +927,32 @@ test("should disambiguate users with invisible characters", () => {
       new Map(),
       mockMediaDevices({}),
       (vm) => {
-        // Skip the null state.
+        expectObservable(vm.memberDisplaynames$).toBe(expectedLayoutMarbles, {
-        expectObservable(vm.memberDisplaynames$.pipe(skip(1))).toBe(
+          // Carol has no displayname - So userId is used.
-          expectedLayoutMarbles,
+          a: new Map([[carolId, carol.userId]]),
-          {
+          // Both Bobs join, and should handle zero width hacks.
-            // Carol has no displayname - So userId is used.
+          b: new Map([
-            a: new Map([[carolId, carol.userId]]),
+            [carolId, carol.userId],
-            // Both Bobs join, and should handle zero width hacks.
+            [bobId, `Bob (${bob.userId})`],
-            b: new Map([
+            [
-              [carolId, carol.userId],
+              bobZeroWidthSpaceId,
-              [bobId, `Bob (${bob.userId})`],
+              `${bobZeroWidthSpace.rawDisplayName} (${bobZeroWidthSpace.userId})`,
-              [
+            ],
-                bobZeroWidthSpaceId,
+          ]),
-                `${bobZeroWidthSpace.rawDisplayName} (${bobZeroWidthSpace.userId})`,
+        });
-              ],
-            ]),
-          },
-        );
       },
     );
   });
 });
 
 test("should strip RTL characters from displayname", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     const scenarioInputMarbles = "ab";
     const expectedLayoutMarbles = "ab";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [daveRtcMember, daveRTLRtcMember],
       }),
@@ -937,22 +960,18 @@ test("should strip RTL characters from displayname", () => {
       new Map(),
       mockMediaDevices({}),
       (vm) => {
-        // Skip the null state.
+        expectObservable(vm.memberDisplaynames$).toBe(expectedLayoutMarbles, {
-        expectObservable(vm.memberDisplaynames$.pipe(skip(1))).toBe(
+          // Carol has no displayname - So userId is used.
-          expectedLayoutMarbles,
+          a: new Map([[carolId, carol.userId]]),
-          {
+          // Both Dave's join. Since after stripping
-            // Carol has no displayname - So userId is used.
+          b: new Map([
-            a: new Map([[carolId, carol.userId]]),
+            [carolId, carol.userId],
-            // Both Dave's join. Since after stripping
+            // Not disambiguated
-            b: new Map([
+            [daveId, "Dave"],
-              [carolId, carol.userId],
+            // This one is, since it's using RTL.
-              // Not disambiguated
+            [daveRTLId, `evaD (${daveRTL.userId})`],
-              [daveId, "Dave"],
+          ]),
-              // This one is, since it's using RTL.
+        });
-              [daveRTLId, `evaD (${daveRTL.userId})`],
-            ]),
-          },
-        );
       },
     );
   });
@@ -964,8 +983,8 @@ it("should rank raised hands above video feeds and below speakers and presenters
     const expectedLayoutMarbles = "ab";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant]),
+      constant([aliceParticipant, bobParticipant]),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
       of(ConnectionState.Connected),
       new Map(),
       mockMediaDevices({}),
@@ -1039,8 +1058,8 @@ test("audio output changes when toggling earpiece mode", () => {
     const expectedTargetStateMarbles = " sese";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      of([]),
+      constant([]),
       of(ConnectionState.Connected),
       new Map(),
       devices,

src/utils/test.ts

@@ -4,7 +4,7 @@ Copyright 2023, 2024 New Vector Ltd.
 SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
-import { map, type Observable, of, type SchedulerLike } from "rxjs";
+import { map, type Observable, of, type SchedulerLike, startWith } from "rxjs";
 import { type RunHelpers, TestScheduler } from "rxjs/testing";
 import { expect, vi, vitest } from "vitest";
 import {
@@ -47,7 +47,8 @@ import {
 } from "../config/ConfigOptions";
 import { Config } from "../config/Config";
 import { type MediaDevices } from "../state/MediaDevices";
-import { constant } from "../state/Behavior";
+import { type Behavior, constant } from "../state/Behavior";
+import { ObservableScope } from "../state/ObservableScope";
 
 export function withFakeTimers(continuation: () => void): void {
   vi.useFakeTimers();
@@ -68,6 +69,11 @@ export interface OurRunHelpers extends RunHelpers {
    * diagram.
    */
   schedule: (marbles: string, actions: Record<string, () => void>) => void;
+  behavior<T = string>(
+    marbles: string,
+    values?: { [marble: string]: T },
+    error?: unknown,
+  ): Behavior<T>;
 }
 
 interface TestRunnerGlobal {
@@ -83,6 +89,7 @@ export function withTestScheduler(
   const scheduler = new TestScheduler((actual, expected) => {
     expect(actual).deep.equals(expected);
   });
+  const scope = new ObservableScope();
   // we set the test scheduler as a global so that you can watch it in a debugger
   // and get the frame number. e.g. `rxjsTestScheduler?.now()`
   (global as unknown as TestRunnerGlobal).rxjsTestScheduler = scheduler;
@@ -99,8 +106,32 @@ export function withTestScheduler(
         // Run the actions and verify that none of them error
         helpers.expectObservable(actionsObservable$).toBe(marbles, results);
       },
+      behavior<T>(
+        marbles: string,
+        values?: { [marble: string]: T },
+        error?: unknown,
+      ) {
+        // Generate a hot Observable with helpers.hot and use it as a Behavior.
+        // To do this, we need to ensure that the initial value emits
+        // synchronously upon subscription. The issue is that helpers.hot emits
+        // frame 0 of the marble diagram *asynchronously*, only once we return
+        // from the continuation, so we need to splice out the initial marble
+        // and turn it into a proper initial value.
+        const initialMarbleIndex = marbles.search(/[^ ]/);
+        if (initialMarbleIndex === -1)
+          throw new Error("Behavior must have an initial value");
+        const initialMarble = marbles[initialMarbleIndex];
+        const initialValue =
+          values === undefined ? (initialMarble as T) : values[initialMarble];
+        // The remainder of the marble diagram should start on frame 1
+        return helpers
+          .hot(`-${marbles.slice(initialMarbleIndex + 1)}`, values, error)
+          .pipe(startWith(initialValue))
+          .behavior(scope);
+      },
     }),
   );
+  scope.end();
 }
 
 interface EmitterMock<T> {