#NEWLINE# def test_old_new_data_api_paths( self ) : #NEWLINE# #INDENT# '\n        Test that the new and old data APIs hit the same code paths.\n\n        We want to ensure that the old data API(data[sid(N)].field and\n        similar)  and the new data API(data.current(sid(N), field) and\n        similar) hit the same code paths on the DataPortal.\n        ' #NEWLINE# test_start_minute = self . env . market_minutes_for_day ( self . sim_params . trading_days [ 0 ] ) [ 1 ] #NEWLINE# test_end_minute = self . env . market_minutes_for_day ( self . sim_params . trading_days [ 0 ] ) [ ( - 1 ) ] #NEWLINE# bar_data = BarData ( self . data_portal , ( lambda : test_end_minute ) , 'minute' ) #NEWLINE# ohlcvp_fields = [ 'open' , 'high' , 'lowclose' , 'volume' , 'price' ] #NEWLINE# spot_value_meth = 'zipline.data.data_portal.DataPortal.get_spot_value' #NEWLINE# def assert_get_spot_value_called( fun , field ) : #NEWLINE# #INDENT# '\n            Assert that get_spot_value was called during the execution of fun.\n\n            Takes in a function fun and a string field.\n            ' #NEWLINE# with patch ( spot_value_meth ) as gsv : #NEWLINE# #INDENT# fun ( ) #NEWLINE# gsv . assert_called_with ( self . asset1 , field , test_end_minute , 'minute' ) #NEWLINE# #UNINDENT# #UNINDENT# for field in ohlcvp_fields : #NEWLINE# #INDENT# assert_get_spot_value_called ( ( lambda : getattr ( bar_data [ self . asset1 ] , field ) ) , field ) #NEWLINE# assert_get_spot_value_called ( ( lambda : bar_data . current ( self . asset1 , field ) ) , field ) #NEWLINE# #UNINDENT# history_meth = 'zipline.data.data_portal.DataPortal.get_history_window' #NEWLINE# def assert_get_history_window_called( fun , is_legacy ) : #NEWLINE# #INDENT# '\n            Assert that get_history_window was called during fun().\n\n            Takes in a function fun and a boolean is_legacy.\n            ' #NEWLINE# with patch ( history_meth ) as ghw : #NEWLINE# #INDENT# fun ( ) #NEWLINE# if is_legacy : #NEWLINE# #INDENT# ghw . assert_called_with ( [ self . asset1 , self . asset2 , self . asset3 ] , test_end_minute , 5 , '1m' , 'volume' , True ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# ghw . assert_called_with ( [ self . asset1 , self . asset2 , self . asset3 ] , test_end_minute , 5 , '1m' , 'volume' ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# test_sim_params = SimulationParameters ( period_start = test_start_minute , period_end = test_end_minute , data_frequency = 'minute' , env = field . env ) #NEWLINE# history_algorithm = self . create_algo ( history_algo , sim_params = test_sim_params ) #NEWLINE# assert_get_history_window_called ( ( lambda : history_algorithm . run ( self . data_portal ) ) , is_legacy = True ) #NEWLINE# assert_get_history_window_called ( ( lambda : bar_data . history ( [ self . asset1 , self . asset2 , self . asset3 ] , 'volume' , 5 , '1m' ) ) , is_legacy = False )