<s> import unittest <EOL> from nose import SkipTest <EOL> import rx <EOL> asyncio = rx . config [ '<STR_LIT>' ] <EOL> if asyncio is None : <EOL> raise SkipTest ( "<STR_LIT>" ) <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 TestToFuture ( unittest . TestCase ) : <EOL> def test_future_success ( self ) : <EOL> loop = asyncio . get_event_loop ( ) <EOL> success = [ False , True ] <EOL> @ asyncio . coroutine <EOL> def go ( ) : <EOL> source = Observable . return_value ( <NUM_LIT> ) <EOL> future = source . to_future ( asyncio . Future ) <EOL> def done ( future ) : <EOL> try : <EOL> value = future . result ( ) <EOL> except Exception as ex : <EOL> success [ <NUM_LIT:1> ] = False <EOL> else : <EOL> success [ <NUM_LIT:0> ] = value == <NUM_LIT> <EOL> future . add_done_callback ( done ) <EOL> loop . run_until_complete ( go ( ) ) <EOL> assert ( all ( success ) ) <EOL> def test_future_failure ( self ) : <EOL> loop = asyncio . get_event_loop ( ) <EOL> success = [ True , False ] <EOL> @ asyncio . coroutine <EOL> def go ( ) : <EOL> error = Exception ( '<STR_LIT>' ) <EOL> source = Observable . throw_exception ( error ) <EOL> future = source . to_future ( asyncio . Future ) <EOL> def done ( future ) : <EOL> try : <EOL> value = future . result ( ) <EOL> except Exception as ex : <EOL> success [ <NUM_LIT:1> ] = str ( ex ) == str ( error ) <EOL> else : <EOL> success [ <NUM_LIT:0> ] = False <EOL> future . add_done_callback ( done ) <EOL> loop . run_until_complete ( go ( ) ) <EOL> assert ( all ( success ) ) </s>
