<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 TestMaxBy ( unittest . TestCase ) : <EOL> def test_maxby_empty ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> msgs = [ <EOL> on_next ( <NUM_LIT> , { "<STR_LIT:key>" : <NUM_LIT:1> , "<STR_LIT:value>" : '<STR_LIT:z>' } ) , <EOL> on_completed ( <NUM_LIT> ) <EOL> ] <EOL> xs = scheduler . create_hot_observable ( msgs ) <EOL> def create ( ) : <EOL> def selector ( x ) : <EOL> return x [ "<STR_LIT:key>" ] <EOL> return xs . max_by ( selector ) <EOL> res = scheduler . start ( create = create ) . messages <EOL> self . assertEqual ( <NUM_LIT:2> , len ( res ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( res [ <NUM_LIT:0> ] . value . value ) ) <EOL> assert ( res [ <NUM_LIT:1> ] . value . kind == '<STR_LIT:C>' and res [ <NUM_LIT:1> ] . time == <NUM_LIT> ) <EOL> def test_maxby_return ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> msgs = [ <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:1> , <EOL> "<STR_LIT:value>" : '<STR_LIT:z>' <EOL> } ) , on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:2> , <EOL> "<STR_LIT:value>" : '<STR_LIT:a>' <EOL> } ) , on_completed ( <NUM_LIT> ) <EOL> ] <EOL> xs = scheduler . create_hot_observable ( msgs ) <EOL> def create ( ) : <EOL> def selector ( x ) : <EOL> return x [ "<STR_LIT:key>" ] <EOL> return xs . max_by ( selector ) <EOL> res = scheduler . start ( create = create ) . messages <EOL> self . assertEqual ( <NUM_LIT:2> , len ( res ) ) <EOL> assert ( res [ <NUM_LIT:0> ] . value . kind == '<STR_LIT:N>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( res [ <NUM_LIT:0> ] . value . value ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , res [ <NUM_LIT:0> ] . value . value [ <NUM_LIT:0> ] [ "<STR_LIT:key>" ] ) <EOL> self . assertEqual ( '<STR_LIT:a>' , res [ <NUM_LIT:0> ] . value . value [ <NUM_LIT:0> ] [ "<STR_LIT:value>" ] ) <EOL> assert ( res [ <NUM_LIT:1> ] . value . kind == '<STR_LIT:C>' and res [ <NUM_LIT:1> ] . time == <NUM_LIT> ) <EOL> def test_maxby_some ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> msgs = [ <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:1> , <EOL> "<STR_LIT:value>" : '<STR_LIT:z>' <EOL> } ) , on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:3> , <EOL> "<STR_LIT:value>" : '<STR_LIT:b>' <EOL> } ) , on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:4> , <EOL> "<STR_LIT:value>" : '<STR_LIT:c>' <EOL> } ) , on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:2> , <EOL> "<STR_LIT:value>" : '<STR_LIT:a>' <EOL> } ) , on_completed ( <NUM_LIT> ) <EOL> ] <EOL> xs = scheduler . create_hot_observable ( msgs ) <EOL> def create ( ) : <EOL> def selector ( x ) : <EOL> return x [ "<STR_LIT:key>" ] <EOL> return xs . max_by ( selector ) <EOL> res = scheduler . start ( create = create ) . messages <EOL> self . assertEqual ( <NUM_LIT:2> , len ( res ) ) <EOL> assert ( res [ <NUM_LIT:0> ] . value . kind == '<STR_LIT:N>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( res [ <NUM_LIT:0> ] . value . value [ <NUM_LIT:0> ] [ "<STR_LIT:value>" ] ) ) <EOL> self . assertEqual ( <NUM_LIT:4> , res [ <NUM_LIT:0> ] . value . value [ <NUM_LIT:0> ] [ "<STR_LIT:key>" ] ) <EOL> self . assertEqual ( '<STR_LIT:c>' , res [ <NUM_LIT:0> ] . value . value [ <NUM_LIT:0> ] [ "<STR_LIT:value>" ] ) <EOL> assert ( res [ <NUM_LIT:1> ] . value . kind == '<STR_LIT:C>' and res [ <NUM_LIT:1> ] . time == <NUM_LIT> ) <EOL> def test_maxby_multiple ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> msgs = [ <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:1> , <EOL> "<STR_LIT:value>" : '<STR_LIT:z>' <EOL> } ) , <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:3> , <EOL> "<STR_LIT:value>" : '<STR_LIT:b>' <EOL> } ) , <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:2> , <EOL> "<STR_LIT:value>" : '<STR_LIT:d>' <EOL> } ) , <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:3> , <EOL> "<STR_LIT:value>" : '<STR_LIT:c>' <EOL> } ) , <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:2> , <EOL> "<STR_LIT:value>" : '<STR_LIT:y>' <EOL> } ) , <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:4> , <EOL> "<STR_LIT:value>" : '<STR_LIT:a>' <EOL> } ) , <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:4> , <EOL> "<STR_LIT:value>" : '<STR_LIT:r>' <EOL> } ) , <EOL> on_completed ( <NUM_LIT> ) <EOL> ] <EOL> xs = scheduler . create_hot_observable ( msgs ) <EOL> def create ( ) : <EOL> return xs . max_by ( lambda x : x [ "<STR_LIT:key>" ] ) <EOL> res = scheduler . start ( create = create ) . messages <EOL> self . assertEqual ( <NUM_LIT:2> , len ( res ) ) <EOL> assert ( res [ <NUM_LIT:0> ] . value . kind == '<STR_LIT:N>' ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( res [ <NUM_LIT:0> ] . value . value ) ) <EOL> self . assertEqual ( <NUM_LIT:4> , res [ <NUM_LIT:0> ] . value . value [ <NUM_LIT:0> ] [ "<STR_LIT:key>" ] ) <EOL> self . assertEqual ( '<STR_LIT:a>' , res [ <NUM_LIT:0> ] . value . value [ <NUM_LIT:0> ] [ "<STR_LIT:value>" ] ) <EOL> self . assertEqual ( <NUM_LIT:4> , res [ <NUM_LIT:0> ] . value . value [ <NUM_LIT:1> ] [ "<STR_LIT:key>" ] ) <EOL> self . assertEqual ( '<STR_LIT:r>' , res [ <NUM_LIT:0> ] . value . value [ <NUM_LIT:1> ] [ "<STR_LIT:value>" ] ) <EOL> assert ( res [ <NUM_LIT:1> ] . value . kind == '<STR_LIT:C>' and res [ <NUM_LIT:1> ] . time == <NUM_LIT> ) <EOL> def test_maxby_throw ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> msgs = [ <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:1> , <EOL> "<STR_LIT:value>" : '<STR_LIT:z>' <EOL> } ) , <EOL> on_error ( <NUM_LIT> , ex ) <EOL> ] <EOL> xs = scheduler . create_hot_observable ( msgs ) <EOL> def create ( ) : <EOL> return xs . max_by ( lambda x : x [ "<STR_LIT:key>" ] ) <EOL> res = scheduler . start ( create = create ) . messages <EOL> res . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_maxby_never ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> msgs = [ <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:1> , <EOL> "<STR_LIT:value>" : '<STR_LIT:z>' <EOL> } ) <EOL> ] <EOL> xs = scheduler . create_hot_observable ( msgs ) <EOL> def create ( ) : <EOL> return xs . max_by ( lambda x : x [ "<STR_LIT:key>" ] ) <EOL> res = scheduler . start ( create = create ) . messages <EOL> res . assert_equal ( ) <EOL> def test_maxby_comparerthrows ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> msgs = [ <EOL> on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:1> , <EOL> "<STR_LIT:value>" : '<STR_LIT:z>' <EOL> } ) , on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:3> , <EOL> "<STR_LIT:value>" : '<STR_LIT:b>' <EOL> } ) , on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:2> , <EOL> "<STR_LIT:value>" : '<STR_LIT:c>' <EOL> } ) , on_next ( <NUM_LIT> , { <EOL> "<STR_LIT:key>" : <NUM_LIT:4> , <EOL> "<STR_LIT:value>" : '<STR_LIT:a>' <EOL> } ) , on_completed ( <NUM_LIT> ) <EOL> ] <EOL> def reverse_comparer ( a , b ) : <EOL> raise Exception ( ex ) <EOL> xs = scheduler . create_hot_observable ( msgs ) <EOL> def create ( ) : <EOL> return xs . max_by ( lambda x : x [ "<STR_LIT:key>" ] , reverse_comparer ) <EOL> res = scheduler . start ( create = create ) . messages <EOL> res . assert_equal ( on_error ( <NUM_LIT> , ex ) ) </s>
