<s> import unittest <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 TestGenerateWithRelativeTime ( unittest . TestCase ) : <EOL> def test_generate_timespan_finite ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : x <= <NUM_LIT:3> , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : x , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> 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> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_generate_timespan_throw_condition ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : _raise ( ex ) , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : x , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_generate_timespan_throw_result_selector ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : True , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : _raise ( ex ) , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_generate_timespan_throw_iterate ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : True , <EOL> lambda x : _raise ( ex ) , <EOL> lambda x : x , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> def test_generate_timespan_throw_timeselector ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : True , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : x , <EOL> lambda x : _raise ( ex ) , <EOL> scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_generate_timespan_dispose ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : True , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : x , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> scheduler = scheduler ) <EOL> results = scheduler . start ( create , disposed = <NUM_LIT> ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) ) <EOL> def test_generate_datetime_offset_finite ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : True , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : x , <EOL> lambda x : scheduler . now ( ) + x + <NUM_LIT:1> , <EOL> 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> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_generate_datetime_offset_throw_condition ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : _raise ( ex ) , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : x , <EOL> lambda x : scheduler . now ( ) + x + <NUM_LIT:1> , <EOL> scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_generate_datetime_offset_throw_result_selector ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : True , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : _raise ( ex ) , <EOL> lambda x : scheduler . now ( ) + x + <NUM_LIT:1> , <EOL> scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_generate_datetime_offset_throw_iterate ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : True , <EOL> lambda x : _raise ( ex ) , <EOL> lambda x : x , <EOL> lambda x : scheduler . now ( ) + x + <NUM_LIT:1> , <EOL> scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> def test_generate_datetime_offset_throw_time_selector ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : True , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : x , <EOL> lambda x : _raise ( ex ) , <EOL> scheduler = scheduler ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_generate_datetime_offset_dispose ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> return Observable . generate_with_relative_time ( <NUM_LIT:0> , <EOL> lambda x : True , <EOL> lambda x : x + <NUM_LIT:1> , <EOL> lambda x : x , <EOL> lambda x : scheduler . now ( ) + x + <NUM_LIT:1> , <EOL> scheduler = scheduler ) <EOL> results = scheduler . start ( create , disposed = <NUM_LIT> ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) ) </s>
