test: fix pyright strict-mode errors in tests/test_observable (batch 2) (#782)

* test: fix pyright strict-mode errors in test_observable (batch 2)

Fix pyright strict-mode type errors in 30 files under tests/test_observable/,
continuing the incremental cleanup of the pyright exclusion list.

Error patterns fixed:
- _raise("string") → _raise(Exception("string"))
- def action(): → def action(_): for on_next callbacks (OnNext[T] requires a param)
- Notification union-attr access annotated with # type: ignore[union-attr]
- subscribe(scheduler="not_set") → subscribe(scheduler=None) for proper typing
- mapper/generate iterate lambdas annotated with # type: ignore[arg-type]
- do_action callback removing spurious return value
- factory(ex: Exception) → factory(ex: Exception | None) for on_error_resume_next
- return lambda: None → return None in subscribe functions
- Hoisted variables to avoid unbound-var errors inside assertRaises blocks
- Various other minor fixes (arg-type, call-arg, misc type: ignore)

260 tests pass. Part of ongoing effort to clean up pyright exclusions.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* refactor(tests): replace type:ignore comments with proper type annotations

Builds on PR #782 to replace the `# type: ignore` suppressions with
real type annotations and structural fixes:

- Use `isinstance(value, OnError|OnNext)` to narrow `Recorded.value` instead
  of `# type: ignore[union-attr]` (test_average, test_elementat, test_reduce,
  test_toiterable).
- Properly annotate test-local `subscribe` functions with
  `ObserverBase[T] / SchedulerBase | None / DisposableBase` and return
  `Disposable()` so `reactivex.create(...)` accepts them without
  `# type: ignore[misc]/[arg-type]` (test_concat, test_observeon, test_map,
  test_skipuntil, test_while_do).
- Type `xs` as `list[ColdObservable[int] | None]` in test_defer instead of
  bare `[None]`, and add `assert xs[0] is not None` narrowing.
- Annotate `_raise` as `NoReturn` in test_map, test_retry, test_generate so
  the surrounding code type-checks without trailing dummy returns.
- Use a typed module-level seed in test_generate to widen `Literal[0]` to
  `int` so `lambda x: x + 1` matches `Mapper[int, int]`.
- `cast(Observable[int], s)` in test_tofuture instead of `# type: ignore`.

Also widens two clearly-incorrect public type signatures (the implementation
already supports the wider input):

- `delay()` and `delay_()` now accept `AbsoluteOrRelativeTime` instead of
  only `RelativeTime`. The implementation already branches on
  `isinstance(duetime, datetime)`, so this just makes the signature match
  reality and removes 5 `# type: ignore[arg-type]` in test_delay.
- `from_callback()` now returns `Callable[..., Observable[Any]]` instead of
  `Callable[[], Observable[Any]]`. The inner `function(*args)` already
  accepts arbitrary args, so this matches reality and removes 2
  `# type: ignore[call-arg]` in test_fromcallback.

Only `test_of.py:51` keeps a `# type: ignore[call-arg]` — it's in a
test that never runs (typo `teest_` in the name) and intentionally calls
`reactivex.of()` with a kwarg that doesn't exist; fixing it is out of
scope.

Pyright strict: 0 errors. All 1494 tests pass.

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Dag Brattli <dag@brattli.net>

Author: 3 people

reactivex/__init__.py

@@ -469,7 +469,7 @@ def from_callable(
 def from_callback(
     func: Callable[..., Callable[..., None]],
     mapper: typing.Mapper[Any, Any] | None = None,
-) -> Callable[[], Observable[Any]]:
+) -> Callable[..., Observable[Any]]:
     """Converts a callback function to an observable sequence.
 
     Args:

reactivex/observable/fromcallback.py

@@ -8,7 +8,7 @@
 def from_callback_(
     func: Callable[..., Callable[..., None]],
     mapper: typing.Mapper[Any, Any] | None = None,
-) -> Callable[[], Observable[Any]]:
+) -> Callable[..., Observable[Any]]:
     """Converts a callback function to an observable sequence.
 
     Args:

reactivex/operators/__init__.py

@@ -707,7 +707,8 @@ def dematerialize() -> Callable[[Observable[Notification[_T]]], Observable[_T]]:
 
 
 def delay(
-    duetime: typing.RelativeTime, scheduler: abc.SchedulerBase | None = None
+    duetime: typing.AbsoluteOrRelativeTime,
+    scheduler: abc.SchedulerBase | None = None,
 ) -> Callable[[Observable[_T]], Observable[_T]]:
     """The delay operator.
 

reactivex/operators/_delay.py

@@ -20,7 +20,7 @@
 
 def observable_delay_timespan(
     source: Observable[_T],
-    duetime: typing.RelativeTime,
+    duetime: typing.AbsoluteOrRelativeTime,
     scheduler: abc.SchedulerBase | None = None,
 ) -> Observable[_T]:
     def subscribe(
@@ -119,7 +119,7 @@ def action(scheduler: abc.SchedulerBase, state: Any = None):
 @curry_flip
 def delay_(
     source: Observable[_T],
-    duetime: typing.RelativeTime,
+    duetime: typing.AbsoluteOrRelativeTime,
     scheduler: abc.SchedulerBase | None = None,
 ) -> Observable[_T]:
     """Time shifts the observable sequence.

tests/test_observable/test_amb.py

@@ -68,7 +68,7 @@ def test_amb_regular_should_dispose_loser(self):
         source_not_disposed = [False]
         o1 = scheduler.create_hot_observable(msgs1)
 
-        def action():
+        def action(_):
             source_not_disposed[0] = True
 
         o2 = scheduler.create_hot_observable(msgs2).pipe(
@@ -91,7 +91,7 @@ def test_amb_winner_throws(self):
         source_not_disposed = [False]
         o1 = scheduler.create_hot_observable(msgs1)
 
-        def action():
+        def action(_):
             source_not_disposed[0] = True
 
         o2 = scheduler.create_hot_observable(msgs2).pipe(
@@ -112,7 +112,7 @@ def test_amb_loser_throws(self):
         msgs2 = [on_next(150, 1), on_next(210, 3), on_completed(250)]
         source_not_disposed = [False]
 
-        def action():
+        def action(_):
             source_not_disposed[0] = True
 
         o1 = scheduler.create_hot_observable(msgs1).pipe(
@@ -136,7 +136,7 @@ def test_amb_throws_before_election(self):
         source_not_disposed = [False]
         o1 = scheduler.create_hot_observable(msgs1)
 
-        def action():
+        def action(_):
             source_not_disposed[0] = True
 
         o2 = scheduler.create_hot_observable(msgs2).pipe(

tests/test_observable/test_average.py

@@ -1,6 +1,7 @@
 import unittest
 
 from reactivex import operators as _
+from reactivex.notification import OnError
 from reactivex.testing import ReactiveTest, TestScheduler
 
 on_next = ReactiveTest.on_next
@@ -20,7 +21,7 @@ def test_average_int32_empty(self):
         res = scheduler.start(create=lambda: xs.pipe(_.average())).messages
 
         assert len(res) == 1
-        assert res[0].value.kind == "E" and res[0].value.exception is not None
+        assert isinstance(res[0].value, OnError) and res[0].value.exception is not None
         assert res[0].time == 250
 
     def test_average_int32_return(self):

tests/test_observable/test_combinelatest.py

@@ -472,7 +472,7 @@ def create():
         ]
 
     def test_combine_latest_mapper_throws(self):
-        ex = "ex"
+        ex = Exception("ex")
         scheduler = TestScheduler()
         msgs1 = [on_next(150, 1), on_next(215, 2), on_completed(230)]
         msgs2 = [on_next(150, 1), on_next(220, 3), on_completed(240)]

tests/test_observable/test_concat.py

@@ -1,7 +1,9 @@
 import unittest
 
 import reactivex
+from reactivex import abc
 from reactivex import operators as ops
+from reactivex.disposable import Disposable
 from reactivex.testing import ReactiveTest, TestScheduler
 
 on_next = ReactiveTest.on_next
@@ -225,15 +227,23 @@ def create():
 
     def test_concat_forward_scheduler(self):
         scheduler = TestScheduler()
-        subscribe_schedulers = {"e1": "unknown", "e2": "unknown"}
+        subscribe_schedulers: dict[str, object] = {"e1": "unknown", "e2": "unknown"}
 
-        def subscribe_e1(observer, scheduler="not_set"):
+        def subscribe_e1(
+            observer: abc.ObserverBase[int],
+            scheduler: abc.SchedulerBase | None = None,
+        ) -> abc.DisposableBase:
             subscribe_schedulers["e1"] = scheduler
             observer.on_completed()
+            return Disposable()
 
-        def subscribe_e2(observer, scheduler="not_set"):
+        def subscribe_e2(
+            observer: abc.ObserverBase[int],
+            scheduler: abc.SchedulerBase | None = None,
+        ) -> abc.DisposableBase:
             subscribe_schedulers["e2"] = scheduler
             observer.on_completed()
+            return Disposable()
 
         e1 = reactivex.create(subscribe_e1)
         e2 = reactivex.create(subscribe_e2)
@@ -245,15 +255,23 @@ def subscribe_e2(observer, scheduler="not_set"):
         assert subscribe_schedulers["e2"] is scheduler
 
     def test_concat_forward_none_scheduler(self):
-        subscribe_schedulers = {"e1": "unknown", "e2": "unknown"}
+        subscribe_schedulers: dict[str, object] = {"e1": "unknown", "e2": "unknown"}
 
-        def subscribe_e1(observer, scheduler="not_set"):
+        def subscribe_e1(
+            observer: abc.ObserverBase[int],
+            scheduler: abc.SchedulerBase | None = None,
+        ) -> abc.DisposableBase:
             subscribe_schedulers["e1"] = scheduler
             observer.on_completed()
+            return Disposable()
 
-        def subscribe_e2(observer, scheduler="not_set"):
+        def subscribe_e2(
+            observer: abc.ObserverBase[int],
+            scheduler: abc.SchedulerBase | None = None,
+        ) -> abc.DisposableBase:
             subscribe_schedulers["e2"] = scheduler
             observer.on_completed()
+            return Disposable()
 
         e1 = reactivex.create(subscribe_e1)
         e2 = reactivex.create(subscribe_e2)

tests/test_observable/test_debounce.py

@@ -359,7 +359,7 @@ def mapper(x):
         assert xs.subscriptions == [subscribe(200, 460)]
 
     def test_debounce_duration_mapper_throws(self):
-        ex = "ex"
+        ex = Exception("ex")
         scheduler = TestScheduler()
         xs = scheduler.create_hot_observable(
             on_next(150, 1),
@@ -376,7 +376,7 @@ def mapper(x):
                         on_next(x * 10, "Ignore"), on_next(x * 10 + 5, "Aargh!")
                     )
                 else:
-                    _raise(ex)
+                    raise ex
 
             return xs.pipe(_.throttle_with_mapper(mapper))
 

tests/test_observable/test_defer.py

@@ -2,6 +2,7 @@
 
 import reactivex
 from reactivex.testing import ReactiveTest, TestScheduler
+from reactivex.testing.coldobservable import ColdObservable
 
 on_next = ReactiveTest.on_next
 on_completed = ReactiveTest.on_completed
@@ -23,7 +24,7 @@ def _raise(ex):
 
 class TestDefer(unittest.TestCase):
     def test_defer_complete(self):
-        xs = [None]
+        xs: list[ColdObservable[int] | None] = [None]
         invoked = [0]
         scheduler = TestScheduler()
 
@@ -40,12 +41,13 @@ def defer(scheduler):
         results = scheduler.start(create)
         assert results.messages == [on_next(300, 200), on_completed(400)]
         assert 1 == invoked[0]
+        assert xs[0] is not None
         assert xs[0].subscriptions == [subscribe(200, 400)]
 
     def test_defer_error(self):
         scheduler = TestScheduler()
         invoked = [0]
-        xs = [None]
+        xs: list[ColdObservable[int] | None] = [None]
         ex = "ex"
 
         def create():
@@ -62,12 +64,13 @@ def defer(scheduler):
 
         assert results.messages == [on_next(300, 200), on_error(400, ex)]
         assert 1 == invoked[0]
+        assert xs[0] is not None
         assert xs[0].subscriptions == [subscribe(200, 400)]
 
     def test_defer_dispose(self):
         scheduler = TestScheduler()
         invoked = [0]
-        xs = [None]
+        xs: list[ColdObservable[int] | None] = [None]
 
         def create():
             def defer(scheduler):
@@ -84,6 +87,7 @@ def defer(scheduler):
         results = scheduler.start(create)
         assert results.messages == [on_next(300, 200), on_next(400, 1)]
         assert 1 == invoked[0]
+        assert xs[0] is not None
         assert xs[0].subscriptions == [subscribe(200, 1000)]
 
     def test_defer_on_error(self):