<s> from __future__ import absolute_import , division , print_function <EOL> import os <EOL> import pytest <EOL> import pandas as pd <EOL> tb = pytest . importorskip ( '<STR_LIT>' ) <EOL> try : <EOL> f = pd . HDFStore ( '<STR_LIT:foo>' ) <EOL> except ( RuntimeError , ImportError ) as e : <EOL> pytest . skip ( '<STR_LIT>' % e ) <EOL> else : <EOL> f . close ( ) <EOL> os . remove ( '<STR_LIT:foo>' ) <EOL> from blaze . compatibility import xfail <EOL> import numpy as np <EOL> from blaze . compute . core import compute <EOL> from blaze . expr import symbol <EOL> from blaze import drop , discover , create_index <EOL> from blaze . utils import tmpfile <EOL> t = symbol ( '<STR_LIT:t>' , '<STR_LIT>' ) <EOL> x = np . array ( [ ( <NUM_LIT:1> , '<STR_LIT>' , <NUM_LIT:100> ) , <EOL> ( <NUM_LIT:2> , '<STR_LIT>' , - <NUM_LIT:200> ) , <EOL> ( <NUM_LIT:3> , '<STR_LIT>' , <NUM_LIT> ) , <EOL> ( <NUM_LIT:4> , '<STR_LIT>' , <NUM_LIT> ) , <EOL> ( <NUM_LIT:5> , '<STR_LIT>' , - <NUM_LIT> ) ] , <EOL> dtype = [ ( '<STR_LIT:id>' , '<STR_LIT>' ) , ( '<STR_LIT:name>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> @ pytest . yield_fixture <EOL> def data ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as filename : <EOL> f = tb . open_file ( filename , mode = '<STR_LIT:w>' ) <EOL> d = f . create_table ( '<STR_LIT:/>' , '<STR_LIT:title>' , x ) <EOL> yield d <EOL> d . close ( ) <EOL> f . close ( ) <EOL> @ pytest . yield_fixture <EOL> def csi_data ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as filename : <EOL> f = tb . open_file ( filename , mode = '<STR_LIT:w>' ) <EOL> d = f . create_table ( '<STR_LIT:/>' , '<STR_LIT:title>' , x ) <EOL> d . cols . amount . create_csindex ( ) <EOL> d . cols . id . create_csindex ( ) <EOL> yield d <EOL> d . close ( ) <EOL> f . close ( ) <EOL> @ pytest . yield_fixture <EOL> def idx_data ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> f = tb . open_file ( fn , mode = '<STR_LIT:w>' ) <EOL> d = f . create_table ( '<STR_LIT:/>' , '<STR_LIT:title>' , x ) <EOL> d . cols . amount . create_index ( ) <EOL> d . cols . id . create_index ( ) <EOL> yield d <EOL> d . close ( ) <EOL> f . close ( ) <EOL> def eq ( a , b ) : <EOL> return ( a == b ) . all ( ) <EOL> def test_discover_datashape ( data ) : <EOL> ds = discover ( data ) <EOL> t = symbol ( '<STR_LIT:t>' , ds ) <EOL> columns = t . fields <EOL> assert columns is not None <EOL> def test_symbol ( data ) : <EOL> assert compute ( t , data ) == data <EOL> assert isinstance ( data , tb . Table ) <EOL> def test_single_column ( data ) : <EOL> assert eq ( compute ( t [ '<STR_LIT:name>' ] , data ) , x [ '<STR_LIT:name>' ] ) <EOL> def test_projection ( data ) : <EOL> assert eq ( compute ( t [ [ '<STR_LIT:name>' , '<STR_LIT>' ] ] , data ) , x [ [ '<STR_LIT:name>' , '<STR_LIT>' ] ] ) <EOL> def test_eq ( data ) : <EOL> assert eq ( compute ( t [ '<STR_LIT>' ] == <NUM_LIT:100> , data ) , x [ '<STR_LIT>' ] == <NUM_LIT:100> ) <EOL> def test_scalar_ops ( data ) : <EOL> from operator import add , sub , mul , truediv <EOL> for op in ( add , sub , mul , truediv ) : <EOL> assert eq ( compute ( op ( t . amount , <NUM_LIT:10> ) , data ) , op ( x [ '<STR_LIT>' ] , <NUM_LIT:10> ) ) <EOL> assert eq ( compute ( op ( t . amount , t . id ) , data ) , op ( x [ '<STR_LIT>' ] , x [ '<STR_LIT:id>' ] ) ) <EOL> assert eq ( compute ( op ( <NUM_LIT> , t . amount ) , data ) , op ( <NUM_LIT> , x [ '<STR_LIT>' ] ) ) <EOL> assert eq ( compute ( op ( <NUM_LIT:10> , t . amount ) , data ) , op ( <NUM_LIT:10> , x [ '<STR_LIT>' ] ) ) <EOL> def test_neg ( data ) : <EOL> assert eq ( compute ( - t . amount , data ) , - x [ '<STR_LIT>' ] ) <EOL> def test_failing_floordiv ( data ) : <EOL> from operator import floordiv as op <EOL> with pytest . raises ( TypeError ) : <EOL> assert eq ( compute ( op ( t . amount , <NUM_LIT:10> ) , data ) , op ( x [ '<STR_LIT>' ] , <NUM_LIT:10> ) ) <EOL> with pytest . raises ( TypeError ) : <EOL> assert eq ( compute ( op ( t . amount , t . id ) , data ) , op ( x [ '<STR_LIT>' ] , x [ '<STR_LIT:id>' ] ) ) <EOL> with pytest . raises ( TypeError ) : <EOL> assert eq ( compute ( op ( <NUM_LIT> , t . amount ) , data ) , op ( <NUM_LIT> , x [ '<STR_LIT>' ] ) ) <EOL> with pytest . raises ( TypeError ) : <EOL> assert eq ( compute ( op ( <NUM_LIT:10> , t . amount ) , data ) , op ( <NUM_LIT:10> , x [ '<STR_LIT>' ] ) ) <EOL> def test_selection ( data ) : <EOL> assert eq ( compute ( t [ t [ '<STR_LIT>' ] == <NUM_LIT:100> ] , data ) , x [ x [ '<STR_LIT>' ] == <NUM_LIT:0> ] ) <EOL> assert eq ( compute ( t [ t [ '<STR_LIT>' ] < <NUM_LIT:0> ] , data ) , x [ x [ '<STR_LIT>' ] < <NUM_LIT:0> ] ) <EOL> def test_arithmetic ( data ) : <EOL> assert eq ( compute ( t [ '<STR_LIT>' ] + t [ '<STR_LIT:id>' ] , data ) , x [ '<STR_LIT>' ] + x [ '<STR_LIT:id>' ] ) <EOL> assert eq ( compute ( t [ '<STR_LIT>' ] * t [ '<STR_LIT:id>' ] , data ) , x [ '<STR_LIT>' ] * x [ '<STR_LIT:id>' ] ) <EOL> assert eq ( compute ( t [ '<STR_LIT>' ] % t [ '<STR_LIT:id>' ] , data ) , x [ '<STR_LIT>' ] % x [ '<STR_LIT:id>' ] ) <EOL> assert eq ( compute ( t [ '<STR_LIT>' ] + t [ '<STR_LIT:id>' ] + <NUM_LIT:3> , data ) , <EOL> x [ '<STR_LIT>' ] + x [ '<STR_LIT:id>' ] + <NUM_LIT:3> ) <EOL> def test_reductions ( data ) : <EOL> assert compute ( t [ '<STR_LIT>' ] . count ( ) , data ) == len ( x [ '<STR_LIT>' ] ) <EOL> assert compute ( t [ '<STR_LIT>' ] . sum ( ) , data ) == x [ '<STR_LIT>' ] . sum ( ) <EOL> assert compute ( t [ '<STR_LIT>' ] . mean ( ) , data ) == x [ '<STR_LIT>' ] . mean ( ) <EOL> assert compute ( t . amount [ <NUM_LIT:0> ] , data ) == x [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> assert compute ( t . amount [ - <NUM_LIT:1> ] , data ) == x [ '<STR_LIT>' ] [ - <NUM_LIT:1> ] <EOL> class TestTopLevelReductions ( object ) : <EOL> def test_count ( self , data ) : <EOL> from blaze import count <EOL> assert compute ( count ( t [ '<STR_LIT>' ] ) , data ) == len ( x [ '<STR_LIT>' ] ) <EOL> def test_sum ( self , data ) : <EOL> from blaze import sum <EOL> assert compute ( sum ( t [ '<STR_LIT>' ] ) , data ) == x [ '<STR_LIT>' ] . sum ( ) <EOL> def test_mean ( self , data ) : <EOL> from blaze import mean <EOL> assert compute ( mean ( t [ '<STR_LIT>' ] ) , data ) == x [ '<STR_LIT>' ] . mean ( ) <EOL> class TestFailingSort ( object ) : <EOL> """<STR_LIT>""" <EOL> def test_basic ( self , data ) : <EOL> with pytest . raises ( ValueError ) : <EOL> compute ( t . sort ( '<STR_LIT:id>' ) , data ) <EOL> @ xfail ( reason = '<STR_LIT>' ) <EOL> def test_multiple_columns ( self , data ) : <EOL> compute ( t . sort ( [ '<STR_LIT>' , '<STR_LIT:id>' ] ) , data ) <EOL> @ xfail ( reason = '<STR_LIT>' ) <EOL> def test_multiple_columns_sorted_data ( self , csi_data ) : <EOL> compute ( t . sort ( [ '<STR_LIT>' , '<STR_LIT:id>' ] ) , csi_data ) <EOL> class TestCSISort ( object ) : <EOL> def test_basic ( self , csi_data ) : <EOL> assert eq ( compute ( t . sort ( '<STR_LIT>' ) , csi_data ) , <EOL> np . sort ( x , order = '<STR_LIT>' ) ) <EOL> assert eq ( compute ( t . sort ( '<STR_LIT:id>' ) , csi_data ) , <EOL> np . sort ( x , order = '<STR_LIT:id>' ) ) <EOL> def test_column_expr ( self , csi_data ) : <EOL> assert eq ( compute ( t . sort ( t . amount ) , csi_data ) , <EOL> np . sort ( x , order = '<STR_LIT>' ) ) <EOL> assert eq ( compute ( t . sort ( t . id ) , csi_data ) , <EOL> np . sort ( x , order = '<STR_LIT:id>' ) ) <EOL> def test_non_existent_column ( self , csi_data ) : <EOL> with pytest . raises ( AssertionError ) : <EOL> compute ( t . sort ( '<STR_LIT>' ) , csi_data ) <EOL> def test_ascending ( self , csi_data ) : <EOL> assert eq ( compute ( t . sort ( '<STR_LIT>' , ascending = False ) , csi_data ) , <EOL> np . sort ( x , order = '<STR_LIT>' ) [ : : - <NUM_LIT:1> ] ) <EOL> assert eq ( compute ( t . sort ( '<STR_LIT>' , ascending = False ) , csi_data ) , <EOL> np . sort ( x , order = '<STR_LIT>' ) [ : : - <NUM_LIT:1> ] ) <EOL> class TestIndexSort ( object ) : <EOL> """<STR_LIT>""" <EOL> @ xfail ( reason = '<STR_LIT>' ) <EOL> def test_basic ( self , idx_data ) : <EOL> compute ( t . sort ( '<STR_LIT>' ) , idx_data ) <EOL> @ xfail ( reason = '<STR_LIT>' ) <EOL> def test_ascending ( self , idx_data ) : <EOL> compute ( t . sort ( '<STR_LIT>' , ascending = False ) , idx_data ) <EOL> def test_head ( data ) : <EOL> assert eq ( compute ( t . head ( <NUM_LIT:2> ) , data ) , x [ : <NUM_LIT:2> ] ) <EOL> assert eq ( compute ( t . amount . head ( <NUM_LIT:2> ) , data ) , x [ '<STR_LIT>' ] [ : <NUM_LIT:2> ] ) <EOL> @ pytest . yield_fixture <EOL> def pyt ( ) : <EOL> tb = pytest . importorskip ( '<STR_LIT>' ) <EOL> fn = '<STR_LIT>' <EOL> f = tb . open_file ( fn , mode = '<STR_LIT:w>' ) <EOL> d = f . create_table ( '<STR_LIT:/>' , '<STR_LIT:test>' , x ) <EOL> yield d <EOL> d . close ( ) <EOL> f . close ( ) <EOL> try : <EOL> os . remove ( fn ) <EOL> except OSError : <EOL> pass <EOL> def test_drop ( pyt ) : <EOL> drop ( pyt ) <EOL> with pytest . raises ( tb . ClosedNodeError ) : <EOL> drop ( pyt ) <EOL> def test_create_index ( pyt ) : <EOL> create_index ( pyt , '<STR_LIT:id>' ) <EOL> assert '<STR_LIT:id>' in pyt . colindexes <EOL> def test_create_multiple_indexes ( pyt ) : <EOL> create_index ( pyt , [ '<STR_LIT:id>' , '<STR_LIT>' ] ) <EOL> assert len ( pyt . colindexes ) == <NUM_LIT:2> <EOL> assert '<STR_LIT:id>' in pyt . colindexes <EOL> assert '<STR_LIT>' in pyt . colindexes <EOL> def test_create_multiple_indexes_fails ( pyt ) : <EOL> with pytest . raises ( ValueError ) : <EOL> create_index ( pyt , [ '<STR_LIT:id>' , '<STR_LIT>' ] ) <EOL> with pytest . raises ( ValueError ) : <EOL> create_index ( pyt , [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] ) <EOL> def test_create_index_fails ( pyt ) : <EOL> with pytest . raises ( AttributeError ) : <EOL> create_index ( pyt , '<STR_LIT>' ) <EOL> def test_nrows ( ) : <EOL> assert compute ( t . nrows , x ) == len ( x ) <EOL> def test_nelements ( ) : <EOL> assert compute ( t . nelements ( axis = <NUM_LIT:0> ) , x ) == len ( x ) <EOL> assert compute ( t . nelements ( ) , x ) == len ( x ) </s>
