<s> import logging <EOL> import unittest <EOL> from datetime import datetime <EOL> from rx . testing import TestScheduler , ReactiveTest , is_prime , MockDisposable <EOL> from rx . disposables import Disposable <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 TestDelayWithSelector ( unittest . TestCase ) : <EOL> def test_delay_duration_simple1 ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:10> ) , on_next ( <NUM_LIT> , <NUM_LIT:30> ) , on_next ( <NUM_LIT> , <NUM_LIT:50> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT:20> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> def selector ( x ) : <EOL> return scheduler . create_cold_observable ( on_next ( x , '<STR_LIT:!>' ) ) <EOL> return xs . delay_with_selector ( selector ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> + <NUM_LIT:10> , <NUM_LIT:10> ) , on_next ( <NUM_LIT> + <NUM_LIT:30> , <NUM_LIT:30> ) , on_next ( <NUM_LIT> + <NUM_LIT:20> , <NUM_LIT:20> ) , on_next ( <NUM_LIT> + <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> + <NUM_LIT:50> , <NUM_LIT:50> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_delay_duration_simple2 ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_completed ( <NUM_LIT> ) ) <EOL> ys = scheduler . create_cold_observable ( on_next ( <NUM_LIT:10> , '<STR_LIT:!>' ) ) <EOL> def create ( ) : <EOL> return xs . delay_with_selector ( lambda _ : ys ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> + <NUM_LIT:10> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> + <NUM_LIT:10> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> + <NUM_LIT:10> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> + <NUM_LIT:10> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> + <NUM_LIT:10> , <NUM_LIT:6> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> ys . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_delay_duration_simple3 ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_completed ( <NUM_LIT> ) ) <EOL> ys = scheduler . create_cold_observable ( on_next ( <NUM_LIT:100> , '<STR_LIT:!>' ) ) <EOL> def create ( ) : <EOL> return xs . delay_with_selector ( lambda _ : ys ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:6> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> ys . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_delay_duration_simple4_inner_empty ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_completed ( <NUM_LIT> ) ) <EOL> ys = scheduler . create_cold_observable ( on_completed ( <NUM_LIT:100> ) ) <EOL> def create ( ) : <EOL> return xs . delay_with_selector ( lambda _ : ys ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> + <NUM_LIT:100> , <NUM_LIT:6> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> ys . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_delay_duration_dispose1 ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_completed ( <NUM_LIT> ) ) <EOL> ys = scheduler . create_cold_observable ( on_next ( <NUM_LIT:200> , '<STR_LIT:!>' ) ) <EOL> def create ( ) : <EOL> return xs . delay_with_selector ( lambda _ : ys ) <EOL> results = scheduler . start ( create , disposed = <NUM_LIT> ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> + <NUM_LIT:200> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> + <NUM_LIT:200> , <NUM_LIT:3> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> ys . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) , subscribe ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_delay_duration_dispose2 ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( 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> ys = scheduler . create_cold_observable ( on_next ( <NUM_LIT:50> , '<STR_LIT:!>' ) ) <EOL> def create ( ) : <EOL> return xs . delay_with_selector ( lambda _ : ys ) <EOL> results = scheduler . start ( create , disposed = <NUM_LIT> ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> + <NUM_LIT:50> , <NUM_LIT:2> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> ys . subscriptions . assert_equal ( subscribe ( <NUM_LIT> , <NUM_LIT> ) ) </s>
