<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 TestCatch ( unittest . TestCase ) : <EOL> def test_catch_no_errors ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ 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> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_completed ( <NUM_LIT> ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> def create ( ) : <EOL> return o1 . catch_exception ( o2 ) <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_catch_never ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_completed ( <NUM_LIT> ) ] <EOL> o1 = Observable . never ( ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> def create ( ) : <EOL> return o1 . catch_exception ( o2 ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( ) <EOL> def test_catch_empty ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_completed ( <NUM_LIT> ) ] <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_completed ( <NUM_LIT> ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> def create ( ) : <EOL> return o1 . catch_exception ( o2 ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_completed ( <NUM_LIT> ) ) <EOL> def test_catch_return ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ] <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_completed ( <NUM_LIT> ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> def create ( ) : <EOL> return o1 . catch_exception ( o2 ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_catch_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_completed ( <NUM_LIT> ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> def create ( ) : <EOL> return o1 . catch_exception ( o2 ) <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:5> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_catch_error_never ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = Observable . never ( ) <EOL> def create ( ) : <EOL> return o1 . catch_exception ( o2 ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) ) <EOL> def test_catch_error_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_error ( <NUM_LIT> , '<STR_LIT>' ) ] <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> def create ( ) : <EOL> return o1 . catch_exception ( o2 ) <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_error ( <NUM_LIT> , ex ) ) <EOL> def test_catch_multiple ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> msgs3 = [ on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_completed ( <NUM_LIT> ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> o3 = scheduler . create_hot_observable ( msgs3 ) <EOL> def create ( ) : <EOL> return Observable . catch_exception ( o1 , o2 , o3 ) <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_completed ( <NUM_LIT> ) ) <EOL> def test_catch_error_specific_caught ( self ) : <EOL> ex = '<STR_LIT>' <EOL> handler_called = [ False ] <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_completed ( <NUM_LIT> ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> def create ( ) : <EOL> def handler ( e ) : <EOL> handler_called [ <NUM_LIT:0> ] = True <EOL> return o2 <EOL> return o1 . catch_exception ( handler ) <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_completed ( <NUM_LIT> ) ) <EOL> assert ( handler_called [ <NUM_LIT:0> ] ) <EOL> def test_catch_error_specific_caught_immediate ( self ) : <EOL> ex = '<STR_LIT>' <EOL> handler_called = [ False ] <EOL> scheduler = TestScheduler ( ) <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_completed ( <NUM_LIT> ) ] <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> def create ( ) : <EOL> def handler ( e ) : <EOL> handler_called [ <NUM_LIT:0> ] = True <EOL> return o2 <EOL> return Observable . throw_exception ( '<STR_LIT>' ) . catch_exception ( handler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_completed ( <NUM_LIT> ) ) <EOL> assert ( handler_called [ <NUM_LIT:0> ] ) <EOL> def test_catch_handler_throws ( self ) : <EOL> ex = '<STR_LIT>' <EOL> ex2 = '<STR_LIT>' <EOL> handler_called = [ False ] <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> def create ( ) : <EOL> def handler ( e ) : <EOL> handler_called [ <NUM_LIT:0> ] = True <EOL> raise Exception ( ex2 ) <EOL> return o1 . catch_exception ( handler ) <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> , ex2 ) ) <EOL> assert ( handler_called [ <NUM_LIT:0> ] ) <EOL> def test_catch_nested_outer_catches ( self ) : <EOL> ex = '<STR_LIT>' <EOL> first_handler_called = [ False ] <EOL> second_handler_called = [ False ] <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_completed ( <NUM_LIT> ) ] <EOL> msgs3 = [ on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_completed ( <NUM_LIT> ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> o3 = scheduler . create_hot_observable ( msgs3 ) <EOL> def create ( ) : <EOL> def handler1 ( e ) : <EOL> first_handler_called [ <NUM_LIT:0> ] = True <EOL> return o2 <EOL> def handler2 ( e ) : <EOL> second_handler_called [ <NUM_LIT:0> ] = True <EOL> return o3 <EOL> return o1 . catch_exception ( handler1 ) . catch_exception ( handler2 ) <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> assert ( first_handler_called [ <NUM_LIT:0> ] ) <EOL> assert ( not second_handler_called [ <NUM_LIT:0> ] ) <EOL> def test_catch_throw_from_nested_catch ( self ) : <EOL> ex = '<STR_LIT>' <EOL> ex2 = '<STR_LIT>' <EOL> first_handler_called = [ False ] <EOL> second_handler_called = [ False ] <EOL> scheduler = TestScheduler ( ) <EOL> msgs1 = [ on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_error ( <NUM_LIT> , ex ) ] <EOL> msgs2 = [ on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_error ( <NUM_LIT> , ex2 ) ] <EOL> msgs3 = [ on_next ( <NUM_LIT> , <NUM_LIT:4> ) , on_completed ( <NUM_LIT> ) ] <EOL> o1 = scheduler . create_hot_observable ( msgs1 ) <EOL> o2 = scheduler . create_hot_observable ( msgs2 ) <EOL> o3 = scheduler . create_hot_observable ( msgs3 ) <EOL> def create ( ) : <EOL> def handler1 ( e ) : <EOL> first_handler_called [ <NUM_LIT:0> ] = True <EOL> assert ( e == ex ) <EOL> return o2 <EOL> def handler2 ( e ) : <EOL> second_handler_called [ <NUM_LIT:0> ] = True <EOL> assert ( e == ex2 ) <EOL> return o3 <EOL> return o1 . catch_exception ( handler1 ) . catch_exception ( handler2 ) <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_completed ( <NUM_LIT> ) ) <EOL> assert ( first_handler_called [ <NUM_LIT:0> ] ) <EOL> assert ( second_handler_called [ <NUM_LIT:0> ] ) </s>
