ceph/qa/tasks/mgr/dashboard/test_rgw.py

   1 # -*- coding: utf-8 -*-
   2 from __future__ import absolute_import
   3
   4 import logging
   5 import urllib
   6 import six
   7
   8 from .helper import DashboardTestCase, JObj, JList, JLeaf
   9
  10 logger = logging.getLogger(__name__)
  11
  12
  13 class RgwTestCase(DashboardTestCase):
  14
  15     maxDiff = None
  16     create_test_user = False
  17
  18     AUTH_ROLES = ['rgw-manager']
  19
  20     @classmethod
  21     def setUpClass(cls):
  22         super(RgwTestCase, cls).setUpClass()
  23         # Create the administrator account.
  24         cls._radosgw_admin_cmd([
  25             'user', 'create', '--uid', 'admin', '--display-name', 'admin',
  26             '--system', '--access-key', 'admin', '--secret', 'admin'
  27         ])
  28         # Update the dashboard configuration.
  29         cls._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', 'admin'])
  30         cls._ceph_cmd(['dashboard', 'set-rgw-api-access-key', 'admin'])
  31         # Create a test user?
  32         if cls.create_test_user:
  33             cls._radosgw_admin_cmd([
  34                 'user', 'create', '--uid', 'teuth-test-user', '--display-name',
  35                 'teuth-test-user'
  36             ])
  37             cls._radosgw_admin_cmd([
  38                 'caps', 'add', '--uid', 'teuth-test-user', '--caps',
  39                 'metadata=write'
  40             ])
  41             cls._radosgw_admin_cmd([
  42                 'subuser', 'create', '--uid', 'teuth-test-user', '--subuser',
  43                 'teuth-test-subuser', '--access', 'full', '--key-type', 's3',
  44                 '--access-key', 'xyz123'
  45             ])
  46             cls._radosgw_admin_cmd([
  47                 'subuser', 'create', '--uid', 'teuth-test-user', '--subuser',
  48                 'teuth-test-subuser2', '--access', 'full', '--key-type',
  49                 'swift'
  50             ])
  51
  52     @classmethod
  53     def tearDownClass(cls):
  54         if cls.create_test_user:
  55             cls._radosgw_admin_cmd(['user', 'rm', '--uid=teuth-test-user'])
  56         super(RgwTestCase, cls).tearDownClass()
  57
  58     def setUp(self):
  59         super(RgwTestCase, self).setUp()
  60
  61     def get_rgw_user(self, uid):
  62         return self._get('/api/rgw/user/{}'.format(uid))
  63
  64
  65 class RgwApiCredentialsTest(RgwTestCase):
  66
  67     AUTH_ROLES = ['rgw-manager']
  68
  69     def setUp(self):
  70         # Restart the Dashboard module to ensure that the connection to the
  71         # RGW Admin Ops API is re-established with the new credentials.
  72         self.logout()
  73         self._ceph_cmd(['mgr', 'module', 'disable', 'dashboard'])
  74         self._ceph_cmd(['mgr', 'module', 'enable', 'dashboard', '--force'])
  75         # Set the default credentials.
  76         self._ceph_cmd(['dashboard', 'set-rgw-api-user-id', ''])
  77         self._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', 'admin'])
  78         self._ceph_cmd(['dashboard', 'set-rgw-api-access-key', 'admin'])
  79         super(RgwApiCredentialsTest, self).setUp()
  80
  81     def test_no_access_secret_key(self):
  82         self._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', ''])
  83         self._ceph_cmd(['dashboard', 'set-rgw-api-access-key', ''])
  84         resp = self._get('/api/rgw/user')
  85         self.assertStatus(500)
  86         self.assertIn('detail', resp)
  87         self.assertIn('component', resp)
  88         self.assertIn('No RGW credentials found', resp['detail'])
  89         self.assertEqual(resp['component'], 'rgw')
  90
  91     def test_success(self):
  92         data = self._get('/api/rgw/status')
  93         self.assertStatus(200)
  94         self.assertIn('available', data)
  95         self.assertIn('message', data)
  96         self.assertTrue(data['available'])
  97
  98     def test_invalid_user_id(self):
  99         self._ceph_cmd(['dashboard', 'set-rgw-api-user-id', 'xyz'])
 100         data = self._get('/api/rgw/status')
 101         self.assertStatus(200)
 102         self.assertIn('available', data)
 103         self.assertIn('message', data)
 104         self.assertFalse(data['available'])
 105         self.assertIn('The user "xyz" is unknown to the Object Gateway.',
 106                       data['message'])
 107
 108
 109 class RgwBucketTest(RgwTestCase):
 110
 111     AUTH_ROLES = ['rgw-manager']
 112
 113     @classmethod
 114     def setUpClass(cls):
 115         cls.create_test_user = True
 116         super(RgwBucketTest, cls).setUpClass()
 117         # Create a tenanted user.
 118         cls._radosgw_admin_cmd([
 119             'user', 'create', '--tenant', 'testx', '--uid', 'teuth-test-user',
 120             '--display-name', 'tenanted teuth-test-user'
 121         ])
 122
 123     @classmethod
 124     def tearDownClass(cls):
 125         cls._radosgw_admin_cmd(
 126             ['user', 'rm', '--tenant', 'testx', '--uid=teuth-test-user'])
 127         super(RgwBucketTest, cls).tearDownClass()
 128
 129     def test_all(self):
 130         # Create a new bucket.
 131         self._post(
 132             '/api/rgw/bucket',
 133             params={
 134                 'bucket': 'teuth-test-bucket',
 135                 'uid': 'admin'
 136             })
 137         self.assertStatus(201)
 138         data = self.jsonBody()
 139         self.assertSchema(data, JObj(sub_elems={
 140             'bucket_info': JObj(sub_elems={
 141                 'bucket': JObj(allow_unknown=True, sub_elems={
 142                     'name': JLeaf(str),
 143                     'bucket_id': JLeaf(str),
 144                     'tenant': JLeaf(str)
 145                 }),
 146                 'quota': JObj(sub_elems={}, allow_unknown=True),
 147                 'creation_time': JLeaf(str)
 148             }, allow_unknown=True)
 149         }, allow_unknown=True))
 150         data = data['bucket_info']['bucket']
 151         self.assertEqual(data['name'], 'teuth-test-bucket')
 152         self.assertEqual(data['tenant'], '')
 153
 154         # List all buckets.
 155         data = self._get('/api/rgw/bucket')
 156         self.assertStatus(200)
 157         self.assertEqual(len(data), 1)
 158         self.assertIn('teuth-test-bucket', data)
 159
 160         # Get the bucket.
 161         data = self._get('/api/rgw/bucket/teuth-test-bucket')
 162         self.assertStatus(200)
 163         self.assertSchema(data, JObj(sub_elems={
 164             'id': JLeaf(str),
 165             'bid': JLeaf(str),
 166             'tenant': JLeaf(str),
 167             'bucket': JLeaf(str),
 168             'bucket_quota': JObj(sub_elems={}, allow_unknown=True),
 169             'owner': JLeaf(str)
 170         }, allow_unknown=True))
 171         self.assertEqual(data['bucket'], 'teuth-test-bucket')
 172         self.assertEqual(data['owner'], 'admin')
 173
 174         # Update the bucket.
 175         self._put(
 176             '/api/rgw/bucket/teuth-test-bucket',
 177             params={
 178                 'bucket_id': data['id'],
 179                 'uid': 'teuth-test-user'
 180             })
 181         self.assertStatus(200)
 182         data = self._get('/api/rgw/bucket/teuth-test-bucket')
 183         self.assertStatus(200)
 184         self.assertSchema(data, JObj(sub_elems={
 185             'owner': JLeaf(str),
 186             'bid': JLeaf(str),
 187             'tenant': JLeaf(str)
 188         }, allow_unknown=True))
 189         self.assertEqual(data['owner'], 'teuth-test-user')
 190
 191         # Delete the bucket.
 192         self._delete('/api/rgw/bucket/teuth-test-bucket')
 193         self.assertStatus(204)
 194         data = self._get('/api/rgw/bucket')
 195         self.assertStatus(200)
 196         self.assertEqual(len(data), 0)
 197
 198     def test_create_get_update_delete_w_tenant(self):
 199         # Create a new bucket. The tenant of the user is used when
 200         # the bucket is created.
 201         self._post(
 202             '/api/rgw/bucket',
 203             params={
 204                 'bucket': 'teuth-test-bucket',
 205                 'uid': 'testx$teuth-test-user'
 206             })
 207         self.assertStatus(201)
 208         # It's not possible to validate the result because there
 209         # IS NO result object returned by the RGW Admin OPS API
 210         # when a tenanted bucket is created.
 211         data = self.jsonBody()
 212         self.assertIsNone(data)
 213
 214         # List all buckets.
 215         data = self._get('/api/rgw/bucket')
 216         self.assertStatus(200)
 217         self.assertEqual(len(data), 1)
 218         self.assertIn('testx/teuth-test-bucket', data)
 219
 220         # Get the bucket.
 221         data = self._get('/api/rgw/bucket/{}'.format(
 222             urllib.quote_plus('testx/teuth-test-bucket')))
 223         self.assertStatus(200)
 224         self.assertSchema(data, JObj(sub_elems={
 225             'owner': JLeaf(str),
 226             'bucket': JLeaf(str),
 227             'tenant': JLeaf(str),
 228             'bid': JLeaf(str)
 229         }, allow_unknown=True))
 230         self.assertEqual(data['owner'], 'testx$teuth-test-user')
 231         self.assertEqual(data['bucket'], 'teuth-test-bucket')
 232         self.assertEqual(data['tenant'], 'testx')
 233         self.assertEqual(data['bid'], 'testx/teuth-test-bucket')
 234
 235         # Update the bucket.
 236         self._put(
 237             '/api/rgw/bucket/{}'.format(
 238                 urllib.quote_plus('testx/teuth-test-bucket')),
 239             params={
 240                 'bucket_id': data['id'],
 241                 'uid': 'admin'
 242             })
 243         self.assertStatus(200)
 244         data = self._get('/api/rgw/bucket/{}'.format(
 245             urllib.quote_plus('testx/teuth-test-bucket')))
 246         self.assertStatus(200)
 247         self.assertIn('owner', data)
 248         self.assertEqual(data['owner'], 'admin')
 249
 250         # Delete the bucket.
 251         self._delete('/api/rgw/bucket/{}'.format(
 252             urllib.quote_plus('testx/teuth-test-bucket')))
 253         self.assertStatus(204)
 254         data = self._get('/api/rgw/bucket')
 255         self.assertStatus(200)
 256         self.assertEqual(len(data), 0)
 257
 258
 259 class RgwDaemonTest(DashboardTestCase):
 260
 261     AUTH_ROLES = ['rgw-manager']
 262
 263     @DashboardTestCase.RunAs('test', 'test', [{
 264         'rgw': ['create', 'update', 'delete']
 265     }])
 266     def test_read_access_permissions(self):
 267         self._get('/api/rgw/daemon')
 268         self.assertStatus(403)
 269         self._get('/api/rgw/daemon/id')
 270         self.assertStatus(403)
 271
 272     def test_list(self):
 273         data = self._get('/api/rgw/daemon')
 274         self.assertStatus(200)
 275         self.assertEqual(len(data), 1)
 276         data = data[0]
 277         self.assertIn('id', data)
 278         self.assertIn('version', data)
 279         self.assertIn('server_hostname', data)
 280
 281     def test_get(self):
 282         data = self._get('/api/rgw/daemon')
 283         self.assertStatus(200)
 284
 285         data = self._get('/api/rgw/daemon/{}'.format(data[0]['id']))
 286         self.assertStatus(200)
 287         self.assertIn('rgw_metadata', data)
 288         self.assertIn('rgw_id', data)
 289         self.assertIn('rgw_status', data)
 290         self.assertTrue(data['rgw_metadata'])
 291
 292     def test_status(self):
 293         self._radosgw_admin_cmd([
 294             'user', 'create', '--uid=admin', '--display-name=admin',
 295             '--system', '--access-key=admin', '--secret=admin'
 296         ])
 297         self._ceph_cmd(['dashboard', 'set-rgw-api-user-id', 'admin'])
 298         self._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', 'admin'])
 299         self._ceph_cmd(['dashboard', 'set-rgw-api-access-key', 'admin'])
 300
 301         data = self._get('/api/rgw/status')
 302         self.assertStatus(200)
 303         self.assertIn('available', data)
 304         self.assertIn('message', data)
 305         self.assertTrue(data['available'])
 306
 307
 308 class RgwUserTest(RgwTestCase):
 309
 310     AUTH_ROLES = ['rgw-manager']
 311
 312     @classmethod
 313     def setUpClass(cls):
 314         super(RgwUserTest, cls).setUpClass()
 315
 316     def _assert_user_data(self, data):
 317         self.assertSchema(data, JObj(sub_elems={
 318             'caps': JList(JObj(sub_elems={}, allow_unknown=True)),
 319             'display_name': JLeaf(str),
 320             'email': JLeaf(str),
 321             'keys': JList(JObj(sub_elems={}, allow_unknown=True)),
 322             'max_buckets': JLeaf(int),
 323             'subusers': JList(JLeaf(str)),
 324             'suspended': JLeaf(int),
 325             'swift_keys': JList(JObj(sub_elems={}, allow_unknown=True)),
 326             'tenant': JLeaf(str),
 327             'user_id': JLeaf(str),
 328             'uid': JLeaf(str)
 329         }, allow_unknown=True))
 330         self.assertGreaterEqual(len(data['keys']), 1)
 331
 332     def test_get(self):
 333         data = self.get_rgw_user('admin')
 334         self.assertStatus(200)
 335         self._assert_user_data(data)
 336         self.assertEqual(data['user_id'], 'admin')
 337
 338     def test_list(self):
 339         data = self._get('/api/rgw/user')
 340         self.assertStatus(200)
 341         self.assertGreaterEqual(len(data), 1)
 342         self.assertIn('admin', data)
 343
 344     def test_create_get_update_delete(self):
 345         # Create a new user.
 346         self._post('/api/rgw/user', params={
 347             'uid': 'teuth-test-user',
 348             'display_name': 'display name'
 349         })
 350         self.assertStatus(201)
 351         data = self.jsonBody()
 352         self._assert_user_data(data)
 353         self.assertEqual(data['user_id'], 'teuth-test-user')
 354         self.assertEqual(data['display_name'], 'display name')
 355
 356         # Get the user.
 357         data = self.get_rgw_user('teuth-test-user')
 358         self.assertStatus(200)
 359         self._assert_user_data(data)
 360         self.assertEqual(data['tenant'], '')
 361         self.assertEqual(data['user_id'], 'teuth-test-user')
 362         self.assertEqual(data['uid'], 'teuth-test-user')
 363
 364         # Update the user.
 365         self._put(
 366             '/api/rgw/user/teuth-test-user',
 367             params={'display_name': 'new name'})
 368         self.assertStatus(200)
 369         data = self.jsonBody()
 370         self._assert_user_data(data)
 371         self.assertEqual(data['display_name'], 'new name')
 372
 373         # Delete the user.
 374         self._delete('/api/rgw/user/teuth-test-user')
 375         self.assertStatus(204)
 376         self.get_rgw_user('teuth-test-user')
 377         self.assertStatus(500)
 378         resp = self.jsonBody()
 379         self.assertIn('detail', resp)
 380         self.assertIn('failed request with status code 404', resp['detail'])
 381         self.assertIn('"Code":"NoSuchUser"', resp['detail'])
 382         self.assertIn('"HostId"', resp['detail'])
 383         self.assertIn('"RequestId"', resp['detail'])
 384
 385     def test_create_get_update_delete_w_tenant(self):
 386         # Create a new user.
 387         self._post(
 388             '/api/rgw/user',
 389             params={
 390                 'uid': 'test01$teuth-test-user',
 391                 'display_name': 'display name'
 392             })
 393         self.assertStatus(201)
 394         data = self.jsonBody()
 395         self._assert_user_data(data)
 396         self.assertEqual(data['user_id'], 'teuth-test-user')
 397         self.assertEqual(data['display_name'], 'display name')
 398
 399         # Get the user.
 400         data = self.get_rgw_user('test01$teuth-test-user')
 401         self.assertStatus(200)
 402         self._assert_user_data(data)
 403         self.assertEqual(data['tenant'], 'test01')
 404         self.assertEqual(data['user_id'], 'teuth-test-user')
 405         self.assertEqual(data['uid'], 'test01$teuth-test-user')
 406
 407         # Update the user.
 408         self._put(
 409             '/api/rgw/user/test01$teuth-test-user',
 410             params={'display_name': 'new name'})
 411         self.assertStatus(200)
 412         data = self.jsonBody()
 413         self._assert_user_data(data)
 414         self.assertEqual(data['display_name'], 'new name')
 415
 416         # Delete the user.
 417         self._delete('/api/rgw/user/test01$teuth-test-user')
 418         self.assertStatus(204)
 419         self.get_rgw_user('test01$teuth-test-user')
 420         self.assertStatus(500)
 421         resp = self.jsonBody()
 422         self.assertIn('detail', resp)
 423         self.assertIn('failed request with status code 404', resp['detail'])
 424         self.assertIn('"Code":"NoSuchUser"', resp['detail'])
 425         self.assertIn('"HostId"', resp['detail'])
 426         self.assertIn('"RequestId"', resp['detail'])
 427
 428
 429 class RgwUserCapabilityTest(RgwTestCase):
 430
 431     AUTH_ROLES = ['rgw-manager']
 432
 433     @classmethod
 434     def setUpClass(cls):
 435         cls.create_test_user = True
 436         super(RgwUserCapabilityTest, cls).setUpClass()
 437
 438     def test_set(self):
 439         self._post(
 440             '/api/rgw/user/teuth-test-user/capability',
 441             params={
 442                 'type': 'usage',
 443                 'perm': 'read'
 444             })
 445         self.assertStatus(201)
 446         data = self.jsonBody()
 447         self.assertEqual(len(data), 1)
 448         data = data[0]
 449         self.assertEqual(data['type'], 'usage')
 450         self.assertEqual(data['perm'], 'read')
 451
 452         # Get the user data to validate the capabilities.
 453         data = self.get_rgw_user('teuth-test-user')
 454         self.assertStatus(200)
 455         self.assertGreaterEqual(len(data['caps']), 1)
 456         self.assertEqual(data['caps'][0]['type'], 'usage')
 457         self.assertEqual(data['caps'][0]['perm'], 'read')
 458
 459     def test_delete(self):
 460         self._delete(
 461             '/api/rgw/user/teuth-test-user/capability',
 462             params={
 463                 'type': 'metadata',
 464                 'perm': 'write'
 465             })
 466         self.assertStatus(204)
 467
 468         # Get the user data to validate the capabilities.
 469         data = self.get_rgw_user('teuth-test-user')
 470         self.assertStatus(200)
 471         self.assertEqual(len(data['caps']), 0)
 472
 473
 474 class RgwUserKeyTest(RgwTestCase):
 475
 476     AUTH_ROLES = ['rgw-manager']
 477
 478     @classmethod
 479     def setUpClass(cls):
 480         cls.create_test_user = True
 481         super(RgwUserKeyTest, cls).setUpClass()
 482
 483     def test_create_s3(self):
 484         self._post(
 485             '/api/rgw/user/teuth-test-user/key',
 486             params={
 487                 'key_type': 's3',
 488                 'generate_key': 'false',
 489                 'access_key': 'abc987',
 490                 'secret_key': 'aaabbbccc'
 491             })
 492         data = self.jsonBody()
 493         self.assertStatus(201)
 494         self.assertGreaterEqual(len(data), 3)
 495         key = self.find_object_in_list('access_key', 'abc987', data)
 496         self.assertIsInstance(key, object)
 497         self.assertEqual(key['secret_key'], 'aaabbbccc')
 498
 499     def test_create_swift(self):
 500         self._post(
 501             '/api/rgw/user/teuth-test-user/key',
 502             params={
 503                 'key_type': 'swift',
 504                 'subuser': 'teuth-test-subuser',
 505                 'generate_key': 'false',
 506                 'secret_key': 'xxxyyyzzz'
 507             })
 508         data = self.jsonBody()
 509         self.assertStatus(201)
 510         self.assertGreaterEqual(len(data), 2)
 511         key = self.find_object_in_list('secret_key', 'xxxyyyzzz', data)
 512         self.assertIsInstance(key, object)
 513
 514     def test_delete_s3(self):
 515         self._delete(
 516             '/api/rgw/user/teuth-test-user/key',
 517             params={
 518                 'key_type': 's3',
 519                 'access_key': 'xyz123'
 520             })
 521         self.assertStatus(204)
 522
 523     def test_delete_swift(self):
 524         self._delete(
 525             '/api/rgw/user/teuth-test-user/key',
 526             params={
 527                 'key_type': 'swift',
 528                 'subuser': 'teuth-test-user:teuth-test-subuser2'
 529             })
 530         self.assertStatus(204)
 531
 532
 533 class RgwUserQuotaTest(RgwTestCase):
 534
 535     AUTH_ROLES = ['rgw-manager']
 536
 537     @classmethod
 538     def setUpClass(cls):
 539         cls.create_test_user = True
 540         super(RgwUserQuotaTest, cls).setUpClass()
 541
 542     def _assert_quota(self, data):
 543         self.assertIn('user_quota', data)
 544         self.assertIn('max_objects', data['user_quota'])
 545         self.assertIn('enabled', data['user_quota'])
 546         self.assertIn('max_size_kb', data['user_quota'])
 547         self.assertIn('max_size', data['user_quota'])
 548         self.assertIn('bucket_quota', data)
 549         self.assertIn('max_objects', data['bucket_quota'])
 550         self.assertIn('enabled', data['bucket_quota'])
 551         self.assertIn('max_size_kb', data['bucket_quota'])
 552         self.assertIn('max_size', data['bucket_quota'])
 553
 554     def test_get_quota(self):
 555         data = self._get('/api/rgw/user/teuth-test-user/quota')
 556         self.assertStatus(200)
 557         self._assert_quota(data)
 558
 559     def test_set_user_quota(self):
 560         self._put(
 561             '/api/rgw/user/teuth-test-user/quota',
 562             params={
 563                 'quota_type': 'user',
 564                 'enabled': 'true',
 565                 'max_size_kb': 2048,
 566                 'max_objects': 101
 567             })
 568         self.assertStatus(200)
 569
 570         data = self._get('/api/rgw/user/teuth-test-user/quota')
 571         self.assertStatus(200)
 572         self._assert_quota(data)
 573         self.assertEqual(data['user_quota']['max_objects'], 101)
 574         self.assertTrue(data['user_quota']['enabled'])
 575         self.assertEqual(data['user_quota']['max_size_kb'], 2048)
 576
 577     def test_set_bucket_quota(self):
 578         self._put(
 579             '/api/rgw/user/teuth-test-user/quota',
 580             params={
 581                 'quota_type': 'bucket',
 582                 'enabled': 'false',
 583                 'max_size_kb': 4096,
 584                 'max_objects': 2000
 585             })
 586         self.assertStatus(200)
 587
 588         data = self._get('/api/rgw/user/teuth-test-user/quota')
 589         self.assertStatus(200)
 590         self._assert_quota(data)
 591         self.assertEqual(data['bucket_quota']['max_objects'], 2000)
 592         self.assertFalse(data['bucket_quota']['enabled'])
 593         self.assertEqual(data['bucket_quota']['max_size_kb'], 4096)
 594
 595
 596 class RgwUserSubuserTest(RgwTestCase):
 597
 598     AUTH_ROLES = ['rgw-manager']
 599
 600     @classmethod
 601     def setUpClass(cls):
 602         cls.create_test_user = True
 603         super(RgwUserSubuserTest, cls).setUpClass()
 604
 605     def test_create_swift(self):
 606         self._post(
 607             '/api/rgw/user/teuth-test-user/subuser',
 608             params={
 609                 'subuser': 'tux',
 610                 'access': 'readwrite',
 611                 'key_type': 'swift'
 612             })
 613         self.assertStatus(201)
 614         data = self.jsonBody()
 615         subuser = self.find_object_in_list('id', 'teuth-test-user:tux', data)
 616         self.assertIsInstance(subuser, object)
 617         self.assertEqual(subuser['permissions'], 'read-write')
 618
 619         # Get the user data to validate the keys.
 620         data = self.get_rgw_user('teuth-test-user')
 621         self.assertStatus(200)
 622         key = self.find_object_in_list('user', 'teuth-test-user:tux',
 623                                        data['swift_keys'])
 624         self.assertIsInstance(key, object)
 625
 626     def test_create_s3(self):
 627         self._post(
 628             '/api/rgw/user/teuth-test-user/subuser',
 629             params={
 630                 'subuser': 'hugo',
 631                 'access': 'write',
 632                 'generate_secret': 'false',
 633                 'access_key': 'yyy',
 634                 'secret_key': 'xxx'
 635             })
 636         self.assertStatus(201)
 637         data = self.jsonBody()
 638         subuser = self.find_object_in_list('id', 'teuth-test-user:hugo', data)
 639         self.assertIsInstance(subuser, object)
 640         self.assertEqual(subuser['permissions'], 'write')
 641
 642         # Get the user data to validate the keys.
 643         data = self.get_rgw_user('teuth-test-user')
 644         self.assertStatus(200)
 645         key = self.find_object_in_list('user', 'teuth-test-user:hugo',
 646                                        data['keys'])
 647         self.assertIsInstance(key, object)
 648         self.assertEqual(key['secret_key'], 'xxx')
 649
 650     def test_delete_w_purge(self):
 651         self._delete(
 652             '/api/rgw/user/teuth-test-user/subuser/teuth-test-subuser2')
 653         self.assertStatus(204)
 654
 655         # Get the user data to check that the keys don't exist anymore.
 656         data = self.get_rgw_user('teuth-test-user')
 657         self.assertStatus(200)
 658         key = self.find_object_in_list(
 659             'user', 'teuth-test-user:teuth-test-subuser2', data['swift_keys'])
 660         self.assertIsNone(key)
 661
 662     def test_delete_wo_purge(self):
 663         self._delete(
 664             '/api/rgw/user/teuth-test-user/subuser/teuth-test-subuser',
 665             params={'purge_keys': 'false'})
 666         self.assertStatus(204)
 667
 668         # Get the user data to check whether they keys still exist.
 669         data = self.get_rgw_user('teuth-test-user')
 670         self.assertStatus(200)
 671         key = self.find_object_in_list(
 672             'user', 'teuth-test-user:teuth-test-subuser', data['keys'])
 673         self.assertIsInstance(key, object)