<s> from __future__ import absolute_import , division , print_function <EOL> import pytest <EOL> h5py = pytest . importorskip ( '<STR_LIT>' ) <EOL> import os <EOL> import sys <EOL> from distutils . version import LooseVersion <EOL> import h5py <EOL> from contextlib import contextmanager <EOL> from odo . backends . h5py import append , create , resource , discover , convert <EOL> from odo . backends . h5py import unicode_dtype <EOL> from odo . utils import tmpfile , ignoring <EOL> from odo . chunks import chunks <EOL> from odo import into , append , convert , discover , drop , odo <EOL> import datashape <EOL> import h5py <EOL> import numpy as np <EOL> @ contextmanager <EOL> def file ( x ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> f = h5py . File ( fn ) <EOL> data = f . create_dataset ( '<STR_LIT>' , data = x , chunks = True , <EOL> maxshape = ( None , ) + x . shape [ <NUM_LIT:1> : ] ) <EOL> try : <EOL> yield fn , f , data <EOL> finally : <EOL> with ignoring ( Exception ) : <EOL> f . close ( ) <EOL> x = np . ones ( ( <NUM_LIT:2> , <NUM_LIT:3> ) , dtype = '<STR_LIT>' ) <EOL> def test_drop_group ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> f = h5py . File ( fn ) <EOL> try : <EOL> f . create_dataset ( '<STR_LIT>' , data = x , chunks = True , <EOL> maxshape = ( None , ) + x . shape [ <NUM_LIT:1> : ] ) <EOL> drop ( f [ '<STR_LIT>' ] ) <EOL> assert '<STR_LIT>' not in f . keys ( ) <EOL> finally : <EOL> with ignoring ( Exception ) : <EOL> f . close ( ) <EOL> def test_drop_dataset ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> f = h5py . File ( fn ) <EOL> try : <EOL> data = f . create_dataset ( '<STR_LIT>' , data = x , chunks = True , <EOL> maxshape = ( None , ) + x . shape [ <NUM_LIT:1> : ] ) <EOL> drop ( data ) <EOL> assert '<STR_LIT>' not in f . keys ( ) <EOL> finally : <EOL> with ignoring ( Exception ) : <EOL> f . close ( ) <EOL> def test_drop_file ( ) : <EOL> with file ( x ) as ( fn , f , data ) : <EOL> drop ( f ) <EOL> assert not os . path . exists ( fn ) <EOL> def test_discover ( ) : <EOL> with file ( x ) as ( fn , f , dset ) : <EOL> assert str ( discover ( dset ) ) == str ( discover ( x ) ) <EOL> assert str ( discover ( f ) ) == str ( discover ( { '<STR_LIT:data>' : x } ) ) <EOL> two_point_five_and_windows_py3 = pytest . mark . skipif ( sys . platform == '<STR_LIT:win32>' and <EOL> h5py . __version__ == LooseVersion ( '<STR_LIT>' ) and <EOL> sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:3> , <EOL> reason = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> @ two_point_five_and_windows_py3 <EOL> def test_discover_on_data_with_object_in_record_name ( ) : <EOL> data = np . array ( [ ( u'<STR_LIT:a>' , <NUM_LIT:1> ) , ( u'<STR_LIT:b>' , <NUM_LIT:2> ) ] , dtype = [ ( '<STR_LIT>' , <EOL> unicode_dtype ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> f = h5py . File ( fn ) <EOL> try : <EOL> f . create_dataset ( '<STR_LIT>' , data = data ) <EOL> except : <EOL> raise <EOL> else : <EOL> assert ( discover ( f [ '<STR_LIT:data>' ] ) == <EOL> datashape . dshape ( '<STR_LIT>' ) ) <EOL> finally : <EOL> with ignoring ( Exception ) : <EOL> f . close ( ) <EOL> def eq ( a , b ) : <EOL> c = a == b <EOL> if isinstance ( c , np . ndarray ) : <EOL> c = c . all ( ) <EOL> return c <EOL> def test_append ( ) : <EOL> with file ( x ) as ( fn , f , dset ) : <EOL> append ( dset , x ) <EOL> assert eq ( dset [ : ] , np . concatenate ( [ x , x ] ) ) <EOL> def test_append_with_uri ( ) : <EOL> with file ( x ) as ( fn , f , dset ) : <EOL> result = odo ( dset , '<STR_LIT>' % ( fn , dset . name ) ) <EOL> assert eq ( result [ : ] , np . concatenate ( [ x , x ] ) ) <EOL> def test_into_resource ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> d = into ( fn + '<STR_LIT>' , x ) <EOL> try : <EOL> assert d . shape == x . shape <EOL> assert eq ( d [ : ] , x [ : ] ) <EOL> finally : <EOL> d . file . close ( ) <EOL> def test_numpy ( ) : <EOL> with file ( x ) as ( fn , f , dset ) : <EOL> assert eq ( convert ( np . ndarray , dset ) , x ) <EOL> def test_chunks ( ) : <EOL> with file ( x ) as ( fn , f , dset ) : <EOL> c = convert ( chunks ( np . ndarray ) , dset ) <EOL> assert eq ( convert ( np . ndarray , c ) , x ) <EOL> def test_append_chunks ( ) : <EOL> with file ( x ) as ( fn , f , dset ) : <EOL> append ( dset , chunks ( np . ndarray ) ( [ x , x ] ) ) <EOL> assert len ( dset ) == len ( x ) * <NUM_LIT:3> <EOL> def test_create ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> ds = datashape . dshape ( '<STR_LIT>' ) <EOL> f = create ( h5py . File , dshape = '<STR_LIT>' , path = fn ) <EOL> try : <EOL> assert isinstance ( f , h5py . File ) <EOL> assert f . filename == fn <EOL> assert discover ( f ) == ds <EOL> finally : <EOL> f . close ( ) <EOL> def test_create_partially_present_dataset ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> ds1 = datashape . dshape ( '<STR_LIT>' ) <EOL> f = create ( h5py . File , dshape = ds1 , path = fn ) <EOL> ds2 = datashape . dshape ( '<STR_LIT>' ) <EOL> f2 = create ( h5py . File , dshape = ds2 , path = fn ) <EOL> try : <EOL> assert f . filename == f2 . filename <EOL> assert list ( f . keys ( ) ) == list ( f2 . keys ( ) ) <EOL> assert f [ '<STR_LIT:y>' ] . dtype == '<STR_LIT>' <EOL> finally : <EOL> f . close ( ) <EOL> f2 . close ( ) <EOL> def test_resource ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> ds = datashape . dshape ( '<STR_LIT>' ) <EOL> r = resource ( fn , dshape = ds ) <EOL> r2 = resource ( fn + '<STR_LIT>' ) <EOL> try : <EOL> assert isinstance ( r , h5py . File ) <EOL> assert discover ( r ) == ds <EOL> assert isinstance ( r2 , h5py . Dataset ) <EOL> finally : <EOL> r . close ( ) <EOL> r2 . file . close ( ) <EOL> def test_resource_with_datapath ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> ds = datashape . dshape ( '<STR_LIT>' ) <EOL> r = resource ( fn + '<STR_LIT>' , dshape = ds ) <EOL> try : <EOL> assert isinstance ( r , h5py . Dataset ) <EOL> assert discover ( r ) == ds <EOL> assert r . file . filename == fn <EOL> assert r . file [ '<STR_LIT>' ] == r <EOL> finally : <EOL> r . file . close ( ) <EOL> def test_resource_with_variable_length ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> ds = datashape . dshape ( '<STR_LIT>' ) <EOL> r = resource ( fn + '<STR_LIT>' , dshape = ds ) <EOL> try : <EOL> assert r . shape == ( <NUM_LIT:0> , <NUM_LIT:4> ) <EOL> finally : <EOL> r . file . close ( ) <EOL> @ two_point_five_and_windows_py3 <EOL> def test_resource_with_option_types ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> ds = datashape . dshape ( '<STR_LIT>' ) <EOL> r = resource ( fn + '<STR_LIT>' , dshape = ds ) <EOL> try : <EOL> assert r . shape == ( <NUM_LIT:4> , ) <EOL> assert r . dtype == [ ( '<STR_LIT:name>' , '<STR_LIT:O>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> finally : <EOL> r . file . close ( ) <EOL> def test_resource_with_h5py_protocol ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> assert isinstance ( resource ( '<STR_LIT>' + fn ) , h5py . File ) <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> assert isinstance ( resource ( '<STR_LIT>' + fn ) , h5py . File ) <EOL> def test_copy_with_into ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> dset = into ( fn + '<STR_LIT>' , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> try : <EOL> assert dset . shape == ( <NUM_LIT:3> , ) <EOL> assert eq ( dset [ : ] , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> finally : <EOL> dset . file . close ( ) <EOL> @ two_point_five_and_windows_py3 <EOL> def test_varlen_dtypes ( ) : <EOL> y = np . array ( [ ( '<STR_LIT>' , <NUM_LIT:100> ) , ( '<STR_LIT>' , <NUM_LIT:200> ) ] , <EOL> dtype = [ ( '<STR_LIT:name>' , '<STR_LIT:O>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> dset = into ( fn + '<STR_LIT>' , y ) <EOL> try : <EOL> assert into ( list , dset ) == into ( list , dset ) <EOL> finally : <EOL> dset . file . close ( ) <EOL> def test_resource_shape ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> r = resource ( fn + '<STR_LIT>' , dshape = '<STR_LIT>' ) <EOL> assert r . shape == ( <NUM_LIT:10> , ) <EOL> r . file . close ( ) <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> r = resource ( fn + '<STR_LIT>' , dshape = '<STR_LIT>' ) <EOL> assert r . shape == ( <NUM_LIT:10> , <NUM_LIT:10> ) <EOL> r . file . close ( ) <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> r = resource ( fn + '<STR_LIT>' , dshape = '<STR_LIT>' ) <EOL> assert r . shape == ( <NUM_LIT:0> , <NUM_LIT:10> ) <EOL> r . file . close ( ) <EOL> @ pytest . mark . parametrize ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_resource_with_hdfstore_extension_works ( ext ) : <EOL> fn = '<STR_LIT>' % ext <EOL> with pytest . raises ( NotImplementedError ) : <EOL> odo ( fn , np . recarray ) <EOL> assert not os . path . exists ( fn ) </s>
