<s> from datetime import datetime <EOL> import json <EOL> from bson import ObjectId <EOL> from flask . ext . pymongo import MongoClient <EOL> import eve <EOL> from eve . auth import BasicAuth <EOL> from eve . tests import TestBase <EOL> from eve . tests . test_settings import MONGO1_PASSWORD , MONGO1_USERNAME , MONGO1_DBNAME , MONGO_DBNAME <EOL> class TestMultiMongo ( TestBase ) : <EOL> def setUp ( self ) : <EOL> super ( TestMultiMongo , self ) . setUp ( ) <EOL> self . setupDB2 ( ) <EOL> schema = { <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT:string>' } , <EOL> '<STR_LIT:title>' : { '<STR_LIT:type>' : '<STR_LIT:string>' } , <EOL> } <EOL> settings = { <EOL> '<STR_LIT>' : schema , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . app . register_resource ( '<STR_LIT>' , settings ) <EOL> def tearDown ( self ) : <EOL> super ( TestMultiMongo , self ) . tearDown ( ) <EOL> self . dropDB2 ( ) <EOL> def setupDB2 ( self ) : <EOL> self . connection = MongoClient ( ) <EOL> self . connection . drop_database ( MONGO1_DBNAME ) <EOL> self . connection [ MONGO1_DBNAME ] . add_user ( MONGO1_USERNAME , <EOL> MONGO1_PASSWORD ) <EOL> self . bulk_insert2 ( ) <EOL> def dropDB2 ( self ) : <EOL> self . connection = MongoClient ( ) <EOL> self . connection . drop_database ( MONGO1_DBNAME ) <EOL> self . connection . close ( ) <EOL> def bulk_insert2 ( self ) : <EOL> _db = self . connection [ MONGO1_DBNAME ] <EOL> works = self . random_works ( self . known_resource_count ) <EOL> _db . works . insert ( works ) <EOL> self . work = _db . works . find_one ( ) <EOL> self . connection . close ( ) <EOL> def random_works ( self , num ) : <EOL> works = [ ] <EOL> for i in range ( num ) : <EOL> dt = datetime . now ( ) <EOL> work = { <EOL> '<STR_LIT>' : self . random_string ( <NUM_LIT:20> ) , <EOL> '<STR_LIT:title>' : self . random_string ( <NUM_LIT:30> ) , <EOL> eve . LAST_UPDATED : dt , <EOL> eve . DATE_CREATED : dt , <EOL> } <EOL> works . append ( work ) <EOL> return works <EOL> class TestMethodsAcrossMultiMongo ( TestMultiMongo ) : <EOL> def test_get_multidb ( self ) : <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> r , s = self . get ( '<STR_LIT>' % self . work [ id_field ] ) <EOL> self . assert200 ( s ) <EOL> self . assertEqual ( r [ '<STR_LIT>' ] , self . work [ '<STR_LIT>' ] ) <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> r , s = self . get ( self . known_resource , item = self . item_id ) <EOL> self . assert200 ( s ) <EOL> self . assertEqual ( r [ id_field ] , self . item_id ) <EOL> def test_post_multidb ( self ) : <EOL> work = self . _save_work ( ) <EOL> db = self . connection [ MONGO1_DBNAME ] <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> new = db . works . find_one ( { id_field : ObjectId ( work [ id_field ] ) } ) <EOL> self . assertTrue ( new is not None ) <EOL> self . connection . close ( ) <EOL> contact = { '<STR_LIT>' : '<STR_LIT>' } <EOL> r , s = self . post ( self . known_resource_url , data = contact ) <EOL> self . assert201 ( s ) <EOL> db = self . connection [ MONGO_DBNAME ] <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> new = db . contacts . find_one ( { id_field : ObjectId ( r [ id_field ] ) } ) <EOL> self . assertTrue ( new is not None ) <EOL> self . connection . close ( ) <EOL> def test_patch_multidb ( self ) : <EOL> work = self . _save_work ( ) <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> id , etag = work [ id_field ] , work [ eve . ETAG ] <EOL> changes = { '<STR_LIT>' : '<STR_LIT>' } <EOL> headers = [ ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) , ( '<STR_LIT>' , etag ) ] <EOL> r = self . test_client . patch ( '<STR_LIT>' % id , data = json . dumps ( changes ) , <EOL> headers = headers ) <EOL> self . assert200 ( r . status_code ) <EOL> db = self . connection [ MONGO1_DBNAME ] <EOL> updated = db . works . find_one ( { id_field : ObjectId ( id ) } ) <EOL> self . assertEqual ( updated [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . connection . close ( ) <EOL> field , value = "<STR_LIT>" , "<STR_LIT>" <EOL> changes = { field : value } <EOL> headers = [ ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) , ( '<STR_LIT>' , <EOL> self . item_etag ) ] <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> r = self . test_client . patch ( self . item_id_url , data = json . dumps ( changes ) , <EOL> headers = headers ) <EOL> self . assert200 ( r . status_code ) <EOL> db = self . connection [ MONGO_DBNAME ] <EOL> updated = db . contacts . find_one ( { id_field : ObjectId ( self . item_id ) } ) <EOL> self . assertEqual ( updated [ field ] , value ) <EOL> self . connection . close ( ) <EOL> def test_put_multidb ( self ) : <EOL> work = self . _save_work ( ) <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> id , etag = work [ id_field ] , work [ eve . ETAG ] <EOL> changes = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:title>' : '<STR_LIT>' } <EOL> headers = [ ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) , ( '<STR_LIT>' , etag ) ] <EOL> r = self . test_client . put ( '<STR_LIT>' % id , data = json . dumps ( changes ) , <EOL> headers = headers ) <EOL> self . assert200 ( r . status_code ) <EOL> db = self . connection [ MONGO1_DBNAME ] <EOL> updated = db . works . find_one ( { id_field : ObjectId ( id ) } ) <EOL> self . assertEqual ( updated [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . connection . close ( ) <EOL> field , value = "<STR_LIT>" , "<STR_LIT>" <EOL> changes = { field : value } <EOL> headers = [ ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) , ( '<STR_LIT>' , <EOL> self . item_etag ) ] <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> r = self . test_client . put ( self . item_id_url , data = json . dumps ( changes ) , <EOL> headers = headers ) <EOL> self . assert200 ( r . status_code ) <EOL> db = self . connection [ MONGO_DBNAME ] <EOL> updated = db . contacts . find_one ( { id_field : ObjectId ( self . item_id ) } ) <EOL> self . assertEqual ( updated [ field ] , value ) <EOL> self . connection . close ( ) <EOL> def test_delete_multidb ( self ) : <EOL> work = self . _save_work ( ) <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> id , etag = work [ id_field ] , work [ eve . ETAG ] <EOL> r = self . test_client . delete ( '<STR_LIT>' % id , headers = [ ( '<STR_LIT>' , <EOL> etag ) ] ) <EOL> self . assert204 ( r . status_code ) <EOL> db = self . connection [ MONGO1_DBNAME ] <EOL> lost = db . works . find_one ( { id_field : ObjectId ( id ) } ) <EOL> self . assertEqual ( lost , None ) <EOL> self . connection . close ( ) <EOL> r = self . test_client . delete ( self . item_id_url , <EOL> headers = [ ( '<STR_LIT>' , self . item_etag ) ] ) <EOL> self . assert204 ( r . status_code ) <EOL> db = self . connection [ MONGO_DBNAME ] <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> lost = db . contacts . find_one ( { id_field : ObjectId ( self . item_id ) } ) <EOL> self . assertEqual ( lost , None ) <EOL> self . connection . close ( ) <EOL> def _save_work ( self ) : <EOL> work = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:title>' : '<STR_LIT>' } <EOL> r , s = self . post ( '<STR_LIT>' , data = work ) <EOL> self . assert201 ( s ) <EOL> return r <EOL> class MyBasicAuth ( BasicAuth ) : <EOL> def check_auth ( self , username , password , allowed_roles , resource , method ) : <EOL> self . set_mongo_prefix ( '<STR_LIT>' ) <EOL> return True <EOL> class TestMultiMongoAuth ( TestMultiMongo ) : <EOL> def test_get_multidb ( self ) : <EOL> self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] = [ ] <EOL> headers = [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> id_field = self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> r = self . test_client . get ( '<STR_LIT>' % self . work [ id_field ] , <EOL> headers = headers ) <EOL> self . assert404 ( r . status_code ) <EOL> self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] = MyBasicAuth <EOL> r = self . test_client . get ( '<STR_LIT>' % self . work [ id_field ] , <EOL> headers = headers ) <EOL> self . assert200 ( r . status_code ) <EOL> payl = json . loads ( r . get_data ( ) . decode ( '<STR_LIT:utf-8>' ) ) <EOL> self . assertEqual ( payl [ '<STR_LIT>' ] , self . work [ '<STR_LIT>' ] ) <EOL> r = self . test_client . get ( '<STR_LIT>' % ( self . known_resource_url , <EOL> self . item_id ) , headers = headers ) <EOL> self . assert200 ( r . status_code ) </s>
