<s> import unittest <EOL> from datetime import datetime , timedelta <EOL> from rx import Observable <EOL> from rx . testing import TestScheduler , ReactiveTest , is_prime , MockDisposable <EOL> from rx . disposables import Disposable , SerialDisposable <EOL> from rx . subjects import Subject <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 Timestamp ( object ) : <EOL> def __init__ ( self , value , timestamp ) : <EOL> if isinstance ( timestamp , datetime ) : <EOL> timestamp = timestamp - datetime . fromtimestamp ( <NUM_LIT:0> ) <EOL> timestamp = int ( timestamp . microseconds / <NUM_LIT:1000> ) <EOL> self . value = value <EOL> self . timestamp = timestamp <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( self . value , self . timestamp ) <EOL> def equals ( other ) : <EOL> return other . timestamp == self . timestamp and other . value == self . value <EOL> class TestTimeInterval ( unittest . TestCase ) : <EOL> def test_timestamp_regular ( 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_next ( <NUM_LIT> , <NUM_LIT:5> ) , on_next ( <NUM_LIT> , <NUM_LIT:6> ) , on_completed ( <NUM_LIT> ) ) <EOL> def create ( ) : <EOL> def selector ( x ) : <EOL> return Timestamp ( x . value , x . timestamp ) <EOL> return xs . timestamp ( scheduler ) . map ( selector ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_next ( <NUM_LIT> , Timestamp ( <NUM_LIT:2> , <NUM_LIT> ) ) , on_next ( <NUM_LIT> , Timestamp ( <NUM_LIT:3> , <NUM_LIT> ) ) , on_next ( <NUM_LIT> , Timestamp ( <NUM_LIT:4> , <NUM_LIT> ) ) , on_next ( <NUM_LIT> , Timestamp ( <NUM_LIT:5> , <NUM_LIT> ) ) , on_next ( <NUM_LIT> , Timestamp ( <NUM_LIT:6> , <NUM_LIT> ) ) , on_completed ( <NUM_LIT> ) ) <EOL> def test_timestamp_empty ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . empty ( scheduler ) . time_interval ( scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_completed ( <NUM_LIT> ) ) <EOL> def test_timestamp_error ( self ) : <EOL> ex = '<STR_LIT>' <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . throw_exception ( ex , scheduler ) . time_interval ( scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( on_error ( <NUM_LIT> , ex ) ) <EOL> def test_timestamp_never ( self ) : <EOL> scheduler = TestScheduler ( ) <EOL> def create ( ) : <EOL> return Observable . never ( ) . time_interval ( scheduler = scheduler ) <EOL> results = scheduler . start ( create ) <EOL> results . messages . assert_equal ( ) </s>
