<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 TestElementAt ( unittest . TestCase ) : <EOL> def test_elementat_first ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . element_at ( <NUM_LIT:0> ) <EOL> results = scheduler . start ( create = create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_elementat_other ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . element_at ( <NUM_LIT:2> ) <EOL> results = scheduler . start ( create = create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_elementat_outofrange ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . element_at ( <NUM_LIT:3> ) <EOL> results = scheduler . start ( create = create ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( results . messages ) ) <EOL> self . assertEqual ( <NUM_LIT> , results . messages [ <NUM_LIT:0> ] . time ) <EOL> self . assertEqual ( '<STR_LIT:E>' , results . messages [ <NUM_LIT:0> ] . value . kind ) <EOL> assert ( results . messages [ <NUM_LIT:0> ] . value . exception ) <EOL> def test_elementat_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> def create ( ) : <EOL> return xs . element_at ( <NUM_LIT:3> ) <EOL> results = scheduler . start ( create = create ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_element_at_or_default_first ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . element_at_or_default ( <NUM_LIT:0> ) <EOL> results = scheduler . start ( create = create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_element_at_or_default_other ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . element_at_or_default ( <NUM_LIT:2> ) <EOL> results = scheduler . start ( create = create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> def test_element_at_or_default_outofrange ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> return xs . element_at_or_default ( <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> results = scheduler . start ( create = create ) <EOL> results . 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_element_at_or_default_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> xs = scheduler . create_hot_observable ( on_next ( <NUM_LIT> , <NUM_LIT> ) , on_next ( <NUM_LIT> , <NUM_LIT> ) , on_error ( <NUM_LIT> , ex ) ) <EOL> def create ( ) : <EOL> return xs . element_at_or_default ( <NUM_LIT:3> ) <EOL> results = scheduler . start ( create = create ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> xs . subscriptions . assert_equal ( subscribe ( <NUM_LIT:200> , <NUM_LIT> ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
