<s> from testrunner import testhelp <EOL> from testutils import sock_utils <EOL> import copy <EOL> import itertools <EOL> import os <EOL> import shutil <EOL> import tempfile <EOL> import time <EOL> import SimpleHTTPServer <EOL> from conary_test import recipes <EOL> from conary_test import rephelp <EOL> from conary import callbacks <EOL> from conary import conarycfg , conaryclient <EOL> from conary import errors as gerrors <EOL> from conary import sqlite3 <EOL> from conary import trove <EOL> from conary import versions <EOL> from conary . cmds import updatecmd <EOL> from conary . deps import arch , deps <EOL> from conary . lib import httputils <EOL> from conary . lib import util <EOL> from conary . repository import changeset , calllog , errors , filecontainer <EOL> from conary . repository import netclient <EOL> from conary . server import server as conaryserver <EOL> class NetclientTest ( rephelp . RepositoryHelper ) : <EOL> def testGetFileContents ( self ) : <EOL> origDir = os . getcwd ( ) <EOL> self . resetRepository ( ) <EOL> self . resetWork ( ) <EOL> os . chdir ( self . workDir ) <EOL> self . newpkg ( '<STR_LIT>' ) <EOL> os . chdir ( '<STR_LIT>' ) <EOL> self . writeFile ( '<STR_LIT>' , recipes . testRecipe1 ) <EOL> self . writeFile ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> self . addfile ( '<STR_LIT>' ) <EOL> self . addfile ( '<STR_LIT:foo>' , binary = True ) <EOL> self . commit ( ) <EOL> self . rename ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> self . commit ( ) <EOL> repos = self . openRepository ( ) <EOL> troveVer = repos . getAllTroveLeaves ( '<STR_LIT:localhost>' , <EOL> { '<STR_LIT>' : None } ) [ '<STR_LIT>' ] . keys ( ) [ <NUM_LIT:0> ] <EOL> barId = None <EOL> for f in repos . iterFilesInTrove ( '<STR_LIT>' , troveVer , <EOL> deps . Flavor ( ) , withFiles = True ) : <EOL> pathId , path , fileId , version , fileObj = f <EOL> if path == '<STR_LIT:bar>' : <EOL> barId = ( fileId , version ) <EOL> elif path == '<STR_LIT>' : <EOL> recipeIds = ( fileId , version ) <EOL> if barId is None : <EOL> self . fail ( '<STR_LIT>' ) <EOL> f = repos . getFileContents ( [ barId ] ) [ <NUM_LIT:0> ] <EOL> assert ( f . get ( ) . read ( ) == '<STR_LIT>' ) <EOL> f , recipe = repos . getFileContents ( [ barId , recipeIds ] ) <EOL> assert ( f . get ( ) . read ( ) == '<STR_LIT>' ) <EOL> assert ( recipe . get ( ) . read ( ) == recipes . testRecipe1 ) <EOL> os . chdir ( origDir ) <EOL> def testFileContentsErrors ( self ) : <EOL> self . openRepository ( <NUM_LIT:1> ) <EOL> t = self . addQuickTestComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . addCollection ( '<STR_LIT:test>' , '<STR_LIT>' , [ '<STR_LIT>' ] ) <EOL> fileList = list ( t . iterFileList ( ) ) <EOL> assert ( len ( fileList ) == <NUM_LIT:1> ) <EOL> fileId , version = fileList [ <NUM_LIT:0> ] [ <NUM_LIT:2> : <NUM_LIT:4> ] <EOL> self . mkbranch ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:test>" , <EOL> shadow = True , binaryOnly = True ) <EOL> shadowVersion = versions . VersionFromString ( <EOL> '<STR_LIT>' ) <EOL> repos = self . openRepository ( ) <EOL> try : <EOL> repos . getFileContents ( [ ( fileId , shadowVersion ) ] ) <EOL> except errors . FileStreamNotFound , e : <EOL> assert ( ( e . fileId , e . fileVer ) == ( fileId , shadowVersion ) ) <EOL> else : <EOL> assert ( <NUM_LIT:0> ) <EOL> bogusV = versions . VersionFromString ( '<STR_LIT>' ) <EOL> bogusFileId = '<STR_LIT:1>' * <NUM_LIT:20> <EOL> bogusPathId = '<STR_LIT:2>' * <NUM_LIT:16> <EOL> try : <EOL> repos . getFileContents ( [ ( bogusFileId , bogusV ) ] ) <EOL> except errors . FileStreamNotFound , e : <EOL> assert ( ( e . fileId , e . fileVer ) == ( bogusFileId , bogusV ) ) <EOL> else : <EOL> assert ( <NUM_LIT:0> ) <EOL> try : <EOL> repos . getFileVersions ( [ ( bogusPathId , bogusFileId , bogusV ) ] ) <EOL> except errors . FileStreamMissing , e : <EOL> assert ( e . fileId == bogusFileId ) <EOL> else : <EOL> assert ( <NUM_LIT:0> ) <EOL> t = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , <EOL> fileContents = [ ( '<STR_LIT>' , <EOL> rephelp . Symlink ( '<STR_LIT>' ) ) ] ) <EOL> fileList = list ( t . iterFileList ( ) ) <EOL> assert ( len ( fileList ) == <NUM_LIT:1> ) <EOL> linkFileId , linkVersion = fileList [ <NUM_LIT:0> ] [ <NUM_LIT:2> : <NUM_LIT:4> ] <EOL> try : <EOL> repos . getFileContents ( [ ( linkFileId , linkVersion ) ] ) <EOL> except errors . FileHasNoContents , e : <EOL> assert ( ( e . fileId , e . fileVer ) == ( linkFileId , linkVersion ) ) <EOL> else : <EOL> assert ( <NUM_LIT:0> ) <EOL> server = self . servers . getServer ( <NUM_LIT:0> ) <EOL> shutil . rmtree ( server . contents . getPath ( ) ) <EOL> try : <EOL> f = repos . getFileContents ( [ ( fileId , version ) ] ) <EOL> except errors . FileContentsNotFound , e : <EOL> assert ( ( e . fileId , e . fileVer ) == ( fileId , version ) ) <EOL> else : <EOL> assert ( <NUM_LIT:0> ) <EOL> def testAnonymousAccess ( self ) : <EOL> ( built , d ) = self . buildRecipe ( recipes . testRecipe1 , "<STR_LIT>" ) <EOL> repos = self . openRepository ( ) <EOL> items = repos . getTroveLeavesByLabel ( <EOL> { '<STR_LIT>' : { self . cfg . buildLabel : None } } ) <EOL> assert ( len ( items ) == <NUM_LIT:1> ) <EOL> port = self . cfg . repositoryMap [ '<STR_LIT:localhost>' ] . split ( '<STR_LIT::>' ) [ - <NUM_LIT:1> ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] <EOL> cfg = copy . copy ( self . cfg ) <EOL> cfg . user = self . cfg . user . __class__ ( ) <EOL> client = conaryclient . ConaryClient ( cfg ) . getRepos ( ) <EOL> items = client . getTroveLeavesByLabel ( <EOL> { '<STR_LIT>' : { self . cfg . buildLabel : None } } ) <EOL> assert ( len ( items ) == <NUM_LIT:1> ) <EOL> def _testBranchPathIds ( self , shadow ) : <EOL> repos = self . openRepository ( ) <EOL> self . makeSourceTrove ( "<STR_LIT>" , recipes . pathIdTest1 ) <EOL> self . logFilter . add ( ) <EOL> version = self . cookFromRepository ( "<STR_LIT>" ) [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> self . logFilter . remove ( ) <EOL> version = versions . VersionFromString ( version ) <EOL> trunk = version . branch ( ) <EOL> if shadow : <EOL> branch = version . createShadow ( <EOL> versions . Label ( "<STR_LIT>" ) ) <EOL> else : <EOL> branch = version . createBranch ( <EOL> versions . Label ( "<STR_LIT>" ) ) <EOL> ids = repos . getPackageBranchPathIds ( "<STR_LIT>" , trunk ) <EOL> assert ( len ( ids ) == <NUM_LIT:3> ) <EOL> self . mkbranch ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> shadow = shadow ) <EOL> self . updateSourceTrove ( "<STR_LIT>" , recipes . pathIdTest2 , <EOL> "<STR_LIT>" ) <EOL> version = self . cookFromRepository ( "<STR_LIT>" , <EOL> buildLabel = versions . Label ( "<STR_LIT>" ) ) [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> branch = versions . VersionFromString ( version ) . branch ( ) <EOL> branchIds = repos . getPackageBranchPathIds ( "<STR_LIT>" , branch ) <EOL> ids = repos . getPackageBranchPathIds ( "<STR_LIT>" , trunk ) <EOL> assert ( len ( ids ) == <NUM_LIT:3> ) <EOL> branchIds = repos . getPackageBranchPathIds ( "<STR_LIT>" , branch ) <EOL> assert ( len ( branchIds ) == <NUM_LIT:2> ) <EOL> assert ( ids [ "<STR_LIT>" ] [ <NUM_LIT:2> ] == branchIds [ "<STR_LIT>" ] [ <NUM_LIT:2> ] ) <EOL> self . updateSourceTrove ( "<STR_LIT>" , recipes . pathIdTest3 ) <EOL> self . cookFromRepository ( "<STR_LIT>" ) <EOL> ids = repos . getPackageBranchPathIds ( "<STR_LIT>" , trunk ) <EOL> assert ( len ( ids ) == <NUM_LIT:5> ) <EOL> branchIds = repos . getPackageBranchPathIds ( "<STR_LIT>" , branch ) <EOL> assert ( len ( branchIds ) == <NUM_LIT:2> ) <EOL> assert ( ids [ "<STR_LIT>" ] [ <NUM_LIT:2> ] == branchIds [ "<STR_LIT>" ] [ <NUM_LIT:2> ] ) <EOL> assert ( ids [ "<STR_LIT>" ] [ <NUM_LIT:2> ] == branchIds [ "<STR_LIT>" ] [ <NUM_LIT:2> ] ) <EOL> self . updateSourceTrove ( "<STR_LIT>" , recipes . pathIdTest4 , <EOL> "<STR_LIT>" ) <EOL> self . cookFromRepository ( "<STR_LIT>" , <EOL> buildLabel = versions . Label ( "<STR_LIT>" ) ) [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> ids = repos . getPackageBranchPathIds ( "<STR_LIT>" , trunk ) <EOL> assert ( len ( ids ) == <NUM_LIT:5> ) <EOL> branchIds = repos . getPackageBranchPathIds ( "<STR_LIT>" , branch ) <EOL> assert ( len ( branchIds ) == <NUM_LIT:4> ) <EOL> assert ( ids [ "<STR_LIT>" ] [ <NUM_LIT:2> ] == branchIds [ "<STR_LIT>" ] [ <NUM_LIT:2> ] ) <EOL> assert ( ids [ "<STR_LIT>" ] [ <NUM_LIT:2> ] == branchIds [ "<STR_LIT>" ] [ <NUM_LIT:2> ] ) <EOL> assert ( ids [ "<STR_LIT>" ] [ <NUM_LIT:2> ] == branchIds [ "<STR_LIT>" ] [ <NUM_LIT:2> ] ) <EOL> def testBranchPathIdsForShadow ( self ) : <EOL> self . _testBranchPathIds ( True ) <EOL> def testBranchPathIdsForBranch ( self ) : <EOL> self . _testBranchPathIds ( False ) <EOL> def testCreateChangeset ( self ) : <EOL> ver = '<STR_LIT>' <EOL> self . addQuickTestComponent ( '<STR_LIT>' , ver ) <EOL> repos = self . openRepository ( ) <EOL> flavor = deps . parseFlavor ( '<STR_LIT>' ) <EOL> version = versions . VersionFromString ( ver ) <EOL> csfile = os . path . join ( self . workDir , '<STR_LIT>' ) <EOL> repos . createChangeSetFile ( <EOL> [ ( '<STR_LIT>' , ( None , flavor ) , ( version , flavor ) , True ) ] , csfile , <EOL> primaryTroveList = [ ( '<STR_LIT>' , version , flavor ) ] ) <EOL> cs = changeset . ChangeSetFromFile ( csfile ) <EOL> assert ( cs . getPrimaryTroveList ( ) == [ ( '<STR_LIT>' , version , flavor ) ] ) <EOL> def testCreateChangeset2 ( self ) : <EOL> self . openRepository ( ) <EOL> repos = self . openRepository ( <NUM_LIT:1> ) <EOL> t1 = self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' , [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> t2 = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> cs = repos . createChangeSet ( [ ( '<STR_LIT>' , <EOL> ( t1 . getVersion ( ) , t1 . getFlavor ( ) ) , <EOL> ( t2 . getVersion ( ) , t2 . getFlavor ( ) ) , <EOL> False ) ] ) <EOL> def testCreateChangesetViaHTTPProxy ( self ) : <EOL> self . openRepository ( ) <EOL> ver = '<STR_LIT>' <EOL> self . addQuickTestComponent ( '<STR_LIT>' , ver ) <EOL> flavor = deps . parseFlavor ( '<STR_LIT>' ) <EOL> version = versions . VersionFromString ( ver ) <EOL> csfile = os . path . join ( self . workDir , '<STR_LIT>' ) <EOL> h = rephelp . HTTPProxy ( os . path . join ( self . workDir , "<STR_LIT>" ) ) <EOL> proxyUri = h . start ( ) <EOL> if proxyUri is None : <EOL> raise testhelp . SkipTestException ( '<STR_LIT>' ) <EOL> proxyUrl = "<STR_LIT>" + proxyUri <EOL> proxies = dict ( http = proxyUrl , https = proxyUrl ) <EOL> self . cfg . configLine ( '<STR_LIT>' % h . port ) <EOL> client = conaryclient . ConaryClient ( self . cfg ) <EOL> repos = client . getRepos ( ) <EOL> repos . createChangeSetFile ( <EOL> [ ( '<STR_LIT>' , ( None , flavor ) , ( version , flavor ) , True ) ] , csfile , <EOL> primaryTroveList = [ ( '<STR_LIT>' , version , flavor ) ] ) <EOL> cs = changeset . ChangeSetFromFile ( csfile ) <EOL> assert ( cs . getPrimaryTroveList ( ) == [ ( '<STR_LIT>' , version , flavor ) ] ) <EOL> def testTimestamps ( self ) : <EOL> """<STR_LIT>""" <EOL> flavor = deps . parseFlavor ( '<STR_LIT>' ) <EOL> cs = changeset . ChangeSet ( ) <EOL> v = versions . VersionFromString ( '<STR_LIT>' , <EOL> timeStamps = [ <NUM_LIT> ] ) <EOL> pkg = trove . Trove ( '<STR_LIT:test>' , v , flavor , None ) <EOL> pkg . troveInfo . sourceName . set ( '<STR_LIT>' ) <EOL> pkg . addTrove ( '<STR_LIT>' , v , flavor , byDefault = True ) <EOL> pkg . computeDigests ( ) <EOL> diff = pkg . diff ( None ) [ <NUM_LIT:0> ] <EOL> cs . newTrove ( diff ) <EOL> cs . addPrimaryTrove ( '<STR_LIT:test>' , v , flavor ) <EOL> foo = trove . Trove ( '<STR_LIT>' , v , flavor , None ) <EOL> pkg . troveInfo . sourceName . set ( '<STR_LIT>' ) <EOL> foo . computeDigests ( ) <EOL> diff = foo . diff ( None ) [ <NUM_LIT:0> ] <EOL> cs . newTrove ( diff ) <EOL> repos = self . openRepository ( ) <EOL> now = time . time ( ) <EOL> from conary . cmds . updatecmd import UpdateCallback <EOL> callback = UpdateCallback ( ) <EOL> repos . commitChangeSet ( cs , callback = callback ) <EOL> trvs = repos . getTroves ( [ ( '<STR_LIT:test>' , v , flavor ) , ( '<STR_LIT>' , v , flavor ) ] ) <EOL> assert ( trvs [ <NUM_LIT:0> ] . getVersion ( ) . trailingRevision ( ) . getTimestamp ( ) == <EOL> trvs [ <NUM_LIT:1> ] . getVersion ( ) . trailingRevision ( ) . getTimestamp ( ) ) <EOL> assert ( trvs [ <NUM_LIT:0> ] . getVersion ( ) . trailingRevision ( ) . getTimestamp ( ) >= now ) <EOL> def testGetTroveLeavesByBranch ( self ) : <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT:1.0>' , '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def testUnknownMethod ( self ) : <EOL> repos = self . openRepository ( ) <EOL> self . assertRaises ( errors . MethodNotSupported , <EOL> repos . c [ '<STR_LIT:localhost>' ] . callWrapper ) <EOL> def testConfigFilesRaisePathIdsConflict ( self ) : <EOL> t1 = self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' , <EOL> [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> t2 = self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' , <EOL> [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> t3 = self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' , <EOL> [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> repos = self . openRepository ( ) <EOL> repos . createChangeSet ( [ <EOL> ( t1 . getName ( ) , ( None , None ) , ( t1 . getVersion ( ) , t1 . getFlavor ( ) ) , <EOL> True ) , <EOL> ( t2 . getName ( ) , ( None , None ) , ( t2 . getVersion ( ) , t2 . getFlavor ( ) ) , <EOL> True ) ] ) <EOL> repos . createChangeSet ( [ <EOL> ( t1 . getName ( ) , ( None , None ) , ( t1 . getVersion ( ) , t1 . getFlavor ( ) ) , <EOL> True ) , <EOL> ( t3 . getName ( ) , ( None , None ) , ( t3 . getVersion ( ) , t3 . getFlavor ( ) ) , <EOL> True ) ] ) <EOL> repos . c [ '<STR_LIT:localhost>' ] . setProtocolVersion ( <NUM_LIT> ) <EOL> try : <EOL> repos . createChangeSet ( [ <EOL> ( t1 . getName ( ) , ( None , None ) , ( t1 . getVersion ( ) , t1 . getFlavor ( ) ) , <EOL> True ) , <EOL> ( t2 . getName ( ) , ( None , None ) , ( t2 . getVersion ( ) , t2 . getFlavor ( ) ) , <EOL> True ) ] ) <EOL> except changeset . PathIdsConflictError , e : <EOL> assert str ( e ) == ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> assert <NUM_LIT:0> , '<STR_LIT>' <EOL> def testFind64BitTroves ( self ) : <EOL> self . cfg . flavorPreferences = arch . getFlavorPreferences ( <EOL> [ [ deps . Dependency ( '<STR_LIT>' ) ] ] ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> trv = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def _getFlavors ( d , name ) : <EOL> return [ str ( x ) for x in itertools . chain ( * d [ name ] . itervalues ( ) ) ] <EOL> repos = self . openRepository ( ) <EOL> n = '<STR_LIT>' <EOL> l = trv . getVersion ( ) . trailingLabel ( ) <EOL> f = deps . parseFlavor ( '<STR_LIT>' ) <EOL> d = repos . getTroveLeavesByLabel ( { n : { l : [ f ] } } , bestFlavor = True ) <EOL> self . assertEqual ( _getFlavors ( d , n ) , [ '<STR_LIT>' ] ) <EOL> self . resetRepository ( ) <EOL> repos = self . openRepository ( ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> trv = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> d = repos . getTroveLeavesByLabel ( { n : { l : [ f ] } } , bestFlavor = True ) <EOL> self . assertEqual ( _getFlavors ( d , n ) , [ '<STR_LIT>' ] ) <EOL> def testFindNegativeFlavor ( self ) : <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' ) <EOL> repos = self . openRepository ( ) <EOL> results = repos . getTroveLeavesByLabel ( { '<STR_LIT>' : <EOL> { self . cfg . buildLabel : None } } ) <EOL> assert ( results ) <EOL> results = repos . getTroveLeavesByLabel ( { '<STR_LIT>' : <EOL> { self . cfg . buildLabel : [ deps . parseFlavor ( '<STR_LIT>' ) ] } } ) <EOL> assert ( results ) <EOL> def testCreateChangeSetWithMissingTroves ( self ) : <EOL> trv = self . addCollection ( '<STR_LIT:foo>' , '<STR_LIT:1>' , [ '<STR_LIT>' ] ) <EOL> repos = self . openRepository ( ) <EOL> cs = repos . createChangeSet ( [ ( '<STR_LIT:foo>' , <EOL> ( None , None ) , <EOL> ( trv . getVersion ( ) , trv . getFlavor ( ) ) , <EOL> True ) ] , <EOL> recurse = True , withFileContents = False ) <EOL> for tcs in cs . iterNewTroveList ( ) : <EOL> if tcs . getName ( ) == '<STR_LIT:foo>' : <EOL> self . assertTrue ( tcs . troveType ( ) == trove . TROVE_TYPE_NORMAL ) <EOL> elif tcs . getName ( ) == '<STR_LIT>' : <EOL> self . assertTrue ( tcs . troveType ( ) == trove . TROVE_TYPE_REMOVED ) <EOL> ti = trove . TroveInfo ( tcs . troveInfoDiff . freeze ( ) ) <EOL> self . assertTrue ( ti . flags . isMissing ( ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> repos = self . openRepository ( <NUM_LIT:1> ) <EOL> trv = self . addCollection ( '<STR_LIT>' , '<STR_LIT>' , <EOL> [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> weakRefList = [ ] ) <EOL> self . assertRaises ( errors . TroveMissing , <EOL> repos . createChangeSet , <EOL> [ ( trv . getName ( ) , <EOL> ( None , None ) , <EOL> ( trv . getVersion ( ) , trv . getFlavor ( ) ) , <EOL> True ) ] , <EOL> recurse = True , withFileContents = False ) <EOL> def testNonpresentTroves ( self ) : <EOL> trv = self . addCollection ( '<STR_LIT:foo>' , '<STR_LIT:1>' , [ '<STR_LIT>' ] ) <EOL> repos = self . openRepository ( ) <EOL> d = repos . getTroveVersionsByLabel ( { '<STR_LIT>' : <EOL> { trv . getVersion ( ) . trailingLabel ( ) : None } } ) <EOL> assert ( not d ) <EOL> d = repos . getTroveVersionsByBranch ( { '<STR_LIT>' : <EOL> { trv . getVersion ( ) . branch ( ) : None } } ) <EOL> assert ( not d ) <EOL> d = repos . getTroveLeavesByLabel ( { '<STR_LIT>' : <EOL> { trv . getVersion ( ) . trailingLabel ( ) : None } } ) <EOL> assert ( not d ) <EOL> d = repos . getTroveLeavesByBranch ( { '<STR_LIT>' : <EOL> { trv . getVersion ( ) . branch ( ) : None } } ) <EOL> assert ( not d ) <EOL> d = repos . getTroveVersionList ( '<STR_LIT:localhost>' , { '<STR_LIT>' : None } ) <EOL> assert ( not d ) <EOL> d = repos . getAllTroveLeaves ( '<STR_LIT:localhost>' , { '<STR_LIT>' : None } ) <EOL> assert ( not d ) <EOL> d = repos . getAllTroveFlavors ( { '<STR_LIT>' : [ trv . getVersion ( ) ] } ) <EOL> assert ( not d ) <EOL> self . assertRaises ( errors . TroveMissing , repos . getTroveLatestVersion , <EOL> '<STR_LIT>' , trv . getVersion ( ) . branch ( ) ) <EOL> assert ( not repos . hasTrove ( '<STR_LIT>' , trv . getVersion ( ) , trv . getFlavor ( ) ) ) <EOL> self . assertRaises ( errors . TroveMissing , repos . getTrove , <EOL> '<STR_LIT>' , trv . getVersion ( ) , trv . getFlavor ( ) ) <EOL> def testGetAllTroveLeavesMultipleServers ( self ) : <EOL> repos = self . openRepository ( serverName = [ '<STR_LIT:localhost>' , '<STR_LIT>' ] ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> q = repos . getAllTroveLeaves ( '<STR_LIT:localhost>' , { } ) <EOL> assert ( q . keys ( ) == [ '<STR_LIT>' ] ) <EOL> assert ( len ( q [ '<STR_LIT>' ] ) == <NUM_LIT:1> ) <EOL> self . stopRepository ( <NUM_LIT:0> ) <EOL> def testPrepareCallbackException ( self ) : <EOL> class Callback ( updatecmd . UpdateCallback ) : <EOL> def preparingChangeSet ( self ) : <EOL> e = Exception ( "<STR_LIT>" ) <EOL> e . errorIsUncatchable = True <EOL> raise e <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' ) <EOL> self . addCollection ( '<STR_LIT>' , '<STR_LIT:1>' , [ '<STR_LIT>' ] ) <EOL> client = conaryclient . ConaryClient ( self . cfg ) <EOL> applyList = conaryclient . cmdline . parseChangeList ( '<STR_LIT>' ) <EOL> client . setUpdateCallback ( Callback ( self . cfg ) ) <EOL> try : <EOL> ( updJob , suggMap ) = client . updateChangeSet ( applyList ) <EOL> except Exception , e : <EOL> self . assertEqual ( str ( e ) , "<STR_LIT>" ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def testPrepareCallbackException2 ( self ) : <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT:2>' ) <EOL> class Callback ( callbacks . UpdateCallback ) : <EOL> def __init__ ( self , cfg ) : <EOL> self . abortEvent = None <EOL> pass <EOL> def preparingChangeSet ( self ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> client = conaryclient . ConaryClient ( self . cfg ) <EOL> applyList = conaryclient . cmdline . parseChangeList ( '<STR_LIT>' ) <EOL> client . setUpdateCallback ( Callback ( self . cfg ) ) <EOL> self . logFilter . add ( ) <EOL> ( updJob , suggMap ) = self . discardOutput ( <EOL> client . updateChangeSet , applyList ) <EOL> self . logFilter . regexpCompare ( '<STR_LIT>' ) <EOL> class Callback ( updatecmd . UpdateCallback ) : <EOL> def preparingChangeSet ( self ) : <EOL> raise gerrors . CancelOperationException ( "<STR_LIT>" ) <EOL> applyList = conaryclient . cmdline . parseChangeList ( '<STR_LIT>' ) <EOL> client . setUpdateCallback ( Callback ( self . cfg ) ) <EOL> self . logFilter . add ( ) <EOL> try : <EOL> ( updJob , suggMap ) = self . discardOutput ( <EOL> client . updateChangeSet , applyList ) <EOL> except errors . RepositoryError : <EOL> pass <EOL> self . logFilter . regexpCompare ( '<STR_LIT>' ) <EOL> self . assertRaises ( errors . RepositoryError , <EOL> client . applyUpdate , updJob ) <EOL> def testChangeSetConversionError ( self ) : <EOL> def errorHttpPutFile ( * args , ** kwargs ) : <EOL> return <NUM_LIT> , "<STR_LIT>" <EOL> cs = changeset . ChangeSet ( ) <EOL> trv = trove . Trove ( '<STR_LIT>' , <EOL> versions . ThawVersion ( '<STR_LIT>' ) , <EOL> deps . parseFlavor ( '<STR_LIT>' ) , None ) <EOL> trv . computeDigests ( ) <EOL> trvCs = trv . diff ( None , absolute = True ) [ <NUM_LIT:0> ] <EOL> cs . newTrove ( trvCs ) <EOL> repos = self . openRepository ( ) <EOL> old = netclient . httpPutFile <EOL> try : <EOL> netclient . httpPutFile = errorHttpPutFile <EOL> self . assertRaises ( errors . CommitError , repos . commitChangeSet , cs ) <EOL> finally : <EOL> netclient . httpPutFile = old <EOL> def testReferences ( self ) : <EOL> if sqlite3 . sqlite_version_info ( ) < ( <NUM_LIT:3> , <NUM_LIT:7> , <NUM_LIT:0> ) : <EOL> raise testhelp . SkipTestException ( "<STR_LIT>" ) <EOL> repos = self . openRepository ( ) <EOL> repos . deleteUserByName ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . addUserAndRole ( repos , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> repos . addAcl ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> allrepo = self . getRepositoryClient ( user = "<STR_LIT>" , password = "<STR_LIT>" ) <EOL> self . addUserAndRole ( repos , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> repos . addAcl ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> qarepo = self . getRepositoryClient ( user = "<STR_LIT>" , password = "<STR_LIT>" ) <EOL> verlist = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] <EOL> for i , v in enumerate ( verlist ) : <EOL> for n1 in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> for n2 in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> self . addComponent ( n1 + n2 , v ) <EOL> self . addCollection ( n1 , v , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . addCollection ( "<STR_LIT>" % i , v , [ ( "<STR_LIT>" , v ) , ( "<STR_LIT>" , v ) ] ) <EOL> self . addCollection ( "<STR_LIT>" % i , v , [ ( "<STR_LIT>" , v ) , ( "<STR_LIT>" , v ) ] , weakRefList = [ ( "<STR_LIT>" , v ) ] ) <EOL> noF = deps . Flavor ( ) <EOL> for i , vStr in enumerate ( verlist ) : <EOL> v = versions . VersionFromString ( vStr ) <EOL> ret = allrepo . getTroveReferences ( "<STR_LIT:localhost>" , [ ( "<STR_LIT>" , v , noF ) ] ) <EOL> self . assertEqual ( ret , [ [ ( "<STR_LIT>" % i , v , noF ) ] ] ) <EOL> ret = allrepo . getTroveReferences ( "<STR_LIT:localhost>" , [ ( "<STR_LIT>" , v , noF ) ] ) <EOL> self . assertEqual ( set ( ret [ <NUM_LIT:0> ] ) , set ( [ ( "<STR_LIT>" , v , noF ) , ( "<STR_LIT>" % i , v , noF ) ] ) ) <EOL> ret = allrepo . getTroveReferences ( "<STR_LIT:localhost>" , [ ( "<STR_LIT>" , v , noF ) ] ) <EOL> self . assertEqual ( set ( ret [ <NUM_LIT:0> ] ) , <EOL> set ( [ ( "<STR_LIT>" , v , noF ) , ( "<STR_LIT>" % i , v , noF ) , ( "<STR_LIT>" % i , v , noF ) ] ) ) <EOL> ret = qarepo . getTroveReferences ( "<STR_LIT:localhost>" , [ ( "<STR_LIT>" , v , noF ) , ( "<STR_LIT>" , v , noF ) ] ) <EOL> if v . branch ( ) . label ( ) . asString ( ) == "<STR_LIT>" : <EOL> self . assertEqual ( set ( ret [ <NUM_LIT:0> ] ) , <EOL> set ( [ ( "<STR_LIT>" , v , noF ) , ( "<STR_LIT>" % i , v , noF ) ] ) ) <EOL> self . assertEqual ( ret [ <NUM_LIT:1> ] , [ ( "<STR_LIT>" , v , noF ) ] ) <EOL> else : <EOL> self . assertEqual ( ret , [ [ ] , [ ] ] ) <EOL> ret = allrepo . getTroveDescendants ( "<STR_LIT:localhost>" , [ <EOL> ( "<STR_LIT>" , versions . VersionFromString ( "<STR_LIT>" ) , noF ) ] ) <EOL> self . assertEqual ( ret , [ [ ] ] ) <EOL> ret = allrepo . getTroveDescendants ( "<STR_LIT:localhost>" , [ <EOL> ( "<STR_LIT>" , versions . VersionFromString ( "<STR_LIT>" ) , noF ) ] ) <EOL> self . assertEqual ( set ( ret [ <NUM_LIT:0> ] ) , <EOL> set ( [ ( versions . VersionFromString ( x ) , noF ) for x in verlist ] ) ) <EOL> qaset = set ( [ <EOL> ( versions . VersionFromString ( "<STR_LIT>" ) , noF ) , <EOL> ( versions . VersionFromString ( "<STR_LIT>" ) , noF ) , <EOL> ] ) <EOL> ret = qarepo . getTroveDescendants ( "<STR_LIT:localhost>" , [ <EOL> ( "<STR_LIT>" , versions . VersionFromString ( "<STR_LIT>" ) , noF ) , <EOL> ( "<STR_LIT>" , versions . VersionFromString ( "<STR_LIT>" ) , noF ) ] ) <EOL> self . assertEqual ( ret [ <NUM_LIT:0> ] , [ ] ) <EOL> self . assertEqual ( set ( ret [ <NUM_LIT:1> ] ) , qaset ) <EOL> branch = "<STR_LIT>" <EOL> retset = set ( ) <EOL> for b in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> branch = branch + "<STR_LIT>" + b <EOL> verStr = branch + "<STR_LIT>" <EOL> self . addComponent ( "<STR_LIT>" , verStr ) <EOL> self . addCollection ( "<STR_LIT>" , verStr , [ "<STR_LIT>" ] ) <EOL> retset . add ( ( versions . VersionFromString ( verStr ) , noF ) ) <EOL> ret = allrepo . getTroveDescendants ( "<STR_LIT:localhost>" , [ <EOL> ( "<STR_LIT>" , versions . VersionFromString ( "<STR_LIT>" ) , noF ) ] ) <EOL> self . assertEqual ( set ( ret [ <NUM_LIT:0> ] ) , retset ) <EOL> def testFlavoredReferences ( self ) : <EOL> repos = self . openRepository ( ) <EOL> v1 = "<STR_LIT>" <EOL> v2 = "<STR_LIT>" <EOL> f1 = deps . parseFlavor ( "<STR_LIT>" ) <EOL> f2 = deps . parseFlavor ( "<STR_LIT>" ) <EOL> self . addComponent ( "<STR_LIT>" , v1 , f1 ) <EOL> self . addComponent ( "<STR_LIT>" , v1 , f2 ) <EOL> self . addCollection ( "<STR_LIT>" , v1 , [ "<STR_LIT>" ] , defaultFlavor = f1 ) <EOL> self . addCollection ( "<STR_LIT>" , v1 , [ "<STR_LIT>" ] , defaultFlavor = f2 ) <EOL> self . addCollection ( "<STR_LIT>" , v2 , [ ( "<STR_LIT>" , v1 , f1 ) ] , defaultFlavor = f1 ) <EOL> ret = repos . getTroveReferences ( "<STR_LIT:localhost>" , [ <EOL> ( "<STR_LIT>" , versions . VersionFromString ( v1 ) , f1 ) , <EOL> ( "<STR_LIT>" , versions . VersionFromString ( v1 ) , f2 ) ] ) <EOL> self . assertEqual ( ret , [ [ ( '<STR_LIT>' , versions . VersionFromString ( v2 ) , f1 ) ] , [ ] ] ) <EOL> self . addCollection ( "<STR_LIT>" , v2 , [ ( "<STR_LIT>" , v1 , f2 ) ] , defaultFlavor = f2 ) <EOL> ret = repos . getTroveReferences ( "<STR_LIT:localhost>" , [ <EOL> ( "<STR_LIT>" , versions . VersionFromString ( v1 ) , f1 ) , <EOL> ( "<STR_LIT>" , versions . VersionFromString ( v1 ) , f2 ) ] ) <EOL> self . assertEqual ( ret , [ <EOL> [ ( '<STR_LIT>' , versions . VersionFromString ( v2 ) , f1 ) ] , <EOL> [ ( '<STR_LIT>' , versions . VersionFromString ( v2 ) , f2 ) ] , <EOL> ] ) <EOL> def testCommitGroupScriptsOldRepo ( self ) : <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT:1.0>' ) <EOL> repos = self . openRepository ( ) <EOL> repos . c [ '<STR_LIT:localhost>' ] . setProtocolVersion ( <NUM_LIT> ) <EOL> try : <EOL> self . addCollection ( '<STR_LIT>' , '<STR_LIT:1.0>' , [ '<STR_LIT>' ] , <EOL> postUpdateScript = '<STR_LIT:foo>' , repos = repos ) <EOL> assert ( not '<STR_LIT>' ) <EOL> except errors . CommitError , e : <EOL> assert ( str ( e ) == '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> except : <EOL> assert ( not '<STR_LIT>' ) <EOL> def testRepositoryMapWildcards ( self ) : <EOL> self . servers . stopAllServers ( ) <EOL> repos = self . openRepository ( serverName = [ '<STR_LIT:localhost>' , '<STR_LIT>' ] ) <EOL> trv1 = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> trv2 = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> urls = list ( set ( x [ <NUM_LIT:1> ] for x in repos . c . map ) ) <EOL> assert ( len ( urls ) == <NUM_LIT:1> ) <EOL> newMap = conarycfg . RepoMap ( ) <EOL> newMap [ '<STR_LIT:*>' ] = urls [ <NUM_LIT:0> ] <EOL> wildcardRepos = self . getRepositoryClient ( repositoryMap = newMap ) <EOL> assert ( wildcardRepos . hasTrove ( * trv1 . getNameVersionFlavor ( ) ) ) <EOL> assert ( wildcardRepos . hasTrove ( * trv2 . getNameVersionFlavor ( ) ) ) <EOL> self . stopRepository ( <NUM_LIT:0> ) <EOL> def testErrorMessagesMentioningProxy ( self ) : <EOL> class RequestHandler ( SimpleHTTPServer . SimpleHTTPRequestHandler ) : <EOL> def log_message ( self , * args , ** kw ) : <EOL> pass <EOL> def do_GET ( self ) : <EOL> self . send_response ( <NUM_LIT:200> ) <EOL> response = '<STR_LIT>' <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . send_header ( '<STR_LIT>' , len ( response ) ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( response ) <EOL> httpServer = rephelp . HTTPServerController ( RequestHandler ) <EOL> sock_utils . tryConnect ( "<STR_LIT:localhost>" , httpServer . port ) <EOL> self . cfg . repositoryMap [ '<STR_LIT:localhost>' ] = '<STR_LIT>' <EOL> self . cfg . configLine ( "<STR_LIT>" % httpServer . port ) <EOL> repos = conaryclient . ConaryClient ( self . cfg ) . getRepos ( ) <EOL> try : <EOL> repos . c [ '<STR_LIT:localhost>' ] . getProtocolVersion ( ) <EOL> except errors . OpenError , e : <EOL> self . assertEqual ( str ( e ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % httpServer . port ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . cfg . resetToDefault ( '<STR_LIT>' ) <EOL> def testClientLog ( self ) : <EOL> p = self . workDir + '<STR_LIT>' <EOL> os . environ [ '<STR_LIT>' ] = p <EOL> repos = self . openRepository ( ) <EOL> del os . environ [ '<STR_LIT>' ] <EOL> log = calllog . ClientCallLogger ( p ) <EOL> methods = [ x . methodName for x in log ] <EOL> self . assertEqual ( methods , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def testRetryChangesetDownload ( self ) : <EOL> self . cfg . downloadRetryTrim = <NUM_LIT:16> <EOL> repos = self . openRepository ( ) <EOL> trv = self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' ) <EOL> job = [ trv . getNameVersionFlavor ( ) . asJob ( ) ] <EOL> host = trv . getVersion ( ) . getHost ( ) <EOL> clean = os . path . join ( self . workDir , '<STR_LIT>' ) <EOL> repos . createChangeSetFile ( job , clean ) <EOL> sp = repos . c [ host ] <EOL> orig_cs = sp . getChangeSet <EOL> did_truncate = [ False ] <EOL> def getChangeSet ( * args , ** kwargs ) : <EOL> rc = orig_cs ( * args , ** kwargs ) <EOL> fobj = rc [ <NUM_LIT:0> ] <EOL> spool = tempfile . TemporaryFile ( dir = self . workDir ) <EOL> size = util . copyfileobj ( fobj , spool ) <EOL> spool . seek ( size // <NUM_LIT:2> ) <EOL> spool . write ( '<STR_LIT>' ) <EOL> spool . truncate ( ) <EOL> spool . seek ( <NUM_LIT:0> ) <EOL> sp . getChangeSet = orig_cs <EOL> did_truncate [ <NUM_LIT:0> ] = True <EOL> return [ spool ] + rc [ <NUM_LIT:1> : ] <EOL> sp . getChangeSet = getChangeSet <EOL> did_truncate = [ False ] <EOL> self . cfg . downloadAttempts = <NUM_LIT:1> <EOL> self . assertRaises ( errors . TruncatedResponseError , <EOL> repos . createChangeSet , job ) <EOL> sp . getChangeSet = getChangeSet <EOL> did_truncate = [ False ] <EOL> self . cfg . downloadAttempts = <NUM_LIT:2> <EOL> retry = os . path . join ( self . workDir , '<STR_LIT>' ) <EOL> repos . createChangeSetFile ( job , retry ) <EOL> assert did_truncate [ <NUM_LIT:0> ] <EOL> self . assertEqual ( open ( clean ) . read ( ) , open ( retry ) . read ( ) ) <EOL> class ServerProxyTest ( rephelp . RepositoryHelper ) : <EOL> def testBadProtocol ( self ) : <EOL> port = self . cfg . repositoryMap [ '<STR_LIT:localhost>' ] = '<STR_LIT>' <EOL> repos = conaryclient . ConaryClient ( self . cfg ) . getRepos ( ) <EOL> try : <EOL> repos . troveNamesOnServer ( '<STR_LIT:localhost>' ) <EOL> except errors . OpenError , e : <EOL> self . assertEqual ( str ( e ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> "<STR_LIT>" ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def testProtocolVersionOverride ( self ) : <EOL> repos = self . openRepository ( ) <EOL> versions = repos . c [ '<STR_LIT:localhost>' ] . checkVersion ( ) <EOL> override = versions [ <NUM_LIT:0> ] - <NUM_LIT:1> <EOL> repos . c [ '<STR_LIT:localhost>' ] . setProtocolVersion ( override ) <EOL> try : <EOL> repos . c [ '<STR_LIT:localhost>' ] . checkVersion ( ) <EOL> except errors . InvalidClientVersion , e : <EOL> assert e . args [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' % override ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def testOverrideChangeSetVersion ( self ) : <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' ) <EOL> self . addCollection ( '<STR_LIT:foo>' , '<STR_LIT:1>' , [ '<STR_LIT>' ] ) <EOL> c = conaryclient . ConaryClient ( self . cfg ) <EOL> dest = os . path . join ( self . workDir , "<STR_LIT>" ) <EOL> ver = versions . VersionFromString ( '<STR_LIT>' % str ( self . defLabel ) ) <EOL> csVersion = changeset . getNativeChangesetVersion ( <NUM_LIT> ) <EOL> c . repos . createChangeSetFile ( [ ( '<STR_LIT:foo>' , ( None , None ) , <EOL> ( ver , deps . parseFlavor ( '<STR_LIT>' ) ) , True ) ] , <EOL> dest , changesetVersion = csVersion ) <EOL> fcont = filecontainer . FileContainer ( util . ExtendedFile ( <EOL> dest , buffering = False ) ) <EOL> self . assertEqual ( fcont . version , csVersion ) <EOL> def testOldClient ( self ) : <EOL> t = self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' ) <EOL> c = conaryclient . ConaryClient ( self . cfg ) <EOL> c . repos . getTrove ( t . getName ( ) , t . getVersion ( ) , t . getFlavor ( ) ) <EOL> c . repos . c [ '<STR_LIT:localhost>' ] . setProtocolVersion ( <NUM_LIT> ) <EOL> self . mock ( filecontainer , '<STR_LIT>' , <EOL> filecontainer . READABLE_VERSIONS [ <NUM_LIT:1> : ] ) <EOL> self . mock ( filecontainer , '<STR_LIT>' , <EOL> max ( filecontainer . READABLE_VERSIONS ) ) <EOL> dest = os . path . join ( self . workDir , "<STR_LIT>" ) <EOL> c . repos . createChangeSetFile ( [ ( t . getName ( ) , ( None , None ) , <EOL> ( t . getVersion ( ) , t . getFlavor ( ) ) , True ) ] , <EOL> dest ) <EOL> fcont = filecontainer . FileContainer ( util . ExtendedFile ( <EOL> dest , buffering = False ) ) <EOL> csVersion = changeset . getNativeChangesetVersion ( <NUM_LIT> ) <EOL> self . assertEqual ( fcont . version , csVersion ) <EOL> def testMultitenantServers ( self ) : <EOL> self . stopRepository ( <NUM_LIT:0> ) <EOL> repos = self . openRepository ( <NUM_LIT:0> , serverName = [ '<STR_LIT:localhost>' , '<STR_LIT>' ] ) <EOL> trv1 = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> trv2 = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> repos . createChangeSet ( [ ( '<STR_LIT>' , <EOL> ( trv1 . getVersion ( ) , trv1 . getFlavor ( ) ) , <EOL> ( trv2 . getVersion ( ) , trv2 . getFlavor ( ) ) , False ) ] ) <EOL> assert ( repos . c [ '<STR_LIT:localhost>' ] is not repos . c [ '<STR_LIT>' ] ) <EOL> self . servers . stopAllServers ( ) <EOL> def testJobSizes ( self ) : <EOL> def _check ( job , size ) : <EOL> fn = self . workDir + '<STR_LIT>' <EOL> cs1 = repos . createChangeSetFile ( job , fn ) <EOL> assert ( os . stat ( fn ) . st_size == size ) <EOL> first = self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' ) <EOL> firstJob = ( '<STR_LIT>' , ( None , None ) , <EOL> ( first . getVersion ( ) , first . getFlavor ( ) ) , True ) <EOL> second = self . addComponent ( '<STR_LIT>' , '<STR_LIT:1>' , <EOL> fileContents = [ '<STR_LIT>' ] ) <EOL> secondJob = ( '<STR_LIT>' , ( None , None ) , <EOL> ( second . getVersion ( ) , second . getFlavor ( ) ) , True ) <EOL> removeJob = ( '<STR_LIT>' , ( second . getVersion ( ) , second . getFlavor ( ) ) , <EOL> ( None , None ) , True ) <EOL> repos = self . openRepository ( ) <EOL> assert ( repos . getChangeSetSize ( [ removeJob ] ) [ <NUM_LIT:0> ] == <NUM_LIT:0> ) <EOL> sizes = repos . getChangeSetSize ( [ firstJob , removeJob , secondJob ] ) <EOL> _check ( [ firstJob ] , sizes [ <NUM_LIT:0> ] ) <EOL> assert ( sizes [ <NUM_LIT:1> ] == <NUM_LIT:0> ) <EOL> _check ( [ secondJob ] , sizes [ <NUM_LIT:2> ] ) <EOL> repos . c [ '<STR_LIT:localhost>' ] . setProtocolVersion ( <NUM_LIT:50> ) <EOL> otherSizes = repos . getChangeSetSize ( [ firstJob , removeJob , secondJob ] ) <EOL> assert ( sizes == otherSizes ) <EOL> def testServerProxyMethods ( self ) : <EOL> sp = netclient . ServerProxy ( '<STR_LIT>' , '<STR_LIT:localhost>' , None , <EOL> None , None ) <EOL> self . assertFalse ( hasattr ( sp , '<STR_LIT>' ) ) <EOL> def testPathIdPermissions ( self ) : <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> foo1 = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , sourceName = '<STR_LIT>' , <EOL> fileContents = [ ( '<STR_LIT>' , rephelp . RegularFile ( pathId = "<STR_LIT:1>" ) ) <EOL> ] ) <EOL> foo2 = self . addComponent ( '<STR_LIT>' , '<STR_LIT>' , sourceName = '<STR_LIT>' , <EOL> fileContents = [ ( '<STR_LIT>' , rephelp . RegularFile ( pathId = "<STR_LIT:2>" ) ) <EOL> ] ) <EOL> branch = foo1 . getVersion ( ) . branch ( ) <EOL> repos = self . openRepository ( ) <EOL> repos . deleteUserByName ( self . cfg . buildLabel , '<STR_LIT>' ) <EOL> repos . addUser ( self . cfg . buildLabel , '<STR_LIT>' , '<STR_LIT>' ) <EOL> repos . addRole ( self . cfg . buildLabel , '<STR_LIT>' ) <EOL> repos . updateRoleMembers ( self . cfg . buildLabel , '<STR_LIT>' , [ '<STR_LIT>' ] ) <EOL> otherClient = self . getRepositoryClient ( user = '<STR_LIT>' , <EOL> password = '<STR_LIT>' ) <EOL> ids = repos . getPackageBranchPathIds ( '<STR_LIT>' , branch ) <EOL> assert ( ids [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == '<STR_LIT>' ) <EOL> assert ( ids [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == '<STR_LIT>' ) <EOL> assert ( len ( ids ) == <NUM_LIT:2> ) <EOL> ids = otherClient . getPackageBranchPathIds ( '<STR_LIT>' , branch ) <EOL> assert ( not ids ) <EOL> repos . addTroveAccess ( '<STR_LIT>' , [ foo1 . getNameVersionFlavor ( ) ] ) <EOL> ids = otherClient . getPackageBranchPathIds ( '<STR_LIT>' , branch ) <EOL> assert ( ids [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == '<STR_LIT>' ) <EOL> assert ( len ( ids ) == <NUM_LIT:1> ) <EOL> repos . addTroveAccess ( '<STR_LIT>' , [ foo2 . getNameVersionFlavor ( ) ] ) <EOL> ids = otherClient . getPackageBranchPathIds ( '<STR_LIT>' , branch ) <EOL> assert ( ids [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == '<STR_LIT>' ) <EOL> assert ( ids [ '<STR_LIT>' ] [ <NUM_LIT:0> ] == '<STR_LIT>' ) <EOL> assert ( len ( ids ) == <NUM_LIT:2> ) <EOL> def testHttpProxyHttpsServer ( self ) : <EOL> if not conaryserver . SSL : <EOL> raise testhelp . SkipTestException ( "<STR_LIT>" ) <EOL> self . stopRepository ( <NUM_LIT:1> ) <EOL> self . openRepository ( <NUM_LIT:1> , useSSL = True ) <EOL> try : <EOL> proxy = self . getConaryProxy ( ) <EOL> proxy . start ( ) <EOL> try : <EOL> proxy . addToConfig ( self . cfg ) <EOL> client = conaryclient . ConaryClient ( self . cfg ) <EOL> repos = client . getRepos ( ) <EOL> repos . c [ '<STR_LIT>' ] . checkVersion ( ) <EOL> finally : <EOL> proxy . stop ( ) <EOL> finally : <EOL> self . stopRepository ( <NUM_LIT:1> ) <EOL> def testUnderUnderMethodName ( self ) : <EOL> repos = self . openRepository ( ) <EOL> self . assertRaises ( AttributeError , <EOL> lambda : repos . c [ '<STR_LIT:localhost>' ] . __foo . bar ) <EOL> self . assertRaises ( AttributeError , <EOL> lambda : repos . c [ '<STR_LIT:localhost>' ] . foo . __bar ) <EOL> self . assertRaises ( AttributeError , <EOL> lambda : repos . c [ '<STR_LIT:localhost>' ] . getTroveLeavesByLabel . __safe_str__ ) <EOL> def testPermissionFallbackCode ( self ) : <EOL> m = ( ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> class MockServerProxy : <EOL> def __init__ ( self ) : <EOL> self . methodCalled = None <EOL> def getProtocolVersion ( self ) : <EOL> return <NUM_LIT> <EOL> def __getattr__ ( self , attr ) : <EOL> self . methodCalled = attr <EOL> def f ( * args , ** kw ) : <EOL> return [ ] <EOL> return f <EOL> empty = conarycfg . ConaryConfiguration ( False ) <EOL> repos = netclient . NetworkRepositoryClient ( cfg = empty ) <EOL> server = MockServerProxy ( ) <EOL> repos . c = { '<STR_LIT:localhost>' : server } <EOL> for method , serverMethod in m : <EOL> f = getattr ( repos , method ) <EOL> args = ( '<STR_LIT:localhost>' , ) + ( [ ] , ) * ( f . func_code . co_argcount - <NUM_LIT:2> ) <EOL> f ( * args ) <EOL> self . assertEqual ( server . methodCalled , serverMethod ) <EOL> def testFakeProtocolVersion ( self ) : <EOL> '''<STR_LIT>''' <EOL> repos = self . openRepository ( ) <EOL> repos . c [ '<STR_LIT:localhost>' ] . checkVersion ( protocolVersion = <NUM_LIT:50> ) <EOL> def testUpdateUsesCache ( self ) : <EOL> ipcache = httputils . IPCache <EOL> ipcache . clear ( ) <EOL> repos = self . openRepository ( ) <EOL> if repos . c . proxyMap : <EOL> return <EOL> repos = self . openRepository ( <NUM_LIT:1> ) <EOL> self . addComponent ( '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' , filePrimer = <NUM_LIT:1> ) <EOL> oldCacheHostLookups = repos . _cacheHostLookups <EOL> def checkCacheHostLookups ( self , hosts ) : <EOL> ipcache . clear ( ) <EOL> self . assertEqual ( set ( hosts ) , set ( [ '<STR_LIT:localhost>' , '<STR_LIT>' ] ) ) <EOL> rv = oldCacheHostLookups ( hosts ) <EOL> self . assertEqual ( ipcache . _cache . _map . keys ( ) , [ ] ) <EOL> return rv <EOL> self . mock ( netclient . NetworkRepositoryClient , <EOL> '<STR_LIT>' , checkCacheHostLookups ) <EOL> def testUpdateUsesCache2 ( self ) : <EOL> ipcache = httputils . IPCache <EOL> ipcache . clear ( ) <EOL> oldLocalHosts = httputils . LocalHosts <EOL> try : <EOL> httputils . LocalHosts = set ( ) <EOL> repos = self . openRepository ( ) <EOL> if repos . c . proxyMap : <EOL> return <EOL> repos = self . openRepository ( <NUM_LIT:1> ) <EOL> self . addComponent ( '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' , filePrimer = <NUM_LIT:1> ) <EOL> self . assertEqual ( ipcache . _cache . _map . keys ( ) , [ '<STR_LIT:localhost>' ] ) <EOL> oldCacheHostLookups = repos . _cacheHostLookups <EOL> def checkCacheHostLookups ( self , hosts ) : <EOL> ipcache . clear ( ) <EOL> self . assertEqual ( set ( hosts ) , set ( [ '<STR_LIT:localhost>' , '<STR_LIT>' ] ) ) <EOL> rv = oldCacheHostLookups ( hosts ) <EOL> self . assertEqual ( ipcache . _cache . _map . keys ( ) , [ '<STR_LIT:localhost>' ] ) <EOL> return rv <EOL> self . mock ( netclient . NetworkRepositoryClient , <EOL> '<STR_LIT>' , checkCacheHostLookups ) <EOL> self . updatePkg ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> finally : <EOL> httputils . LocalHosts = oldLocalHosts <EOL> def testGetChangeSetFingerprints ( self ) : <EOL> self . openRepository ( <NUM_LIT:0> ) <EOL> repos = self . openRepository ( <NUM_LIT:1> ) <EOL> v1 = versions . VersionFromString ( '<STR_LIT>' ) <EOL> v2 = versions . VersionFromString ( '<STR_LIT>' ) <EOL> v3 = versions . VersionFromString ( '<STR_LIT>' ) <EOL> v4 = versions . VersionFromString ( '<STR_LIT>' ) <EOL> f = deps . parseFlavor ( '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' , str ( v1 ) ) <EOL> self . addComponent ( '<STR_LIT>' , str ( v2 ) ) <EOL> self . addComponent ( '<STR_LIT>' , str ( v3 ) ) <EOL> self . addComponent ( '<STR_LIT>' , str ( v4 ) ) <EOL> jobList = [ ( '<STR_LIT>' , ( None , None ) , ( v1 , f ) , False ) , <EOL> ( '<STR_LIT>' , ( None , None ) , ( v2 , f ) , False ) , <EOL> ( '<STR_LIT>' , ( None , None ) , ( v3 , f ) , False ) , <EOL> ( '<STR_LIT>' , ( None , None ) , ( v4 , f ) , False ) ] <EOL> expectedList = [ <EOL> ( dict ( recurse = False , <EOL> withFiles = False , <EOL> withFileContents = False , <EOL> excludeAutoSource = False , <EOL> mirrorMode = False ) , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> ( dict ( recurse = True , <EOL> withFiles = False , <EOL> withFileContents = False , <EOL> excludeAutoSource = False , <EOL> mirrorMode = False ) , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> ( dict ( recurse = False , <EOL> withFiles = True , <EOL> withFileContents = False , <EOL> excludeAutoSource = False , <EOL> mirrorMode = False ) , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> ( dict ( recurse = False , <EOL> withFiles = False , <EOL> withFileContents = True , <EOL> excludeAutoSource = False , <EOL> mirrorMode = False ) , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> ( dict ( recurse = False , <EOL> withFiles = False , <EOL> withFileContents = False , <EOL> excludeAutoSource = True , <EOL> mirrorMode = False ) , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> ( dict ( recurse = False , <EOL> withFiles = False , <EOL> withFileContents = False , <EOL> excludeAutoSource = False , <EOL> mirrorMode = True ) , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> ] <EOL> failed = False <EOL> for i , ( kwargs , expected ) in enumerate ( expectedList ) : <EOL> csf = repos . getChangeSetFingerprints ( jobList , <EOL> ** kwargs ) <EOL> if csf != expected : <EOL> if not failed : <EOL> print <EOL> print "<STR_LIT>" , i <EOL> print "<STR_LIT>" , csf <EOL> print "<STR_LIT>" , expected <EOL> failed = True <EOL> if failed : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def testGetFileContentsFromTrove ( self ) : <EOL> trv = self . addComponent ( '<STR_LIT>' , [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> repos = self . openRepository ( ) <EOL> n , v , f = trv . getNameVersionFlavor ( ) <EOL> contents = repos . getFileContentsFromTrove ( n , v , f , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> assert ( contents [ <NUM_LIT:0> ] . get ( ) . read ( ) == '<STR_LIT>' ) <EOL> assert ( contents [ <NUM_LIT:1> ] . get ( ) . read ( ) == '<STR_LIT>' ) <EOL> self . assertRaises ( errors . PathsNotFound , <EOL> repos . getFileContentsFromTrove , <EOL> n , v , f , <EOL> [ '<STR_LIT>' ] ) <EOL> repos . c [ '<STR_LIT:localhost>' ] . getProtocolVersion = lambda : <NUM_LIT> <EOL> contents = repos . getFileContentsFromTrove ( n , v , f , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> assert ( contents [ <NUM_LIT:0> ] . get ( ) . read ( ) == '<STR_LIT>' ) <EOL> assert ( contents [ <NUM_LIT:1> ] . get ( ) . read ( ) == '<STR_LIT>' ) <EOL> def testGetLabelsForHost ( self ) : <EOL> try : <EOL> self . openRepository ( <NUM_LIT:0> , serverName = [ '<STR_LIT:localhost>' , '<STR_LIT>' ] ) <EOL> self . addComponent ( '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' ) <EOL> self . addComponent ( '<STR_LIT>' ) <EOL> self . addComponent ( <EOL> '<STR_LIT>' ) <EOL> labels = self . openRepository ( ) . getLabelsForHost ( '<STR_LIT:localhost>' ) <EOL> assert ( sorted ( [ str ( x ) for x in labels ] ) == <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> labels = self . openRepository ( ) . getLabelsForHost ( '<STR_LIT>' ) <EOL> assert ( sorted ( [ str ( x ) for x in labels ] ) == <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> finally : <EOL> self . resetRepository ( <NUM_LIT:0> ) <EOL> self . stopRepository ( <NUM_LIT:0> ) </s>
