<s> from __future__ import print_function <EOL> import pytest <EOL> import sys <EOL> pytestmark = pytest . mark . skipif ( sys . platform == '<STR_LIT:win32>' , <EOL> reason = '<STR_LIT>' ) <EOL> boto = pytest . importorskip ( '<STR_LIT>' ) <EOL> import os <EOL> import itertools <EOL> import json <EOL> from contextlib import contextmanager , closing <EOL> import datashape <EOL> from datashape import string , float64 , int64 <EOL> from datashape . util . testing import assert_dshape_equal <EOL> import pandas as pd <EOL> import pandas . util . testing as tm <EOL> from odo import into , resource , S3 , discover , CSV , drop , append , odo <EOL> from odo . backends . aws import get_s3_connection <EOL> from odo . utils import tmpfile <EOL> from odo . compatibility import urlopen <EOL> from boto . exception import S3ResponseError , NoAuthHandlerFound <EOL> tips_uri = '<STR_LIT>' <EOL> df = pd . DataFrame ( { <EOL> '<STR_LIT:a>' : list ( '<STR_LIT:abc>' ) , <EOL> '<STR_LIT:b>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , <EOL> '<STR_LIT:c>' : [ <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ] <EOL> } ) [ [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] ] <EOL> js = pd . io . json . loads ( pd . io . json . dumps ( df , orient = '<STR_LIT>' ) ) <EOL> is_authorized = False <EOL> tried = False <EOL> with closing ( urlopen ( '<STR_LIT>' ) ) as url : <EOL> public_ip = json . loads ( url . read ( ) . decode ( ) ) [ '<STR_LIT>' ] <EOL> cidrip = public_ip + '<STR_LIT>' <EOL> @ pytest . yield_fixture <EOL> def tmpcsv ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> with open ( fn , mode = '<STR_LIT:w>' ) as f : <EOL> df . to_csv ( f , index = False ) <EOL> yield fn <EOL> @ contextmanager <EOL> def s3_bucket ( extension ) : <EOL> with conn ( ) : <EOL> b = '<STR_LIT>' % ( test_bucket_name , next ( _tmps ) , extension ) <EOL> try : <EOL> yield b <EOL> finally : <EOL> drop ( resource ( b ) ) <EOL> @ contextmanager <EOL> def conn ( ) : <EOL> try : <EOL> conn = get_s3_connection ( ) <EOL> except S3ResponseError : <EOL> pytest . skip ( '<STR_LIT>' ) <EOL> else : <EOL> try : <EOL> grants = conn . get_bucket ( test_bucket_name ) . get_acl ( ) . acl . grants <EOL> except S3ResponseError : <EOL> pytest . skip ( '<STR_LIT>' % <EOL> test_bucket_name ) <EOL> else : <EOL> if not any ( g . permission == '<STR_LIT>' or <EOL> g . permission == '<STR_LIT>' for g in grants ) : <EOL> pytest . skip ( '<STR_LIT>' % <EOL> test_bucket_name ) <EOL> else : <EOL> yield conn <EOL> test_bucket_name = '<STR_LIT>' <EOL> _tmps = ( '<STR_LIT>' % i for i in itertools . count ( ) ) <EOL> def test_s3_resource ( ) : <EOL> csv = resource ( tips_uri ) <EOL> assert isinstance ( csv , S3 ( CSV ) ) <EOL> def test_s3_discover ( ) : <EOL> csv = resource ( tips_uri ) <EOL> assert isinstance ( discover ( csv ) , datashape . DataShape ) <EOL> def test_s3_to_local_csv ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> csv = into ( fn , tips_uri ) <EOL> path = os . path . abspath ( csv . path ) <EOL> assert os . path . exists ( path ) <EOL> def test_csv_to_s3_append ( ) : <EOL> df = tm . makeMixedDataFrame ( ) <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> with s3_bucket ( '<STR_LIT>' ) as b : <EOL> s3 = resource ( b ) <EOL> df . to_csv ( fn , index = False ) <EOL> append ( s3 , CSV ( fn ) ) <EOL> result = into ( pd . DataFrame , s3 ) <EOL> tm . assert_frame_equal ( df , result ) <EOL> def test_csv_to_s3_into ( ) : <EOL> df = tm . makeMixedDataFrame ( ) <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> with s3_bucket ( '<STR_LIT>' ) as b : <EOL> df . to_csv ( fn , index = False ) <EOL> s3 = into ( b , CSV ( fn ) ) <EOL> result = into ( pd . DataFrame , s3 ) <EOL> tm . assert_frame_equal ( df , result ) <EOL> def test_frame_to_s3_to_frame ( ) : <EOL> with s3_bucket ( '<STR_LIT>' ) as b : <EOL> s3_csv = into ( b , df ) <EOL> result = into ( pd . DataFrame , s3_csv ) <EOL> tm . assert_frame_equal ( result , df ) <EOL> def test_textfile_to_s3 ( ) : <EOL> text = '<STR_LIT>' <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> with s3_bucket ( '<STR_LIT>' ) as b : <EOL> with open ( fn , mode = '<STR_LIT:w>' ) as f : <EOL> f . write ( os . linesep . join ( text . split ( ) ) ) <EOL> result = into ( b , resource ( fn ) ) <EOL> assert discover ( result ) == datashape . dshape ( '<STR_LIT>' ) <EOL> def test_jsonlines_to_s3 ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> with open ( fn , mode = '<STR_LIT:w>' ) as f : <EOL> for row in js : <EOL> f . write ( pd . io . json . dumps ( row ) ) <EOL> f . write ( os . linesep ) <EOL> with s3_bucket ( '<STR_LIT>' ) as b : <EOL> result = into ( b , resource ( fn ) ) <EOL> assert discover ( result ) == discover ( js ) <EOL> def test_s3_jsonlines_discover ( ) : <EOL> json_dshape = discover ( resource ( '<STR_LIT>' ) ) <EOL> names = list ( map ( str , sorted ( json_dshape . measure . names ) ) ) <EOL> assert names == [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:size>' , '<STR_LIT>' , '<STR_LIT:time>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> types = [ json_dshape . measure [ name ] for name in names ] <EOL> assert types == [ string , string , int64 , string , string , float64 , float64 ] <EOL> def test_s3_csv_discover ( ) : <EOL> result = discover ( resource ( '<STR_LIT>' ) ) <EOL> expected = datashape . dshape ( """<STR_LIT>""" ) <EOL> assert_dshape_equal ( result , expected ) <EOL> def test_s3_gz_csv_discover ( ) : <EOL> result = discover ( S3 ( CSV ) ( '<STR_LIT>' ) ) <EOL> expected = datashape . dshape ( """<STR_LIT>""" ) <EOL> assert_dshape_equal ( result , expected ) <EOL> def test_s3_to_sqlite ( ) : <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> tb = into ( '<STR_LIT>' % fn , tips_uri , <EOL> dshape = discover ( resource ( tips_uri ) ) ) <EOL> lhs = into ( list , tb ) <EOL> assert lhs == into ( list , tips_uri ) <EOL> def test_csv_to_s3__using_multipart_upload ( ) : <EOL> df = pd . DataFrame ( { '<STR_LIT:a>' : [ "<STR_LIT:*>" * <NUM_LIT:5> * <NUM_LIT> ** <NUM_LIT:2> ] } ) <EOL> with tmpfile ( '<STR_LIT>' ) as fn : <EOL> with s3_bucket ( '<STR_LIT>' ) as b : <EOL> df . to_csv ( fn , index = False ) <EOL> s3 = into ( b , CSV ( fn ) , multipart = True ) <EOL> result = into ( pd . DataFrame , s3 ) <EOL> tm . assert_frame_equal ( df , result ) <EOL> @ pytest . mark . parametrize ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ <EOL> pytest . mark . xfail ( ( '<STR_LIT>' , '<STR_LIT>' ) , raises = NotImplementedError ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ] <EOL> ) <EOL> def test_chunks_of_s3 ( prefix , suffix ) : <EOL> uri = '<STR_LIT>' . format ( prefix , suffix ) <EOL> result = resource ( uri ) <EOL> assert len ( result . data ) == <NUM_LIT:2> <EOL> expected = odo ( tips_uri , pd . DataFrame ) <EOL> tm . assert_frame_equal ( odo ( result , pd . DataFrame ) , expected ) </s>
