<s> import unittest <EOL> import eve <EOL> import string <EOL> import random <EOL> import os <EOL> import simplejson as json <EOL> from datetime import datetime , timedelta <EOL> from flask . ext . pymongo import MongoClient <EOL> from bson import ObjectId <EOL> from eve . tests . test_settings import MONGO_PASSWORD , MONGO_USERNAME , MONGO_DBNAME , DOMAIN , MONGO_HOST , MONGO_PORT <EOL> from eve import ISSUES , ETAG <EOL> from eve . utils import date_to_str <EOL> try : <EOL> from urlparse import parse_qs , urlparse <EOL> except ImportError : <EOL> from urllib . parse import parse_qs , urlparse <EOL> class ValueStack ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , on_delete ) : <EOL> """<STR_LIT>""" <EOL> self . elements = [ ] <EOL> self . on_delete = on_delete <EOL> def __set__ ( self , obj , val ) : <EOL> self . elements . append ( val ) <EOL> def __get__ ( self , obj , objtype ) : <EOL> return self . elements [ - <NUM_LIT:1> ] if self . elements else None <EOL> def __delete__ ( self , obj ) : <EOL> for item in self . elements : <EOL> self . on_delete ( item ) <EOL> self . elements = [ ] <EOL> def close_pymongo_connection ( app ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' not in app . extensions : <EOL> return <EOL> del app . extensions [ '<STR_LIT>' ] <EOL> del app . media <EOL> class TestMinimal ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> app = ValueStack ( close_pymongo_connection ) <EOL> def setUp ( self , settings_file = None , url_converters = None ) : <EOL> """<STR_LIT>""" <EOL> self . this_directory = os . path . dirname ( os . path . realpath ( __file__ ) ) <EOL> if settings_file is None : <EOL> settings_file = os . path . join ( self . this_directory , <EOL> '<STR_LIT>' ) <EOL> self . connection = None <EOL> self . known_resource_count = <NUM_LIT> <EOL> self . setupDB ( ) <EOL> self . settings_file = settings_file <EOL> self . app = eve . Eve ( settings = self . settings_file , <EOL> url_converters = url_converters ) <EOL> self . test_client = self . app . test_client ( ) <EOL> self . domain = self . app . config [ '<STR_LIT>' ] <EOL> def tearDown ( self ) : <EOL> del self . app <EOL> self . dropDB ( ) <EOL> def assert200 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT:200> ) <EOL> def assert201 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert204 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert301 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert304 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert404 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert422 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def get ( self , resource , query = '<STR_LIT>' , item = None ) : <EOL> if resource in self . domain : <EOL> resource = self . domain [ resource ] [ '<STR_LIT:url>' ] <EOL> if item : <EOL> request = '<STR_LIT>' % ( resource , item , query ) <EOL> else : <EOL> request = '<STR_LIT>' % ( resource , query ) <EOL> r = self . test_client . get ( request ) <EOL> return self . parse_response ( r ) <EOL> def post ( self , url , data , headers = None , content_type = '<STR_LIT:application/json>' ) : <EOL> if headers is None : <EOL> headers = [ ] <EOL> headers . append ( ( '<STR_LIT:Content-Type>' , content_type ) ) <EOL> r = self . test_client . post ( url , data = json . dumps ( data ) , headers = headers ) <EOL> return self . parse_response ( r ) <EOL> def put ( self , url , data , headers = None ) : <EOL> if headers is None : <EOL> headers = [ ] <EOL> headers . append ( ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) ) <EOL> r = self . test_client . put ( url , data = json . dumps ( data ) , headers = headers ) <EOL> return self . parse_response ( r ) <EOL> def patch ( self , url , data , headers = None ) : <EOL> if headers is None : <EOL> headers = [ ] <EOL> headers . append ( ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) ) <EOL> r = self . test_client . patch ( url , data = json . dumps ( data ) , headers = headers ) <EOL> return self . parse_response ( r ) <EOL> def delete ( self , url , headers = None ) : <EOL> r = self . test_client . delete ( url , headers = headers ) <EOL> return self . parse_response ( r ) <EOL> def parse_response ( self , r ) : <EOL> try : <EOL> v = json . loads ( r . get_data ( ) ) <EOL> except json . JSONDecodeError : <EOL> v = None <EOL> return v , r . status_code <EOL> def assertValidationErrorStatus ( self , status ) : <EOL> self . assertEqual ( status , <EOL> self . app . config . get ( '<STR_LIT>' ) ) <EOL> def assertValidationError ( self , response , matches ) : <EOL> self . assertTrue ( eve . STATUS in response ) <EOL> self . assertTrue ( eve . STATUS_ERR in response [ eve . STATUS ] ) <EOL> self . assertTrue ( ISSUES in response ) <EOL> issues = response [ ISSUES ] <EOL> self . assertTrue ( len ( issues ) ) <EOL> for k , v in matches . items ( ) : <EOL> self . assertTrue ( k in issues ) <EOL> self . assertTrue ( v in issues [ k ] ) <EOL> def assertExpires ( self , resource ) : <EOL> r = self . test_client . get ( resource ) <EOL> expires = r . headers . get ( '<STR_LIT>' ) <EOL> self . assertTrue ( expires is not None ) <EOL> def assertCacheControl ( self , resource ) : <EOL> r = self . test_client . get ( resource ) <EOL> cache_control = r . headers . get ( '<STR_LIT>' ) <EOL> self . assertTrue ( cache_control is not None ) <EOL> self . assertEqual ( cache_control , <EOL> self . domain [ self . known_resource ] [ '<STR_LIT>' ] ) <EOL> def assertIfModifiedSince ( self , resource ) : <EOL> r = self . test_client . get ( resource ) <EOL> last_modified = r . headers . get ( '<STR_LIT>' ) <EOL> self . assertTrue ( last_modified is not None ) <EOL> r = self . test_client . get ( resource , headers = [ ( '<STR_LIT>' , <EOL> last_modified ) ] ) <EOL> self . assert304 ( r . status_code ) <EOL> self . assertTrue ( not r . get_data ( ) ) <EOL> def assertItem ( self , item , resource ) : <EOL> self . assertEqual ( type ( item ) , dict ) <EOL> updated_on = item . get ( self . app . config [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( updated_on is not None ) <EOL> try : <EOL> datetime . strptime ( updated_on , self . app . config [ '<STR_LIT>' ] ) <EOL> except Exception as e : <EOL> self . fail ( '<STR_LIT>' % <EOL> ( self . app . config [ '<STR_LIT>' ] , e ) ) <EOL> created_on = item . get ( self . app . config [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( updated_on is not None ) <EOL> try : <EOL> datetime . strptime ( created_on , self . app . config [ '<STR_LIT>' ] ) <EOL> except Exception as e : <EOL> self . fail ( '<STR_LIT>' % <EOL> ( self . app . config [ '<STR_LIT>' ] , e ) ) <EOL> link = item . get ( '<STR_LIT>' ) <EOL> _id = item . get ( self . domain [ resource ] [ '<STR_LIT>' ] ) <EOL> self . assertItemLink ( link , _id ) <EOL> def assertPagination ( self , response , page , total , max_results ) : <EOL> p_key , mr_key = self . app . config [ '<STR_LIT>' ] , self . app . config [ '<STR_LIT>' ] <EOL> self . assertTrue ( self . app . config [ '<STR_LIT>' ] in response ) <EOL> meta = response . get ( self . app . config [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( p_key in meta ) <EOL> self . assertTrue ( mr_key in meta ) <EOL> self . assertTrue ( '<STR_LIT>' in meta ) <EOL> self . assertEqual ( meta [ p_key ] , page ) <EOL> self . assertEqual ( meta [ mr_key ] , max_results ) <EOL> self . assertEqual ( meta [ '<STR_LIT>' ] , total ) <EOL> def assertHomeLink ( self , links ) : <EOL> self . assertTrue ( '<STR_LIT>' in links ) <EOL> link = links [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT:title>' in link ) <EOL> self . assertTrue ( '<STR_LIT>' in link ) <EOL> self . assertEqual ( '<STR_LIT>' , link [ '<STR_LIT:title>' ] ) <EOL> self . assertEqual ( "<STR_LIT:/>" , link [ '<STR_LIT>' ] ) <EOL> def assertResourceLink ( self , links , resource ) : <EOL> self . assertTrue ( '<STR_LIT>' in links ) <EOL> link = links [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT:title>' in link ) <EOL> self . assertTrue ( '<STR_LIT>' in link ) <EOL> url = self . domain [ resource ] [ '<STR_LIT:url>' ] <EOL> self . assertEqual ( url , link [ '<STR_LIT:title>' ] ) <EOL> self . assertEqual ( "<STR_LIT:%s>" % url , link [ '<STR_LIT>' ] ) <EOL> def assertCollectionLink ( self , links , resource ) : <EOL> self . assertTrue ( '<STR_LIT>' in links ) <EOL> link = links [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT:title>' in link ) <EOL> self . assertTrue ( '<STR_LIT>' in link ) <EOL> url = self . domain [ resource ] [ '<STR_LIT:url>' ] <EOL> self . assertEqual ( url , link [ '<STR_LIT:title>' ] ) <EOL> self . assertEqual ( "<STR_LIT:%s>" % url , link [ '<STR_LIT>' ] ) <EOL> def assertNextLink ( self , links , page ) : <EOL> self . assertTrue ( '<STR_LIT>' in links ) <EOL> link = links [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT:title>' in link ) <EOL> self . assertTrue ( '<STR_LIT>' in link ) <EOL> self . assertEqual ( '<STR_LIT>' , link [ '<STR_LIT:title>' ] ) <EOL> self . assertTrue ( "<STR_LIT>" % ( self . app . config [ '<STR_LIT>' ] , page ) <EOL> in link [ '<STR_LIT>' ] ) <EOL> def assertPrevLink ( self , links , page ) : <EOL> self . assertTrue ( '<STR_LIT>' in links ) <EOL> link = links [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT:title>' in link ) <EOL> self . assertTrue ( '<STR_LIT>' in link ) <EOL> self . assertEqual ( '<STR_LIT>' , link [ '<STR_LIT:title>' ] ) <EOL> if page > <NUM_LIT:1> : <EOL> self . assertTrue ( "<STR_LIT>" % ( self . app . config [ '<STR_LIT>' ] , page ) <EOL> in link [ '<STR_LIT>' ] ) <EOL> def assertItemLink ( self , links , item_id ) : <EOL> self . assertTrue ( '<STR_LIT>' in links ) <EOL> link = links [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT:title>' in link ) <EOL> self . assertTrue ( '<STR_LIT>' in link ) <EOL> self . assertTrue ( '<STR_LIT>' % item_id in link [ '<STR_LIT>' ] ) <EOL> def assertLastLink ( self , links , page ) : <EOL> if page : <EOL> self . assertTrue ( '<STR_LIT>' in links ) <EOL> link = links [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT:title>' in link ) <EOL> self . assertTrue ( '<STR_LIT>' in link ) <EOL> self . assertEqual ( '<STR_LIT>' , link [ '<STR_LIT:title>' ] ) <EOL> self . assertTrue ( "<STR_LIT>" % ( self . app . config [ '<STR_LIT>' ] , page ) <EOL> in link [ '<STR_LIT>' ] ) <EOL> else : <EOL> self . assertTrue ( '<STR_LIT>' not in links ) <EOL> def assertCustomParams ( self , link , params ) : <EOL> self . assertTrue ( '<STR_LIT>' in link ) <EOL> url_params = parse_qs ( urlparse ( link [ '<STR_LIT>' ] ) . query ) <EOL> for param , values in params . lists ( ) : <EOL> self . assertTrue ( param in url_params ) <EOL> for value in values : <EOL> self . assertTrue ( value in url_params [ param ] ) <EOL> def assert400 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert401 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert401or405 ( self , status ) : <EOL> self . assertTrue ( status == <NUM_LIT> or <NUM_LIT> ) <EOL> def assert403 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert405 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert412 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert428 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def assert500 ( self , status ) : <EOL> self . assertEqual ( status , <NUM_LIT> ) <EOL> def setupDB ( self ) : <EOL> self . connection = MongoClient ( MONGO_HOST , MONGO_PORT ) <EOL> self . connection . drop_database ( MONGO_DBNAME ) <EOL> if MONGO_USERNAME : <EOL> self . connection [ MONGO_DBNAME ] . add_user ( MONGO_USERNAME , <EOL> MONGO_PASSWORD ) <EOL> self . bulk_insert ( ) <EOL> def bulk_insert ( self ) : <EOL> pass <EOL> def dropDB ( self ) : <EOL> self . connection = MongoClient ( MONGO_HOST , MONGO_PORT ) <EOL> self . connection . drop_database ( MONGO_DBNAME ) <EOL> self . connection . close ( ) <EOL> class TestBase ( TestMinimal ) : <EOL> def setUp ( self , url_converters = None ) : <EOL> super ( TestBase , self ) . setUp ( url_converters = url_converters ) <EOL> self . disabled_bulk = '<STR_LIT>' <EOL> self . disabled_bulk_url = ( '<STR_LIT>' % <EOL> self . domain [ self . disabled_bulk ] [ '<STR_LIT:url>' ] ) <EOL> self . known_resource = '<STR_LIT>' <EOL> self . known_resource_url = ( '<STR_LIT>' % <EOL> self . domain [ self . known_resource ] [ '<STR_LIT:url>' ] ) <EOL> self . empty_resource = '<STR_LIT>' <EOL> self . empty_resource_url = '<STR_LIT>' % self . empty_resource <EOL> self . unknown_resource = '<STR_LIT>' <EOL> self . unknown_resource_url = '<STR_LIT>' % self . unknown_resource <EOL> self . unknown_item_id = '<STR_LIT>' <EOL> self . unknown_item_name = '<STR_LIT>' <EOL> self . unknown_item_id_url = ( '<STR_LIT>' % <EOL> ( self . domain [ self . known_resource ] [ '<STR_LIT:url>' ] , <EOL> self . unknown_item_id ) ) <EOL> self . unknown_item_name_url = ( '<STR_LIT>' % <EOL> ( self . domain [ self . known_resource ] [ '<STR_LIT:url>' ] , <EOL> self . unknown_item_name ) ) <EOL> self . readonly_resource = '<STR_LIT>' <EOL> self . readonly_resource_url = ( <EOL> '<STR_LIT>' % self . domain [ self . readonly_resource ] [ '<STR_LIT:url>' ] ) <EOL> self . different_resource = '<STR_LIT>' <EOL> self . different_resource_url = ( '<STR_LIT>' % <EOL> self . domain [ <EOL> self . different_resource ] [ '<STR_LIT:url>' ] ) <EOL> response , _ = self . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> contact = self . response_item ( response ) <EOL> self . item = contact <EOL> self . item_id = contact [ self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] ] <EOL> self . item_name = contact [ '<STR_LIT>' ] <EOL> self . item_tid = contact [ '<STR_LIT>' ] <EOL> self . item_etag = contact [ ETAG ] <EOL> self . item_ref = contact [ '<STR_LIT>' ] <EOL> self . item_id_url = ( '<STR_LIT>' % <EOL> ( self . domain [ self . known_resource ] [ '<STR_LIT:url>' ] , <EOL> self . item_id ) ) <EOL> self . item_name_url = ( '<STR_LIT>' % <EOL> ( self . domain [ self . known_resource ] [ '<STR_LIT:url>' ] , <EOL> self . item_name ) ) <EOL> self . alt_ref = self . response_item ( response , <NUM_LIT:1> ) [ '<STR_LIT>' ] <EOL> response , _ = self . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . readonly_id = self . response_item ( response ) [ '<STR_LIT>' ] <EOL> self . readonly_id_url = ( '<STR_LIT>' % ( self . readonly_resource_url , <EOL> self . readonly_id ) ) <EOL> response , _ = self . get ( '<STR_LIT>' ) <EOL> user = self . response_item ( response ) <EOL> self . user_id = user [ self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] ] <EOL> self . user_username = user [ '<STR_LIT:username>' ] <EOL> self . user_name = user [ '<STR_LIT>' ] <EOL> self . user_etag = user [ ETAG ] <EOL> self . user_id_url = ( '<STR_LIT>' % <EOL> ( self . domain [ self . different_resource ] [ '<STR_LIT:url>' ] , <EOL> self . user_id ) ) <EOL> self . user_username_url = ( <EOL> '<STR_LIT>' % ( self . domain [ self . different_resource ] [ '<STR_LIT:url>' ] , <EOL> self . user_username ) <EOL> ) <EOL> response , _ = self . get ( '<STR_LIT>' ) <EOL> invoice = self . response_item ( response ) <EOL> self . invoice_id = invoice [ self . domain [ '<STR_LIT>' ] [ '<STR_LIT>' ] ] <EOL> self . invoice_etag = invoice [ ETAG ] <EOL> self . invoice_id_url = ( '<STR_LIT>' % <EOL> ( self . domain [ '<STR_LIT>' ] [ '<STR_LIT:url>' ] , <EOL> self . invoice_id ) ) <EOL> self . epoch = date_to_str ( datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) ) <EOL> def response_item ( self , response , i = <NUM_LIT:0> ) : <EOL> if self . app . config [ '<STR_LIT>' ] : <EOL> return response [ '<STR_LIT>' ] [ i ] <EOL> else : <EOL> return response [ i ] <EOL> def random_contacts ( self , num , standard_date_fields = True ) : <EOL> schema = DOMAIN [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> contacts = [ ] <EOL> for i in range ( num ) : <EOL> dt = datetime . utcnow ( ) . replace ( microsecond = <NUM_LIT:0> ) <EOL> contact = { <EOL> '<STR_LIT>' : self . random_string ( schema [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : i , <EOL> '<STR_LIT>' : random . choice ( schema [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : self . random_rows ( random . randint ( <NUM_LIT:0> , <NUM_LIT:5> ) ) , <EOL> '<STR_LIT>' : self . random_list ( random . randint ( <NUM_LIT:0> , <NUM_LIT:5> ) ) , <EOL> '<STR_LIT:location>' : { <EOL> '<STR_LIT:address>' : '<STR_LIT>' + self . random_string ( <NUM_LIT:5> ) , <EOL> '<STR_LIT>' : '<STR_LIT>' + self . random_string ( <NUM_LIT:3> ) , <EOL> } , <EOL> '<STR_LIT>' : datetime . today ( ) + timedelta ( <EOL> days = random . randint ( - <NUM_LIT:10> , <NUM_LIT:10> ) ) , <EOL> '<STR_LIT>' : ObjectId ( ) , <EOL> '<STR_LIT>' : schema [ '<STR_LIT>' ] [ '<STR_LIT:default>' ] <EOL> } <EOL> if standard_date_fields : <EOL> contact [ eve . LAST_UPDATED ] = dt <EOL> contact [ eve . DATE_CREATED ] = dt <EOL> contacts . append ( contact ) <EOL> return contacts <EOL> def random_users ( self , num ) : <EOL> users = self . random_contacts ( num ) <EOL> for user in users : <EOL> user [ '<STR_LIT:username>' ] = self . random_string ( <NUM_LIT:10> ) <EOL> return users <EOL> def random_payments ( self , num ) : <EOL> payments = [ ] <EOL> for i in range ( num ) : <EOL> dt = datetime . utcnow ( ) . replace ( microsecond = <NUM_LIT:0> ) <EOL> payment = { <EOL> '<STR_LIT>' : self . random_string ( <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : i , <EOL> eve . LAST_UPDATED : dt , <EOL> eve . DATE_CREATED : dt , <EOL> } <EOL> payments . append ( payment ) <EOL> return payments <EOL> def random_invoices ( self , num ) : <EOL> invoices = [ ] <EOL> for _ in range ( num ) : <EOL> dt = datetime . utcnow ( ) . replace ( microsecond = <NUM_LIT:0> ) <EOL> invoice = { <EOL> '<STR_LIT>' : self . random_string ( <NUM_LIT:10> ) , <EOL> eve . LAST_UPDATED : dt , <EOL> eve . DATE_CREATED : dt , <EOL> } <EOL> invoices . append ( invoice ) <EOL> return invoices <EOL> def random_products ( self , num ) : <EOL> schema = DOMAIN [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> products = [ ] <EOL> for _ in range ( num ) : <EOL> products . append ( <EOL> { <EOL> '<STR_LIT>' : self . random_string ( schema [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT:title>' : ( "<STR_LIT>" + self . random_string ( <NUM_LIT:2> ) + <EOL> str ( random . randint ( <NUM_LIT:100> , <NUM_LIT:1000> ) ) ) <EOL> } <EOL> ) <EOL> return products <EOL> def random_string ( self , num ) : <EOL> return ( '<STR_LIT>' . join ( random . choice ( string . ascii_uppercase ) <EOL> for x in range ( num ) ) ) <EOL> def random_list ( self , num ) : <EOL> alist = [ ] <EOL> for i in range ( num ) : <EOL> alist . append ( [ '<STR_LIT:string>' + str ( i ) , random . randint ( <NUM_LIT:1000> , <NUM_LIT> ) ] ) <EOL> return alist <EOL> def random_rows ( self , num ) : <EOL> schema = DOMAIN [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> rows = [ ] <EOL> for _ in range ( num ) : <EOL> rows . append ( <EOL> { <EOL> '<STR_LIT>' : self . random_string ( schema [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : random . randint ( <NUM_LIT:100> , <NUM_LIT:1000> ) , <EOL> } <EOL> ) <EOL> return rows <EOL> def random_internal_transactions ( self , num ) : <EOL> transactions = [ ] <EOL> for i in range ( num ) : <EOL> dt = datetime . utcnow ( ) . replace ( microsecond = <NUM_LIT:0> ) <EOL> transaction = { <EOL> '<STR_LIT>' : self . random_string ( <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : i , <EOL> eve . LAST_UPDATED : dt , <EOL> eve . DATE_CREATED : dt , <EOL> } <EOL> transactions . append ( transaction ) <EOL> return transactions <EOL> def bulk_insert ( self ) : <EOL> _db = self . connection [ MONGO_DBNAME ] <EOL> _db . contacts . insert ( self . random_contacts ( self . known_resource_count ) ) <EOL> _db . contacts . insert ( self . random_users ( <NUM_LIT:2> ) ) <EOL> _db . payments . insert ( self . random_payments ( <NUM_LIT:10> ) ) <EOL> _db . invoices . insert ( self . random_invoices ( <NUM_LIT:1> ) ) <EOL> _db . internal_transactions . insert ( self . random_internal_transactions ( <NUM_LIT:4> ) ) <EOL> _db . products . insert ( self . random_products ( <NUM_LIT:2> ) ) <EOL> self . connection . close ( ) </s>
