<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 TestCount ( unittest . TestCase ) : <EOL> def test_count_empty ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_completed ( <NUM_LIT> ) ) <EOL> res = scheduler . start ( create = lambda : xs . count ( ) ) . messages <EOL> res . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_count_empty_ii ( 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_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . count ( ) <EOL> res = scheduler . start ( create = create ) . messages <EOL> res . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_count_some ( 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_completed ( <NUM_LIT> ) ) <EOL> res = scheduler . start ( create = lambda : xs . count ( ) ) . messages <EOL> res . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_count_throw ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> res = scheduler . start ( create = lambda : xs . count ( ) ) . messages <EOL> res . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_count_never ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) ) <EOL> res = scheduler . start ( create = lambda : xs . count ( ) ) . messages <EOL> res . assert_equal ( ) <EOL> def test_count_predicate_empty_true ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda _ : True ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_count_predicate_empty_false ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda _ : False ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_count_predicate_return_true ( 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_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda _ : True ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_count_predicate_return_false ( 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_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda _ : False ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_count_predicate_some_all ( 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_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda x : x < <NUM_LIT:10> ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:3> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_count_predicate_some_none ( 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_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda x : x > <NUM_LIT:10> ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:0> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_count_predicate_some_even ( 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_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda x : x % <NUM_LIT:2> == <NUM_LIT:0> ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT:2> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_count_predicate_throw_true ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda _ : True ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_count_predicate_throw_false ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda _ : False ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_count_predicate_never ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT:1> ) ) <EOL> def create ( ) : <EOL> return xs . count ( lambda _ : True ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT:1000> ) ) <EOL> def test_count_predicate_predicate_throws ( self ) : <EOL> ex = '<STR_LIT>' <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> def create ( ) : <EOL> def predicate ( x ) : <EOL> if x == <NUM_LIT:3> : <EOL> raise Exception ( ex ) <EOL> else : <EOL> return True <EOL> return xs . count ( predicate ) <EOL> res = scheduler . start ( create = create ) <EOL> res . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) </s>
