<s> from __future__ import unicode_literals <EOL> """<STR_LIT>""" <EOL> import datetime <EOL> import unittest <EOL> from functools import partial <EOL> from bson . objectid import ObjectId <EOL> from gridfs . errors import NoFile <EOL> from tornado import gen <EOL> from tornado . testing import gen_test <EOL> from pymongo . errors import InvalidOperation <EOL> import motor <EOL> from test import MockRequestHandler <EOL> from test . tornado_tests import MotorTest <EOL> class MotorGridFileTest ( MotorTest ) : <EOL> @ gen . coroutine <EOL> def _reset ( self ) : <EOL> yield self . db . drop_collection ( "<STR_LIT>" ) <EOL> yield self . db . drop_collection ( "<STR_LIT>" ) <EOL> yield self . db . drop_collection ( "<STR_LIT>" ) <EOL> yield self . db . drop_collection ( "<STR_LIT>" ) <EOL> def tearDown ( self ) : <EOL> self . io_loop . run_sync ( self . _reset ) <EOL> super ( MotorGridFileTest , self ) . tearDown ( ) <EOL> @ gen_test <EOL> def test_grid_in_callback ( self ) : <EOL> f = motor . MotorGridIn ( self . db . fs , filename = "<STR_LIT:test>" ) <EOL> yield self . check_optional_callback ( partial ( f . set , '<STR_LIT:name>' , '<STR_LIT:value>' ) ) <EOL> yield self . check_optional_callback ( partial ( f . write , b'<STR_LIT:a>' ) ) <EOL> yield self . check_optional_callback ( partial ( f . writelines , [ b'<STR_LIT:a>' ] ) ) <EOL> self . assertRaises ( TypeError , f . close , callback = '<STR_LIT:foo>' ) <EOL> self . assertRaises ( TypeError , f . close , callback = <NUM_LIT:1> ) <EOL> f . close ( callback = None ) <EOL> @ gen_test <EOL> def test_grid_out_callback ( self ) : <EOL> f = motor . MotorGridIn ( self . db . fs , filename = "<STR_LIT:test>" ) <EOL> yield f . close ( ) <EOL> g = motor . MotorGridOut ( self . db . fs , f . _id ) <EOL> yield self . check_optional_callback ( g . open ) <EOL> g = yield motor . MotorGridOut ( self . db . fs , f . _id ) . open ( ) <EOL> yield self . check_optional_callback ( g . read ) <EOL> yield self . check_optional_callback ( g . readline ) <EOL> @ gen_test <EOL> def test_attributes ( self ) : <EOL> f = motor . MotorGridIn ( <EOL> self . db . fs , <EOL> filename = "<STR_LIT:test>" , <EOL> foo = "<STR_LIT:bar>" , <EOL> content_type = "<STR_LIT:text>" ) <EOL> yield f . close ( ) <EOL> g = motor . MotorGridOut ( self . db . fs , f . _id ) <EOL> attr_names = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:filename>' , <EOL> '<STR_LIT:name>' , <EOL> '<STR_LIT:name>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> for attr_name in attr_names : <EOL> self . assertRaises ( InvalidOperation , getattr , g , attr_name ) <EOL> yield g . open ( ) <EOL> for attr_name in attr_names : <EOL> getattr ( g , attr_name ) <EOL> @ gen_test <EOL> def test_iteration ( self ) : <EOL> fs = motor . MotorGridFS ( self . db ) <EOL> _id = yield fs . put ( b'<STR_LIT:foo>' ) <EOL> g = motor . MotorGridOut ( self . db . fs , _id ) <EOL> self . assertRaises ( TypeError , iter , g ) <EOL> @ gen_test <EOL> def test_basic ( self ) : <EOL> f = motor . MotorGridIn ( self . db . fs , filename = "<STR_LIT:test>" ) <EOL> yield f . write ( b"<STR_LIT>" ) <EOL> yield f . close ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , ( yield self . db . fs . files . find ( ) . count ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , ( yield self . db . fs . chunks . find ( ) . count ( ) ) ) <EOL> g = motor . MotorGridOut ( self . db . fs , f . _id ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield g . read ( ) ) ) <EOL> f = motor . MotorGridIn ( self . db . fs , filename = "<STR_LIT:test>" ) <EOL> yield f . close ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , ( yield self . db . fs . files . find ( ) . count ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , ( yield self . db . fs . chunks . find ( ) . count ( ) ) ) <EOL> g = motor . MotorGridOut ( self . db . fs , f . _id ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield g . read ( ) ) ) <EOL> @ gen_test <EOL> def test_readchunk ( self ) : <EOL> in_data = b'<STR_LIT:a>' * <NUM_LIT:10> <EOL> f = motor . MotorGridIn ( self . db . fs , chunkSize = <NUM_LIT:3> ) <EOL> yield f . write ( in_data ) <EOL> yield f . close ( ) <EOL> g = motor . MotorGridOut ( self . db . fs , f . _id ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( ( yield g . readchunk ( ) ) ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ( yield g . read ( <NUM_LIT:2> ) ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( ( yield g . readchunk ( ) ) ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( ( yield g . read ( <NUM_LIT:3> ) ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( ( yield g . readchunk ( ) ) ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( ( yield g . readchunk ( ) ) ) ) <EOL> @ gen_test <EOL> def test_alternate_collection ( self ) : <EOL> yield self . db . alt . files . remove ( ) <EOL> yield self . db . alt . chunks . remove ( ) <EOL> f = motor . MotorGridIn ( self . db . alt ) <EOL> yield f . write ( b"<STR_LIT>" ) <EOL> yield f . close ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , ( yield self . db . alt . files . find ( ) . count ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , ( yield self . db . alt . chunks . find ( ) . count ( ) ) ) <EOL> g = motor . MotorGridOut ( self . db . alt , f . _id ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield g . read ( ) ) ) <EOL> self . assertEqual ( "<STR_LIT>" , g . md5 ) <EOL> @ gen_test <EOL> def test_grid_in_default_opts ( self ) : <EOL> self . assertRaises ( TypeError , motor . MotorGridIn , "<STR_LIT:foo>" ) <EOL> a = motor . MotorGridIn ( self . db . fs ) <EOL> self . assertTrue ( isinstance ( a . _id , ObjectId ) ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertEqual ( None , a . filename ) <EOL> def setter ( ) : <EOL> a . filename = "<STR_LIT>" <EOL> self . assertRaises ( AttributeError , setter ) <EOL> yield a . set ( "<STR_LIT:filename>" , "<STR_LIT>" ) <EOL> self . assertEqual ( "<STR_LIT>" , a . filename ) <EOL> self . assertEqual ( None , a . content_type ) <EOL> yield a . set ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEqual ( "<STR_LIT>" , a . content_type ) <EOL> self . assertRaises ( AttributeError , getattr , a , "<STR_LIT>" ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertEqual ( <NUM_LIT:255> * <NUM_LIT> , a . chunk_size ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertRaises ( AttributeError , getattr , a , "<STR_LIT>" ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertRaises ( AttributeError , getattr , a , "<STR_LIT>" ) <EOL> yield a . set ( "<STR_LIT>" , [ "<STR_LIT:foo>" ] ) <EOL> self . assertEqual ( [ "<STR_LIT:foo>" ] , a . aliases ) <EOL> self . assertRaises ( AttributeError , getattr , a , "<STR_LIT>" ) <EOL> yield a . set ( "<STR_LIT>" , { "<STR_LIT:foo>" : <NUM_LIT:1> } ) <EOL> self . assertEqual ( { "<STR_LIT:foo>" : <NUM_LIT:1> } , a . metadata ) <EOL> self . assertRaises ( AttributeError , getattr , a , "<STR_LIT>" ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> yield a . close ( ) <EOL> self . assertTrue ( isinstance ( a . _id , ObjectId ) ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertEqual ( "<STR_LIT>" , a . filename ) <EOL> self . assertEqual ( "<STR_LIT>" , a . content_type ) <EOL> self . assertEqual ( <NUM_LIT:0> , a . length ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertEqual ( <NUM_LIT:255> * <NUM_LIT> , a . chunk_size ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertTrue ( isinstance ( a . upload_date , datetime . datetime ) ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertEqual ( [ "<STR_LIT:foo>" ] , a . aliases ) <EOL> self . assertEqual ( { "<STR_LIT:foo>" : <NUM_LIT:1> } , a . metadata ) <EOL> self . assertEqual ( "<STR_LIT>" , a . md5 ) <EOL> self . assertRaises ( AttributeError , setattr , a , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> @ gen_test <EOL> def test_grid_in_custom_opts ( self ) : <EOL> self . assertRaises ( TypeError , motor . MotorGridIn , "<STR_LIT:foo>" ) <EOL> a = motor . MotorGridIn ( <EOL> self . db . fs , _id = <NUM_LIT:5> , filename = "<STR_LIT>" , <EOL> contentType = "<STR_LIT>" , chunkSize = <NUM_LIT:1000> , aliases = [ "<STR_LIT:foo>" ] , <EOL> metadata = { "<STR_LIT:foo>" : <NUM_LIT:1> , "<STR_LIT:bar>" : <NUM_LIT:2> } , bar = <NUM_LIT:3> , baz = "<STR_LIT:hello>" ) <EOL> self . assertEqual ( <NUM_LIT:5> , a . _id ) <EOL> self . assertEqual ( "<STR_LIT>" , a . filename ) <EOL> self . assertEqual ( "<STR_LIT>" , a . content_type ) <EOL> self . assertEqual ( <NUM_LIT:1000> , a . chunk_size ) <EOL> self . assertEqual ( [ "<STR_LIT:foo>" ] , a . aliases ) <EOL> self . assertEqual ( { "<STR_LIT:foo>" : <NUM_LIT:1> , "<STR_LIT:bar>" : <NUM_LIT:2> } , a . metadata ) <EOL> self . assertEqual ( <NUM_LIT:3> , a . bar ) <EOL> self . assertEqual ( "<STR_LIT:hello>" , a . baz ) <EOL> self . assertRaises ( AttributeError , getattr , a , "<STR_LIT>" ) <EOL> b = motor . MotorGridIn ( <EOL> self . db . fs , <EOL> content_type = "<STR_LIT>" , <EOL> chunk_size = <NUM_LIT:1000> , <EOL> baz = <NUM_LIT:100> ) <EOL> self . assertEqual ( "<STR_LIT>" , b . content_type ) <EOL> self . assertEqual ( <NUM_LIT:1000> , b . chunk_size ) <EOL> self . assertEqual ( <NUM_LIT:100> , b . baz ) <EOL> @ gen_test <EOL> def test_grid_out_default_opts ( self ) : <EOL> self . assertRaises ( TypeError , motor . MotorGridOut , "<STR_LIT:foo>" ) <EOL> gout = motor . MotorGridOut ( self . db . fs , <NUM_LIT:5> ) <EOL> with self . assertRaises ( NoFile ) : <EOL> yield gout . open ( ) <EOL> a = motor . MotorGridIn ( self . db . fs ) <EOL> yield a . close ( ) <EOL> b = yield motor . MotorGridOut ( self . db . fs , a . _id ) . open ( ) <EOL> self . assertEqual ( a . _id , b . _id ) <EOL> self . assertEqual ( <NUM_LIT:0> , b . length ) <EOL> self . assertEqual ( None , b . content_type ) <EOL> self . assertEqual ( <NUM_LIT:255> * <NUM_LIT> , b . chunk_size ) <EOL> self . assertTrue ( isinstance ( b . upload_date , datetime . datetime ) ) <EOL> self . assertEqual ( None , b . aliases ) <EOL> self . assertEqual ( None , b . metadata ) <EOL> self . assertEqual ( "<STR_LIT>" , b . md5 ) <EOL> @ gen_test <EOL> def test_grid_out_custom_opts ( self ) : <EOL> one = motor . MotorGridIn ( <EOL> self . db . fs , _id = <NUM_LIT:5> , filename = "<STR_LIT>" , <EOL> contentType = "<STR_LIT>" , chunkSize = <NUM_LIT:1000> , aliases = [ "<STR_LIT:foo>" ] , <EOL> metadata = { "<STR_LIT:foo>" : <NUM_LIT:1> , "<STR_LIT:bar>" : <NUM_LIT:2> } , bar = <NUM_LIT:3> , baz = "<STR_LIT:hello>" ) <EOL> yield one . write ( b"<STR_LIT>" ) <EOL> yield one . close ( ) <EOL> two = yield motor . MotorGridOut ( self . db . fs , <NUM_LIT:5> ) . open ( ) <EOL> self . assertEqual ( <NUM_LIT:5> , two . _id ) <EOL> self . assertEqual ( <NUM_LIT:11> , two . length ) <EOL> self . assertEqual ( "<STR_LIT>" , two . content_type ) <EOL> self . assertEqual ( <NUM_LIT:1000> , two . chunk_size ) <EOL> self . assertTrue ( isinstance ( two . upload_date , datetime . datetime ) ) <EOL> self . assertEqual ( [ "<STR_LIT:foo>" ] , two . aliases ) <EOL> self . assertEqual ( { "<STR_LIT:foo>" : <NUM_LIT:1> , "<STR_LIT:bar>" : <NUM_LIT:2> } , two . metadata ) <EOL> self . assertEqual ( <NUM_LIT:3> , two . bar ) <EOL> self . assertEqual ( "<STR_LIT>" , two . md5 ) <EOL> @ gen_test <EOL> def test_grid_out_file_document ( self ) : <EOL> one = motor . MotorGridIn ( self . db . fs ) <EOL> yield one . write ( b"<STR_LIT>" ) <EOL> yield one . close ( ) <EOL> file_document = yield self . db . fs . files . find_one ( ) <EOL> two = motor . MotorGridOut ( <EOL> self . db . fs , file_document = file_document ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield two . read ( ) ) ) <EOL> file_document = yield self . db . fs . files . find_one ( ) <EOL> three = motor . MotorGridOut ( self . db . fs , <NUM_LIT:5> , file_document ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield three . read ( ) ) ) <EOL> gridout = motor . MotorGridOut ( self . db . fs , file_document = { } ) <EOL> with self . assertRaises ( NoFile ) : <EOL> yield gridout . open ( ) <EOL> @ gen_test <EOL> def test_write_file_like ( self ) : <EOL> one = motor . MotorGridIn ( self . db . fs ) <EOL> yield one . write ( b"<STR_LIT>" ) <EOL> yield one . close ( ) <EOL> two = motor . MotorGridOut ( self . db . fs , one . _id ) <EOL> three = motor . MotorGridIn ( self . db . fs ) <EOL> yield three . write ( two ) <EOL> yield three . close ( ) <EOL> four = motor . MotorGridOut ( self . db . fs , three . _id ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield four . read ( ) ) ) <EOL> @ gen_test <EOL> def test_set_after_close ( self ) : <EOL> f = motor . MotorGridIn ( self . db . fs , _id = "<STR_LIT:foo>" , bar = "<STR_LIT>" ) <EOL> self . assertEqual ( "<STR_LIT:foo>" , f . _id ) <EOL> self . assertEqual ( "<STR_LIT>" , f . bar ) <EOL> self . assertRaises ( AttributeError , getattr , f , "<STR_LIT>" ) <EOL> self . assertRaises ( AttributeError , getattr , f , "<STR_LIT>" ) <EOL> self . assertRaises ( AttributeError , setattr , f , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> f . bar = "<STR_LIT:foo>" <EOL> f . baz = <NUM_LIT:5> <EOL> self . assertEqual ( "<STR_LIT:foo>" , f . bar ) <EOL> self . assertEqual ( <NUM_LIT:5> , f . baz ) <EOL> self . assertRaises ( AttributeError , getattr , f , "<STR_LIT>" ) <EOL> yield f . close ( ) <EOL> self . assertEqual ( "<STR_LIT:foo>" , f . _id ) <EOL> self . assertEqual ( "<STR_LIT:foo>" , f . bar ) <EOL> self . assertEqual ( <NUM_LIT:5> , f . baz ) <EOL> self . assertTrue ( f . uploadDate ) <EOL> self . assertRaises ( AttributeError , setattr , f , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> yield f . set ( "<STR_LIT:bar>" , "<STR_LIT:a>" ) <EOL> yield f . set ( "<STR_LIT>" , "<STR_LIT:b>" ) <EOL> self . assertRaises ( AttributeError , setattr , f , "<STR_LIT>" , <NUM_LIT:5> ) <EOL> g = yield motor . MotorGridOut ( self . db . fs , f . _id ) . open ( ) <EOL> self . assertEqual ( "<STR_LIT:a>" , g . bar ) <EOL> self . assertEqual ( "<STR_LIT:b>" , g . baz ) <EOL> @ gen_test <EOL> def test_stream_to_handler ( self ) : <EOL> fs = motor . MotorGridFS ( self . db ) <EOL> for content_length in ( <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:100> , <NUM_LIT:100> * <NUM_LIT:1000> ) : <EOL> _id = yield fs . put ( b'<STR_LIT:a>' * content_length ) <EOL> gridout = yield fs . get ( _id ) <EOL> handler = MockRequestHandler ( ) <EOL> yield gridout . stream_to_handler ( handler ) <EOL> self . assertEqual ( content_length , handler . n_written ) <EOL> yield fs . delete ( _id ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
