Add tests for absolute first emit

tests/test_observable/test_timer.py

@@ -128,6 +128,60 @@ def create():
         results = scheduler.start(create)
         assert results.messages == [on_next(500, 0), on_next(800, 1)]
 
+    def test_periodic_timer_repeat(self):
+        scheduler = TestScheduler()
+        t = reactivex.timer(duetime=130, period=200, scheduler=scheduler)
+
+        def create():
+            return t.pipe(operators.take(3), operators.repeat())
+
+        results = scheduler.start(create)
+        assert results.messages == [
+            on_next(330, 0),
+            on_next(530, 1),
+            on_next(730, 2),
+            on_next(860, 0),
+        ]
+
+    def test_periodic_timer_repeat_with_absolute_datetime(self):
+        scheduler = TestScheduler()
+        t = reactivex.timer(
+            duetime=scheduler.to_datetime(360), period=200, scheduler=scheduler
+        )  # here we have an absolute first value, so on second subscription, the timer should emit immediately
+
+        def create():
+            return t.pipe(operators.take(3), operators.repeat())
+
+        results = scheduler.start(create)
+        assert results.messages == [
+            on_next(360, 0),
+            on_next(560, 1),
+            on_next(760, 2),
+            on_next(
+                760, 0
+            ),  # our duetime is absolute and in the past so new sub emits immediately
+            on_next(960, 1),
+        ]
+
+    def test_periodic_timer_repeat_with_relative_timespan(self):
+        scheduler = TestScheduler()
+        t = reactivex.timer(
+            duetime=scheduler.to_timedelta(130),
+            period=scheduler.to_timedelta(250),
+            scheduler=scheduler,
+        )
+
+        def create():
+            return t.pipe(operators.take(3), operators.repeat())
+
+        results = scheduler.start(create)
+        assert results.messages == [
+            on_next(330, 0),
+            on_next(580, 1),
+            on_next(830, 2),
+            on_next(960, 0),
+        ]
+
     def test_periodic_timer_second_subscription(self):
         scheduler = TestScheduler()
         t = reactivex.timer(duetime=200, period=300, scheduler=scheduler)
@@ -149,7 +203,7 @@ def create():
             on_next(800, (1, "second")),
         ]
 
-    def test_on_off_timer_repeat(self):
+    def test_one_off_timer_repeat(self):
         scheduler = TestScheduler()
         t = reactivex.timer(duetime=230, scheduler=scheduler)
 
@@ -162,18 +216,3 @@ def create():
             on_next(660, 0),
             on_next(890, 0),
         ]
-
-    def test_periodic_timer_repeat(self):
-        scheduler = TestScheduler()
-        t = reactivex.timer(duetime=130, period=200, scheduler=scheduler)
-
-        def create():
-            return t.pipe(operators.take(3), operators.repeat())
-
-        results = scheduler.start(create)
-        assert results.messages == [
-            on_next(330, 0),
-            on_next(530, 1),
-            on_next(730, 2),
-            on_next(860, 0),
-        ]