import logging from rx.testing import ReactiveTest FORMAT = '%(asctime)-15s %(threadName)s %(message)s' logging.basicConfig(filename='rx.log', format=FORMAT, level=logging.DEBUG) log = logging.getLogger('Rx') on_next = ReactiveTest.on_next on_completed = ReactiveTest.on_completed on_error = ReactiveTest.on_error subscribe = ReactiveTest.subscribe subscribed = ReactiveTest.subscribed disposed = ReactiveTest.disposed created = ReactiveTest.created class RxException(Exception): pass # Helper function for raising exceptions within lambdas def _raise(ex): raise RxException(ex) # // TakeLastBuffer # def test_takeLastBuffer_with_time_Zero1(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_completed(230)) # res = scheduler.start(create) # return xs.takeLastBuffer_with_time(0, scheduler) # res.messages.assert_equal(on_next(230, function (lst) { # return lst.length === 0 # }), on_completed(230)) # assert xs.subscriptions == [subscribe(200, 230)] # def test_takeLastBuffer_with_time_Zero2(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_next(230, 3), on_completed(230)) # res = scheduler.start(create) # return xs.takeLastBuffer_with_time(0, scheduler) # res.messages.assert_equal(on_next(230, function (lst) { # return lst.length === 0 # }), on_completed(230)) # assert xs.subscriptions == [subscribe(200, 230)] # function arrayEqual(arr1, arr2) { # if (arr1.length != arr2.length) return false # for (var i = 0, len = arr1.length i < len i++) { # if (arr1[i] != arr2[i]) return false # } # return true # } # def test_takeLastBuffer_with_time_Some1(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_next(230, 3), on_completed(240)) # res = scheduler.start(create) # return xs.takeLastBuffer_with_time(25, scheduler) # res.messages.assert_equal(on_next(240, function (lst) { # return arrayEqual(lst, [2, 3]) # }), on_completed(240)) # assert xs.subscriptions == [subscribe(200, 240)] # def test_takeLastBuffer_with_time_Some2(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_next(230, 3), on_completed(300)) # res = scheduler.start(create) # return xs.takeLastBuffer_with_time(25, scheduler) # res.messages.assert_equal(on_next(300, function (lst) { # return lst.length === 0 # }), on_completed(300)) # assert xs.subscriptions == [subscribe(200, 300)] # def test_takeLastBuffer_with_time_Some3(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_next(230, 3), on_next(240, 4), on_next(250, 5), on_next(260, 6), on_next(270, 7), on_next(280, 8), on_next(290, 9), on_completed(300)) # res = scheduler.start(create) # return xs.takeLastBuffer_with_time(45, scheduler) # res.messages.assert_equal(on_next(300, function (lst) { # return arrayEqual(lst, [6, 7, 8, 9]) # }), on_completed(300)) # assert xs.subscriptions == [subscribe(200, 300)] # def test_takeLastBuffer_with_time_Some4(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(240, 2), on_next(250, 3), on_next(280, 4), on_next(290, 5), on_next(300, 6), on_completed(350)) # res = scheduler.start(create) # return xs.takeLastBuffer_with_time(25, scheduler) # res.messages.assert_equal(on_next(350, function (lst) { # return lst.length === 0 # }), on_completed(350)) # assert xs.subscriptions == [subscribe(200, 350)] # def test_takeLastBuffer_with_time_All(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_completed(230)) # res = scheduler.start(create) # return xs.takeLastBuffer_with_time(50, scheduler) # res.messages.assert_equal(on_next(230, function (lst) { # return arrayEqual(lst, [1, 2]) # }), on_completed(230)) # assert xs.subscriptions == [subscribe(200, 230)] # def test_takeLastBuffer_with_time_Error(): # var ex, res, scheduler, xs # scheduler = TestScheduler() # ex = 'ex' # xs = scheduler.create_hot_observable(on_error(210, ex)) # res = scheduler.start(create) # return xs.takeLastBuffer_with_time(50, scheduler) # assert res.messages == [on_error(210, ex)] # assert xs.subscriptions == [subscribe(200, 210)] # def test_takeLastBuffer_with_time_Never(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable() # res = scheduler.start(create) # return xs.takeLastBuffer_with_time(50, scheduler) # assert res.messages == [] # assert xs.subscriptions == [subscribe(200, 1000)] # def test_Take_Zero(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_completed(230)) # res = scheduler.start(create) # return xs.takeWithTime(0, scheduler) # assert res.messages == [on_completed(201)] # assert xs.subscriptions == [subscribe(200, 201)] # def test_Take_Some(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_next(230, 3), on_completed(240)) # res = scheduler.start(create) # return xs.takeWithTime(25, scheduler) # assert res.messages == [on_next(210, 1), on_next(220, 2), on_completed(225)] # assert xs.subscriptions == [subscribe(200, 225)] # def test_Take_Late(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_completed(230)) # res = scheduler.start(create) # return xs.takeWithTime(50, scheduler) # assert res.messages == [on_next(210, 1), on_next(220, 2), on_completed(230)] # assert xs.subscriptions == [subscribe(200, 230)] # def test_Take_Error(): # var ex, res, scheduler, xs # scheduler = TestScheduler() # ex = 'ex' # xs = scheduler.create_hot_observable(on_error(210, ex)) # res = scheduler.start(create) # return xs.takeWithTime(50, scheduler) # assert res.messages == [on_error(210, ex)] # assert xs.subscriptions == [subscribe(200, 210)] # def test_Take_Never(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable() # res = scheduler.start(create) # return xs.takeWithTime(50, scheduler) # assert res.messages == [on_completed(250)] # assert xs.subscriptions == [subscribe(200, 250)] # def test_Take_Twice1(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_next(230, 3), on_next(240, 4), on_next(250, 5), on_next(260, 6), on_completed(270)) # res = scheduler.start(create) # return xs.takeWithTime(55, scheduler).takeWithTime(35, scheduler) # assert res.messages == [on_next(210, 1), on_next(220, 2), on_next(230, 3), on_completed(235)] # assert xs.subscriptions == [subscribe(200, 235)] # def test_Take_Twice2(): # var res, scheduler, xs # scheduler = TestScheduler() # xs = scheduler.create_hot_observable(on_next(210, 1), on_next(220, 2), on_next(230, 3), on_next(240, 4), on_next(250, 5), on_next(260, 6), on_completed(270)) # res = scheduler.start(create) # return xs.takeWithTime(35, scheduler).takeWithTime(55, scheduler) # assert res.messages == [on_next(210, 1), on_next(220, 2), on_next(230, 3), on_completed(235)] # assert xs.subscriptions == [subscribe(200, 235)] # // TakeLast if __name__ == '__main__': test_buffer_with_time_or_count_basic()