#NEWLINE# def create_test_method( type_ , op , extracond , func = None ) : #NEWLINE# #INDENT# sctype = sctype_from_type [ type_ ] #NEWLINE# condvars = { #NEWLINE# #INDENT# 'bound' : right_bound , #NEWLINE# 'lbound' : left_bound , #NEWLINE# 'rbound' : right_bound , #NEWLINE# 'func_bound' : func_bound , #NEWLINE# #UNINDENT# } #NEWLINE# for ( bname , bvalue ) in six . iteritems ( condvars ) : #NEWLINE# #INDENT# if ( type_ == 'string' ) : #NEWLINE# #INDENT# bvalue = ( str_format % bvalue ) #NEWLINE# #UNINDENT# bvalue = nxtype_from_type [ type_ ] ( bvalue ) #NEWLINE# condvars [ bname ] = colname #NEWLINE# #UNINDENT# colname = ( 'c_%s' % type_ ) #NEWLINE# ncolname = ( 'c_nested/%s' % colname ) #NEWLINE# if ( not op ) : #NEWLINE# #INDENT# cond = colname #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( op == '~' ) : #NEWLINE# #INDENT# cond = ( '~(%s)' % colname ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( ( op == '<' ) and ( func is None ) ) : #NEWLINE# #INDENT# cond = ( '%s %s %s' % ( colname , op , repr ( condvars [ 'bound' ] ) ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if isinstance ( op , tuple ) : #NEWLINE# #INDENT# cond = ( '(lbound %s %s) & (%s %s rbound)' % ( op [ 0 ] , colname , colname , op [ 1 ] ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( func is not None ) : #NEWLINE# #INDENT# cond = ( '%s(%s) %s func_bound' % ( func , colname , op ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# cond = ( '%s %s bound' % ( colname , op ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# if extracond : #NEWLINE# #INDENT# cond = ( '(%s) %s' % ( cond , extracond ) ) #NEWLINE# #UNINDENT# def ignore_skipped( oldmethod ) : #NEWLINE# #INDENT# @ functools . wraps ( oldmethod ) #NEWLINE# def newmethod( self , * args , **kwargs ) : #NEWLINE# #INDENT# self . _verboseHeader ( ) #NEWLINE# try : #NEWLINE# #INDENT# return oldmethod ( self , * args , ** kwargs ) #NEWLINE# #UNINDENT# except SilentlySkipTest as se : #NEWLINE# #INDENT# if se . args : #NEWLINE# #INDENT# msg = se . args [ 0 ] #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# msg = '<skipped>' #NEWLINE# #UNINDENT# common . verbosePrint ( ( '\nSkipped test: %s' % msg ) ) #NEWLINE# #UNINDENT# finally : #NEWLINE# #INDENT# common . verbosePrint ( '' ) #NEWLINE# #UNINDENT# #UNINDENT# return newmethod #NEWLINE# #UNINDENT# @ ignore_skipped #NEWLINE# def test_method( self ) : #NEWLINE# #INDENT# vprint ( ( '* Condition is ``%s``.' % cond ) ) #NEWLINE# pycond = cond #NEWLINE# for ( ptop , pyop ) in [ ( '&' , 'and' ) , ( '|' , 'or' ) , ( '~' , 'not' ) ] : #NEWLINE# #INDENT# pycond = pycond . replace ( ptop , pyop ) #NEWLINE# #UNINDENT# pycond = compile ( pycond , '<string>' , 'eval' ) #NEWLINE# table = self . table #NEWLINE# self . create_indexes ( colname , ncolname , 'c_idxextra' ) #NEWLINE# table_slice = dict ( start = 1 , stop = ( table . nrows - 5 ) , step = 3 ) #NEWLINE# ( rownos , fvalues ) = ( None , None ) #NEWLINE# for acolname in [ colname , ncolname ] : #NEWLINE# #INDENT# ( pyrownos , pyfvalues , pyvars ) = ( [ ] , [ ] , condvars . copy ( ) ) #NEWLINE# for row in table . iterrows ( ** table_slice ) : #NEWLINE# #INDENT# pyvars [ colname ] = row [ acolname ] #NEWLINE# pyvars [ 'c_extra' ] = row [ 'c_extra' ] #NEWLINE# pyvars [ 'c_idxextra' ] = row [ 'c_idxextra' ] #NEWLINE# try : #NEWLINE# #INDENT# isvalidrow = eval ( pycond , func_info , pyvars ) #NEWLINE# #UNINDENT# except TypeError : #NEWLINE# #INDENT# raise SilentlySkipTest ( 'The Python type does not support the operation.' ) #NEWLINE# #UNINDENT# if isvalidrow : #NEWLINE# #INDENT# pyrownos . append ( row . nrow ) #NEWLINE# pyfvalues . append ( row [ acolname ] ) #NEWLINE# #UNINDENT# #UNINDENT# pyrownos = numpy . array ( pyrownos ) #NEWLINE# pyfvalues = numpy . array ( pyfvalues , dtype = sctype ) #NEWLINE# pyfvalues . sort ( ) #NEWLINE# vprint ( ( '* %d rows selected by Python from ``%s``.' % ( len ( pyrownos ) , acolname ) ) ) #NEWLINE# if ( rownos is None ) : #NEWLINE# #INDENT# rownos = pyrownos #NEWLINE# fvalues = pyfvalues #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# self . assertTrue ( numpy . all ( ( pyrownos == rownos ) ) ) #NEWLINE# self . assertTrue ( numpy . all ( ( pyfvalues == fvalues ) ) ) #NEWLINE# #UNINDENT# ptvars = condvars . copy ( ) #NEWLINE# ptvars [ colname ] = table . colinstances [ acolname ] #NEWLINE# ptvars [ 'c_extra' ] = table . colinstances [ 'c_extra' ] #NEWLINE# ptvars [ 'c_idxextra' ] = table . colinstances [ 'c_idxextra' ] #NEWLINE# try : #NEWLINE# #INDENT# isidxq = table . will_query_use_indexing ( cond , ptvars ) #NEWLINE# ptrownos = [ table . get_where_list ( cond , condvars , sort = True , ** table_slice ) for _ in range ( 2 ) ] #NEWLINE# ptfvalues = [ table . read_where ( cond , condvars , field = acolname , ** table_slice ) for _ in range ( 2 ) ] #NEWLINE# #UNINDENT# except TypeError as te : #NEWLINE# #INDENT# if self . condNotBoolean_re . search ( str ( te ) ) : #NEWLINE# #INDENT# raise SilentlySkipTest ( 'The condition is not boolean.' ) #NEWLINE# #UNINDENT# raise #NEWLINE# #UNINDENT# except NotImplementedError : #NEWLINE# #INDENT# raise SilentlySkipTest ( 'The PyTables type does not support the operation.' ) #NEWLINE# #UNINDENT# for ptfvals in ptfvalues : #NEWLINE# #INDENT# ptfvals . sort ( ) #NEWLINE# #UNINDENT# vprint ( ( '* %d rows selected by PyTables from ``%s``' % ( len ( ptrownos [ 0 ] ) , acolname ) ) , nonl = True ) #NEWLINE# vprint ( ( '(indexing: %s).' % [ 'no' , 'yes' ] [ bool ( isidxq ) ] ) ) #NEWLINE# self . assertTrue ( numpy . all ( ( ptrownos [ 0 ] == rownos ) ) ) #NEWLINE# self . assertTrue ( numpy . all ( ( ptfvalues [ 0 ] == fvalues ) ) ) #NEWLINE# self . assertTrue ( numpy . all ( ( ptrownos [ 0 ] == ptrownos [ 1 ] ) ) ) #NEWLINE# self . assertTrue ( numpy . all ( ( ptfvalues [ 0 ] == ptfvalues [ 1 ] ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# test_method . __doc__ = ( 'Testing ``%s``.' % cond ) #NEWLINE# return test_method