<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 <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 TestTakeUntil ( unittest . TestCase ) : <EOL> def test_take_until_preempt_somedata_next ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> l_msgs = [ 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_completed ( <NUM_LIT> ) ] <EOL> r_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ] <EOL> l = scheduler . create_hot_observable ( l_msgs ) <EOL> r = scheduler . create_hot_observable ( r_msgs ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_take_until_preempt_somedata_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> l_msgs = [ 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_completed ( <NUM_LIT> ) ] <EOL> r_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> l = scheduler . create_hot_observable ( l_msgs ) <EOL> r = scheduler . create_hot_observable ( r_msgs ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> def test_take_until_nopreempt_somedata_empty ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> l_msgs = [ 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_completed ( <NUM_LIT> ) ] <EOL> r_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_completed ( <NUM_LIT> ) ] <EOL> l = scheduler . create_hot_observable ( l_msgs ) <EOL> r = scheduler . create_hot_observable ( r_msgs ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( 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_completed ( <NUM_LIT> ) ) <EOL> def test_take_until_nopreempt_somedata_never ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> l_msgs = [ 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_completed ( <NUM_LIT> ) ] <EOL> l = scheduler . create_hot_observable ( l_msgs ) <EOL> r = Observable . never ( ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( 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_completed ( <NUM_LIT> ) ) <EOL> def test_take_until_preempt_never_next ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> r_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ] <EOL> l = Observable . never ( ) <EOL> r = scheduler . create_hot_observable ( r_msgs ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_completed ( <NUM_LIT> ) ) <EOL> def test_take_until_preempt_never_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> r_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> l = Observable . never ( ) <EOL> r = scheduler . create_hot_observable ( r_msgs ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_take_until_nopreempt_never_empty ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> r_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_completed ( <NUM_LIT> ) ] <EOL> l = Observable . never ( ) <EOL> r = scheduler . create_hot_observable ( r_msgs ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( ) <EOL> def test_take_until_nopreempt_never_never ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> l = Observable . never ( ) <EOL> r = Observable . never ( ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( ) <EOL> def test_take_until_preempt_beforefirstproduced ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> l_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ] <EOL> r_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ] <EOL> l = scheduler . create_hot_observable ( l_msgs ) <EOL> r = scheduler . create_hot_observable ( r_msgs ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_completed ( <NUM_LIT> ) ) <EOL> def test_take_until_preempt_beforefirstproduced_remain_silent_and_proper_disposed ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> l_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_error ( <NUM_LIT> , '<STR_LIT>' ) , on_completed ( <NUM_LIT> ) ] <EOL> r_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ] <EOL> source_not_disposed = [ False ] <EOL> def action ( ) : <EOL> source_not_disposed [ <NUM_LIT:0> ] = True <EOL> l = scheduler . create_hot_observable ( l_msgs ) . do_action ( on_next = action ) <EOL> r = scheduler . create_hot_observable ( r_msgs ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_completed ( <NUM_LIT> ) ) <EOL> assert ( not source_not_disposed [ <NUM_LIT:0> ] ) <EOL> def test_take_until_nopreempt_afterlastproduced_proper_disposed_signal ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> l_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ] <EOL> r_msgs = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ] <EOL> signal_not_disposed = [ False ] <EOL> l = scheduler . create_hot_observable ( l_msgs ) <EOL> def action ( ) : <EOL> signal_not_disposed [ <NUM_LIT:0> ] = True <EOL> r = scheduler . create_hot_observable ( r_msgs ) . do_action ( on_next = action ) <EOL> def create ( ) : <EOL> return l . take_until ( r ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ) <EOL> assert ( not signal_not_disposed [ <NUM_LIT:0> ] ) </s>
