ceph/src/test/pybind/test_rados.py

   1 from __future__ import print_function
   2 from nose import SkipTest
   3 from nose.tools import eq_ as eq, ok_ as ok, assert_raises
   4 from rados import (Rados, Error, RadosStateError, Object, ObjectExists,
   5                    ObjectNotFound, ObjectBusy, requires, opt,
   6                    LIBRADOS_ALL_NSPACES, WriteOpCtx, ReadOpCtx, LIBRADOS_CREATE_EXCLUSIVE,
   7                    LIBRADOS_SNAP_HEAD, LIBRADOS_OPERATION_BALANCE_READS, LIBRADOS_OPERATION_SKIPRWLOCKS, MonitorLog)
   8 import time
   9 import threading
  10 import json
  11 import errno
  12 import os
  13 import re
  14 import sys
  15
  16 # Are we running Python 2.x
  17 _python2 = sys.version_info[0] < 3
  18
  19 def test_rados_init_error():
  20     assert_raises(Error, Rados, conffile='', rados_id='admin',
  21                   name='client.admin')
  22     assert_raises(Error, Rados, conffile='', name='invalid')
  23     assert_raises(Error, Rados, conffile='', name='bad.invalid')
  24
  25 def test_rados_init():
  26     with Rados(conffile='', rados_id='admin'):
  27         pass
  28     with Rados(conffile='', name='client.admin'):
  29         pass
  30     with Rados(conffile='', name='client.admin'):
  31         pass
  32     with Rados(conffile='', name='client.admin'):
  33         pass
  34
  35 def test_ioctx_context_manager():
  36     with Rados(conffile='', rados_id='admin') as conn:
  37         with conn.open_ioctx('rbd') as ioctx:
  38             pass
  39
  40 def test_parse_argv():
  41     args = ['osd', 'pool', 'delete', 'foobar', 'foobar', '--yes-i-really-really-mean-it']
  42     r = Rados()
  43     eq(args, r.conf_parse_argv(args))
  44
  45 def test_parse_argv_empty_str():
  46     args = ['']
  47     r = Rados()
  48     eq(args, r.conf_parse_argv(args))
  49
  50 class TestRequires(object):
  51     @requires(('foo', str), ('bar', int), ('baz', int))
  52     def _method_plain(self, foo, bar, baz):
  53         ok(isinstance(foo, str))
  54         ok(isinstance(bar, int))
  55         ok(isinstance(baz, int))
  56         return (foo, bar, baz)
  57
  58     def test_method_plain(self):
  59         assert_raises(TypeError, self._method_plain, 42, 42, 42)
  60         assert_raises(TypeError, self._method_plain, '42', '42', '42')
  61         assert_raises(TypeError, self._method_plain, foo='42', bar='42', baz='42')
  62         eq(self._method_plain('42', 42, 42), ('42', 42, 42))
  63         eq(self._method_plain(foo='42', bar=42, baz=42), ('42', 42, 42))
  64
  65     @requires(('opt_foo', opt(str)), ('opt_bar', opt(int)), ('baz', int))
  66     def _method_with_opt_arg(self, foo, bar, baz):
  67         ok(isinstance(foo, str) or foo is None)
  68         ok(isinstance(bar, int) or bar is None)
  69         ok(isinstance(baz, int))
  70         return (foo, bar, baz)
  71
  72     def test_method_with_opt_args(self):
  73         assert_raises(TypeError, self._method_with_opt_arg, 42, 42, 42)
  74         assert_raises(TypeError, self._method_with_opt_arg, '42', '42', 42)
  75         assert_raises(TypeError, self._method_with_opt_arg, None, None, None)
  76         eq(self._method_with_opt_arg(None, 42, 42), (None, 42, 42))
  77         eq(self._method_with_opt_arg('42', None, 42), ('42', None, 42))
  78         eq(self._method_with_opt_arg(None, None, 42), (None, None, 42))
  79
  80
  81 class TestRadosStateError(object):
  82     def _requires_configuring(self, rados):
  83         assert_raises(RadosStateError, rados.connect)
  84
  85     def _requires_configuring_or_connected(self, rados):
  86         assert_raises(RadosStateError, rados.conf_read_file)
  87         assert_raises(RadosStateError, rados.conf_parse_argv, None)
  88         assert_raises(RadosStateError, rados.conf_parse_env)
  89         assert_raises(RadosStateError, rados.conf_get, 'opt')
  90         assert_raises(RadosStateError, rados.conf_set, 'opt', 'val')
  91         assert_raises(RadosStateError, rados.ping_monitor, 0)
  92
  93     def _requires_connected(self, rados):
  94         assert_raises(RadosStateError, rados.pool_exists, 'foo')
  95         assert_raises(RadosStateError, rados.pool_lookup, 'foo')
  96         assert_raises(RadosStateError, rados.pool_reverse_lookup, 0)
  97         assert_raises(RadosStateError, rados.create_pool, 'foo')
  98         assert_raises(RadosStateError, rados.get_pool_base_tier, 0)
  99         assert_raises(RadosStateError, rados.delete_pool, 'foo')
 100         assert_raises(RadosStateError, rados.list_pools)
 101         assert_raises(RadosStateError, rados.get_fsid)
 102         assert_raises(RadosStateError, rados.open_ioctx, 'foo')
 103         assert_raises(RadosStateError, rados.mon_command, '', b'')
 104         assert_raises(RadosStateError, rados.osd_command, 0, '', b'')
 105         assert_raises(RadosStateError, rados.pg_command, '', '', b'')
 106         assert_raises(RadosStateError, rados.wait_for_latest_osdmap)
 107         assert_raises(RadosStateError, rados.blacklist_add, '127.0.0.1/123', 0)
 108
 109     def test_configuring(self):
 110         rados = Rados(conffile='')
 111         eq('configuring', rados.state)
 112         self._requires_connected(rados)
 113
 114     def test_connected(self):
 115         rados = Rados(conffile='')
 116         with rados:
 117             eq('connected', rados.state)
 118             self._requires_configuring(rados)
 119
 120     def test_shutdown(self):
 121         rados = Rados(conffile='')
 122         with rados:
 123             pass
 124         eq('shutdown', rados.state)
 125         self._requires_configuring(rados)
 126         self._requires_configuring_or_connected(rados)
 127         self._requires_connected(rados)
 128
 129
 130 class TestRados(object):
 131
 132     def setUp(self):
 133         self.rados = Rados(conffile='')
 134         self.rados.conf_parse_env('FOO_DOES_NOT_EXIST_BLAHBLAH')
 135         self.rados.conf_parse_env()
 136         self.rados.connect()
 137
 138         # Assume any pre-existing pools are the cluster's defaults
 139         self.default_pools = self.rados.list_pools()
 140
 141     def tearDown(self):
 142         self.rados.shutdown()
 143
 144     def test_ping_monitor(self):
 145         assert_raises(ObjectNotFound, self.rados.ping_monitor, 'not_exists_monitor')
 146         cmd = {'prefix': 'mon dump', 'format':'json'}
 147         ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'')
 148         for mon in json.loads(buf.decode('utf8'))['mons']:
 149             while True:
 150                 output = self.rados.ping_monitor(mon['name'])
 151                 if output is None:
 152                     continue
 153                 buf = json.loads(output)
 154                 if buf.get('health'):
 155                     break
 156
 157     def test_create(self):
 158         self.rados.create_pool('foo')
 159         self.rados.delete_pool('foo')
 160
 161     def test_create_utf8(self):
 162         if _python2:
 163             # Use encoded bytestring
 164             poolname = b"\351\273\204"
 165         else:
 166             poolname = "\u9ec4"
 167         self.rados.create_pool(poolname)
 168         assert self.rados.pool_exists(u"\u9ec4")
 169         self.rados.delete_pool(poolname)
 170
 171     def test_pool_lookup_utf8(self):
 172         if _python2:
 173             poolname = u'\u9ec4'
 174         else:
 175             poolname = '\u9ec4'
 176         self.rados.create_pool(poolname)
 177         try:
 178             poolid = self.rados.pool_lookup(poolname)
 179             eq(poolname, self.rados.pool_reverse_lookup(poolid))
 180         finally:
 181             self.rados.delete_pool(poolname)
 182
 183     def test_eexist(self):
 184         self.rados.create_pool('foo')
 185         assert_raises(ObjectExists, self.rados.create_pool, 'foo')
 186         self.rados.delete_pool('foo')
 187
 188     def list_non_default_pools(self):
 189         pools = self.rados.list_pools()
 190         for p in self.default_pools:
 191             pools.remove(p)
 192         return set(pools)
 193
 194     def test_list_pools(self):
 195         eq(set(), self.list_non_default_pools())
 196         self.rados.create_pool('foo')
 197         eq(set(['foo']), self.list_non_default_pools())
 198         self.rados.create_pool('bar')
 199         eq(set(['foo', 'bar']), self.list_non_default_pools())
 200         self.rados.create_pool('baz')
 201         eq(set(['foo', 'bar', 'baz']), self.list_non_default_pools())
 202         self.rados.delete_pool('foo')
 203         eq(set(['bar', 'baz']), self.list_non_default_pools())
 204         self.rados.delete_pool('baz')
 205         eq(set(['bar']), self.list_non_default_pools())
 206         self.rados.delete_pool('bar')
 207         eq(set(), self.list_non_default_pools())
 208         self.rados.create_pool('a' * 500)
 209         eq(set(['a' * 500]), self.list_non_default_pools())
 210         self.rados.delete_pool('a' * 500)
 211
 212     def test_get_pool_base_tier(self):
 213         self.rados.create_pool('foo')
 214         try:
 215             self.rados.create_pool('foo-cache')
 216             try:
 217                 pool_id = self.rados.pool_lookup('foo')
 218                 tier_pool_id = self.rados.pool_lookup('foo-cache')
 219
 220                 cmd = {"prefix":"osd tier add", "pool":"foo", "tierpool":"foo-cache", "force_nonempty":""}
 221                 ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30)
 222                 eq(ret, 0)
 223
 224                 try:
 225                     cmd = {"prefix":"osd tier cache-mode", "pool":"foo-cache", "tierpool":"foo-cache", "mode":"readonly", "yes_i_really_mean_it": True}
 226                     ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30)
 227                     eq(ret, 0)
 228
 229                     eq(self.rados.wait_for_latest_osdmap(), 0)
 230
 231                     eq(pool_id, self.rados.get_pool_base_tier(pool_id))
 232                     eq(pool_id, self.rados.get_pool_base_tier(tier_pool_id))
 233                 finally:
 234                     cmd = {"prefix":"osd tier remove", "pool":"foo", "tierpool":"foo-cache"}
 235                     ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30)
 236                     eq(ret, 0)
 237             finally:
 238                 self.rados.delete_pool('foo-cache')
 239         finally:
 240             self.rados.delete_pool('foo')
 241
 242     def test_get_fsid(self):
 243         fsid = self.rados.get_fsid()
 244         assert re.match('[0-9a-f\-]{36}', fsid, re.I)
 245
 246     def test_blacklist_add(self):
 247         self.rados.blacklist_add("1.2.3.4/123", 1)
 248
 249     def test_get_cluster_stats(self):
 250         stats = self.rados.get_cluster_stats()
 251         assert stats['kb'] > 0
 252         assert stats['kb_avail'] > 0
 253         assert stats['kb_used'] > 0
 254         assert stats['num_objects'] >= 0
 255
 256     def test_monitor_log(self):
 257         lock = threading.Condition()
 258         def cb(arg, line, who, sec, nsec, seq, level, msg):
 259             # NOTE(sileht): the old pyrados API was received the pointer as int
 260             # instead of the value of arg
 261             eq(arg, "arg")
 262             with lock:
 263                 lock.notify()
 264             return 0
 265
 266         # NOTE(sileht): force don't save the monitor into local var
 267         # to ensure all references are correctly tracked into the lib
 268         MonitorLog(self.rados, "debug", cb, "arg")
 269         with lock:
 270             lock.wait()
 271         MonitorLog(self.rados, "debug", None, None)
 272         eq(None, self.rados.monitor_callback)
 273
 274 class TestIoctx(object):
 275
 276     def setUp(self):
 277         self.rados = Rados(conffile='')
 278         self.rados.connect()
 279         self.rados.create_pool('test_pool')
 280         assert self.rados.pool_exists('test_pool')
 281         self.ioctx = self.rados.open_ioctx('test_pool')
 282
 283     def tearDown(self):
 284         cmd = {"prefix":"osd unset", "key":"noup"}
 285         self.rados.mon_command(json.dumps(cmd), b'')
 286         self.ioctx.close()
 287         self.rados.delete_pool('test_pool')
 288         self.rados.shutdown()
 289
 290     def test_get_last_version(self):
 291         version = self.ioctx.get_last_version()
 292         assert version >= 0
 293
 294     def test_get_stats(self):
 295         stats = self.ioctx.get_stats()
 296         eq(stats, {'num_objects_unfound': 0,
 297                    'num_objects_missing_on_primary': 0,
 298                    'num_object_clones': 0,
 299                    'num_objects': 0,
 300                    'num_object_copies': 0,
 301                    'num_bytes': 0,
 302                    'num_rd_kb': 0,
 303                    'num_wr_kb': 0,
 304                    'num_kb': 0,
 305                    'num_wr': 0,
 306                    'num_objects_degraded': 0,
 307                    'num_rd': 0})
 308
 309     def test_write(self):
 310         self.ioctx.write('abc', b'abc')
 311         eq(self.ioctx.read('abc'), b'abc')
 312
 313     def test_write_full(self):
 314         self.ioctx.write('abc', b'abc')
 315         eq(self.ioctx.read('abc'), b'abc')
 316         self.ioctx.write_full('abc', b'd')
 317         eq(self.ioctx.read('abc'), b'd')
 318
 319     def test_writesame(self):
 320         self.ioctx.writesame('ob', b'rzx', 9)
 321         eq(self.ioctx.read('ob'), b'rzxrzxrzx')
 322
 323     def test_append(self):
 324         self.ioctx.write('abc', b'a')
 325         self.ioctx.append('abc', b'b')
 326         self.ioctx.append('abc', b'c')
 327         eq(self.ioctx.read('abc'), b'abc')
 328
 329     def test_write_zeros(self):
 330         self.ioctx.write('abc', b'a\0b\0c')
 331         eq(self.ioctx.read('abc'), b'a\0b\0c')
 332
 333     def test_trunc(self):
 334         self.ioctx.write('abc', b'abc')
 335         self.ioctx.trunc('abc', 2)
 336         eq(self.ioctx.read('abc'), b'ab')
 337         size = self.ioctx.stat('abc')[0]
 338         eq(size, 2)
 339
 340     def test_list_objects_empty(self):
 341         eq(list(self.ioctx.list_objects()), [])
 342
 343     def test_list_objects(self):
 344         self.ioctx.write('a', b'')
 345         self.ioctx.write('b', b'foo')
 346         self.ioctx.write_full('c', b'bar')
 347         self.ioctx.append('d', b'jazz')
 348         object_names = [obj.key for obj in self.ioctx.list_objects()]
 349         eq(sorted(object_names), ['a', 'b', 'c', 'd'])
 350
 351     def test_list_ns_objects(self):
 352         self.ioctx.write('a', b'')
 353         self.ioctx.write('b', b'foo')
 354         self.ioctx.write_full('c', b'bar')
 355         self.ioctx.append('d', b'jazz')
 356         self.ioctx.set_namespace("ns1")
 357         self.ioctx.write('ns1-a', b'')
 358         self.ioctx.write('ns1-b', b'foo')
 359         self.ioctx.write_full('ns1-c', b'bar')
 360         self.ioctx.append('ns1-d', b'jazz')
 361         self.ioctx.append('d', b'jazz')
 362         self.ioctx.set_namespace(LIBRADOS_ALL_NSPACES)
 363         object_names = [(obj.nspace, obj.key) for obj in self.ioctx.list_objects()]
 364         eq(sorted(object_names), [('', 'a'), ('','b'), ('','c'), ('','d'),\
 365                 ('ns1', 'd'), ('ns1', 'ns1-a'), ('ns1', 'ns1-b'),\
 366                 ('ns1', 'ns1-c'), ('ns1', 'ns1-d')])
 367
 368     def test_xattrs(self):
 369         xattrs = dict(a=b'1', b=b'2', c=b'3', d=b'a\0b', e=b'\0', f=b'')
 370         self.ioctx.write('abc', b'')
 371         for key, value in xattrs.items():
 372             self.ioctx.set_xattr('abc', key, value)
 373             eq(self.ioctx.get_xattr('abc', key), value)
 374         stored_xattrs = {}
 375         for key, value in self.ioctx.get_xattrs('abc'):
 376             stored_xattrs[key] = value
 377         eq(stored_xattrs, xattrs)
 378
 379     def test_obj_xattrs(self):
 380         xattrs = dict(a=b'1', b=b'2', c=b'3', d=b'a\0b', e=b'\0', f=b'')
 381         self.ioctx.write('abc', b'')
 382         obj = list(self.ioctx.list_objects())[0]
 383         for key, value in xattrs.items():
 384             obj.set_xattr(key, value)
 385             eq(obj.get_xattr(key), value)
 386         stored_xattrs = {}
 387         for key, value in obj.get_xattrs():
 388             stored_xattrs[key] = value
 389         eq(stored_xattrs, xattrs)
 390
 391     def test_get_pool_id(self):
 392         eq(self.ioctx.get_pool_id(), self.rados.pool_lookup('test_pool'))
 393
 394     def test_get_pool_name(self):
 395         eq(self.ioctx.get_pool_name(), 'test_pool')
 396
 397     def test_create_snap(self):
 398         assert_raises(ObjectNotFound, self.ioctx.remove_snap, 'foo')
 399         self.ioctx.create_snap('foo')
 400         self.ioctx.remove_snap('foo')
 401
 402     def test_list_snaps_empty(self):
 403         eq(list(self.ioctx.list_snaps()), [])
 404
 405     def test_list_snaps(self):
 406         snaps = ['snap1', 'snap2', 'snap3']
 407         for snap in snaps:
 408             self.ioctx.create_snap(snap)
 409         listed_snaps = [snap.name for snap in self.ioctx.list_snaps()]
 410         eq(snaps, listed_snaps)
 411
 412     def test_lookup_snap(self):
 413         self.ioctx.create_snap('foo')
 414         snap = self.ioctx.lookup_snap('foo')
 415         eq(snap.name, 'foo')
 416
 417     def test_snap_timestamp(self):
 418         self.ioctx.create_snap('foo')
 419         snap = self.ioctx.lookup_snap('foo')
 420         snap.get_timestamp()
 421
 422     def test_remove_snap(self):
 423         self.ioctx.create_snap('foo')
 424         (snap,) = self.ioctx.list_snaps()
 425         eq(snap.name, 'foo')
 426         self.ioctx.remove_snap('foo')
 427         eq(list(self.ioctx.list_snaps()), [])
 428
 429     def test_snap_rollback(self):
 430         self.ioctx.write("insnap", b"contents1")
 431         self.ioctx.create_snap("snap1")
 432         self.ioctx.remove_object("insnap")
 433         self.ioctx.snap_rollback("insnap", "snap1")
 434         eq(self.ioctx.read("insnap"), b"contents1")
 435         self.ioctx.remove_snap("snap1")
 436         self.ioctx.remove_object("insnap")
 437
 438     def test_snap_read(self):
 439         self.ioctx.write("insnap", b"contents1")
 440         self.ioctx.create_snap("snap1")
 441         self.ioctx.remove_object("insnap")
 442         snap = self.ioctx.lookup_snap("snap1")
 443         self.ioctx.set_read(snap.snap_id)
 444         eq(self.ioctx.read("insnap"), b"contents1")
 445         self.ioctx.set_read(LIBRADOS_SNAP_HEAD)
 446         self.ioctx.write("inhead", b"contents2")
 447         eq(self.ioctx.read("inhead"), b"contents2")
 448         self.ioctx.remove_snap("snap1")
 449         self.ioctx.remove_object("inhead")
 450
 451     def test_set_omap(self):
 452         keys = ("1", "2", "3", "4")
 453         values = (b"aaa", b"bbb", b"ccc", b"\x04\x04\x04\x04")
 454         with WriteOpCtx() as write_op:
 455             self.ioctx.set_omap(write_op, keys, values)
 456             write_op.set_flags(LIBRADOS_OPERATION_SKIPRWLOCKS)
 457             self.ioctx.operate_write_op(write_op, "hw")
 458         with ReadOpCtx() as read_op:
 459             iter, ret = self.ioctx.get_omap_vals(read_op, "", "", 4)
 460             eq(ret, 0)
 461             self.ioctx.operate_read_op(read_op, "hw")
 462             next(iter)
 463             eq(list(iter), [("2", b"bbb"), ("3", b"ccc"), ("4", b"\x04\x04\x04\x04")])
 464         with ReadOpCtx() as read_op:
 465             iter, ret = self.ioctx.get_omap_vals(read_op, "2", "", 4)
 466             eq(ret, 0)
 467             self.ioctx.operate_read_op(read_op, "hw")
 468             eq(("3", b"ccc"), next(iter))
 469             eq(list(iter), [("4", b"\x04\x04\x04\x04")])
 470         with ReadOpCtx() as read_op:
 471             iter, ret = self.ioctx.get_omap_vals(read_op, "", "2", 4)
 472             eq(ret, 0)
 473             read_op.set_flags(LIBRADOS_OPERATION_BALANCE_READS)
 474             self.ioctx.operate_read_op(read_op, "hw")
 475             eq(list(iter), [("2", b"bbb")])
 476
 477     def test_set_omap_aio(self):
 478         lock = threading.Condition()
 479         count = [0]
 480         def cb(blah):
 481             with lock:
 482                 count[0] += 1
 483                 lock.notify()
 484             return 0
 485
 486         keys = ("1", "2", "3", "4")
 487         values = (b"aaa", b"bbb", b"ccc", b"\x04\x04\x04\x04")
 488         with WriteOpCtx() as write_op:
 489             self.ioctx.set_omap(write_op, keys, values)
 490             comp = self.ioctx.operate_aio_write_op(write_op, "hw", cb, cb)
 491             comp.wait_for_complete()
 492             with lock:
 493                 while count[0] < 2:
 494                     lock.wait()
 495             eq(comp.get_return_value(), 0)
 496
 497         with ReadOpCtx() as read_op:
 498             iter, ret = self.ioctx.get_omap_vals(read_op, "", "", 4)
 499             eq(ret, 0)
 500             comp = self.ioctx.operate_aio_read_op(read_op, "hw", cb, cb)
 501             comp.wait_for_complete()
 502             with lock:
 503                 while count[0] < 4:
 504                     lock.wait()
 505             eq(comp.get_return_value(), 0)
 506             next(iter)
 507             eq(list(iter), [("2", b"bbb"), ("3", b"ccc"), ("4", b"\x04\x04\x04\x04")])
 508
 509     def test_write_ops(self):
 510         with WriteOpCtx() as write_op:
 511             write_op.new(0)
 512             self.ioctx.operate_write_op(write_op, "write_ops")
 513             eq(self.ioctx.read('write_ops'), b'')
 514             write_op.write_full(b'1')
 515             write_op.append(b'2')
 516             self.ioctx.operate_write_op(write_op, "write_ops")
 517             eq(self.ioctx.read('write_ops'), b'12')
 518             write_op.write_full(b'12345')
 519             write_op.write(b'x', 2)
 520             self.ioctx.operate_write_op(write_op, "write_ops")
 521             eq(self.ioctx.read('write_ops'), b'12x45')
 522             write_op.write_full(b'12345')
 523             write_op.zero(2, 2)
 524             self.ioctx.operate_write_op(write_op, "write_ops")
 525             eq(self.ioctx.read('write_ops'), b'12\x00\x005')
 526             write_op.write_full(b'12345')
 527             write_op.truncate(2)
 528             self.ioctx.operate_write_op(write_op, "write_ops")
 529             eq(self.ioctx.read('write_ops'), b'12')
 530             write_op.remove()
 531             self.ioctx.operate_write_op(write_op, "write_ops")
 532             with assert_raises(ObjectNotFound):
 533                 self.ioctx.read('write_ops')
 534
 535     def test_execute_op(self):
 536         with WriteOpCtx() as write_op:
 537             write_op.execute("hello", "record_hello", b"ebs")
 538             self.ioctx.operate_write_op(write_op, "object")
 539         eq(self.ioctx.read('object'), b"Hello, ebs!")
 540
 541     def test_writesame_op(self):
 542         with WriteOpCtx() as write_op:
 543             write_op.writesame(b'rzx', 9)
 544             self.ioctx.operate_write_op(write_op, 'abc')
 545             eq(self.ioctx.read('abc'), b'rzxrzxrzx')
 546
 547     def test_get_omap_vals_by_keys(self):
 548         keys = ("1", "2", "3", "4")
 549         values = (b"aaa", b"bbb", b"ccc", b"\x04\x04\x04\x04")
 550         with WriteOpCtx() as write_op:
 551             self.ioctx.set_omap(write_op, keys, values)
 552             self.ioctx.operate_write_op(write_op, "hw")
 553         with ReadOpCtx() as read_op:
 554             iter, ret = self.ioctx.get_omap_vals_by_keys(read_op,("3","4",))
 555             eq(ret, 0)
 556             self.ioctx.operate_read_op(read_op, "hw")
 557             eq(list(iter), [("3", b"ccc"), ("4", b"\x04\x04\x04\x04")])
 558         with ReadOpCtx() as read_op:
 559             iter, ret = self.ioctx.get_omap_vals_by_keys(read_op,("3","4",))
 560             eq(ret, 0)
 561             with assert_raises(ObjectNotFound):
 562                 self.ioctx.operate_read_op(read_op, "no_such")
 563
 564     def test_get_omap_keys(self):
 565         keys = ("1", "2", "3")
 566         values = (b"aaa", b"bbb", b"ccc")
 567         with WriteOpCtx() as write_op:
 568             self.ioctx.set_omap(write_op, keys, values)
 569             self.ioctx.operate_write_op(write_op, "hw")
 570         with ReadOpCtx() as read_op:
 571             iter, ret = self.ioctx.get_omap_keys(read_op,"",2)
 572             eq(ret, 0)
 573             self.ioctx.operate_read_op(read_op, "hw")
 574             eq(list(iter), [("1", None), ("2", None)])
 575         with ReadOpCtx() as read_op:
 576             iter, ret = self.ioctx.get_omap_keys(read_op,"",2)
 577             eq(ret, 0)
 578             with assert_raises(ObjectNotFound):
 579                 self.ioctx.operate_read_op(read_op, "no_such")
 580
 581     def test_clear_omap(self):
 582         keys = ("1", "2", "3")
 583         values = (b"aaa", b"bbb", b"ccc")
 584         with WriteOpCtx() as write_op:
 585             self.ioctx.set_omap(write_op, keys, values)
 586             self.ioctx.operate_write_op(write_op, "hw")
 587         with WriteOpCtx() as write_op_1:
 588             self.ioctx.clear_omap(write_op_1)
 589             self.ioctx.operate_write_op(write_op_1, "hw")
 590         with ReadOpCtx() as read_op:
 591             iter, ret = self.ioctx.get_omap_vals_by_keys(read_op,("1",))
 592             eq(ret, 0)
 593             self.ioctx.operate_read_op(read_op, "hw")
 594             eq(list(iter), [])
 595
 596     def test_xattrs_op(self):
 597         xattrs = dict(a=b'1', b=b'2', c=b'3', d=b'a\0b', e=b'\0')
 598         with WriteOpCtx() as write_op:
 599             write_op.new(LIBRADOS_CREATE_EXCLUSIVE)
 600             for key, value in xattrs.items():
 601                 write_op.set_xattr(key, value)
 602                 self.ioctx.operate_write_op(write_op, "abc")
 603                 eq(self.ioctx.get_xattr('abc', key), value)
 604             stored_xattrs_1 = {}
 605             for key, value in self.ioctx.get_xattrs('abc'):
 606                 stored_xattrs_1[key] = value
 607             eq(stored_xattrs_1, xattrs)
 608             for key in xattrs.keys():
 609                 write_op.rm_xattr(key)
 610                 self.ioctx.operate_write_op(write_op, "abc")
 611             stored_xattrs_2 = {}
 612             for key, value in self.ioctx.get_xattrs('abc'):
 613                 stored_xattrs_2[key] = value
 614             eq(stored_xattrs_2, {})
 615             write_op.remove()
 616
 617     def test_locator(self):
 618         self.ioctx.set_locator_key("bar")
 619         self.ioctx.write('foo', b'contents1')
 620         objects = [i for i in self.ioctx.list_objects()]
 621         eq(len(objects), 1)
 622         eq(self.ioctx.get_locator_key(), "bar")
 623         self.ioctx.set_locator_key("")
 624         objects[0].seek(0)
 625         objects[0].write(b"contents2")
 626         eq(self.ioctx.get_locator_key(), "")
 627         self.ioctx.set_locator_key("bar")
 628         contents = self.ioctx.read("foo")
 629         eq(contents, b"contents2")
 630         eq(self.ioctx.get_locator_key(), "bar")
 631         objects[0].remove()
 632         objects = [i for i in self.ioctx.list_objects()]
 633         eq(objects, [])
 634         self.ioctx.set_locator_key("")
 635
 636     def test_operate_aio_write_op(self):
 637         lock = threading.Condition()
 638         count = [0]
 639         def cb(blah):
 640             with lock:
 641                 count[0] += 1
 642                 lock.notify()
 643             return 0
 644         with WriteOpCtx() as write_op:
 645             write_op.write(b'rzx')
 646             comp = self.ioctx.operate_aio_write_op(write_op, "object", cb, cb)
 647             comp.wait_for_complete()
 648             with lock:
 649                 while count[0] < 2:
 650                     lock.wait()
 651             eq(comp.get_return_value(), 0)
 652             eq(self.ioctx.read('object'), b'rzx')
 653
 654     def test_aio_write(self):
 655         lock = threading.Condition()
 656         count = [0]
 657         def cb(blah):
 658             with lock:
 659                 count[0] += 1
 660                 lock.notify()
 661             return 0
 662         comp = self.ioctx.aio_write("foo", b"bar", 0, cb, cb)
 663         comp.wait_for_complete()
 664         with lock:
 665             while count[0] < 2:
 666                 lock.wait()
 667         eq(comp.get_return_value(), 0)
 668         contents = self.ioctx.read("foo")
 669         eq(contents, b"bar")
 670         [i.remove() for i in self.ioctx.list_objects()]
 671
 672     def test_aio_write_no_comp_ref(self):
 673         lock = threading.Condition()
 674         count = [0]
 675         def cb(blah):
 676             with lock:
 677                 count[0] += 1
 678                 lock.notify()
 679             return 0
 680         # NOTE(sileht): force don't save the comp into local var
 681         # to ensure all references are correctly tracked into the lib
 682         self.ioctx.aio_write("foo", b"bar", 0, cb, cb)
 683         with lock:
 684             while count[0] < 2:
 685                 lock.wait()
 686         contents = self.ioctx.read("foo")
 687         eq(contents, b"bar")
 688         [i.remove() for i in self.ioctx.list_objects()]
 689
 690     def test_aio_append(self):
 691         lock = threading.Condition()
 692         count = [0]
 693         def cb(blah):
 694             with lock:
 695                 count[0] += 1
 696                 lock.notify()
 697             return 0
 698         comp = self.ioctx.aio_write("foo", b"bar", 0, cb, cb)
 699         comp2 = self.ioctx.aio_append("foo", b"baz", cb, cb)
 700         comp.wait_for_complete()
 701         contents = self.ioctx.read("foo")
 702         eq(contents, b"barbaz")
 703         with lock:
 704             while count[0] < 4:
 705                 lock.wait()
 706         eq(comp.get_return_value(), 0)
 707         eq(comp2.get_return_value(), 0)
 708         [i.remove() for i in self.ioctx.list_objects()]
 709
 710     def test_aio_write_full(self):
 711         lock = threading.Condition()
 712         count = [0]
 713         def cb(blah):
 714             with lock:
 715                 count[0] += 1
 716                 lock.notify()
 717             return 0
 718         self.ioctx.aio_write("foo", b"barbaz", 0, cb, cb)
 719         comp = self.ioctx.aio_write_full("foo", b"bar", cb, cb)
 720         comp.wait_for_complete()
 721         with lock:
 722             while count[0] < 2:
 723                 lock.wait()
 724         eq(comp.get_return_value(), 0)
 725         contents = self.ioctx.read("foo")
 726         eq(contents, b"bar")
 727         [i.remove() for i in self.ioctx.list_objects()]
 728
 729     def test_aio_writesame(self):
 730         lock = threading.Condition()
 731         count = [0]
 732         def cb(blah):
 733             with lock:
 734                 count[0] += 1
 735                 lock.notify()
 736             return 0
 737         comp = self.ioctx.aio_writesame("abc", b"rzx", 9, 0, cb)
 738         comp.wait_for_complete()
 739         with lock:
 740             while count[0] < 1:
 741                 lock.wait()
 742         eq(comp.get_return_value(), 0)
 743         eq(self.ioctx.read("abc"), b"rzxrzxrzx")
 744         [i.remove() for i in self.ioctx.list_objects()]
 745
 746     def test_aio_stat(self):
 747         lock = threading.Condition()
 748         count = [0]
 749         def cb(_, size, mtime):
 750             with lock:
 751                 count[0] += 1
 752                 lock.notify()
 753
 754         comp = self.ioctx.aio_stat("foo", cb)
 755         comp.wait_for_complete()
 756         with lock:
 757             while count[0] < 1:
 758                 lock.wait()
 759         eq(comp.get_return_value(), -2)
 760
 761         self.ioctx.write("foo", b"bar")
 762
 763         comp = self.ioctx.aio_stat("foo", cb)
 764         comp.wait_for_complete()
 765         with lock:
 766             while count[0] < 2:
 767                 lock.wait()
 768         eq(comp.get_return_value(), 0)
 769
 770         [i.remove() for i in self.ioctx.list_objects()]
 771
 772     def test_aio_remove(self):
 773         lock = threading.Condition()
 774         count = [0]
 775         def cb(blah):
 776             with lock:
 777                 count[0] += 1
 778                 lock.notify()
 779             return 0
 780         self.ioctx.write('foo', b'wrx')
 781         eq(self.ioctx.read('foo'), b'wrx')
 782         comp = self.ioctx.aio_remove('foo', cb, cb)
 783         comp.wait_for_complete()
 784         with lock:
 785             while count[0] < 2:
 786                 lock.wait()
 787         eq(comp.get_return_value(), 0)
 788         eq(list(self.ioctx.list_objects()), [])
 789
 790     def _take_down_acting_set(self, pool, objectname):
 791         # find acting_set for pool:objectname and take it down; used to
 792         # verify that async reads don't complete while acting set is missing
 793         cmd = {
 794             "prefix":"osd map",
 795             "pool":pool,
 796             "object":objectname,
 797             "format":"json",
 798         }
 799         r, jsonout, _ = self.rados.mon_command(json.dumps(cmd), b'')
 800         objmap = json.loads(jsonout.decode("utf-8"))
 801         acting_set = objmap['acting']
 802         cmd = {"prefix":"osd set", "key":"noup"}
 803         r, _, _ = self.rados.mon_command(json.dumps(cmd), b'')
 804         eq(r, 0)
 805         cmd = {"prefix":"osd down", "ids":[str(i) for i in acting_set]}
 806         r, _, _ = self.rados.mon_command(json.dumps(cmd), b'')
 807         eq(r, 0)
 808
 809         # wait for OSDs to acknowledge the down
 810         eq(self.rados.wait_for_latest_osdmap(), 0)
 811
 812     def _let_osds_back_up(self):
 813         cmd = {"prefix":"osd unset", "key":"noup"}
 814         r, _, _ = self.rados.mon_command(json.dumps(cmd), b'')
 815         eq(r, 0)
 816
 817     def test_aio_read(self):
 818         # this is a list so that the local cb() can modify it
 819         retval = [None]
 820         lock = threading.Condition()
 821         def cb(_, buf):
 822             with lock:
 823                 retval[0] = buf
 824                 lock.notify()
 825         payload = b"bar\000frob"
 826         self.ioctx.write("foo", payload)
 827
 828         # test1: use wait_for_complete() and wait for cb by
 829         # watching retval[0]
 830         self._take_down_acting_set('test_pool', 'foo')
 831         comp = self.ioctx.aio_read("foo", len(payload), 0, cb)
 832         eq(False, comp.is_complete())
 833         time.sleep(3)
 834         eq(False, comp.is_complete())
 835         with lock:
 836             eq(None, retval[0])
 837         self._let_osds_back_up()
 838         comp.wait_for_complete()
 839         loops = 0
 840         with lock:
 841             while retval[0] is None and loops <= 10:
 842                 lock.wait(timeout=5)
 843                 loops += 1
 844         assert(loops <= 10)
 845
 846         eq(retval[0], payload)
 847         eq(sys.getrefcount(comp), 2)
 848
 849         # test2: use wait_for_complete_and_cb(), verify retval[0] is
 850         # set by the time we regain control
 851
 852         retval[0] = None
 853         self._take_down_acting_set('test_pool', 'foo')
 854         comp = self.ioctx.aio_read("foo", len(payload), 0, cb)
 855         eq(False, comp.is_complete())
 856         time.sleep(3)
 857         eq(False, comp.is_complete())
 858         with lock:
 859             eq(None, retval[0])
 860         self._let_osds_back_up()
 861
 862         comp.wait_for_complete_and_cb()
 863         assert(retval[0] is not None)
 864         eq(retval[0], payload)
 865         eq(sys.getrefcount(comp), 2)
 866
 867         # test3: error case, use wait_for_complete_and_cb(), verify retval[0] is
 868         # set by the time we regain control
 869
 870         retval[0] = 1
 871         self._take_down_acting_set('test_pool', 'bar')
 872         comp = self.ioctx.aio_read("bar", len(payload), 0, cb)
 873         eq(False, comp.is_complete())
 874         time.sleep(3)
 875         eq(False, comp.is_complete())
 876         with lock:
 877             eq(1, retval[0])
 878         self._let_osds_back_up()
 879
 880         comp.wait_for_complete_and_cb()
 881         eq(None, retval[0])
 882         assert(comp.get_return_value() < 0)
 883         eq(sys.getrefcount(comp), 2)
 884
 885         [i.remove() for i in self.ioctx.list_objects()]
 886
 887     def test_lock(self):
 888         self.ioctx.lock_exclusive("foo", "lock", "locker", "desc_lock",
 889                                   10000, 0)
 890         assert_raises(ObjectExists,
 891                       self.ioctx.lock_exclusive,
 892                       "foo", "lock", "locker", "desc_lock", 10000, 0)
 893         self.ioctx.unlock("foo", "lock", "locker")
 894         assert_raises(ObjectNotFound, self.ioctx.unlock, "foo", "lock", "locker")
 895
 896         self.ioctx.lock_shared("foo", "lock", "locker1", "tag", "desc_lock",
 897                                10000, 0)
 898         self.ioctx.lock_shared("foo", "lock", "locker2", "tag", "desc_lock",
 899                                10000, 0)
 900         assert_raises(ObjectBusy,
 901                       self.ioctx.lock_exclusive,
 902                       "foo", "lock", "locker3", "desc_lock", 10000, 0)
 903         self.ioctx.unlock("foo", "lock", "locker1")
 904         self.ioctx.unlock("foo", "lock", "locker2")
 905         assert_raises(ObjectNotFound, self.ioctx.unlock, "foo", "lock", "locker1")
 906         assert_raises(ObjectNotFound, self.ioctx.unlock, "foo", "lock", "locker2")
 907
 908     def test_execute(self):
 909         self.ioctx.write("foo", b"") # ensure object exists
 910
 911         ret, buf = self.ioctx.execute("foo", "hello", "say_hello", b"")
 912         eq(buf, b"Hello, world!")
 913
 914         ret, buf = self.ioctx.execute("foo", "hello", "say_hello", b"nose")
 915         eq(buf, b"Hello, nose!")
 916
 917     def test_aio_execute(self):
 918         count = [0]
 919         retval = [None]
 920         lock = threading.Condition()
 921         def cb(_, buf):
 922             with lock:
 923                 if retval[0] is None:
 924                     retval[0] = buf
 925                 count[0] += 1
 926                 lock.notify()
 927         self.ioctx.write("foo", b"") # ensure object exists
 928
 929         comp = self.ioctx.aio_execute("foo", "hello", "say_hello", b"", 32, cb, cb)
 930         comp.wait_for_complete()
 931         with lock:
 932             while count[0] < 2:
 933                 lock.wait()
 934         eq(comp.get_return_value(), 13)
 935         eq(retval[0], b"Hello, world!")
 936
 937         retval[0] = None
 938         comp = self.ioctx.aio_execute("foo", "hello", "say_hello", b"nose", 32, cb, cb)
 939         comp.wait_for_complete()
 940         with lock:
 941             while count[0] < 4:
 942                 lock.wait()
 943         eq(comp.get_return_value(), 12)
 944         eq(retval[0], b"Hello, nose!")
 945
 946         [i.remove() for i in self.ioctx.list_objects()]
 947
 948     def test_applications(self):
 949         cmd = {"prefix":"osd dump", "format":"json"}
 950         ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'')
 951         eq(ret, 0)
 952         assert len(buf) > 0
 953         release = json.loads(buf.decode("utf-8")).get("require_osd_release",
 954                                                       None)
 955         if not release or release[0] < 'l':
 956             raise SkipTest
 957
 958         eq([], self.ioctx.application_list())
 959
 960         self.ioctx.application_enable("app1")
 961         assert_raises(Error, self.ioctx.application_enable, "app2")
 962         self.ioctx.application_enable("app2", True)
 963
 964         assert_raises(Error, self.ioctx.application_metadata_list, "dne")
 965         eq([], self.ioctx.application_metadata_list("app1"))
 966
 967         assert_raises(Error, self.ioctx.application_metadata_set, "dne", "key",
 968                       "key")
 969         self.ioctx.application_metadata_set("app1", "key1", "val1")
 970         eq("val1", self.ioctx.application_metadata_get("app1", "key1"))
 971         self.ioctx.application_metadata_set("app1", "key2", "val2")
 972         eq("val2", self.ioctx.application_metadata_get("app1", "key2"))
 973         self.ioctx.application_metadata_set("app2", "key1", "val1")
 974         eq("val1", self.ioctx.application_metadata_get("app2", "key1"))
 975
 976         eq([("key1", "val1"), ("key2", "val2")],
 977            self.ioctx.application_metadata_list("app1"))
 978
 979         self.ioctx.application_metadata_remove("app1", "key1")
 980         eq([("key2", "val2")], self.ioctx.application_metadata_list("app1"))
 981
 982     def test_service_daemon(self):
 983         name = "pid-" + str(os.getpid())
 984         metadata = {'version': '3.14', 'memory': '42'}
 985         self.rados.service_daemon_register("laundry", name, metadata)
 986         status = {'result': 'unknown', 'test': 'running'}
 987         self.rados.service_daemon_update(status)
 988
 989     def test_alignment(self):
 990         eq(self.ioctx.alignment(), None)
 991
 992
 993 class TestIoctxEc(object):
 994
 995     def setUp(self):
 996         self.rados = Rados(conffile='')
 997         self.rados.connect()
 998         self.pool = 'test-ec'
 999         self.profile = 'testprofile-%s' % self.pool
