diff options
Diffstat (limited to 'src/test/pybind/test_rados.py')
| -rw-r--r-- | src/test/pybind/test_rados.py | 1538 |
1 files changed, 1538 insertions, 0 deletions
diff --git a/src/test/pybind/test_rados.py b/src/test/pybind/test_rados.py new file mode 100644 index 000000000..ff6fee3db --- /dev/null +++ b/src/test/pybind/test_rados.py @@ -0,0 +1,1538 @@ +from __future__ import print_function +from nose import SkipTest +from nose.plugins.attrib import attr +from nose.tools import eq_ as eq, ok_ as ok, assert_raises +from rados import (Rados, Error, RadosStateError, Object, ObjectExists, + ObjectNotFound, ObjectBusy, NotConnected, + LIBRADOS_ALL_NSPACES, WriteOpCtx, ReadOpCtx, LIBRADOS_CREATE_EXCLUSIVE, + LIBRADOS_CMPXATTR_OP_EQ, LIBRADOS_CMPXATTR_OP_GT, LIBRADOS_CMPXATTR_OP_LT, OSError, + LIBRADOS_SNAP_HEAD, LIBRADOS_OPERATION_BALANCE_READS, LIBRADOS_OPERATION_SKIPRWLOCKS, MonitorLog, MAX_ERRNO, NoData, ExtendMismatch) +from datetime import timedelta +import time +import threading +import json +import errno +import os +import re +import sys + +def test_rados_init_error(): + assert_raises(Error, Rados, conffile='', rados_id='admin', + name='client.admin') + assert_raises(Error, Rados, conffile='', name='invalid') + assert_raises(Error, Rados, conffile='', name='bad.invalid') + +def test_rados_init(): + with Rados(conffile='', rados_id='admin'): + pass + with Rados(conffile='', name='client.admin'): + pass + with Rados(conffile='', name='client.admin'): + pass + with Rados(conffile='', name='client.admin'): + pass + +def test_ioctx_context_manager(): + with Rados(conffile='', rados_id='admin') as conn: + with conn.open_ioctx('rbd') as ioctx: + pass + +def test_parse_argv(): + args = ['osd', 'pool', 'delete', 'foobar', 'foobar', '--yes-i-really-really-mean-it'] + r = Rados() + eq(args, r.conf_parse_argv(args)) + +def test_parse_argv_empty_str(): + args = [''] + r = Rados() + eq(args, r.conf_parse_argv(args)) + +class TestRadosStateError(object): + def _requires_configuring(self, rados): + assert_raises(RadosStateError, rados.connect) + + def _requires_configuring_or_connected(self, rados): + assert_raises(RadosStateError, rados.conf_read_file) + assert_raises(RadosStateError, rados.conf_parse_argv, None) + assert_raises(RadosStateError, rados.conf_parse_env) + assert_raises(RadosStateError, rados.conf_get, 'opt') + assert_raises(RadosStateError, rados.conf_set, 'opt', 'val') + assert_raises(RadosStateError, rados.ping_monitor, '0') + + def _requires_connected(self, rados): + assert_raises(RadosStateError, rados.pool_exists, 'foo') + assert_raises(RadosStateError, rados.pool_lookup, 'foo') + assert_raises(RadosStateError, rados.pool_reverse_lookup, 0) + assert_raises(RadosStateError, rados.create_pool, 'foo') + assert_raises(RadosStateError, rados.get_pool_base_tier, 0) + assert_raises(RadosStateError, rados.delete_pool, 'foo') + assert_raises(RadosStateError, rados.list_pools) + assert_raises(RadosStateError, rados.get_fsid) + assert_raises(RadosStateError, rados.open_ioctx, 'foo') + assert_raises(RadosStateError, rados.mon_command, '', b'') + assert_raises(RadosStateError, rados.osd_command, 0, '', b'') + assert_raises(RadosStateError, rados.pg_command, '', '', b'') + assert_raises(RadosStateError, rados.wait_for_latest_osdmap) + assert_raises(RadosStateError, rados.blocklist_add, '127.0.0.1/123', 0) + + def test_configuring(self): + rados = Rados(conffile='') + eq('configuring', rados.state) + self._requires_connected(rados) + + def test_connected(self): + rados = Rados(conffile='') + with rados: + eq('connected', rados.state) + self._requires_configuring(rados) + + def test_shutdown(self): + rados = Rados(conffile='') + with rados: + pass + eq('shutdown', rados.state) + self._requires_configuring(rados) + self._requires_configuring_or_connected(rados) + self._requires_connected(rados) + + +class TestRados(object): + + def setUp(self): + self.rados = Rados(conffile='') + self.rados.conf_parse_env('FOO_DOES_NOT_EXIST_BLAHBLAH') + self.rados.conf_parse_env() + self.rados.connect() + + # Assume any pre-existing pools are the cluster's defaults + self.default_pools = self.rados.list_pools() + + def tearDown(self): + self.rados.shutdown() + + def test_ping_monitor(self): + assert_raises(ObjectNotFound, self.rados.ping_monitor, 'not_exists_monitor') + cmd = {'prefix': 'mon dump', 'format':'json'} + ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'') + for mon in json.loads(buf.decode('utf8'))['mons']: + while True: + output = self.rados.ping_monitor(mon['name']) + if output is None: + continue + buf = json.loads(output) + if buf.get('health'): + break + + def test_annotations(self): + with assert_raises(TypeError): + self.rados.create_pool(0xf00) + + def test_create(self): + self.rados.create_pool('foo') + self.rados.delete_pool('foo') + + def test_create_utf8(self): + poolname = "\u9ec4" + self.rados.create_pool(poolname) + assert self.rados.pool_exists(u"\u9ec4") + self.rados.delete_pool(poolname) + + def test_pool_lookup_utf8(self): + poolname = '\u9ec4' + self.rados.create_pool(poolname) + try: + poolid = self.rados.pool_lookup(poolname) + eq(poolname, self.rados.pool_reverse_lookup(poolid)) + finally: + self.rados.delete_pool(poolname) + + def test_eexist(self): + self.rados.create_pool('foo') + assert_raises(ObjectExists, self.rados.create_pool, 'foo') + self.rados.delete_pool('foo') + + def list_non_default_pools(self): + pools = self.rados.list_pools() + for p in self.default_pools: + pools.remove(p) + return set(pools) + + def test_list_pools(self): + eq(set(), self.list_non_default_pools()) + self.rados.create_pool('foo') + eq(set(['foo']), self.list_non_default_pools()) + self.rados.create_pool('bar') + eq(set(['foo', 'bar']), self.list_non_default_pools()) + self.rados.create_pool('baz') + eq(set(['foo', 'bar', 'baz']), self.list_non_default_pools()) + self.rados.delete_pool('foo') + eq(set(['bar', 'baz']), self.list_non_default_pools()) + self.rados.delete_pool('baz') + eq(set(['bar']), self.list_non_default_pools()) + self.rados.delete_pool('bar') + eq(set(), self.list_non_default_pools()) + self.rados.create_pool('a' * 500) + eq(set(['a' * 500]), self.list_non_default_pools()) + self.rados.delete_pool('a' * 500) + + @attr('tier') + def test_get_pool_base_tier(self): + self.rados.create_pool('foo') + try: + self.rados.create_pool('foo-cache') + try: + pool_id = self.rados.pool_lookup('foo') + tier_pool_id = self.rados.pool_lookup('foo-cache') + + cmd = {"prefix":"osd tier add", "pool":"foo", "tierpool":"foo-cache", "force_nonempty":""} + ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30) + eq(ret, 0) + + try: + cmd = {"prefix":"osd tier cache-mode", "pool":"foo-cache", "tierpool":"foo-cache", "mode":"readonly", "yes_i_really_mean_it": True} + ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30) + eq(ret, 0) + + eq(self.rados.wait_for_latest_osdmap(), 0) + + eq(pool_id, self.rados.get_pool_base_tier(pool_id)) + eq(pool_id, self.rados.get_pool_base_tier(tier_pool_id)) + finally: + cmd = {"prefix":"osd tier remove", "pool":"foo", "tierpool":"foo-cache"} + ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30) + eq(ret, 0) + finally: + self.rados.delete_pool('foo-cache') + finally: + self.rados.delete_pool('foo') + + def test_get_fsid(self): + fsid = self.rados.get_fsid() + assert re.match('[0-9a-f\-]{36}', fsid, re.I) + + def test_blocklist_add(self): + self.rados.blocklist_add("1.2.3.4/123", 1) + + @attr('stats') + def test_get_cluster_stats(self): + stats = self.rados.get_cluster_stats() + assert stats['kb'] > 0 + assert stats['kb_avail'] > 0 + assert stats['kb_used'] > 0 + assert stats['num_objects'] >= 0 + + def test_monitor_log(self): + lock = threading.Condition() + def cb(arg, line, who, sec, nsec, seq, level, msg): + # NOTE(sileht): the old pyrados API was received the pointer as int + # instead of the value of arg + eq(arg, "arg") + with lock: + lock.notify() + return 0 + + # NOTE(sileht): force don't save the monitor into local var + # to ensure all references are correctly tracked into the lib + MonitorLog(self.rados, "debug", cb, "arg") + with lock: + lock.wait() + MonitorLog(self.rados, "debug", None, None) + eq(None, self.rados.monitor_callback) + +class TestIoctx(object): + + def setUp(self): + self.rados = Rados(conffile='') + self.rados.connect() + self.rados.create_pool('test_pool') + assert self.rados.pool_exists('test_pool') + self.ioctx = self.rados.open_ioctx('test_pool') + + def tearDown(self): + cmd = {"prefix":"osd unset", "key":"noup"} + self.rados.mon_command(json.dumps(cmd), b'') + self.ioctx.close() + self.rados.delete_pool('test_pool') + self.rados.shutdown() + + def test_get_last_version(self): + version = self.ioctx.get_last_version() + assert version >= 0 + + def test_get_stats(self): + stats = self.ioctx.get_stats() + eq(stats, {'num_objects_unfound': 0, + 'num_objects_missing_on_primary': 0, + 'num_object_clones': 0, + 'num_objects': 0, + 'num_object_copies': 0, + 'num_bytes': 0, + 'num_rd_kb': 0, + 'num_wr_kb': 0, + 'num_kb': 0, + 'num_wr': 0, + 'num_objects_degraded': 0, + 'num_rd': 0}) + + def test_write(self): + self.ioctx.write('abc', b'abc') + eq(self.ioctx.read('abc'), b'abc') + + def test_write_full(self): + self.ioctx.write('abc', b'abc') + eq(self.ioctx.read('abc'), b'abc') + self.ioctx.write_full('abc', b'd') + eq(self.ioctx.read('abc'), b'd') + + def test_writesame(self): + self.ioctx.writesame('ob', b'rzx', 9) + eq(self.ioctx.read('ob'), b'rzxrzxrzx') + + def test_append(self): + self.ioctx.write('abc', b'a') + self.ioctx.append('abc', b'b') + self.ioctx.append('abc', b'c') + eq(self.ioctx.read('abc'), b'abc') + + def test_write_zeros(self): + self.ioctx.write('abc', b'a\0b\0c') + eq(self.ioctx.read('abc'), b'a\0b\0c') + + def test_trunc(self): + self.ioctx.write('abc', b'abc') + self.ioctx.trunc('abc', 2) + eq(self.ioctx.read('abc'), b'ab') + size = self.ioctx.stat('abc')[0] + eq(size, 2) + + def test_cmpext(self): + self.ioctx.write('test_object', b'abcdefghi') + eq(0, self.ioctx.cmpext('test_object', b'abcdefghi', 0)) + eq(-MAX_ERRNO - 4, self.ioctx.cmpext('test_object', b'abcdxxxxx', 0)) + + def test_list_objects_empty(self): + eq(list(self.ioctx.list_objects()), []) + + def test_list_objects(self): + self.ioctx.write('a', b'') + self.ioctx.write('b', b'foo') + self.ioctx.write_full('c', b'bar') + self.ioctx.append('d', b'jazz') + object_names = [obj.key for obj in self.ioctx.list_objects()] + eq(sorted(object_names), ['a', 'b', 'c', 'd']) + + def test_list_ns_objects(self): + self.ioctx.write('a', b'') + self.ioctx.write('b', b'foo') + self.ioctx.write_full('c', b'bar') + self.ioctx.append('d', b'jazz') + self.ioctx.set_namespace("ns1") + self.ioctx.write('ns1-a', b'') + self.ioctx.write('ns1-b', b'foo') + self.ioctx.write_full('ns1-c', b'bar') + self.ioctx.append('ns1-d', b'jazz') + self.ioctx.append('d', b'jazz') + self.ioctx.set_namespace(LIBRADOS_ALL_NSPACES) + object_names = [(obj.nspace, obj.key) for obj in self.ioctx.list_objects()] + eq(sorted(object_names), [('', 'a'), ('','b'), ('','c'), ('','d'),\ + ('ns1', 'd'), ('ns1', 'ns1-a'), ('ns1', 'ns1-b'),\ + ('ns1', 'ns1-c'), ('ns1', 'ns1-d')]) + + def test_xattrs(self): + xattrs = dict(a=b'1', b=b'2', c=b'3', d=b'a\0b', e=b'\0', f=b'') + self.ioctx.write('abc', b'') + for key, value in xattrs.items(): + self.ioctx.set_xattr('abc', key, value) + eq(self.ioctx.get_xattr('abc', key), value) + stored_xattrs = {} + for key, value in self.ioctx.get_xattrs('abc'): + stored_xattrs[key] = value + eq(stored_xattrs, xattrs) + + def test_obj_xattrs(self): + xattrs = dict(a=b'1', b=b'2', c=b'3', d=b'a\0b', e=b'\0', f=b'') + self.ioctx.write('abc', b'') + obj = list(self.ioctx.list_objects())[0] + for key, value in xattrs.items(): + obj.set_xattr(key, value) + eq(obj.get_xattr(key), value) + stored_xattrs = {} + for key, value in obj.get_xattrs(): + stored_xattrs[key] = value + eq(stored_xattrs, xattrs) + + def test_get_pool_id(self): + eq(self.ioctx.get_pool_id(), self.rados.pool_lookup('test_pool')) + + def test_get_pool_name(self): + eq(self.ioctx.get_pool_name(), 'test_pool') + + @attr('snap') + def test_create_snap(self): + assert_raises(ObjectNotFound, self.ioctx.remove_snap, 'foo') + self.ioctx.create_snap('foo') + self.ioctx.remove_snap('foo') + + @attr('snap') + def test_list_snaps_empty(self): + eq(list(self.ioctx.list_snaps()), []) + + @attr('snap') + def test_list_snaps(self): + snaps = ['snap1', 'snap2', 'snap3'] + for snap in snaps: + self.ioctx.create_snap(snap) + listed_snaps = [snap.name for snap in self.ioctx.list_snaps()] + eq(snaps, listed_snaps) + + @attr('snap') + def test_lookup_snap(self): + self.ioctx.create_snap('foo') + snap = self.ioctx.lookup_snap('foo') + eq(snap.name, 'foo') + + @attr('snap') + def test_snap_timestamp(self): + self.ioctx.create_snap('foo') + snap = self.ioctx.lookup_snap('foo') + snap.get_timestamp() + + @attr('snap') + def test_remove_snap(self): + self.ioctx.create_snap('foo') + (snap,) = self.ioctx.list_snaps() + eq(snap.name, 'foo') + self.ioctx.remove_snap('foo') + eq(list(self.ioctx.list_snaps()), []) + + @attr('snap') + def test_snap_rollback(self): + self.ioctx.write("insnap", b"contents1") + self.ioctx.create_snap("snap1") + self.ioctx.remove_object("insnap") + self.ioctx.snap_rollback("insnap", "snap1") + eq(self.ioctx.read("insnap"), b"contents1") + self.ioctx.remove_snap("snap1") + self.ioctx.remove_object("insnap") + + @attr('snap') + def test_snap_read(self): + self.ioctx.write("insnap", b"contents1") + self.ioctx.create_snap("snap1") + self.ioctx.remove_object("insnap") + snap = self.ioctx.lookup_snap("snap1") + self.ioctx.set_read(snap.snap_id) + eq(self.ioctx.read("insnap"), b"contents1") + self.ioctx.set_read(LIBRADOS_SNAP_HEAD) + self.ioctx.write("inhead", b"contents2") + eq(self.ioctx.read("inhead"), b"contents2") + self.ioctx.remove_snap("snap1") + self.ioctx.remove_object("inhead") + + def test_set_omap(self): + keys = ("1", "2", "3", "4") + values = (b"aaa", b"bbb", b"ccc", b"\x04\x04\x04\x04") + with WriteOpCtx() as write_op: + self.ioctx.set_omap(write_op, keys, values) + write_op.set_flags(LIBRADOS_OPERATION_SKIPRWLOCKS) + self.ioctx.operate_write_op(write_op, "hw") + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals(read_op, "", "", 4) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, "hw") + next(iter) + eq(list(iter), [("2", b"bbb"), ("3", b"ccc"), ("4", b"\x04\x04\x04\x04")]) + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals(read_op, "2", "", 4) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, "hw") + eq(("3", b"ccc"), next(iter)) + eq(list(iter), [("4", b"\x04\x04\x04\x04")]) + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals(read_op, "", "2", 4) + eq(ret, 0) + read_op.set_flags(LIBRADOS_OPERATION_BALANCE_READS) + self.ioctx.operate_read_op(read_op, "hw") + eq(list(iter), [("2", b"bbb")]) + + def test_set_omap_aio(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + + keys = ("1", "2", "3", "4") + values = (b"aaa", b"bbb", b"ccc", b"\x04\x04\x04\x04") + with WriteOpCtx() as write_op: + self.ioctx.set_omap(write_op, keys, values) + comp = self.ioctx.operate_aio_write_op(write_op, "hw", cb, cb) + comp.wait_for_complete() + with lock: + while count[0] < 2: + lock.wait() + eq(comp.get_return_value(), 0) + + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals(read_op, "", "", 4) + eq(ret, 0) + comp = self.ioctx.operate_aio_read_op(read_op, "hw", cb, cb) + comp.wait_for_complete() + with lock: + while count[0] < 4: + lock.wait() + eq(comp.get_return_value(), 0) + next(iter) + eq(list(iter), [("2", b"bbb"), ("3", b"ccc"), ("4", b"\x04\x04\x04\x04")]) + + def test_write_ops(self): + with WriteOpCtx() as write_op: + write_op.new(0) + self.ioctx.operate_write_op(write_op, "write_ops") + eq(self.ioctx.read('write_ops'), b'') + + write_op.write_full(b'1') + write_op.append(b'2') + self.ioctx.operate_write_op(write_op, "write_ops") + eq(self.ioctx.read('write_ops'), b'12') + + write_op.write_full(b'12345') + write_op.write(b'x', 2) + self.ioctx.operate_write_op(write_op, "write_ops") + eq(self.ioctx.read('write_ops'), b'12x45') + + write_op.write_full(b'12345') + write_op.zero(2, 2) + self.ioctx.operate_write_op(write_op, "write_ops") + eq(self.ioctx.read('write_ops'), b'12\x00\x005') + + write_op.write_full(b'12345') + write_op.truncate(2) + self.ioctx.operate_write_op(write_op, "write_ops") + eq(self.ioctx.read('write_ops'), b'12') + + write_op.remove() + self.ioctx.operate_write_op(write_op, "write_ops") + with assert_raises(ObjectNotFound): + self.ioctx.read('write_ops') + + def test_execute_op(self): + with WriteOpCtx() as write_op: + write_op.execute("hello", "record_hello", b"ebs") + self.ioctx.operate_write_op(write_op, "object") + eq(self.ioctx.read('object'), b"Hello, ebs!") + + def test_writesame_op(self): + with WriteOpCtx() as write_op: + write_op.writesame(b'rzx', 9) + self.ioctx.operate_write_op(write_op, 'abc') + eq(self.ioctx.read('abc'), b'rzxrzxrzx') + + def test_get_omap_vals_by_keys(self): + keys = ("1", "2", "3", "4") + values = (b"aaa", b"bbb", b"ccc", b"\x04\x04\x04\x04") + with WriteOpCtx() as write_op: + self.ioctx.set_omap(write_op, keys, values) + self.ioctx.operate_write_op(write_op, "hw") + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals_by_keys(read_op,("3","4",)) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, "hw") + eq(list(iter), [("3", b"ccc"), ("4", b"\x04\x04\x04\x04")]) + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals_by_keys(read_op,("3","4",)) + eq(ret, 0) + with assert_raises(ObjectNotFound): + self.ioctx.operate_read_op(read_op, "no_such") + + def test_get_omap_keys(self): + keys = ("1", "2", "3") + values = (b"aaa", b"bbb", b"ccc") + with WriteOpCtx() as write_op: + self.ioctx.set_omap(write_op, keys, values) + self.ioctx.operate_write_op(write_op, "hw") + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_keys(read_op,"",2) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, "hw") + eq(list(iter), [("1", None), ("2", None)]) + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_keys(read_op,"",2) + eq(ret, 0) + with assert_raises(ObjectNotFound): + self.ioctx.operate_read_op(read_op, "no_such") + + def test_clear_omap(self): + keys = ("1", "2", "3") + values = (b"aaa", b"bbb", b"ccc") + with WriteOpCtx() as write_op: + self.ioctx.set_omap(write_op, keys, values) + self.ioctx.operate_write_op(write_op, "hw") + with WriteOpCtx() as write_op_1: + self.ioctx.clear_omap(write_op_1) + self.ioctx.operate_write_op(write_op_1, "hw") + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals_by_keys(read_op,("1",)) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, "hw") + eq(list(iter), []) + + def test_remove_omap_ramge2(self): + keys = ("1", "2", "3", "4") + values = (b"a", b"bb", b"ccc", b"dddd") + with WriteOpCtx() as write_op: + self.ioctx.set_omap(write_op, keys, values) + self.ioctx.operate_write_op(write_op, "test_obj") + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals_by_keys(read_op, keys) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, "test_obj") + eq(list(iter), list(zip(keys, values))) + with WriteOpCtx() as write_op: + self.ioctx.remove_omap_range2(write_op, "1", "4") + self.ioctx.operate_write_op(write_op, "test_obj") + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals_by_keys(read_op, keys) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, "test_obj") + eq(list(iter), [("4", b"dddd")]) + + def test_omap_cmp(self): + object_id = 'test' + self.ioctx.write(object_id, b'omap_cmp') + with WriteOpCtx() as write_op: + self.ioctx.set_omap(write_op, ('key1',), ('1',)) + self.ioctx.operate_write_op(write_op, object_id) + with WriteOpCtx() as write_op: + write_op.omap_cmp('key1', '1', LIBRADOS_CMPXATTR_OP_EQ) + self.ioctx.set_omap(write_op, ('key1',), ('2',)) + self.ioctx.operate_write_op(write_op, object_id) + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals_by_keys(read_op, ('key1',)) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, object_id) + eq(list(iter), [('key1', b'2')]) + with WriteOpCtx() as write_op: + write_op.omap_cmp('key1', '1', LIBRADOS_CMPXATTR_OP_GT) + self.ioctx.set_omap(write_op, ('key1',), ('3',)) + self.ioctx.operate_write_op(write_op, object_id) + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals_by_keys(read_op, ('key1',)) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, object_id) + eq(list(iter), [('key1', b'3')]) + with WriteOpCtx() as write_op: + write_op.omap_cmp('key1', '4', LIBRADOS_CMPXATTR_OP_LT) + self.ioctx.set_omap(write_op, ('key1',), ('4',)) + self.ioctx.operate_write_op(write_op, object_id) + with ReadOpCtx() as read_op: + iter, ret = self.ioctx.get_omap_vals_by_keys(read_op, ('key1',)) + eq(ret, 0) + self.ioctx.operate_read_op(read_op, object_id) + eq(list(iter), [('key1', b'4')]) + with WriteOpCtx() as write_op: + write_op.omap_cmp('key1', '1', LIBRADOS_CMPXATTR_OP_EQ) + self.ioctx.set_omap(write_op, ('key1',), ('5',)) + try: + self.ioctx.operate_write_op(write_op, object_id) + except (OSError, ExtendMismatch) as e: + eq(e.errno, 125) + else: + message = "omap_cmp did not raise Exception when omap content does not match" + raise AssertionError(message) + + def test_cmpext_op(self): + object_id = 'test' + with WriteOpCtx() as write_op: + write_op.write(b'12345', 0) + self.ioctx.operate_write_op(write_op, object_id) + with WriteOpCtx() as write_op: + write_op.cmpext(b'12345', 0) + write_op.write(b'54321', 0) + self.ioctx.operate_write_op(write_op, object_id) + eq(self.ioctx.read(object_id), b'54321') + with WriteOpCtx() as write_op: + write_op.cmpext(b'56789', 0) + write_op.write(b'12345', 0) + try: + self.ioctx.operate_write_op(write_op, object_id) + except ExtendMismatch as e: + # the cmpext_result compare with expected error number, it should be (-MAX_ERRNO - 1) + # where "1" is the offset of the first unmatched byte + eq(-e.errno, -MAX_ERRNO - 1) + eq(e.offset, 1) + else: + message = "cmpext did not raise Exception when object content does not match" + raise AssertionError(message) + with ReadOpCtx() as read_op: + read_op.cmpext(b'54321', 0) + self.ioctx.operate_read_op(read_op, object_id) + with ReadOpCtx() as read_op: + read_op.cmpext(b'54789', 0) + try: + self.ioctx.operate_read_op(read_op, object_id) + except ExtendMismatch as e: + # the cmpext_result compare with expected error number, it should be (-MAX_ERRNO - 2) + # where "2" is the offset of the first unmatched byte + eq(-e.errno, -MAX_ERRNO - 2) + eq(e.offset, 2) + else: + message = "cmpext did not raise Exception when object content does not match" + raise AssertionError(message) + + def test_xattrs_op(self): + xattrs = dict(a=b'1', b=b'2', c=b'3', d=b'a\0b', e=b'\0') + with WriteOpCtx() as write_op: + write_op.new(LIBRADOS_CREATE_EXCLUSIVE) + for key, value in xattrs.items(): + write_op.set_xattr(key, value) + self.ioctx.operate_write_op(write_op, 'abc') + eq(self.ioctx.get_xattr('abc', key), value) + + stored_xattrs_1 = {} + for key, value in self.ioctx.get_xattrs('abc'): + stored_xattrs_1[key] = value + eq(stored_xattrs_1, xattrs) + + for key in xattrs.keys(): + write_op.rm_xattr(key) + self.ioctx.operate_write_op(write_op, 'abc') + stored_xattrs_2 = {} + for key, value in self.ioctx.get_xattrs('abc'): + stored_xattrs_2[key] = value + eq(stored_xattrs_2, {}) + + write_op.remove() + self.ioctx.operate_write_op(write_op, 'abc') + + def test_locator(self): + self.ioctx.set_locator_key("bar") + self.ioctx.write('foo', b'contents1') + objects = [i for i in self.ioctx.list_objects()] + eq(len(objects), 1) + eq(self.ioctx.get_locator_key(), "bar") + self.ioctx.set_locator_key("") + objects[0].seek(0) + objects[0].write(b"contents2") + eq(self.ioctx.get_locator_key(), "") + self.ioctx.set_locator_key("bar") + contents = self.ioctx.read("foo") + eq(contents, b"contents2") + eq(self.ioctx.get_locator_key(), "bar") + objects[0].remove() + objects = [i for i in self.ioctx.list_objects()] + eq(objects, []) + self.ioctx.set_locator_key("") + + def test_operate_aio_write_op(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + with WriteOpCtx() as write_op: + write_op.write(b'rzx') + comp = self.ioctx.operate_aio_write_op(write_op, "object", cb, cb) + comp.wait_for_complete() + with lock: + while count[0] < 2: + lock.wait() + eq(comp.get_return_value(), 0) + eq(self.ioctx.read('object'), b'rzx') + + def test_aio_write(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + comp = self.ioctx.aio_write("foo", b"bar", 0, cb, cb) + comp.wait_for_complete() + with lock: + while count[0] < 2: + lock.wait() + eq(comp.get_return_value(), 0) + contents = self.ioctx.read("foo") + eq(contents, b"bar") + [i.remove() for i in self.ioctx.list_objects()] + + def test_aio_cmpext(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + + self.ioctx.write('test_object', b'abcdefghi') + comp = self.ioctx.aio_cmpext('test_object', b'abcdefghi', 0, cb) + comp.wait_for_complete() + with lock: + while count[0] < 1: + lock.wait() + eq(comp.get_return_value(), 0) + + def test_aio_rmxattr(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + self.ioctx.set_xattr("xyz", "key", b'value') + eq(self.ioctx.get_xattr("xyz", "key"), b'value') + comp = self.ioctx.aio_rmxattr("xyz", "key", cb) + comp.wait_for_complete() + with lock: + while count[0] < 1: + lock.wait() + eq(comp.get_return_value(), 0) + with assert_raises(NoData): + self.ioctx.get_xattr("xyz", "key") + + def test_aio_write_no_comp_ref(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + # NOTE(sileht): force don't save the comp into local var + # to ensure all references are correctly tracked into the lib + self.ioctx.aio_write("foo", b"bar", 0, cb, cb) + with lock: + while count[0] < 2: + lock.wait() + contents = self.ioctx.read("foo") + eq(contents, b"bar") + [i.remove() for i in self.ioctx.list_objects()] + + def test_aio_append(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + comp = self.ioctx.aio_write("foo", b"bar", 0, cb, cb) + comp2 = self.ioctx.aio_append("foo", b"baz", cb, cb) + comp.wait_for_complete() + contents = self.ioctx.read("foo") + eq(contents, b"barbaz") + with lock: + while count[0] < 4: + lock.wait() + eq(comp.get_return_value(), 0) + eq(comp2.get_return_value(), 0) + [i.remove() for i in self.ioctx.list_objects()] + + def test_aio_write_full(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + self.ioctx.aio_write("foo", b"barbaz", 0, cb, cb) + comp = self.ioctx.aio_write_full("foo", b"bar", cb, cb) + comp.wait_for_complete() + with lock: + while count[0] < 2: + lock.wait() + eq(comp.get_return_value(), 0) + contents = self.ioctx.read("foo") + eq(contents, b"bar") + [i.remove() for i in self.ioctx.list_objects()] + + def test_aio_writesame(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + comp = self.ioctx.aio_writesame("abc", b"rzx", 9, 0, cb) + comp.wait_for_complete() + with lock: + while count[0] < 1: + lock.wait() + eq(comp.get_return_value(), 0) + eq(self.ioctx.read("abc"), b"rzxrzxrzx") + [i.remove() for i in self.ioctx.list_objects()] + + def test_aio_stat(self): + lock = threading.Condition() + count = [0] + def cb(_, size, mtime): + with lock: + count[0] += 1 + lock.notify() + + comp = self.ioctx.aio_stat("foo", cb) + comp.wait_for_complete() + with lock: + while count[0] < 1: + lock.wait() + eq(comp.get_return_value(), -2) + + self.ioctx.write("foo", b"bar") + + comp = self.ioctx.aio_stat("foo", cb) + comp.wait_for_complete() + with lock: + while count[0] < 2: + lock.wait() + eq(comp.get_return_value(), 0) + + [i.remove() for i in self.ioctx.list_objects()] + + def test_aio_remove(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + self.ioctx.write('foo', b'wrx') + eq(self.ioctx.read('foo'), b'wrx') + comp = self.ioctx.aio_remove('foo', cb, cb) + comp.wait_for_complete() + with lock: + while count[0] < 2: + lock.wait() + eq(comp.get_return_value(), 0) + eq(list(self.ioctx.list_objects()), []) + + def _take_down_acting_set(self, pool, objectname): + # find acting_set for pool:objectname and take it down; used to + # verify that async reads don't complete while acting set is missing + cmd = { + "prefix":"osd map", + "pool":pool, + "object":objectname, + "format":"json", + } + r, jsonout, _ = self.rados.mon_command(json.dumps(cmd), b'') + objmap = json.loads(jsonout.decode("utf-8")) + acting_set = objmap['acting'] + cmd = {"prefix":"osd set", "key":"noup"} + r, _, _ = self.rados.mon_command(json.dumps(cmd), b'') + eq(r, 0) + cmd = {"prefix":"osd down", "ids":[str(i) for i in acting_set]} + r, _, _ = self.rados.mon_command(json.dumps(cmd), b'') + eq(r, 0) + + # wait for OSDs to acknowledge the down + eq(self.rados.wait_for_latest_osdmap(), 0) + + def _let_osds_back_up(self): + cmd = {"prefix":"osd unset", "key":"noup"} + r, _, _ = self.rados.mon_command(json.dumps(cmd), b'') + eq(r, 0) + + def test_aio_read_wait_for_complete(self): + # use wait_for_complete() and wait for cb by + # watching retval[0] + + # this is a list so that the local cb() can modify it + payload = b"bar\000frob" + self.ioctx.write("foo", payload) + self._take_down_acting_set('test_pool', 'foo') + + retval = [None] + lock = threading.Condition() + def cb(_, buf): + with lock: + retval[0] = buf + lock.notify() + + comp = self.ioctx.aio_read("foo", len(payload), 0, cb) + eq(False, comp.is_complete()) + time.sleep(3) + eq(False, comp.is_complete()) + with lock: + eq(None, retval[0]) + + self._let_osds_back_up() + comp.wait_for_complete() + loops = 0 + with lock: + while retval[0] is None and loops <= 10: + lock.wait(timeout=5) + loops += 1 + assert(loops <= 10) + + eq(retval[0], payload) + eq(sys.getrefcount(comp), 2) + + def test_aio_read_wait_for_complete_and_cb(self): + # use wait_for_complete_and_cb(), verify retval[0] is + # set by the time we regain control + payload = b"bar\000frob" + self.ioctx.write("foo", payload) + + self._take_down_acting_set('test_pool', 'foo') + # this is a list so that the local cb() can modify it + retval = [None] + lock = threading.Condition() + def cb(_, buf): + with lock: + retval[0] = buf + lock.notify() + comp = self.ioctx.aio_read("foo", len(payload), 0, cb) + eq(False, comp.is_complete()) + time.sleep(3) + eq(False, comp.is_complete()) + with lock: + eq(None, retval[0]) + + self._let_osds_back_up() + comp.wait_for_complete_and_cb() + assert(retval[0] is not None) + eq(retval[0], payload) + eq(sys.getrefcount(comp), 2) + + def test_aio_read_wait_for_complete_and_cb_error(self): + # error case, use wait_for_complete_and_cb(), verify retval[0] is + # set by the time we regain control + self._take_down_acting_set('test_pool', 'bar') + + # this is a list so that the local cb() can modify it + retval = [1] + lock = threading.Condition() + def cb(_, buf): + with lock: + retval[0] = buf + lock.notify() + + # read from a DNE object + comp = self.ioctx.aio_read("bar", 3, 0, cb) + eq(False, comp.is_complete()) + time.sleep(3) + eq(False, comp.is_complete()) + with lock: + eq(1, retval[0]) + self._let_osds_back_up() + + comp.wait_for_complete_and_cb() + eq(None, retval[0]) + assert(comp.get_return_value() < 0) + eq(sys.getrefcount(comp), 2) + + def test_lock(self): + self.ioctx.lock_exclusive("foo", "lock", "locker", "desc_lock", + 10000, 0) + assert_raises(ObjectExists, + self.ioctx.lock_exclusive, + "foo", "lock", "locker", "desc_lock", 10000, 0) + self.ioctx.unlock("foo", "lock", "locker") + assert_raises(ObjectNotFound, self.ioctx.unlock, "foo", "lock", "locker") + + self.ioctx.lock_shared("foo", "lock", "locker1", "tag", "desc_lock", + 10000, 0) + self.ioctx.lock_shared("foo", "lock", "locker2", "tag", "desc_lock", + 10000, 0) + assert_raises(ObjectBusy, + self.ioctx.lock_exclusive, + "foo", "lock", "locker3", "desc_lock", 10000, 0) + self.ioctx.unlock("foo", "lock", "locker1") + self.ioctx.unlock("foo", "lock", "locker2") + assert_raises(ObjectNotFound, self.ioctx.unlock, "foo", "lock", "locker1") + assert_raises(ObjectNotFound, self.ioctx.unlock, "foo", "lock", "locker2") + + def test_execute(self): + self.ioctx.write("foo", b"") # ensure object exists + + ret, buf = self.ioctx.execute("foo", "hello", "say_hello", b"") + eq(buf, b"Hello, world!") + + ret, buf = self.ioctx.execute("foo", "hello", "say_hello", b"nose") + eq(buf, b"Hello, nose!") + + def test_aio_execute(self): + count = [0] + retval = [None] + lock = threading.Condition() + def cb(_, buf): + with lock: + if retval[0] is None: + retval[0] = buf + count[0] += 1 + lock.notify() + self.ioctx.write("foo", b"") # ensure object exists + + comp = self.ioctx.aio_execute("foo", "hello", "say_hello", b"", 32, cb, cb) + comp.wait_for_complete() + with lock: + while count[0] < 2: + lock.wait() + eq(comp.get_return_value(), 13) + eq(retval[0], b"Hello, world!") + + retval[0] = None + comp = self.ioctx.aio_execute("foo", "hello", "say_hello", b"nose", 32, cb, cb) + comp.wait_for_complete() + with lock: + while count[0] < 4: + lock.wait() + eq(comp.get_return_value(), 12) + eq(retval[0], b"Hello, nose!") + + [i.remove() for i in self.ioctx.list_objects()] + + def test_aio_setxattr(self): + lock = threading.Condition() + count = [0] + def cb(blah): + with lock: + count[0] += 1 + lock.notify() + return 0 + comp = self.ioctx.aio_setxattr("obj", "key", b'value', cb) + comp.wait_for_complete() + with lock: + while count[0] < 1: + lock.wait() + eq(comp.get_return_value(), 0) + eq(self.ioctx.get_xattr("obj", "key"), b'value') + + def test_applications(self): + cmd = {"prefix":"osd dump", "format":"json"} + ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'') + eq(ret, 0) + assert len(buf) > 0 + release = json.loads(buf.decode("utf-8")).get("require_osd_release", + None) + if not release or release[0] < 'l': + raise SkipTest + + eq([], self.ioctx.application_list()) + + self.ioctx.application_enable("app1") + assert_raises(Error, self.ioctx.application_enable, "app2") + self.ioctx.application_enable("app2", True) + + assert_raises(Error, self.ioctx.application_metadata_list, "dne") + eq([], self.ioctx.application_metadata_list("app1")) + + assert_raises(Error, self.ioctx.application_metadata_set, "dne", "key", + "key") + self.ioctx.application_metadata_set("app1", "key1", "val1") + eq("val1", self.ioctx.application_metadata_get("app1", "key1")) + self.ioctx.application_metadata_set("app1", "key2", "val2") + eq("val2", self.ioctx.application_metadata_get("app1", "key2")) + self.ioctx.application_metadata_set("app2", "key1", "val1") + eq("val1", self.ioctx.application_metadata_get("app2", "key1")) + + eq([("key1", "val1"), ("key2", "val2")], + self.ioctx.application_metadata_list("app1")) + + self.ioctx.application_metadata_remove("app1", "key1") + eq([("key2", "val2")], self.ioctx.application_metadata_list("app1")) + + def test_service_daemon(self): + name = "pid-" + str(os.getpid()) + metadata = {'version': '3.14', 'memory': '42'} + self.rados.service_daemon_register("laundry", name, metadata) + status = {'result': 'unknown', 'test': 'running'} + self.rados.service_daemon_update(status) + + def test_alignment(self): + eq(self.ioctx.alignment(), None) + + +@attr('ec') +class TestIoctxEc(object): + + def setUp(self): + self.rados = Rados(conffile='') + self.rados.connect() + self.pool = 'test-ec' + self.profile = 'testprofile-%s' % self.pool + cmd = {"prefix": "osd erasure-code-profile set", + "name": self.profile, "profile": ["k=2", "m=1", "crush-failure-domain=osd"]} + ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'', timeout=30) + eq(ret, 0, msg=out) + # create ec pool with profile created above + cmd = {'prefix': 'osd pool create', 'pg_num': 8, 'pgp_num': 8, + 'pool': self.pool, 'pool_type': 'erasure', + 'erasure_code_profile': self.profile} + ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'', timeout=30) + eq(ret, 0, msg=out) + assert self.rados.pool_exists(self.pool) + self.ioctx = self.rados.open_ioctx(self.pool) + + def tearDown(self): + cmd = {"prefix": "osd unset", "key": "noup"} + self.rados.mon_command(json.dumps(cmd), b'') + self.ioctx.close() + self.rados.delete_pool(self.pool) + self.rados.shutdown() + + def test_alignment(self): + eq(self.ioctx.alignment(), 8192) + + +class TestIoctx2(object): + + def setUp(self): + self.rados = Rados(conffile='') + self.rados.connect() + self.rados.create_pool('test_pool') + assert self.rados.pool_exists('test_pool') + pool_id = self.rados.pool_lookup('test_pool') + assert pool_id > 0 + self.ioctx2 = self.rados.open_ioctx2(pool_id) + + def tearDown(self): + cmd = {"prefix": "osd unset", "key": "noup"} + self.rados.mon_command(json.dumps(cmd), b'') + self.ioctx2.close() + self.rados.delete_pool('test_pool') + self.rados.shutdown() + + def test_get_last_version(self): + version = self.ioctx2.get_last_version() + assert version >= 0 + + def test_get_stats(self): + stats = self.ioctx2.get_stats() + eq(stats, {'num_objects_unfound': 0, + 'num_objects_missing_on_primary': 0, + 'num_object_clones': 0, + 'num_objects': 0, + 'num_object_copies': 0, + 'num_bytes': 0, + 'num_rd_kb': 0, + 'num_wr_kb': 0, + 'num_kb': 0, + 'num_wr': 0, + 'num_objects_degraded': 0, + 'num_rd': 0}) + + +class TestObject(object): + + def setUp(self): + self.rados = Rados(conffile='') + self.rados.connect() + self.rados.create_pool('test_pool') + assert self.rados.pool_exists('test_pool') + self.ioctx = self.rados.open_ioctx('test_pool') + self.ioctx.write('foo', b'bar') + self.object = Object(self.ioctx, 'foo') + + def tearDown(self): + self.ioctx.close() + self.ioctx = None + self.rados.delete_pool('test_pool') + self.rados.shutdown() + self.rados = None + + def test_read(self): + eq(self.object.read(3), b'bar') + eq(self.object.read(100), b'') + + def test_seek(self): + self.object.write(b'blah') + self.object.seek(0) + eq(self.object.read(4), b'blah') + self.object.seek(1) + eq(self.object.read(3), b'lah') + + def test_write(self): + self.object.write(b'barbaz') + self.object.seek(0) + eq(self.object.read(3), b'bar') + eq(self.object.read(3), b'baz') + +@attr('snap') +class TestIoCtxSelfManagedSnaps(object): + def setUp(self): + self.rados = Rados(conffile='') + self.rados.connect() + self.rados.create_pool('test_pool') + assert self.rados.pool_exists('test_pool') + self.ioctx = self.rados.open_ioctx('test_pool') + + def tearDown(self): + cmd = {"prefix":"osd unset", "key":"noup"} + self.rados.mon_command(json.dumps(cmd), b'') + self.ioctx.close() + self.rados.delete_pool('test_pool') + self.rados.shutdown() + + def test(self): + # cannot mix-and-match pool and self-managed snapshot mode + self.ioctx.set_self_managed_snap_write([]) + self.ioctx.write('abc', b'abc') + snap_id_1 = self.ioctx.create_self_managed_snap() + self.ioctx.set_self_managed_snap_write([snap_id_1]) + + self.ioctx.write('abc', b'def') + snap_id_2 = self.ioctx.create_self_managed_snap() + self.ioctx.set_self_managed_snap_write([snap_id_1, snap_id_2]) + + self.ioctx.write('abc', b'ghi') + + self.ioctx.rollback_self_managed_snap('abc', snap_id_1) + eq(self.ioctx.read('abc'), b'abc') + + self.ioctx.rollback_self_managed_snap('abc', snap_id_2) + eq(self.ioctx.read('abc'), b'def') + + self.ioctx.remove_self_managed_snap(snap_id_1) + self.ioctx.remove_self_managed_snap(snap_id_2) + +class TestCommand(object): + + def setUp(self): + self.rados = Rados(conffile='') + self.rados.connect() + + def tearDown(self): + self.rados.shutdown() + + def test_monmap_dump(self): + + # check for success and some plain output with epoch in it + cmd = {"prefix":"mon dump"} + ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30) + eq(ret, 0) + assert len(buf) > 0 + assert(b'epoch' in buf) + + # JSON, and grab current epoch + cmd['format'] = 'json' + ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30) + eq(ret, 0) + assert len(buf) > 0 + d = json.loads(buf.decode("utf-8")) + assert('epoch' in d) + epoch = d['epoch'] + + # assume epoch + 1000 does not exist; test for ENOENT + cmd['epoch'] = epoch + 1000 + ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30) + eq(ret, -errno.ENOENT) + eq(len(buf), 0) + del cmd['epoch'] + + # send to specific target by name, rank + cmd = {"prefix": "version"} + + target = d['mons'][0]['name'] + print(target) + ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30, + target=target) + eq(ret, 0) + assert len(buf) > 0 + e = json.loads(buf.decode("utf-8")) + assert('release' in e) + + target = d['mons'][0]['rank'] + print(target) + ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30, + target=target) + eq(ret, 0) + assert len(buf) > 0 + e = json.loads(buf.decode("utf-8")) + assert('release' in e) + + @attr('bench') + def test_osd_bench(self): + cmd = dict(prefix='bench', size=4096, count=8192) + ret, buf, err = self.rados.osd_command(0, json.dumps(cmd), b'', + timeout=30) + eq(ret, 0) + assert len(buf) > 0 + out = json.loads(buf.decode('utf-8')) + eq(out['blocksize'], cmd['size']) + eq(out['bytes_written'], cmd['count']) + + def test_ceph_osd_pool_create_utf8(self): + poolname = "\u9ec5" + + cmd = {"prefix": "osd pool create", "pg_num": 16, "pool": poolname} + ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'') + eq(ret, 0) + assert len(out) > 0 + eq(u"pool '\u9ec5' created", out) + + +@attr('watch') +class TestWatchNotify(object): + OID = "test_watch_notify" + + def setUp(self): + self.rados = Rados(conffile='') + self.rados.connect() + self.rados.create_pool('test_pool') + assert self.rados.pool_exists('test_pool') + self.ioctx = self.rados.open_ioctx('test_pool') + self.ioctx.write(self.OID, b'test watch notify') + self.lock = threading.Condition() + self.notify_cnt = {} + self.notify_data = {} + self.notify_error = {} + # aio related + self.ack_cnt = {} + self.ack_data = {} + self.instance_id = self.rados.get_instance_id() + + def tearDown(self): + self.ioctx.close() + self.rados.delete_pool('test_pool') + self.rados.shutdown() + + def make_callback(self): + def callback(notify_id, notifier_id, watch_id, data): + with self.lock: + if watch_id not in self.notify_cnt: + self.notify_cnt[watch_id] = 1 + elif self.notify_data[watch_id] != data: + self.notify_cnt[watch_id] += 1 + self.notify_data[watch_id] = data + return callback + + def make_error_callback(self): + def callback(watch_id, error): + with self.lock: + self.notify_error[watch_id] = error + return callback + + + def test(self): + with self.ioctx.watch(self.OID, self.make_callback(), + self.make_error_callback()) as watch1: + watch_id1 = watch1.get_id() + assert(watch_id1 > 0) + + with self.rados.open_ioctx('test_pool') as ioctx: + watch2 = ioctx.watch(self.OID, self.make_callback(), + self.make_error_callback()) + watch_id2 = watch2.get_id() + assert(watch_id2 > 0) + + assert(self.ioctx.notify(self.OID, 'test')) + with self.lock: + assert(watch_id1 in self.notify_cnt) + assert(watch_id2 in self.notify_cnt) + eq(self.notify_cnt[watch_id1], 1) + eq(self.notify_cnt[watch_id2], 1) + eq(self.notify_data[watch_id1], b'test') + eq(self.notify_data[watch_id2], b'test') + + assert(watch1.check() >= timedelta()) + assert(watch2.check() >= timedelta()) + + assert(self.ioctx.notify(self.OID, 'best')) + with self.lock: + eq(self.notify_cnt[watch_id1], 2) + eq(self.notify_cnt[watch_id2], 2) + eq(self.notify_data[watch_id1], b'best') + eq(self.notify_data[watch_id2], b'best') + + watch2.close() + + assert(self.ioctx.notify(self.OID, 'rest')) + with self.lock: + eq(self.notify_cnt[watch_id1], 3) + eq(self.notify_cnt[watch_id2], 2) + eq(self.notify_data[watch_id1], b'rest') + eq(self.notify_data[watch_id2], b'best') + + assert(watch1.check() >= timedelta()) + + self.ioctx.remove_object(self.OID) + + for i in range(10): + with self.lock: + if watch_id1 in self.notify_error: + break + time.sleep(1) + eq(self.notify_error[watch_id1], -errno.ENOTCONN) + assert_raises(NotConnected, watch1.check) + + assert_raises(ObjectNotFound, self.ioctx.notify, self.OID, 'test') + + def make_callback_reply(self): + def callback(notify_id, notifier_id, watch_id, data): + with self.lock: + return data + return callback + + def notify_callback(self, _, r, ack_list, timeout_list): + eq(r, 0) + with self.lock: + for notifier_id, _, notifier_data in ack_list: + if notifier_id not in self.ack_cnt: + self.ack_cnt[notifier_id] = 0 + self.ack_cnt[notifier_id] += 1 + self.ack_data[notifier_id] = notifier_data + + def notify_callback_err(self, _, r, ack_list, timeout_list): + eq(r, -errno.ENOENT) + + def test_aio_notify(self): + with self.ioctx.watch(self.OID, self.make_callback_reply(), + self.make_error_callback()) as watch1: + watch_id1 = watch1.get_id() + ok(watch_id1 > 0) + + with self.rados.open_ioctx('test_pool') as ioctx: + watch2 = ioctx.watch(self.OID, self.make_callback_reply(), + self.make_error_callback()) + watch_id2 = watch2.get_id() + ok(watch_id2 > 0) + + comp = self.ioctx.aio_notify(self.OID, self.notify_callback, msg='test') + comp.wait_for_complete_and_cb() + with self.lock: + ok(self.instance_id in self.ack_cnt) + eq(self.ack_cnt[self.instance_id], 2) + eq(self.ack_data[self.instance_id], b'test') + + ok(watch1.check() >= timedelta()) + ok(watch2.check() >= timedelta()) + + comp = self.ioctx.aio_notify(self.OID, self.notify_callback, msg='best') + comp.wait_for_complete_and_cb() + with self.lock: + eq(self.ack_cnt[self.instance_id], 4) + eq(self.ack_data[self.instance_id], b'best') + + watch2.close() + + comp = self.ioctx.aio_notify(self.OID, self.notify_callback, msg='rest') + comp.wait_for_complete_and_cb() + with self.lock: + eq(self.ack_cnt[self.instance_id], 5) + eq(self.ack_data[self.instance_id], b'rest') + + assert(watch1.check() >= timedelta()) + self.ioctx.remove_object(self.OID) + + for i in range(10): + with self.lock: + if watch_id1 in self.notify_error: + break + time.sleep(1) + eq(self.notify_error[watch_id1], -errno.ENOTCONN) + assert_raises(NotConnected, watch1.check) + + comp = self.ioctx.aio_notify(self.OID, self.notify_callback_err, msg='test') + comp.wait_for_complete_and_cb() |