test(operators): add regression tests for scheduler propagation

Cover pairwise, to_marbles, and delay_with_mapper (subscription-delay
path): each test installs a source whose subscribe captures the
scheduler it received, pipes through the operator, subscribes with a
known scheduler, and asserts the captured scheduler matches. Verified
to fail on the pre-fix code.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: dbrattli

tests/test_observable/test_delaywithmapper.py

@@ -1,6 +1,9 @@
 import unittest
 
+from reactivex import Observable, abc
 from reactivex import operators as ops
+from reactivex.disposable import Disposable
+from reactivex.scheduler import ImmediateScheduler
 from reactivex.testing import ReactiveTest, TestScheduler
 
 on_next = ReactiveTest.on_next
@@ -199,3 +202,37 @@ def create():
         assert results.messages == [on_next(210 + 50, 2)]
         assert xs.subscriptions == [subscribe(200, 300)]
         assert ys.subscriptions == [subscribe(210, 260)]
+
+    def test_delay_with_mapper_forwards_scheduler_to_subscription_delay(
+        self,
+    ) -> None:
+        captured: dict[str, abc.SchedulerBase | None] = {}
+        expected: abc.SchedulerBase = ImmediateScheduler()
+
+        def sub_delay_subscribe(
+            observer: abc.ObserverBase[int],
+            scheduler: abc.SchedulerBase | None = None,
+        ) -> abc.DisposableBase:
+            captured["scheduler"] = scheduler
+            observer.on_completed()
+            return Disposable()
+
+        sub_delay: Observable[int] = Observable(sub_delay_subscribe)
+
+        def source_subscribe(
+            observer: abc.ObserverBase[int],
+            scheduler: abc.SchedulerBase | None = None,
+        ) -> abc.DisposableBase:
+            observer.on_completed()
+            return Disposable()
+
+        source: Observable[int] = Observable(source_subscribe)
+
+        def mapper(_: int) -> Observable[int]:
+            return source
+
+        source.pipe(ops.delay_with_mapper(sub_delay, mapper)).subscribe(
+            scheduler=expected
+        )
+
+        assert captured["scheduler"] is expected

tests/test_observable/test_marbles.py

@@ -2,8 +2,11 @@
 import unittest
 
 import reactivex
-from reactivex import notification
+from reactivex import Observable, abc, notification
+from reactivex import operators as ops
+from reactivex.disposable import Disposable
 from reactivex.observable.marbles import parse
+from reactivex.scheduler import ImmediateScheduler
 from reactivex.testing import TestScheduler
 from reactivex.testing.reactivetest import ReactiveTest
 
@@ -578,3 +581,22 @@ def test_hot_marble_with_timedelta(self):
             ReactiveTest.on_completed(300.7),
         ]
         assert results == expected
+
+
+class TestToMarbles(unittest.TestCase):
+    def test_to_marbles_forwards_scheduler_to_source(self) -> None:
+        captured: dict[str, abc.SchedulerBase | None] = {}
+        expected: abc.SchedulerBase = ImmediateScheduler()
+
+        def source_subscribe(
+            observer: abc.ObserverBase[int],
+            scheduler: abc.SchedulerBase | None = None,
+        ) -> abc.DisposableBase:
+            captured["scheduler"] = scheduler
+            observer.on_completed()
+            return Disposable()
+
+        source: Observable[int] = Observable(source_subscribe)
+        source.pipe(ops.to_marbles()).subscribe(scheduler=expected)
+
+        assert captured["scheduler"] is expected

tests/test_observable/test_pairwise.py

@@ -1,6 +1,9 @@
 import unittest
 
+from reactivex import Observable, abc
 from reactivex import operators as ops
+from reactivex.disposable import Disposable
+from reactivex.scheduler import ImmediateScheduler
 from reactivex.testing import ReactiveTest, TestScheduler
 
 on_next = ReactiveTest.on_next
@@ -134,3 +137,20 @@ def create():
         assert results.messages == [on_next(240, (4, 3))]
 
         assert xs.subscriptions == [subscribe(200, 280)]
+
+    def test_pairwise_forwards_scheduler_to_source(self) -> None:
+        captured: dict[str, abc.SchedulerBase | None] = {}
+        expected: abc.SchedulerBase = ImmediateScheduler()
+
+        def source_subscribe(
+            observer: abc.ObserverBase[int],
+            scheduler: abc.SchedulerBase | None = None,
+        ) -> abc.DisposableBase:
+            captured["scheduler"] = scheduler
+            observer.on_completed()
+            return Disposable()
+
+        source: Observable[int] = Observable(source_subscribe)
+        source.pipe(ops.pairwise()).subscribe(scheduler=expected)
+
+        assert captured["scheduler"] is expected