<s> import unittest <EOL> from rx . observable import Observable <EOL> from rx . testing import TestScheduler , ReactiveTest <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 TestDo ( unittest . TestCase ) : <EOL> def test_do_should_see_all_values ( 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_completed ( <NUM_LIT> ) ) <EOL> i = [ <NUM_LIT:0> ] <EOL> sum = [ <NUM_LIT:2> + <NUM_LIT:3> + <NUM_LIT:4> + <NUM_LIT:5> ] <EOL> def create ( ) : <EOL> def action ( x ) : <EOL> i [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> sum [ <NUM_LIT:0> ] -= x <EOL> return sum [ <NUM_LIT:0> ] <EOL> return xs . do_action ( action ) <EOL> scheduler . start ( create ) <EOL> self . assertEqual ( <NUM_LIT:4> , i [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( <NUM_LIT:0> , sum [ <NUM_LIT:0> ] ) <EOL> def test_do_plain_action ( 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_completed ( <NUM_LIT> ) ) <EOL> i = [ <NUM_LIT:0> ] <EOL> def create ( ) : <EOL> def action ( x ) : <EOL> i [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> return i [ <NUM_LIT:0> ] <EOL> return xs . do_action ( action ) <EOL> scheduler . start ( create ) <EOL> self . assertEqual ( <NUM_LIT:4> , i [ <NUM_LIT:0> ] ) <EOL> def test_do_next_completed ( 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_completed ( <NUM_LIT> ) ) <EOL> i = [ <NUM_LIT:0> ] <EOL> sum = [ <NUM_LIT:2> + <NUM_LIT:3> + <NUM_LIT:4> + <NUM_LIT:5> ] <EOL> completed = [ False ] <EOL> def create ( ) : <EOL> def on_next ( x ) : <EOL> i [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> sum [ <NUM_LIT:0> ] -= x <EOL> def on_completed ( ) : <EOL> completed [ <NUM_LIT:0> ] = True <EOL> return xs . do_action ( on_next = on_next , on_completed = on_completed ) <EOL> scheduler . start ( create ) <EOL> self . assertEqual ( <NUM_LIT:4> , i [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( <NUM_LIT:0> , sum [ <NUM_LIT:0> ] ) <EOL> assert ( completed [ <NUM_LIT:0> ] ) <EOL> def test_do_next_completed_never ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> i = [ <NUM_LIT:0> ] <EOL> completed = [ False ] <EOL> def create ( ) : <EOL> def on_next ( x ) : <EOL> i [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> def on_completed ( ) : <EOL> completed = True <EOL> return Observable . never ( ) . do_action ( on_next = on_next , on_completed = on_completed ) <EOL> scheduler . start ( create ) <EOL> self . assertEqual ( <NUM_LIT:0> , i [ <NUM_LIT:0> ] ) <EOL> assert ( not completed [ <NUM_LIT:0> ] ) </s>