1000         cmd = {"prefix": "osd erasure-code-profile set",
1001                "name": self.profile, "profile": ["k=2", "m=1", "crush-failure-domain=osd"]}
1002         ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'', timeout=30)
1003         eq(ret, 0, msg=out)
1004         # create ec pool with profile created above
1005         cmd = {'prefix': 'osd pool create', 'pg_num': 8, 'pgp_num': 8,
1006                'pool': self.pool, 'pool_type': 'erasure',
1007                'erasure_code_profile': self.profile}
1008         ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'', timeout=30)
1009         eq(ret, 0, msg=out)
1010         assert self.rados.pool_exists(self.pool)
1011         self.ioctx = self.rados.open_ioctx(self.pool)
1012
1013     def tearDown(self):
1014         cmd = {"prefix": "osd unset", "key": "noup"}
1015         self.rados.mon_command(json.dumps(cmd), b'')
1016         self.ioctx.close()
1017         self.rados.delete_pool(self.pool)
1018         self.rados.shutdown()
1019
1020     def test_alignment(self):
1021         eq(self.ioctx.alignment(), 8192)
1022
1023
1024 class TestIoctx2(object):
1025
1026     def setUp(self):
1027         self.rados = Rados(conffile='')
1028         self.rados.connect()
1029         self.rados.create_pool('test_pool')
1030         assert self.rados.pool_exists('test_pool')
1031         pool_id = self.rados.pool_lookup('test_pool')
1032         assert pool_id > 0
1033         self.ioctx2 = self.rados.open_ioctx2(pool_id)
1034
1035     def tearDown(self):
1036         cmd = {"prefix": "osd unset", "key": "noup"}
1037         self.rados.mon_command(json.dumps(cmd), b'')
1038         self.ioctx2.close()
1039         self.rados.delete_pool('test_pool')
1040         self.rados.shutdown()
1041
1042     def test_get_last_version(self):
1043         version = self.ioctx2.get_last_version()
1044         assert version >= 0
1045
1046     def test_get_stats(self):
1047         stats = self.ioctx2.get_stats()
1048         eq(stats, {'num_objects_unfound': 0,
1049                    'num_objects_missing_on_primary': 0,
1050                    'num_object_clones': 0,
1051                    'num_objects': 0,
1052                    'num_object_copies': 0,
1053                    'num_bytes': 0,
1054                    'num_rd_kb': 0,
1055                    'num_wr_kb': 0,
1056                    'num_kb': 0,
1057                    'num_wr': 0,
1058                    'num_objects_degraded': 0,
1059                    'num_rd': 0})
1060
1061
1062 class TestObject(object):
1063
1064     def setUp(self):
1065         self.rados = Rados(conffile='')
1066         self.rados.connect()
1067         self.rados.create_pool('test_pool')
1068         assert self.rados.pool_exists('test_pool')
1069         self.ioctx = self.rados.open_ioctx('test_pool')
1070         self.ioctx.write('foo', b'bar')
1071         self.object = Object(self.ioctx, 'foo')
1072
1073     def tearDown(self):
1074         self.ioctx.close()
1075         self.ioctx = None
1076         self.rados.delete_pool('test_pool')
1077         self.rados.shutdown()
1078         self.rados = None
1079
1080     def test_read(self):
1081         eq(self.object.read(3), b'bar')
1082         eq(self.object.read(100), b'')
1083
1084     def test_seek(self):
1085         self.object.write(b'blah')
1086         self.object.seek(0)
1087         eq(self.object.read(4), b'blah')
1088         self.object.seek(1)
1089         eq(self.object.read(3), b'lah')
1090
1091     def test_write(self):
1092         self.object.write(b'barbaz')
1093         self.object.seek(0)
1094         eq(self.object.read(3), b'bar')
1095         eq(self.object.read(3), b'baz')
1096
1097 class TestIoCtxSelfManagedSnaps(object):
1098     def setUp(self):
1099         self.rados = Rados(conffile='')
1100         self.rados.connect()
1101         self.rados.create_pool('test_pool')
1102         assert self.rados.pool_exists('test_pool')
1103         self.ioctx = self.rados.open_ioctx('test_pool')
1104
1105     def tearDown(self):
1106         cmd = {"prefix":"osd unset", "key":"noup"}
1107         self.rados.mon_command(json.dumps(cmd), b'')
1108         self.ioctx.close()
1109         self.rados.delete_pool('test_pool')
1110         self.rados.shutdown()
1111
1112     def test(self):
1113         # cannot mix-and-match pool and self-managed snapshot mode
1114         self.ioctx.set_self_managed_snap_write([])
1115         self.ioctx.write('abc', b'abc')
1116         snap_id_1 = self.ioctx.create_self_managed_snap()
1117         self.ioctx.set_self_managed_snap_write([snap_id_1])
1118
1119         self.ioctx.write('abc', b'def')
1120         snap_id_2 = self.ioctx.create_self_managed_snap()
1121         self.ioctx.set_self_managed_snap_write([snap_id_1, snap_id_2])
1122
1123         self.ioctx.write('abc', b'ghi')
1124
1125         self.ioctx.rollback_self_managed_snap('abc', snap_id_1)
1126         eq(self.ioctx.read('abc'), b'abc')
1127
1128         self.ioctx.rollback_self_managed_snap('abc', snap_id_2)
1129         eq(self.ioctx.read('abc'), b'def')
1130
1131         self.ioctx.remove_self_managed_snap(snap_id_1)
1132         self.ioctx.remove_self_managed_snap(snap_id_2)
1133
1134 class TestCommand(object):
1135
1136     def setUp(self):
1137         self.rados = Rados(conffile='')
1138         self.rados.connect()
1139
1140     def tearDown(self):
1141         self.rados.shutdown()
1142
1143     def test_monmap_dump(self):
1144
1145         # check for success and some plain output with epoch in it
1146         cmd = {"prefix":"mon dump"}
1147         ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30)
1148         eq(ret, 0)
1149         assert len(buf) > 0
1150         assert(b'epoch' in buf)
1151
1152         # JSON, and grab current epoch
1153         cmd['format'] = 'json'
1154         ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30)
1155         eq(ret, 0)
1156         assert len(buf) > 0
1157         d = json.loads(buf.decode("utf-8"))
1158         assert('epoch' in d)
1159         epoch = d['epoch']
1160
1161         # assume epoch + 1000 does not exist; test for ENOENT
1162         cmd['epoch'] = epoch + 1000
1163         ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30)
1164         eq(ret, -errno.ENOENT)
1165         eq(len(buf), 0)
1166         del cmd['epoch']
1167
1168         # send to specific target by name, rank
1169         cmd = {"prefix": "version"}
1170
1171         target = d['mons'][0]['name']
1172         print(target)
1173         ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30,
1174                                                 target=target)
1175         eq(ret, 0)
1176         assert len(buf) > 0
1177         e = json.loads(buf.decode("utf-8"))
1178         assert('release' in e)
1179
1180         target = d['mons'][0]['rank']
1181         print(target)
1182         ret, buf, errs = self.rados.mon_command(json.dumps(cmd), b'', timeout=30,
1183                                                 target=target)
1184         eq(ret, 0)
1185         assert len(buf) > 0
1186         e = json.loads(buf.decode("utf-8"))
1187         assert('release' in e)
1188
1189     def test_osd_bench(self):
1190         cmd = dict(prefix='bench', size=4096, count=8192)
1191         ret, buf, err = self.rados.osd_command(0, json.dumps(cmd), b'',
1192                                                timeout=30)
1193         eq(ret, 0)
1194         assert len(buf) > 0
1195         out = json.loads(buf.decode('utf-8'))
1196         eq(out['blocksize'], cmd['size'])
1197         eq(out['bytes_written'], cmd['count'])
1198
1199     def test_ceph_osd_pool_create_utf8(self):
1200         if _python2:
1201             # Use encoded bytestring
1202             poolname = b"\351\273\205"
1203         else:
1204             poolname = "\u9ec5"
1205
1206         cmd = {"prefix": "osd pool create", "pg_num": 16, "pool": poolname}
1207         ret, buf, out = self.rados.mon_command(json.dumps(cmd), b'')
1208         eq(ret, 0)
1209         assert len(out) > 0
1210         eq(u"pool '\u9ec5' created", out)