[Repo Assist] fix(operators): forward scheduler in pairwise, to_marbles, delay_with_mapper

changes.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+- Operators: Fixed scheduler forwarding in `pairwise`, `to_marbles`, and
+  `delay_with_mapper` (subscription-delay path). These operators now pass the
+  `scheduler` argument through to `source.subscribe(...)` and, in the case of
+  `delay_with_mapper`, to the subscription-delay observable, consistent with
+  all other pipeable operators. Closes #480 (partial — the operators listed
+  in the issue that were not yet fixed).
+
 - Testing: Fixed ruff lint issues in `tests/test_scheduler/` (import ordering,
   f-string upgrades, `object` base class removal) and removed it from the
   ruff exclude list. `tests/test_subject/` also removed from the ruff exclude

reactivex/operators/_delaywithmapper.py

@@ -95,7 +95,7 @@ def on_completed() -> None:
                 start()
             else:
                 subscription.disposable = sub_delay.subscribe(
-                    lambda _: start(), observer.on_error, start
+                    lambda _: start(), observer.on_error, start, scheduler=scheduler
                 )
 
             return CompositeDisposable(subscription, delays)

reactivex/operators/_pairwise.py

@@ -49,7 +49,9 @@ def on_next(x: _T) -> None:
             if pair:
                 observer.on_next(pair)
 
-        return source.subscribe(on_next, observer.on_error, observer.on_completed)
+        return source.subscribe(
+            on_next, observer.on_error, observer.on_completed, scheduler=scheduler
+        )
 
     return Observable(subscribe)
 

reactivex/operators/_tomarbles.py

@@ -59,7 +59,9 @@ def on_completed():
                 observer.on_next("".join(n for n in result))
                 observer.on_completed()
 
-            return source.subscribe(on_next, on_error, on_completed)
+            return source.subscribe(
+                on_next, on_error, on_completed, scheduler=scheduler
+            )
 
         return Observable(subscribe)
 

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