conddbCopyTest Namespace Reference

Classes
class	CopyTest
class	DB

Functions
def	copy (sourceTag, destTag, since, logFileName)
def	main ()
def	prepareFile (inputTag, sourceTag, startingSince)

Variables
string	fileName = 'copy_test.db'

Function Documentation

copy()

def conddbCopyTest.copy	(		sourceTag,
			destTag,
			since,
			logFileName
	)

Definition at line 35 of file conddbCopyTest.py.

References print().

Referenced by conddbCopyTest.CopyTest.execute().

def copy( sourceTag, destTag, since, logFileName ):
    command = "conddb --yes --db %s copy %s %s -f %s --synchronize" %(fileName,sourceTag,destTag,since)
    pipe = subprocess.Popen( command, shell=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
    out = pipe.communicate()[0]
    lines = out.split('\n')
    ret = pipe.returncode
    for line in lines:
        print(line)
    with open(logFileName,'a') as logFile:
        logFile.write(out)
    return ret==0

main()

def conddbCopyTest.main

(

)

Definition at line 104 of file conddbCopyTest.py.

References prepareFile(), and print().

def main():
    print('Testing...')
    bfile0 = fileName
    bfile1 = fileName
    db = DB()
    inputTag = 'runinfo_31X_mc'
    inputTag0  ='runinfo_0'
    inputTag1 = 'runinfo_1'
    prepareFile( inputTag,inputTag0,1)
    prepareFile( inputTag,inputTag1,100)
    test = CopyTest( db )
    # test with synch=any
    tag = 'test_CondUpload_any'
    test.execute( inputTag0, bfile0, tag, 'any', 1, True, 'CREATE' )
    test.execute( inputTag0, bfile0, tag, 'any', 200, True, 'APPEND' )  
    test.execute( inputTag0, bfile0, tag, 'any', 100, True, 'INSERT')  
    test.execute( inputTag0, bfile0, tag, 'any', 200, True, 'INSERT')  
    # test with synch=validation
    tag = 'test_CondUpload_validation'
    test.execute( inputTag0, bfile0, tag, 'validation', 1, True, 'CREATE')  
    db.setSynchronizationType( tag, 'validation' ) 
    test.execute( inputTag0, bfile0, tag, 'validation', 1, True, 'INSERT')  
    test.execute( inputTag0, bfile0, tag, 'validation', 200, True, 'APPEND')  
    test.execute( inputTag0, bfile0, tag, 'validation', 100, True, 'INSERT')  
    # test with synch=mc
    tag = 'test_CondUpload_mc'
    test.execute( inputTag0, bfile0, tag, 'mc', 1, True, 'CREATE')  
    db.setSynchronizationType( tag, 'mc' ) 
    test.execute( inputTag1, bfile1, tag, 'mc', 1, False, 'FAIL')  
    test.execute( inputTag0, bfile0, tag, 'mc', 1, False, 'FAIL')  
    test.execute( inputTag0, bfile0, tag, 'mc', 200, False, 'FAIL') 
    # test with synch=hlt
    tag = 'test_CondUpload_hlt'
    test.execute( inputTag0, bfile0, tag, 'hlt', 1, True, 'CREATE')  
    db.setSynchronizationType( tag, 'hlt' ) 
    test.execute( inputTag0, bfile0, tag, 'hlt', 200, True, 'SYNCHRONIZE')  
    fcsr = test.execute( inputTag0, bfile0, tag, 'hlt', 100, True, 'SYNCHRONIZE')  
    if not fcsr is None:
        since = fcsr + 200
        test.execute( inputTag0, bfile0, tag, 'hlt', since, True, 'APPEND')  
        since = fcsr + 100
        test.execute( inputTag0, bfile0, tag, 'hlt', since, True, 'INSERT')  
    # test with synch=express
    tag = 'test_CondUpload_express'
    test.execute( inputTag0, bfile0, tag, 'express', 1, True, 'CREATE')  
    db.setSynchronizationType( tag, 'express' ) 
    test.execute( inputTag0, bfile0, tag, 'express', 200, True, 'SYNCHRONIZE')  
    fcsr = test.execute( inputTag0, bfile0, tag, 'express', 100, True, 'SYNCHRONIZE')  
    if not fcsr is None:
        since = fcsr + 200
        test.execute( inputTag0, bfile0, tag, 'express', since, True, 'APPEND')  
        since = fcsr + 100
        test.execute( inputTag0, bfile0, tag, 'express', since, True, 'INSERT')  
    # test with synch=prompt
    tag = 'test_CondUpload_prompt'
    test.execute( inputTag0, bfile0, tag, 'prompt', 1, True, 'CREATE')  
    db.setSynchronizationType( tag, 'prompt' ) 
    test.execute( inputTag0, bfile0, tag, 'prompt', 200, True, 'SYNCHRONIZE')  
    fcsr = test.execute( inputTag0, bfile0, tag, 'prompt', 100, True, 'SYNCHRONIZE')  
    if not fcsr is None:
        since = fcsr + 200
        test.execute( inputTag0, bfile0, tag, 'prompt', since, True, 'APPEND')  
        since = fcsr + 100
        test.execute( inputTag0, bfile0, tag, 'prompt', since, True, 'INSERT')  
    # test with synch=pcl
    tag = 'test_CondUpload_pcl'
    test.execute( inputTag0, bfile0, tag, 'pcl', 1, True, 'CREATE')  
    db.setSynchronizationType( tag, 'pcl' ) 
    test.execute( inputTag0, bfile0, tag, 'pcl', 200, False, 'FAIL')  
    if not fcsr is None:
        since = fcsr + 200
        test.execute( inputTag0, bfile0, tag, 'pcl', since, True, 'APPEND')  
        since = fcsr + 100
        test.execute( inputTag0, bfile0, tag, 'pcl', since, True, 'INSERT')  
    # test with synch=offline
    tag = 'test_CondUpload_offline'
    test.execute( inputTag0, bfile0, tag, 'offline', 1, True, 'CREATE')  
    db.setSynchronizationType( tag, 'offline' ) 
    test.execute( inputTag0, bfile0, tag, 'offline', 1000, True, 'APPEND')
    test.execute( inputTag0, bfile0, tag, 'offline', 500, False, 'FAIL' ) 
    test.execute( inputTag0, bfile0, tag, 'offline', 1000, False, 'FAIL' ) 
    test.execute( inputTag0, bfile0, tag, 'offline', 2000, True, 'APPEND' ) 
    # test with synch=runmc
    tag = 'test_CondUpload_runmc'
    test.execute( inputTag0, bfile0, tag, 'runmc', 1, True, 'CREATE')  
    db.setSynchronizationType( tag, 'runmc' ) 
    test.execute( inputTag0, bfile0, tag, 'runmc', 1000, True, 'APPEND')
    test.execute( inputTag0, bfile0, tag, 'runmc', 500, False, 'FAIL' ) 
    test.execute( inputTag0, bfile0, tag, 'runmc', 1000, False, 'FAIL' ) 
    test.execute( inputTag0, bfile0, tag, 'runmc', 2000, True, 'APPEND' )
    os.remove( fileName )
    print('Done. Errors: %s' %test.errors)

    

prepareFile()

def conddbCopyTest.prepareFile	(		inputTag,
			sourceTag,
			startingSince
	)

Definition at line 30 of file conddbCopyTest.py.

Referenced by main().

def prepareFile( inputTag, sourceTag, startingSince ):
    command = "conddb --yes copy %s %s --destdb %s -f %s" %(inputTag,sourceTag,fileName,startingSince)
    pipe = subprocess.Popen( command, shell=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
    out = pipe.communicate()[0]

Variable Documentation

fileName

string conddbCopyTest.fileName = 'copy_test.db'

Definition at line 11 of file conddbCopyTest.py.