<s> from __future__ import unicode_literals <EOL> """<STR_LIT>""" <EOL> import unittest <EOL> from functools import partial <EOL> from bson import ObjectId <EOL> from gridfs . errors import FileExists , NoFile <EOL> from pymongo . errors import AutoReconnect , ConfigurationError <EOL> from pymongo . read_preferences import ReadPreference <EOL> from tornado import gen <EOL> from tornado . testing import gen_test <EOL> import motor <EOL> import test <EOL> from test . test_environment import db_password , db_user <EOL> from motor . motor_py3_compat import StringIO <EOL> from test . tornado_tests import MotorTest , MotorReplicaSetTestBase <EOL> class MotorGridfsTest ( 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 setUp ( self ) : <EOL> super ( MotorGridfsTest , self ) . setUp ( ) <EOL> self . fs = motor . MotorGridFS ( self . db ) <EOL> def tearDown ( self ) : <EOL> self . io_loop . run_sync ( self . _reset ) <EOL> super ( MotorGridfsTest , self ) . tearDown ( ) <EOL> @ gen_test <EOL> def test_gridfs ( self ) : <EOL> self . assertRaises ( TypeError , motor . MotorGridFS , "<STR_LIT:foo>" ) <EOL> self . assertRaises ( TypeError , motor . MotorGridFS , <NUM_LIT:5> ) <EOL> @ gen_test <EOL> def test_get_version ( self ) : <EOL> gin = yield self . fs . new_file ( _id = <NUM_LIT:1> , filename = '<STR_LIT:foo>' , field = <NUM_LIT:0> ) <EOL> yield gin . write ( b'<STR_LIT:a>' ) <EOL> yield gin . close ( ) <EOL> yield self . fs . put ( b'<STR_LIT>' , filename = '<STR_LIT:foo>' , field = <NUM_LIT:1> ) <EOL> yield self . fs . put ( b'<STR_LIT>' , filename = '<STR_LIT:foo>' , field = <NUM_LIT:2> ) <EOL> gout = yield self . fs . get_version ( '<STR_LIT:foo>' ) <EOL> self . assertEqual ( <NUM_LIT:2> , gout . field ) <EOL> gout = yield self . fs . get_version ( '<STR_LIT:foo>' , - <NUM_LIT:3> ) <EOL> self . assertEqual ( <NUM_LIT:0> , gout . field ) <EOL> gout = yield self . fs . get_last_version ( '<STR_LIT:foo>' ) <EOL> self . assertEqual ( <NUM_LIT:2> , gout . field ) <EOL> @ gen_test <EOL> def test_gridfs_callback ( self ) : <EOL> yield self . check_optional_callback ( self . fs . new_file ) <EOL> yield self . check_optional_callback ( partial ( self . fs . put , b'<STR_LIT:a>' ) ) <EOL> yield self . fs . put ( b'<STR_LIT:foo>' , _id = <NUM_LIT:1> , filename = '<STR_LIT:f>' ) <EOL> yield self . check_optional_callback ( self . fs . get , <NUM_LIT:1> ) <EOL> yield self . check_optional_callback ( self . fs . get_version , '<STR_LIT:f>' ) <EOL> yield self . check_optional_callback ( self . fs . get_last_version , '<STR_LIT:f>' ) <EOL> yield self . check_optional_callback ( partial ( self . fs . delete , <NUM_LIT:1> ) ) <EOL> yield self . check_optional_callback ( self . fs . list ) <EOL> yield self . check_optional_callback ( self . fs . exists ) <EOL> @ gen_test <EOL> def test_basic ( self ) : <EOL> oid = yield self . fs . put ( b"<STR_LIT>" ) <EOL> out = yield self . fs . get ( oid ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield out . read ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , ( yield self . db . fs . files . count ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , ( yield self . db . fs . chunks . count ( ) ) ) <EOL> yield self . fs . delete ( oid ) <EOL> with self . assertRaises ( NoFile ) : <EOL> yield self . fs . get ( oid ) <EOL> self . assertEqual ( <NUM_LIT:0> , ( yield self . db . fs . files . count ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , ( yield self . db . fs . chunks . count ( ) ) ) <EOL> with self . assertRaises ( NoFile ) : <EOL> yield self . fs . get ( "<STR_LIT:foo>" ) <EOL> self . assertEqual ( <EOL> "<STR_LIT:foo>" , ( yield self . fs . put ( b"<STR_LIT>" , _id = "<STR_LIT:foo>" ) ) ) <EOL> gridout = yield self . fs . get ( "<STR_LIT:foo>" ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield gridout . read ( ) ) ) <EOL> @ gen_test <EOL> def test_list ( self ) : <EOL> self . assertEqual ( [ ] , ( yield self . fs . list ( ) ) ) <EOL> yield self . fs . put ( b"<STR_LIT>" ) <EOL> self . assertEqual ( [ ] , ( yield self . fs . list ( ) ) ) <EOL> yield self . fs . put ( b"<STR_LIT>" , filename = "<STR_LIT>" ) <EOL> yield self . fs . put ( b"<STR_LIT:foo>" , filename = "<STR_LIT:test>" ) <EOL> yield self . fs . put ( b"<STR_LIT>" , filename = "<STR_LIT>" ) <EOL> self . assertEqual ( set ( [ "<STR_LIT>" , "<STR_LIT:test>" , "<STR_LIT>" ] ) , <EOL> set ( ( yield self . fs . list ( ) ) ) ) <EOL> @ gen_test <EOL> def test_alt_collection ( self ) : <EOL> db = self . db <EOL> alt = motor . MotorGridFS ( db , '<STR_LIT>' ) <EOL> oid = yield alt . put ( b"<STR_LIT>" ) <EOL> gridout = yield alt . get ( oid ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield gridout . read ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , ( yield self . db . alt . files . count ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , ( yield self . db . alt . chunks . count ( ) ) ) <EOL> yield alt . delete ( oid ) <EOL> with self . assertRaises ( NoFile ) : <EOL> yield alt . get ( oid ) <EOL> self . assertEqual ( <NUM_LIT:0> , ( yield self . db . alt . files . count ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , ( yield self . db . alt . chunks . count ( ) ) ) <EOL> with self . assertRaises ( NoFile ) : <EOL> yield alt . get ( "<STR_LIT:foo>" ) <EOL> oid = yield alt . put ( b"<STR_LIT>" , _id = "<STR_LIT:foo>" ) <EOL> self . assertEqual ( "<STR_LIT:foo>" , oid ) <EOL> gridout = yield alt . get ( "<STR_LIT:foo>" ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield gridout . read ( ) ) ) <EOL> yield alt . put ( b"<STR_LIT>" , filename = "<STR_LIT>" ) <EOL> yield alt . put ( b"<STR_LIT:foo>" , filename = "<STR_LIT:test>" ) <EOL> yield alt . put ( b"<STR_LIT>" , filename = "<STR_LIT>" ) <EOL> self . assertEqual ( set ( [ "<STR_LIT>" , "<STR_LIT:test>" , "<STR_LIT>" ] ) , <EOL> set ( ( yield alt . list ( ) ) ) ) <EOL> @ gen_test <EOL> def test_put_filelike ( self ) : <EOL> oid = yield self . fs . put ( StringIO ( b"<STR_LIT>" ) , chunk_size = <NUM_LIT:1> ) <EOL> self . assertEqual ( <NUM_LIT:11> , ( yield self . cx . motor_test . fs . chunks . count ( ) ) ) <EOL> gridout = yield self . fs . get ( oid ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield gridout . read ( ) ) ) <EOL> @ gen_test <EOL> def test_put_callback ( self ) : <EOL> ( oid , error ) , _ = yield gen . Task ( self . fs . put , b"<STR_LIT:hello>" ) <EOL> self . assertTrue ( isinstance ( oid , ObjectId ) ) <EOL> self . assertEqual ( None , error ) <EOL> ( result , error ) , _ = yield gen . Task ( self . fs . put , b"<STR_LIT:hello>" , _id = oid ) <EOL> self . assertEqual ( None , result ) <EOL> self . assertTrue ( isinstance ( error , FileExists ) ) <EOL> @ gen_test <EOL> def test_put_duplicate ( self ) : <EOL> oid = yield self . fs . put ( b"<STR_LIT:hello>" ) <EOL> with self . assertRaises ( FileExists ) : <EOL> yield self . fs . put ( b"<STR_LIT>" , _id = oid ) <EOL> @ gen_test <EOL> def test_put_kwargs ( self ) : <EOL> oid = yield self . fs . put ( b"<STR_LIT:hello>" , foo = '<STR_LIT:bar>' , w = <NUM_LIT:0> ) <EOL> gridout = yield self . fs . get ( oid ) <EOL> self . assertEqual ( '<STR_LIT:bar>' , gridout . foo ) <EOL> self . assertEqual ( <NUM_LIT:0> , gridout . w ) <EOL> @ gen_test <EOL> def test_put_unacknowledged ( self ) : <EOL> client = self . motor_client ( w = <NUM_LIT:0> ) <EOL> fs = motor . MotorGridFS ( client . motor_test ) <EOL> with self . assertRaises ( ConfigurationError ) : <EOL> yield fs . put ( b"<STR_LIT:hello>" ) <EOL> client . close ( ) <EOL> @ gen_test <EOL> def test_gridfs_find ( self ) : <EOL> yield self . fs . put ( b"<STR_LIT>" , filename = "<STR_LIT>" ) <EOL> yield self . fs . put ( b"<STR_LIT>" , filename = "<STR_LIT>" ) <EOL> yield self . fs . put ( b"<STR_LIT>" , filename = "<STR_LIT>" ) <EOL> yield self . fs . put ( b"<STR_LIT>" , filename = "<STR_LIT>" ) <EOL> cursor = self . fs . find ( ) . sort ( "<STR_LIT>" , - <NUM_LIT:1> ) . skip ( <NUM_LIT:1> ) . limit ( <NUM_LIT:2> ) <EOL> self . assertTrue ( ( yield cursor . fetch_next ) ) <EOL> grid_out = cursor . next_object ( ) <EOL> self . assertTrue ( isinstance ( grid_out , motor . MotorGridOut ) ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield grid_out . read ( ) ) ) <EOL> cursor . rewind ( ) <EOL> self . assertTrue ( ( yield cursor . fetch_next ) ) <EOL> grid_out = cursor . next_object ( ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield grid_out . read ( ) ) ) <EOL> self . assertTrue ( ( yield cursor . fetch_next ) ) <EOL> grid_out = cursor . next_object ( ) <EOL> self . assertEqual ( b"<STR_LIT>" , ( yield grid_out . read ( ) ) ) <EOL> self . assertFalse ( ( yield cursor . fetch_next ) ) <EOL> self . assertEqual ( None , cursor . next_object ( ) ) <EOL> self . assertRaises ( TypeError , self . fs . find , { } , { "<STR_LIT>" : True } ) <EOL> class TestGridfsReplicaSet ( MotorReplicaSetTestBase ) : <EOL> @ gen_test <EOL> def test_gridfs_secondary ( self ) : <EOL> primary_host , primary_port = test . env . primary <EOL> primary = self . motor_client ( primary_host , primary_port ) <EOL> if test . env . auth : <EOL> yield primary . admin . authenticate ( test . db_user , test . db_password ) <EOL> secondary_host , secondary_port = test . env . secondaries [ <NUM_LIT:0> ] <EOL> secondary = self . motor_client ( <EOL> secondary_host , secondary_port , <EOL> read_preference = ReadPreference . SECONDARY ) <EOL> if test . env . auth : <EOL> yield secondary . admin . authenticate ( db_user , db_password ) <EOL> yield primary . motor_test . drop_collection ( "<STR_LIT>" ) <EOL> yield primary . motor_test . drop_collection ( "<STR_LIT>" ) <EOL> fs = motor . MotorGridFS ( secondary . motor_test ) <EOL> with self . assertRaises ( AutoReconnect ) : <EOL> yield fs . put ( b'<STR_LIT:foo>' ) <EOL> def tearDown ( self ) : <EOL> self . rsc . motor_test . drop_collection ( '<STR_LIT>' ) <EOL> self . rsc . motor_test . drop_collection ( '<STR_LIT>' ) <EOL> super ( TestGridfsReplicaSet , self ) . tearDown ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
