<s> from conary_test import rephelp <EOL> import copy <EOL> import gzip <EOL> import os <EOL> from conary . repository import calllog , errors , shimclient <EOL> from conary . repository . netrepos import netserver <EOL> from conary import conarycfg <EOL> class ShimNetClientTest ( rephelp . RepositoryHelper ) : <EOL> def _setupShim ( self , log = None , authToken = None ) : <EOL> if authToken is None : <EOL> authToken = ( '<STR_LIT>' , '<STR_LIT>' , None , None ) <EOL> self . openRepository ( ) <EOL> testserver = self . servers . getServer ( ) <EOL> cfg = netserver . ServerConfig ( ) <EOL> cfg . serverName = testserver . getName ( ) <EOL> cfg . tmpDir = self . tmpDir <EOL> cfg . contentsDir = ( '<STR_LIT>' , [ testserver . contents . getPath ( ) ] ) <EOL> cfg . configLine ( '<STR_LIT>' % testserver . reposDB . getDriver ( ) ) <EOL> server = shimclient . NetworkRepositoryServer ( cfg , <EOL> self . cfg . repositoryMap [ '<STR_LIT:localhost>' ] ) <EOL> if log : <EOL> os . environ [ '<STR_LIT>' ] = log <EOL> cliCfg = copy . copy ( self . cfg ) <EOL> rm = conarycfg . RepoMap ( ) <EOL> for host , url in self . cfg . repositoryMap : <EOL> if host != '<STR_LIT:localhost>' : <EOL> rm . append ( ( host , url ) ) <EOL> cliCfg . repositoryMap = rm <EOL> shim = shimclient . ShimNetClient ( server , '<STR_LIT:http>' , <NUM_LIT> , authToken , cliCfg ) <EOL> if log : <EOL> del os . environ [ '<STR_LIT>' ] <EOL> return shim , cfg <EOL> def testShimClient ( self ) : <EOL> t = self . addQuickTestComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> shim , cfg = self . _setupShim ( ) <EOL> label = t . getVersion ( ) . branch ( ) . label ( ) <EOL> self . assertEqual ( shim . troveNames ( label ) , [ '<STR_LIT>' ] ) <EOL> n , v , f = t . getName ( ) , t . getVersion ( ) , t . getFlavor ( ) <EOL> trove = shim . getTroves ( [ ( n , v , f ) ] ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( trove . getName ( ) , n ) <EOL> server = shimclient . NetworkRepositoryServer ( cfg , <EOL> self . cfg . repositoryMap [ '<STR_LIT:localhost>' ] ) <EOL> empty = conarycfg . ConaryConfiguration ( False ) <EOL> shim = shimclient . ShimNetClient ( server , '<STR_LIT:http>' , <NUM_LIT> , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ ] ) , cfg = empty ) <EOL> trove = shim . getTroves ( [ ( n , v , f ) ] ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( trove . getName ( ) , n ) <EOL> self . assertRaises ( errors . InsufficientPermission , <EOL> shim . updateRoleMembers , '<STR_LIT:localhost>' , '<STR_LIT>' , [ ] ) <EOL> def testShimClientFileContents ( self ) : <EOL> self . openRepository ( <NUM_LIT:1> ) <EOL> shim = self . _setupShim ( ) [ <NUM_LIT:0> ] <EOL> t = self . addQuickTestComponent ( '<STR_LIT>' , '<STR_LIT>' , <EOL> fileContents = [ ( '<STR_LIT>' , '<STR_LIT:hello>' ) ] ) <EOL> label = t . getVersion ( ) . branch ( ) . label ( ) <EOL> self . assertEqual ( shim . troveNames ( label ) , [ '<STR_LIT>' ] ) <EOL> n , v , f = t . getName ( ) , t . getVersion ( ) , t . getFlavor ( ) <EOL> trove = shim . getTroves ( [ ( n , v , f ) ] ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( trove . getName ( ) , n ) <EOL> fileObj = shim . getFileContents ( [ ( fileId , version ) <EOL> for pathId , path , fileId , version in trove . iterFileList ( ) ] ) [ <NUM_LIT:0> ] <EOL> assert ( not fileObj . isCompressed ( ) ) <EOL> assert ( fileObj . get ( ) . readlines ( ) == [ '<STR_LIT:hello>' ] ) <EOL> fileObj = shim . getFileContents ( [ ( fileId , version ) <EOL> for pathId , path , fileId , version in trove . iterFileList ( ) ] , <EOL> compressed = True ) [ <NUM_LIT:0> ] <EOL> assert ( fileObj . isCompressed ( ) ) <EOL> f = gzip . GzipFile ( fileobj = fileObj . get ( ) , mode = "<STR_LIT:r>" ) <EOL> assert ( f . readlines ( ) == [ '<STR_LIT:hello>' ] ) <EOL> t = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> shim . getFileContents ( [ ( x [ <NUM_LIT:2> ] , x [ <NUM_LIT:3> ] ) for x in t . iterFileList ( ) ] ) <EOL> def testShimLog ( self ) : <EOL> p = self . workDir + '<STR_LIT>' <EOL> shim = self . _setupShim ( p ) [ <NUM_LIT:0> ] <EOL> trv = self . addComponent ( '<STR_LIT>' , '<STR_LIT:1.0>' ) <EOL> log = calllog . ClientCallLogger ( p ) <EOL> shim . troveNames ( trv . getVersion ( ) . trailingLabel ( ) ) <EOL> logEntries = list ( log ) <EOL> assert ( len ( logEntries ) == <NUM_LIT:1> ) <EOL> assert ( logEntries [ <NUM_LIT:0> ] . methodName == '<STR_LIT>' ) <EOL> assert ( logEntries [ <NUM_LIT:0> ] . url == '<STR_LIT>' ) <EOL> def testShimCommit ( self ) : <EOL> self . openRepository ( <NUM_LIT:1> ) <EOL> p = self . workDir + '<STR_LIT>' <EOL> shim = self . _setupShim ( p , authToken = ( '<STR_LIT:test>' , '<STR_LIT:foo>' , None , None ) ) [ <NUM_LIT:0> ] <EOL> trv , cs = self . Component ( '<STR_LIT>' , '<STR_LIT:1.0>' ) <EOL> shim . commitChangeSet ( cs ) <EOL> repos = self . openRepository ( ) <EOL> trv2 = repos . getTrove ( * trv . getNameVersionFlavor ( ) ) <EOL> assert ( trv == trv2 ) <EOL> del trv , trv2 <EOL> trv , cs = self . Component ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> cs . writeToFile ( self . workDir + '<STR_LIT>' ) <EOL> shim . commitChangeSetFile ( self . workDir + '<STR_LIT>' ) <EOL> trv , cs = self . Component ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> shim . commitChangeSet ( cs ) </s>
