|
#!/usr/bin/python
|
|
|
|
import rados
|
|
import rbd
|
|
import uuid
|
|
import hashlib
|
|
import unittest
|
|
|
|
class RBDTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.cluster = rados.Rados(conffile='/etc/ceph/ceph.conf')
|
|
self.cluster.connect()
|
|
self.ioctx = self.cluster.open_ioctx('rbd')
|
|
self.rbd_inst = rbd.RBD()
|
|
self.image_name = str(uuid.uuid4())
|
|
image_size = 16 * 1024**2 # 16 MiB
|
|
self.rbd_inst.create(self.ioctx, self.image_name, image_size)
|
|
|
|
|
|
def test_first_object(self):
|
|
with rbd.Image(self.ioctx, self.image_name) as image:
|
|
# Overwrite the entire image
|
|
data = 'a' * image.size()
|
|
image.write(data,0)
|
|
# Grab the first object and its size
|
|
prefix = image.stat()['block_name_prefix']
|
|
o = rados.Object(self.ioctx, '%s.000000000000' % prefix)
|
|
length = o.stat()[0]
|
|
# Make sure that what we read back from the first object
|
|
# matches the first X bytes of the data we put in,
|
|
# where X is the length of the first object
|
|
image_data = image.read(0,length)
|
|
object_data = o.read(length)
|
|
image_digest = hashlib.md5(image_data).hexdigest()
|
|
object_digest = hashlib.md5(object_data).hexdigest()
|
|
self.assertEqual(image_digest, object_digest)
|
|
|
|
|
|
def test_remove_object(self):
|
|
with rbd.Image(self.ioctx, self.image_name) as image:
|
|
# Overwrite the entire image
|
|
data = 'a' * image.size()
|
|
image.write(data,0)
|
|
# Grab the first object and its size
|
|
prefix = image.stat()['block_name_prefix']
|
|
o = rados.Object(self.ioctx, '%s.000000000000' % prefix)
|
|
length = o.stat()[0]
|
|
# Zap the object
|
|
o.remove()
|
|
# Now read back the length of the object from the image,
|
|
# hitting the just-deleted object
|
|
image.read(0, length)
|
|
self.assertRaises(rados.ObjectNotFound, image.read, 0, length)
|
|
|
|
|
|
def tearDown(self):
|
|
self.rbd_inst.remove(self.ioctx, self.image_name)
|
|
self.ioctx.close()
|
|
self.cluster.shutdown()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|