Commit 6649801e authored by Andrea Valassi's avatar Andrea Valassi Committed by cdelort
Browse files

Add a test for Blobs in PyCool (CORALCOOL-2997)

git-svn-id: https://svnweb.cern.ch/trac/lcgcool/changeset/20302 4525493e-7705-40b1-a816-d608a930855b
parent 6b0b0a2c
......@@ -22,6 +22,7 @@ include(CORALConfigScripts)
copy_and_install_python_test(test_ChannelSelection.py)
copy_and_install_python_test(test_Channels.py)
copy_and_install_python_test(test_CoolFunctionality.py)
copy_and_install_python_test(test_coolBlob.py)
copy_and_install_python_test(test_coralAttributeList.py)
copy_and_install_python_test(test_coralBlob.py)
copy_and_install_python_test(test_FolderSpecification.py)
......
#!/usr/bin/env python
import unittest, sys, os, time
from PyCool import cool, coral
import traceback
connectString = None
class TestCoolBlob( unittest.TestCase ):
rspec = None
unittest.TestCase.shared_db = None
def setUp(self):
try:
if self.rspec is None:
self.rspec = cool.RecordSpecification()
self.rspec.extend( "I", cool.StorageType.Int32 )
self.rspec.extend( "B", cool.StorageType.Blob64k )
self.fspecSV = cool.FolderSpecification( cool.FolderVersioning.SINGLE_VERSION, self.rspec )
if unittest.TestCase.shared_db is None:
dbSvc = cool.DatabaseSvcFactory.databaseService()
dbSvc.dropDatabase(connectString)
unittest.TestCase.shared_db = dbSvc.createDatabase(connectString)
else:
unittest.TestCase.shared_db.refreshDatabase()
self.db = unittest.TestCase.shared_db
except Exception, e:
print e
print "could not recreate the database"
print "check your seal.opts and authentication.xml"
sys.exit(-1)
except:
print "Unexpected error:", sys.exc_info()[0]
print traceback.print_tb(sys.exc_info()[2])
sys.exit(-1)
def tearDown(self):
#del self.db
pass
# Test the PyCool Blob file-like interface
# NB: This is a PyCool specific interface, it is not the CoralBase API!
# See https://docs.python.org/2.4/lib/bltin-file-objects.html
# - seek(offset[,whence])
# Set current position, like stdio's fseek(). The whence argument is
# optional and defaults to 0 (absolute positioning); other values are 1
# (seek relative to current position) and 2 (seek relative to blob end).
# - read([size])
# Read at most size bytes from blob (less if hits the blob end before).
# If size is negative or omitted, read all data until the blob end.
# - write(str)
# Write a string to the blob.
# - tell()
# Return the blob's current position, like stdio's ftell().
def test_coolBlob(self):
f = self.db.createFolder( '/a', self.fspecSV )
spec = f.payloadSpecification()
self.assertEquals( 2, spec.size() )
self.assertEquals( cool.FolderVersioning.SINGLE_VERSION,
f.versioningMode() )
# Store payload: 20 characters '0123456789abcdefghij'
payload = cool.Record( self.rspec )
payload["I"] = 1
payload['B'].write('0123456789')
payload['B'].write('abcdefghij') # WARNING! This does NOT append!
self.assert_( 20 != len(payload['B']) )
self.assert_( 10 == len(payload['B']) )
blob = coral.Blob()
blob.write('0123456789')
blob.write('abcdefghij') # This DOES append...
payload['B'] = blob
self.assert_( 20 == len(payload['B']) )
f.storeObject( 0, 2, payload, 0 )
# Read back payload
obj = f.findObject( 1, 0 )
self.assert_( obj.payload()['I'] == 1 )
print
# Read from obj.payload()['B']... get a copy each time...
print 'obj.payload()[\'B\'] =', obj.payload()['B']
print 'obj.payload()[\'B\'].read() [initial] =', obj.payload()['B'].read() #1st read...
print 'obj.payload()[\'B\'].read() [after 1st read] =', obj.payload()['B'].read() #2nd
self.assert_( obj.payload()['B'].read() == '0123456789abcdefghij' ) #3rd
self.assert_( obj.payload()['B'].read() == '0123456789abcdefghij' ) #4th
# Get a single copy into a coral::Blob...
blob = obj.payload()['B']
print 'blob =', blob
print 'str(blob) =', str(blob)
self.assert_( blob != '0123456789abcdefghij' ) # NOT a str!
s = blob.read()
print 'blob.read() [initial] =', s
self.assert_( s == '0123456789abcdefghij' )
s = blob.read()
print 'blob.read() [after 1st read] =', s
self.assert_( s == '' ) # Nothing else to read!
#######################################################################
envKey = "COOLTESTDB"
if __name__ == '__main__':
if ( len(sys.argv) == 2
and not sys.argv[1].startswith( 'TestCoolBlob' ) ):
connectString = sys.argv[1]
elif envKey in os.environ:
connectString = os.environ[envKey]
else:
print 'usage:', sys.argv[0], '<connect string>'
print '<connect string>: a COOL (RAL) compatible connect string, e.g.'
print ( ' "oracle://devdb10;schema=atlas_cool_sas;'
'user=atlas_cool_sas;dbname=COOLTEST"' )
print 'or set the environment variable %s'%(envKey)
sys.exit(-1)
#print 'connect string', connectString
unittest.main( testRunner =
unittest.TextTestRunner(stream=sys.stdout,verbosity=2) )
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment