<s> import unittest <EOL> from datetime import datetime , timedelta <EOL> from rx import Observable <EOL> from rx . testing import TestScheduler , ReactiveTest , is_prime , MockDisposable <EOL> from rx . disposables import Disposable , SerialDisposable , BooleanDisposable <EOL> on_next = ReactiveTest . on_next <EOL> on_completed = ReactiveTest . on_completed <EOL> on_error = ReactiveTest . on_error <EOL> subscribe = ReactiveTest . subscribe <EOL> subscribed = ReactiveTest . subscribed <EOL> disposed = ReactiveTest . disposed <EOL> created = ReactiveTest . created <EOL> class RxException ( Exception ) : <EOL> pass <EOL> def _raise ( ex ) : <EOL> raise RxException ( ex ) <EOL> class TestTimer ( unittest . TestCase ) : <EOL> def test_oneshot_timer_timespan_basic ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . timer ( duetime = <NUM_LIT> , scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_oneshot_timer_timespan_zero ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . timer ( <NUM_LIT:0> , scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_oneshot_timer_timespan_negative ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . timer ( - <NUM_LIT:1> , scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_oneshot_timer_timespan_disposed ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . timer ( <NUM_LIT:1000> , scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( ) <EOL> def test_oneshot_timer_timespan_observer_throws ( self ) : <EOL> scheduler1 = TestScheduler ( ) <EOL> xs = Observable . timer ( <NUM_LIT:1> , scheduler = scheduler1 ) <EOL> xs . subscribe ( lambda x : _raise ( "<STR_LIT>" ) ) <EOL> self . assertRaises ( RxException , scheduler1 . start ) <EOL> scheduler2 = TestScheduler ( ) <EOL> ys = Observable . timer ( <NUM_LIT:1> , period = None , scheduler = scheduler2 ) <EOL> ys . subscribe ( on_completed = lambda : _raise ( "<STR_LIT>" ) ) <EOL> self . assertRaises ( RxException , scheduler2 . start ) <EOL> def test_periodic_timer_basic ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . timer ( duetime = <NUM_LIT> , period = <NUM_LIT> , scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) ) <EOL> def test_periodic_timer_equal_time_and_period ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . timer ( duetime = <NUM_LIT> , period = <NUM_LIT> , scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) ) </s>
