<s> import datetime <EOL> import json <EOL> import os <EOL> from tests import base <EOL> from girder . constants import AssetstoreType , SettingKey <EOL> from girder . models . model_base import ValidationException <EOL> from girder . utility . system import formatSize <EOL> def setUpModule ( ) : <EOL> base . enabledPlugins . append ( '<STR_LIT>' ) <EOL> base . startServer ( ) <EOL> def tearDownModule ( ) : <EOL> base . stopServer ( ) <EOL> class QuotaTestCase ( base . TestCase ) : <EOL> def setUp ( self ) : <EOL> base . TestCase . setUp ( self ) <EOL> admin = { <EOL> '<STR_LIT:email>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True <EOL> } <EOL> self . admin = self . model ( '<STR_LIT:user>' ) . createUser ( ** admin ) <EOL> user = { <EOL> '<STR_LIT:email>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False <EOL> } <EOL> self . user = self . model ( '<STR_LIT:user>' ) . createUser ( ** user ) <EOL> coll = { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . admin <EOL> } <EOL> self . collection = self . model ( '<STR_LIT>' ) . createCollection ( ** coll ) <EOL> self . model ( '<STR_LIT>' ) . createFolder ( <EOL> parent = self . collection , parentType = '<STR_LIT>' , name = '<STR_LIT>' , <EOL> public = True , creator = self . admin ) <EOL> def _uploadFile ( self , name , parent , parentType = '<STR_LIT>' , size = <NUM_LIT> , <EOL> error = None , partial = False ) : <EOL> """<STR_LIT>""" <EOL> if parentType != '<STR_LIT:file>' : <EOL> resp = self . request ( <EOL> path = '<STR_LIT>' , method = '<STR_LIT:POST>' , user = self . admin , params = { <EOL> '<STR_LIT>' : parentType , <EOL> '<STR_LIT>' : parent [ '<STR_LIT>' ] , <EOL> '<STR_LIT:name>' : name , <EOL> '<STR_LIT:size>' : size , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , exception = error is not None ) <EOL> else : <EOL> resp = self . request ( <EOL> path = '<STR_LIT>' % str ( parent [ '<STR_LIT>' ] ) , method = '<STR_LIT>' , <EOL> user = self . admin , params = { <EOL> '<STR_LIT:size>' : size , <EOL> } , exception = error is not None ) <EOL> if error : <EOL> self . assertStatus ( resp , <NUM_LIT> ) <EOL> self . assertEqual ( resp . json [ '<STR_LIT:type>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . json [ '<STR_LIT:message>' ] [ : len ( error ) ] , error ) <EOL> return None <EOL> contents = os . urandom ( size ) <EOL> self . assertStatusOk ( resp ) <EOL> upload = resp . json <EOL> fields = [ ( '<STR_LIT>' , <NUM_LIT:0> ) , ( '<STR_LIT>' , upload [ '<STR_LIT>' ] ) ] <EOL> if not partial : <EOL> files = [ ( '<STR_LIT>' , name , contents ) ] <EOL> else : <EOL> files = [ ( '<STR_LIT>' , name , contents [ : - <NUM_LIT:1> ] ) ] <EOL> resp = self . multipartRequest ( path = '<STR_LIT>' , user = self . admin , <EOL> fields = fields , files = files ) <EOL> self . assertStatusOk ( resp ) <EOL> if partial : <EOL> fields = [ ( '<STR_LIT>' , len ( contents ) - <NUM_LIT:1> ) , ( '<STR_LIT>' , upload [ '<STR_LIT>' ] ) ] <EOL> files = [ ( '<STR_LIT>' , name , contents [ - <NUM_LIT:1> : ] ) ] <EOL> kwargs = { '<STR_LIT:path>' : '<STR_LIT>' , '<STR_LIT:user>' : self . admin , <EOL> '<STR_LIT>' : fields , '<STR_LIT>' : files } <EOL> return kwargs <EOL> return resp . json <EOL> def _setPolicy ( self , policy , model , resource , user , error = None , <EOL> results = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( policy , dict ) : <EOL> policyJSON = json . dumps ( policy ) <EOL> else : <EOL> policyJSON = policy <EOL> path = '<STR_LIT>' % ( model , resource [ '<STR_LIT>' ] ) <EOL> resp = self . request ( path = path , method = '<STR_LIT>' , user = user , <EOL> params = { '<STR_LIT>' : policyJSON } ) <EOL> if error : <EOL> self . assertStatus ( resp , <NUM_LIT> ) <EOL> self . assertTrue ( error in resp . json [ '<STR_LIT:message>' ] ) <EOL> return <EOL> self . assertStatusOk ( resp ) <EOL> resp = self . request ( path = path , method = '<STR_LIT:GET>' , user = user ) <EOL> self . assertStatusOk ( resp ) <EOL> currentPolicy = resp . json [ '<STR_LIT>' ] <EOL> if not results : <EOL> results = policy <EOL> for key in results : <EOL> if results [ key ] or results [ key ] is False : <EOL> self . assertEqual ( currentPolicy [ key ] , results [ key ] ) <EOL> else : <EOL> self . assertEqual ( currentPolicy [ key ] , None ) <EOL> def _setQuotaDefault ( self , model , value , testVal = '<STR_LIT>' , <EOL> error = None ) : <EOL> """<STR_LIT>""" <EOL> from girder . plugins . user_quota import constants <EOL> if model == '<STR_LIT:user>' : <EOL> key = constants . PluginSettings . QUOTA_DEFAULT_USER_QUOTA <EOL> elif model == '<STR_LIT>' : <EOL> key = constants . PluginSettings . QUOTA_DEFAULT_COLLECTION_QUOTA <EOL> try : <EOL> self . model ( '<STR_LIT>' ) . set ( key , value ) <EOL> except ValidationException as err : <EOL> if not error : <EOL> raise <EOL> if error not in err . args [ <NUM_LIT:0> ] : <EOL> raise <EOL> return <EOL> if testVal is not '<STR_LIT>' : <EOL> newVal = self . model ( '<STR_LIT>' ) . get ( key ) <EOL> self . assertEqual ( newVal , testVal ) <EOL> def _testAssetstores ( self , model , resource , user ) : <EOL> """<STR_LIT>""" <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' , user = user , <EOL> params = { '<STR_LIT>' : model , <EOL> '<STR_LIT>' : resource [ '<STR_LIT>' ] } ) <EOL> self . assertStatusOk ( resp ) <EOL> self . assertGreaterEqual ( len ( resp . json ) , <NUM_LIT:1> ) <EOL> folder = resp . json [ <NUM_LIT:0> ] <EOL> curAssetstoreId = str ( self . currentAssetstore [ '<STR_LIT>' ] ) <EOL> altAssetstoreId = str ( self . alternateAssetstore [ '<STR_LIT>' ] ) <EOL> badAssetstoreId = str ( self . brokenAssetstore [ '<STR_LIT>' ] ) <EOL> invalidAssetstoreId = str ( folder [ '<STR_LIT>' ] ) <EOL> file = self . _uploadFile ( '<STR_LIT>' , folder ) <EOL> self . assertEqual ( file [ '<STR_LIT>' ] , curAssetstoreId ) <EOL> self . _setPolicy ( { '<STR_LIT>' : altAssetstoreId } , <EOL> model , resource , user ) <EOL> file = self . _uploadFile ( '<STR_LIT>' , folder ) <EOL> self . assertEqual ( file [ '<STR_LIT>' ] , altAssetstoreId ) <EOL> self . _setPolicy ( { '<STR_LIT>' : invalidAssetstoreId } , <EOL> model , resource , user ) <EOL> file = self . _uploadFile ( '<STR_LIT>' , folder ) <EOL> self . assertEqual ( file [ '<STR_LIT>' ] , curAssetstoreId ) <EOL> self . _setPolicy ( { '<STR_LIT>' : badAssetstoreId } , <EOL> model , resource , user ) <EOL> file = self . _uploadFile ( '<STR_LIT>' , folder ) <EOL> self . assertEqual ( file [ '<STR_LIT>' ] , curAssetstoreId ) <EOL> self . _setPolicy ( { '<STR_LIT>' : altAssetstoreId } , <EOL> model , resource , user ) <EOL> file = self . _uploadFile ( '<STR_LIT>' , folder ) <EOL> self . assertEqual ( file [ '<STR_LIT>' ] , altAssetstoreId ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT:none>' } , <EOL> model , resource , user ) <EOL> self . _uploadFile ( '<STR_LIT>' , folder , <EOL> error = '<STR_LIT>' ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> model , resource , user ) <EOL> file = self . _uploadFile ( '<STR_LIT>' , folder ) <EOL> self . assertEqual ( file [ '<STR_LIT>' ] , curAssetstoreId ) <EOL> def _testQuota ( self , model , resource , user ) : <EOL> """<STR_LIT>""" <EOL> self . model ( '<STR_LIT>' ) . set ( SettingKey . UPLOAD_MINIMUM_CHUNK_SIZE , <NUM_LIT:0> ) <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' , user = user , <EOL> params = { '<STR_LIT>' : model , <EOL> '<STR_LIT>' : resource [ '<STR_LIT>' ] } ) <EOL> self . assertStatusOk ( resp ) <EOL> self . assertGreaterEqual ( len ( resp . json ) , <NUM_LIT:1> ) <EOL> folder = resp . json [ <NUM_LIT:0> ] <EOL> self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> ) <EOL> self . _setPolicy ( { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : False } , <EOL> model , resource , user ) <EOL> self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> , <EOL> error = '<STR_LIT>' ) <EOL> file = self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> ) <EOL> self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> , <EOL> error = '<STR_LIT>' ) <EOL> file1kwargs = self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> , <EOL> partial = True ) <EOL> file2kwargs = self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> , <EOL> partial = True ) <EOL> resp = self . multipartRequest ( ** file1kwargs ) <EOL> self . assertStatusOk ( resp ) <EOL> try : <EOL> resp = self . multipartRequest ( ** file2kwargs ) <EOL> self . assertStatus ( resp , <NUM_LIT> ) <EOL> except AssertionError as exc : <EOL> self . assertTrue ( '<STR_LIT>' in exc . args [ <NUM_LIT:0> ] ) <EOL> self . _setPolicy ( { '<STR_LIT>' : <NUM_LIT> } , model , resource , user ) <EOL> self . _uploadFile ( '<STR_LIT>' , file , '<STR_LIT:file>' , size = <NUM_LIT> ) <EOL> self . _setPolicy ( { '<STR_LIT>' : True } , model , resource , user ) <EOL> self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> ) <EOL> self . _setQuotaDefault ( model , <NUM_LIT> ) <EOL> self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> , <EOL> error = '<STR_LIT>' ) <EOL> file = self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> ) <EOL> self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> , <EOL> error = '<STR_LIT>' ) <EOL> self . _setQuotaDefault ( model , <NUM_LIT:5> * <NUM_LIT> ** <NUM_LIT:3> ) <EOL> file = self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT> ) <EOL> self . _uploadFile ( '<STR_LIT>' , folder , size = <NUM_LIT:6> * <NUM_LIT> ** <NUM_LIT:3> , <EOL> error = '<STR_LIT>' ) <EOL> def testAssetstorePolicy ( self ) : <EOL> """<STR_LIT>""" <EOL> base . dropGridFSDatabase ( '<STR_LIT>' ) <EOL> params = { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : AssetstoreType . GRIDFS , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:POST>' , user = self . admin , <EOL> params = params ) <EOL> self . assertStatusOk ( resp ) <EOL> self . model ( '<STR_LIT>' ) . save ( { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : AssetstoreType . FILESYSTEM , <EOL> '<STR_LIT:root>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : datetime . datetime . utcnow ( ) <EOL> } , validate = False ) <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' , user = self . admin , <EOL> params = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertStatusOk ( resp ) <EOL> assetstores = resp . json <EOL> self . assertEqual ( len ( assetstores ) , <NUM_LIT:3> ) <EOL> self . currentAssetstore = assetstores [ <NUM_LIT:0> ] <EOL> self . alternateAssetstore = assetstores [ <NUM_LIT:1> ] <EOL> self . brokenAssetstore = assetstores [ <NUM_LIT:2> ] <EOL> self . _testAssetstores ( '<STR_LIT:user>' , self . user , self . user ) <EOL> self . _testAssetstores ( '<STR_LIT>' , self . collection , self . admin ) <EOL> def testQuotaPolicy ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _testQuota ( '<STR_LIT:user>' , self . user , self . user ) <EOL> self . _testQuota ( '<STR_LIT>' , self . collection , self . admin ) <EOL> def testPolicySettings ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _setPolicy ( '<STR_LIT>' , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> error = '<STR_LIT>' ) <EOL> self . _setPolicy ( json . dumps ( [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> error = '<STR_LIT>' ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> error = '<STR_LIT>' ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> results = { '<STR_LIT>' : None } ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> error = '<STR_LIT>' ) <EOL> self . _setPolicy ( { '<STR_LIT>' : None } , <EOL> '<STR_LIT:user>' , self . user , self . user ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> results = { '<STR_LIT>' : None } ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT:none>' } , <EOL> '<STR_LIT:user>' , self . user , self . user ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> error = '<STR_LIT>' ) <EOL> self . _setPolicy ( { '<STR_LIT>' : None } , <EOL> '<STR_LIT:user>' , self . user , self . user ) <EOL> self . _setPolicy ( { '<STR_LIT>' : <NUM_LIT:0> } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> results = { '<STR_LIT>' : None } ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> results = { '<STR_LIT>' : None } ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> results = { '<STR_LIT>' : None } ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> error = '<STR_LIT>' ) <EOL> self . _setPolicy ( { '<STR_LIT>' : - <NUM_LIT:1> } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> error = '<STR_LIT>' ) <EOL> self . _setPolicy ( { '<STR_LIT>' : None } , <EOL> '<STR_LIT:user>' , self . user , self . user ) <EOL> valdict = { None : True , True : True , '<STR_LIT:true>' : True , <NUM_LIT:1> : True , '<STR_LIT:True>' : True , <EOL> False : False , '<STR_LIT:false>' : False , <NUM_LIT:0> : False , '<STR_LIT:False>' : False } <EOL> for val in valdict : <EOL> self . _setPolicy ( { '<STR_LIT>' : val } , <EOL> '<STR_LIT:user>' , self . user , self . user , <EOL> results = { '<STR_LIT>' : valdict [ val ] } ) <EOL> self . _setPolicy ( { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT:user>' , <EOL> self . user , self . user , error = '<STR_LIT>' ) <EOL> self . _setQuotaDefault ( '<STR_LIT:user>' , <NUM_LIT:0> , None ) <EOL> self . _setQuotaDefault ( '<STR_LIT:user>' , '<STR_LIT>' , None ) <EOL> self . _setQuotaDefault ( '<STR_LIT:user>' , '<STR_LIT>' , None ) <EOL> self . _setQuotaDefault ( '<STR_LIT:user>' , '<STR_LIT>' , error = '<STR_LIT>' ) <EOL> self . _setQuotaDefault ( '<STR_LIT:user>' , - <NUM_LIT:1> , error = '<STR_LIT>' ) <EOL> self . _setQuotaDefault ( '<STR_LIT:user>' , None ) <EOL> def testFormatSize ( self ) : <EOL> """<STR_LIT>""" <EOL> testList = [ <EOL> ( <NUM_LIT:1000> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , '<STR_LIT>' ) , <EOL> ] <EOL> for testItem in testList : <EOL> self . assertEqual ( testItem [ <NUM_LIT:1> ] , formatSize ( testItem [ <NUM_LIT:0> ] ) ) </s>
