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

   1 # -*- coding: utf-8 -*-
   2 from __future__ import absolute_import
   3
   4 import base64
   5 import logging
   6 import time
   7
   8 from cryptography.hazmat.backends import default_backend
   9 from cryptography.hazmat.primitives.twofactor.totp import TOTP
  10 from cryptography.hazmat.primitives.hashes import SHA1
  11 from six.moves.urllib import parse
  12
  13 from .helper import DashboardTestCase, JObj, JList, JLeaf
  14
  15 logger = logging.getLogger(__name__)
  16
  17
  18 class RgwTestCase(DashboardTestCase):
  19
  20     maxDiff = None
  21     create_test_user = False
  22
  23     AUTH_ROLES = ['rgw-manager']
  24
  25     @classmethod
  26     def setUpClass(cls):
  27         super(RgwTestCase, cls).setUpClass()
  28         # Create the administrator account.
  29         cls._radosgw_admin_cmd([
  30             'user', 'create', '--uid', 'admin', '--display-name', 'admin',
  31             '--system', '--access-key', 'admin', '--secret', 'admin'
  32         ])
  33         # Update the dashboard configuration.
  34         cls._ceph_cmd(['dashboard', 'set-rgw-api-user-id', 'admin'])
  35         cls._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', 'admin'])
  36         cls._ceph_cmd(['dashboard', 'set-rgw-api-access-key', 'admin'])
  37         # Create a test user?
  38         if cls.create_test_user:
  39             cls._radosgw_admin_cmd([
  40                 'user', 'create', '--uid', 'teuth-test-user', '--display-name',
  41                 'teuth-test-user'
  42             ])
  43             cls._radosgw_admin_cmd([
  44                 'caps', 'add', '--uid', 'teuth-test-user', '--caps',
  45                 'metadata=write'
  46             ])
  47             cls._radosgw_admin_cmd([
  48                 'subuser', 'create', '--uid', 'teuth-test-user', '--subuser',
  49                 'teuth-test-subuser', '--access', 'full', '--key-type', 's3',
  50                 '--access-key', 'xyz123'
  51             ])
  52             cls._radosgw_admin_cmd([
  53                 'subuser', 'create', '--uid', 'teuth-test-user', '--subuser',
  54                 'teuth-test-subuser2', '--access', 'full', '--key-type',
  55                 'swift'
  56             ])
  57
  58     @classmethod
  59     def tearDownClass(cls):
  60         # Delete administrator account.
  61         cls._radosgw_admin_cmd(['user', 'rm', '--uid', 'admin'])
  62         if cls.create_test_user:
  63             cls._radosgw_admin_cmd(['user', 'rm', '--uid=teuth-test-user', '--purge-data'])
  64         super(RgwTestCase, cls).tearDownClass()
  65
  66     def get_rgw_user(self, uid):
  67         return self._get('/api/rgw/user/{}'.format(uid))
  68
  69
  70 class RgwApiCredentialsTest(RgwTestCase):
  71
  72     AUTH_ROLES = ['rgw-manager']
  73
  74     def setUp(self):
  75         super(RgwApiCredentialsTest, self).setUp()
  76         # Restart the Dashboard module to ensure that the connection to the
  77         # RGW Admin Ops API is re-established with the new credentials.
  78         self.logout()
  79         self._ceph_cmd(['mgr', 'module', 'disable', 'dashboard'])
  80         self._ceph_cmd(['mgr', 'module', 'enable', 'dashboard', '--force'])
  81         # Set the default credentials.
  82         self._ceph_cmd(['dashboard', 'set-rgw-api-user-id', ''])
  83         self._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', 'admin'])
  84         self._ceph_cmd(['dashboard', 'set-rgw-api-access-key', 'admin'])
  85         super(RgwApiCredentialsTest, self).setUp()
  86
  87     def test_no_access_secret_key(self):
  88         self._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', ''])
  89         self._ceph_cmd(['dashboard', 'set-rgw-api-access-key', ''])
  90         resp = self._get('/api/rgw/user')
  91         self.assertStatus(500)
  92         self.assertIn('detail', resp)
  93         self.assertIn('component', resp)
  94         self.assertIn('No RGW credentials found', resp['detail'])
  95         self.assertEqual(resp['component'], 'rgw')
  96
  97     def test_success(self):
  98         data = self._get('/api/rgw/status')
  99         self.assertStatus(200)
 100         self.assertIn('available', data)
 101         self.assertIn('message', data)
 102         self.assertTrue(data['available'])
 103
 104     def test_invalid_user_id(self):
 105         self._ceph_cmd(['dashboard', 'set-rgw-api-user-id', 'xyz'])
 106         data = self._get('/api/rgw/status')
 107         self.assertStatus(200)
 108         self.assertIn('available', data)
 109         self.assertIn('message', data)
 110         self.assertFalse(data['available'])
 111         self.assertIn('The user "xyz" is unknown to the Object Gateway.',
 112                       data['message'])
 113
 114
 115 class RgwSiteTest(RgwTestCase):
 116
 117     AUTH_ROLES = ['rgw-manager']
 118
 119     def test_get_placement_targets(self):
 120         data = self._get('/api/rgw/site?query=placement-targets')
 121         self.assertStatus(200)
 122         self.assertSchema(data, JObj({
 123             'zonegroup': str,
 124             'placement_targets': JList(JObj({
 125                 'name': str,
 126                 'data_pool': str
 127             }))
 128         }))
 129
 130     def test_get_realms(self):
 131         data = self._get('/api/rgw/site?query=realms')
 132         self.assertStatus(200)
 133         self.assertSchema(data, JList(str))
 134
 135
 136 class RgwBucketTest(RgwTestCase):
 137
 138     _mfa_token_serial = '1'
 139     _mfa_token_seed = '23456723'
 140     _mfa_token_time_step = 2
 141
 142     AUTH_ROLES = ['rgw-manager']
 143
 144     @classmethod
 145     def setUpClass(cls):
 146         cls.create_test_user = True
 147         super(RgwBucketTest, cls).setUpClass()
 148         # Create MFA TOTP token for test user.
 149         cls._radosgw_admin_cmd([
 150             'mfa', 'create', '--uid', 'teuth-test-user', '--totp-serial', cls._mfa_token_serial,
 151             '--totp-seed', cls._mfa_token_seed, '--totp-seed-type', 'base32',
 152             '--totp-seconds', str(cls._mfa_token_time_step), '--totp-window', '1'
 153         ])
 154         # Create tenanted users.
 155         cls._radosgw_admin_cmd([
 156             'user', 'create', '--tenant', 'testx', '--uid', 'teuth-test-user',
 157             '--display-name', 'tenanted teuth-test-user'
 158         ])
 159         cls._radosgw_admin_cmd([
 160             'user', 'create', '--tenant', 'testx2', '--uid', 'teuth-test-user2',
 161             '--display-name', 'tenanted teuth-test-user 2'
 162         ])
 163
 164     @classmethod
 165     def tearDownClass(cls):
 166         cls._radosgw_admin_cmd(
 167             ['user', 'rm', '--tenant', 'testx', '--uid=teuth-test-user', '--purge-data'])
 168         cls._radosgw_admin_cmd(
 169             ['user', 'rm', '--tenant', 'testx2', '--uid=teuth-test-user2', '--purge-data'])
 170         super(RgwBucketTest, cls).tearDownClass()
 171
 172     def _get_mfa_token_pin(self):
 173         totp_key = base64.b32decode(self._mfa_token_seed)
 174         totp = TOTP(totp_key, 6, SHA1(), self._mfa_token_time_step, backend=default_backend(),
 175                     enforce_key_length=False)
 176         time_value = int(time.time())
 177         return totp.generate(time_value)
 178
 179     def test_all(self):
 180         # Create a new bucket.
 181         self._post(
 182             '/api/rgw/bucket',
 183             params={
 184                 'bucket': 'teuth-test-bucket',
 185                 'uid': 'admin',
 186                 'zonegroup': 'default',
 187                 'placement_target': 'default-placement'
 188             })
 189         self.assertStatus(201)
 190         data = self.jsonBody()
 191         self.assertSchema(data, JObj(sub_elems={
 192             'bucket_info': JObj(sub_elems={
 193                 'bucket': JObj(allow_unknown=True, sub_elems={
 194                     'name': JLeaf(str),
 195                     'bucket_id': JLeaf(str),
 196                     'tenant': JLeaf(str)
 197                 }),
 198                 'quota': JObj(sub_elems={}, allow_unknown=True),
 199                 'creation_time': JLeaf(str)
 200             }, allow_unknown=True)
 201         }, allow_unknown=True))
 202         data = data['bucket_info']['bucket']
 203         self.assertEqual(data['name'], 'teuth-test-bucket')
 204         self.assertEqual(data['tenant'], '')
 205
 206         # List all buckets.
 207         data = self._get('/api/rgw/bucket')
 208         self.assertStatus(200)
 209         self.assertEqual(len(data), 1)
 210         self.assertIn('teuth-test-bucket', data)
 211
 212         # List all buckets with stats.
 213         data = self._get('/api/rgw/bucket?stats=true')
 214         self.assertStatus(200)
 215         self.assertEqual(len(data), 1)
 216         self.assertSchema(data[0], JObj(sub_elems={
 217             'bid': JLeaf(str),
 218             'bucket': JLeaf(str),
 219             'bucket_quota': JObj(sub_elems={}, allow_unknown=True),
 220             'id': JLeaf(str),
 221             'owner': JLeaf(str),
 222             'usage': JObj(sub_elems={}, allow_unknown=True),
 223             'tenant': JLeaf(str),
 224         }, allow_unknown=True))
 225
 226         # Get the bucket.
 227         data = self._get('/api/rgw/bucket/teuth-test-bucket')
 228         self.assertStatus(200)
 229         self.assertSchema(data, JObj(sub_elems={
 230             'id': JLeaf(str),
 231             'bid': JLeaf(str),
 232             'tenant': JLeaf(str),
 233             'bucket': JLeaf(str),
 234             'bucket_quota': JObj(sub_elems={}, allow_unknown=True),
 235             'owner': JLeaf(str),
 236             'mfa_delete': JLeaf(str),
 237             'usage': JObj(sub_elems={}, allow_unknown=True),
 238             'versioning': JLeaf(str)
 239         }, allow_unknown=True))
 240         self.assertEqual(data['bucket'], 'teuth-test-bucket')
 241         self.assertEqual(data['owner'], 'admin')
 242         self.assertEqual(data['placement_rule'], 'default-placement')
 243         self.assertEqual(data['versioning'], 'Suspended')
 244
 245         # Update bucket: change owner, enable versioning.
 246         self._put(
 247             '/api/rgw/bucket/teuth-test-bucket',
 248             params={
 249                 'bucket_id': data['id'],
 250                 'uid': 'teuth-test-user',
 251                 'versioning_state': 'Enabled'
 252             })
 253         self.assertStatus(200)
 254         data = self._get('/api/rgw/bucket/teuth-test-bucket')
 255         self.assertStatus(200)
 256         self.assertSchema(data, JObj(sub_elems={
 257             'owner': JLeaf(str),
 258             'bid': JLeaf(str),
 259             'tenant': JLeaf(str)
 260         }, allow_unknown=True))
 261         self.assertEqual(data['owner'], 'teuth-test-user')
 262         self.assertEqual(data['versioning'], 'Enabled')
 263
 264         # Update bucket: enable MFA Delete.
 265         self._put(
 266             '/api/rgw/bucket/teuth-test-bucket',
 267             params={
 268                 'bucket_id': data['id'],
 269                 'uid': 'teuth-test-user',
 270                 'versioning_state': 'Enabled',
 271                 'mfa_delete': 'Enabled',
 272                 'mfa_token_serial': self._mfa_token_serial,
 273                 'mfa_token_pin': self._get_mfa_token_pin()
 274             })
 275         self.assertStatus(200)
 276         data = self._get('/api/rgw/bucket/teuth-test-bucket')
 277         self.assertStatus(200)
 278         self.assertEqual(data['versioning'], 'Enabled')
 279         self.assertEqual(data['mfa_delete'], 'Enabled')
 280
 281         # Update bucket: disable versioning & MFA Delete.
 282         time.sleep(self._mfa_token_time_step * 3)  # Required to get new TOTP pin.
 283         self._put(
 284             '/api/rgw/bucket/teuth-test-bucket',
 285             params={
 286                 'bucket_id': data['id'],
 287                 'uid': 'teuth-test-user',
 288                 'versioning_state': 'Suspended',
 289                 'mfa_delete': 'Disabled',
 290                 'mfa_token_serial': self._mfa_token_serial,
 291                 'mfa_token_pin': self._get_mfa_token_pin()
 292             })
 293         self.assertStatus(200)
 294         data = self._get('/api/rgw/bucket/teuth-test-bucket')
 295         self.assertStatus(200)
 296         self.assertEqual(data['versioning'], 'Suspended')
 297         self.assertEqual(data['mfa_delete'], 'Disabled')
 298
 299         # Delete the bucket.
 300         self._delete('/api/rgw/bucket/teuth-test-bucket')
 301         self.assertStatus(204)
 302         data = self._get('/api/rgw/bucket')
 303         self.assertStatus(200)
 304         self.assertEqual(len(data), 0)
 305
 306     def test_crud_w_tenant(self):
 307         # Create a new bucket. The tenant of the user is used when
 308         # the bucket is created.
 309         self._post(
 310             '/api/rgw/bucket',
 311             params={
 312                 'bucket': 'teuth-test-bucket',
 313                 'uid': 'testx$teuth-test-user',
 314                 'zonegroup': 'default',
 315                 'placement_target': 'default-placement'
 316             })
 317         self.assertStatus(201)
 318         # It's not possible to validate the result because there
 319         # IS NO result object returned by the RGW Admin OPS API
 320         # when a tenanted bucket is created.
 321         data = self.jsonBody()
 322         self.assertIsNone(data)
 323
 324         # List all buckets.
 325         data = self._get('/api/rgw/bucket')
 326         self.assertStatus(200)
 327         self.assertEqual(len(data), 1)
 328         self.assertIn('testx/teuth-test-bucket', data)
 329
 330         def _verify_tenant_bucket(bucket, tenant, uid):
 331             full_bucket_name = '{}/{}'.format(tenant, bucket)
 332             _data = self._get('/api/rgw/bucket/{}'.format(
 333                 parse.quote_plus(full_bucket_name)))
 334             self.assertStatus(200)
 335             self.assertSchema(_data, JObj(sub_elems={
 336                 'owner': JLeaf(str),
 337                 'bucket': JLeaf(str),
 338                 'tenant': JLeaf(str),
 339                 'bid': JLeaf(str)
 340             }, allow_unknown=True))
 341             self.assertEqual(_data['owner'], '{}${}'.format(tenant, uid))
 342             self.assertEqual(_data['bucket'], bucket)
 343             self.assertEqual(_data['tenant'], tenant)
 344             self.assertEqual(_data['bid'], full_bucket_name)
 345             return _data
 346
 347         # Get the bucket.
 348         data = _verify_tenant_bucket('teuth-test-bucket', 'testx', 'teuth-test-user')
 349         self.assertEqual(data['placement_rule'], 'default-placement')
 350         self.assertEqual(data['versioning'], 'Suspended')
 351
 352         # Update bucket: different user with different tenant, enable versioning.
 353         self._put(
 354             '/api/rgw/bucket/{}'.format(
 355                 parse.quote_plus('testx/teuth-test-bucket')),
 356             params={
 357                 'bucket_id': data['id'],
 358                 'uid': 'testx2$teuth-test-user2',
 359                 'versioning_state': 'Enabled'
 360             })
 361         data = _verify_tenant_bucket('teuth-test-bucket', 'testx2', 'teuth-test-user2')
 362         self.assertEqual(data['versioning'], 'Enabled')
 363
 364         # Change owner to a non-tenanted user
 365         self._put(
 366             '/api/rgw/bucket/{}'.format(
 367                 parse.quote_plus('testx2/teuth-test-bucket')),
 368             params={
 369                 'bucket_id': data['id'],
 370                 'uid': 'admin'
 371             })
 372         self.assertStatus(200)
 373         data = self._get('/api/rgw/bucket/teuth-test-bucket')
 374         self.assertStatus(200)
 375         self.assertIn('owner', data)
 376         self.assertEqual(data['owner'], 'admin')
 377         self.assertEqual(data['tenant'], '')
 378         self.assertEqual(data['bucket'], 'teuth-test-bucket')
 379         self.assertEqual(data['bid'], 'teuth-test-bucket')
 380         self.assertEqual(data['versioning'], 'Enabled')
 381
 382         # Change owner back to tenanted user, suspend versioning.
 383         self._put(
 384             '/api/rgw/bucket/teuth-test-bucket',
 385             params={
 386                 'bucket_id': data['id'],
 387                 'uid': 'testx$teuth-test-user',
 388                 'versioning_state': 'Suspended'
 389             })
 390         self.assertStatus(200)
 391         data = _verify_tenant_bucket('teuth-test-bucket', 'testx', 'teuth-test-user')
 392         self.assertEqual(data['versioning'], 'Suspended')
 393
 394         # Delete the bucket.
 395         self._delete('/api/rgw/bucket/{}'.format(
 396             parse.quote_plus('testx/teuth-test-bucket')))
 397         self.assertStatus(204)
 398         data = self._get('/api/rgw/bucket')
 399         self.assertStatus(200)
 400         self.assertEqual(len(data), 0)
 401
 402     def test_crud_w_locking(self):
 403         # Create
 404         self._post('/api/rgw/bucket',
 405                    params={
 406                        'bucket': 'teuth-test-bucket',
 407                        'uid': 'teuth-test-user',
 408                        'zonegroup': 'default',
 409                        'placement_target': 'default-placement',
 410                        'lock_enabled': 'true',
 411                        'lock_mode': 'GOVERNANCE',
 412                        'lock_retention_period_days': '0',
 413                        'lock_retention_period_years': '1'
 414                    })
 415         self.assertStatus(201)
 416         # Read
 417         data = self._get('/api/rgw/bucket/teuth-test-bucket')
 418         self.assertStatus(200)
 419         self.assertSchema(
 420             data,
 421             JObj(sub_elems={
 422                 'lock_enabled': JLeaf(bool),
 423                 'lock_mode': JLeaf(str),
 424                 'lock_retention_period_days': JLeaf(int),
 425                 'lock_retention_period_years': JLeaf(int)
 426             },
 427                  allow_unknown=True))
 428         self.assertTrue(data['lock_enabled'])
 429         self.assertEqual(data['lock_mode'], 'GOVERNANCE')
 430         self.assertEqual(data['lock_retention_period_days'], 0)
 431         self.assertEqual(data['lock_retention_period_years'], 1)
 432         # Update
 433         self._put('/api/rgw/bucket/teuth-test-bucket',
 434                   params={
 435                       'bucket_id': data['id'],
 436                       'uid': 'teuth-test-user',
 437                       'lock_mode': 'COMPLIANCE',
 438                       'lock_retention_period_days': '15',
 439                       'lock_retention_period_years': '0'
 440                   })
 441         self.assertStatus(200)
 442         data = self._get('/api/rgw/bucket/teuth-test-bucket')
 443         self.assertTrue(data['lock_enabled'])
 444         self.assertEqual(data['lock_mode'], 'COMPLIANCE')
 445         self.assertEqual(data['lock_retention_period_days'], 15)
 446         self.assertEqual(data['lock_retention_period_years'], 0)
 447         self.assertStatus(200)
 448         # Delete
 449         self._delete('/api/rgw/bucket/teuth-test-bucket')
 450         self.assertStatus(204)
 451
 452
 453 class RgwDaemonTest(RgwTestCase):
 454
 455     AUTH_ROLES = ['rgw-manager']
 456
 457     @DashboardTestCase.RunAs('test', 'test', [{
 458         'rgw': ['create', 'update', 'delete']
 459     }])
 460     def test_read_access_permissions(self):
 461         self._get('/api/rgw/daemon')
 462         self.assertStatus(403)
 463         self._get('/api/rgw/daemon/id')
 464         self.assertStatus(403)
 465
 466     def test_list(self):
 467         data = self._get('/api/rgw/daemon')
 468         self.assertStatus(200)
 469         self.assertEqual(len(data), 1)
 470         data = data[0]
 471         self.assertIn('id', data)
 472         self.assertIn('version', data)
 473         self.assertIn('server_hostname', data)
 474
 475     def test_get(self):
 476         data = self._get('/api/rgw/daemon')
 477         self.assertStatus(200)
 478
 479         data = self._get('/api/rgw/daemon/{}'.format(data[0]['id']))
 480         self.assertStatus(200)
 481         self.assertIn('rgw_metadata', data)
 482         self.assertIn('rgw_id', data)
 483         self.assertIn('rgw_status', data)
 484         self.assertTrue(data['rgw_metadata'])
 485
 486     def test_status(self):
 487         data = self._get('/api/rgw/status')
 488         self.assertStatus(200)
 489         self.assertIn('available', data)
 490         self.assertIn('message', data)
 491         self.assertTrue(data['available'])
 492
 493
 494 class RgwUserTest(RgwTestCase):
 495
 496     AUTH_ROLES = ['rgw-manager']
 497
 498     @classmethod
 499     def setUpClass(cls):
 500         super(RgwUserTest, cls).setUpClass()
 501
 502     def _assert_user_data(self, data):
 503         self.assertSchema(data, JObj(sub_elems={
 504             'caps': JList(JObj(sub_elems={}, allow_unknown=True)),
 505             'display_name': JLeaf(str),
 506             'email': JLeaf(str),
 507             'keys': JList(JObj(sub_elems={}, allow_unknown=True)),
 508             'max_buckets': JLeaf(int),
 509             'subusers': JList(JLeaf(str)),
 510             'suspended': JLeaf(int),
 511             'swift_keys': JList(JObj(sub_elems={}, allow_unknown=True)),
 512             'tenant': JLeaf(str),
 513             'user_id': JLeaf(str),
 514             'uid': JLeaf(str)
 515         }, allow_unknown=True))
 516         self.assertGreaterEqual(len(data['keys']), 1)
 517
 518     def test_get(self):
 519         data = self.get_rgw_user('admin')
 520         self.assertStatus(200)
 521         self._assert_user_data(data)
 522         self.assertEqual(data['user_id'], 'admin')
 523
 524     def test_list(self):
 525         data = self._get('/api/rgw/user')
 526         self.assertStatus(200)
 527         self.assertGreaterEqual(len(data), 1)
 528         self.assertIn('admin', data)
 529
 530     def test_get_emails(self):
 531         data = self._get('/api/rgw/user/get_emails')
 532         self.assertStatus(200)
 533         self.assertSchema(data, JList(str))
 534
 535     def test_create_get_update_delete(self):
 536         # Create a new user.
 537         self._post('/api/rgw/user', params={
 538             'uid': 'teuth-test-user',
 539             'display_name': 'display name'
 540         })
 541         self.assertStatus(201)
 542         data = self.jsonBody()
 543         self._assert_user_data(data)
 544         self.assertEqual(data['user_id'], 'teuth-test-user')
 545         self.assertEqual(data['display_name'], 'display name')
 546
 547         # Get the user.
 548         data = self.get_rgw_user('teuth-test-user')
 549         self.assertStatus(200)
 550         self._assert_user_data(data)
 551         self.assertEqual(data['tenant'], '')
 552         self.assertEqual(data['user_id'], 'teuth-test-user')
 553         self.assertEqual(data['uid'], 'teuth-test-user')
 554
 555         # Update the user.
 556         self._put(
 557             '/api/rgw/user/teuth-test-user',
 558             params={'display_name': 'new name'})
 559         self.assertStatus(200)
 560         data = self.jsonBody()
 561         self._assert_user_data(data)
 562         self.assertEqual(data['display_name'], 'new name')
 563
 564         # Delete the user.
 565         self._delete('/api/rgw/user/teuth-test-user')
 566         self.assertStatus(204)
 567         self.get_rgw_user('teuth-test-user')
 568         self.assertStatus(500)
 569         resp = self.jsonBody()
 570         self.assertIn('detail', resp)
 571         self.assertIn('failed request with status code 404', resp['detail'])
 572         self.assertIn('"Code":"NoSuchUser"', resp['detail'])
 573         self.assertIn('"HostId"', resp['detail'])
 574         self.assertIn('"RequestId"', resp['detail'])
 575
 576     def test_create_get_update_delete_w_tenant(self):
 577         # Create a new user.
 578         self._post(
 579             '/api/rgw/user',
 580             params={
 581                 'uid': 'test01$teuth-test-user',
 582                 'display_name': 'display name'
 583             })
 584         self.assertStatus(201)
 585         data = self.jsonBody()
 586         self._assert_user_data(data)
 587         self.assertEqual(data['user_id'], 'teuth-test-user')
 588         self.assertEqual(data['display_name'], 'display name')
 589
 590         # Get the user.
 591         data = self.get_rgw_user('test01$teuth-test-user')
 592         self.assertStatus(200)
 593         self._assert_user_data(data)
 594         self.assertEqual(data['tenant'], 'test01')
 595         self.assertEqual(data['user_id'], 'teuth-test-user')
 596         self.assertEqual(data['uid'], 'test01$teuth-test-user')
 597
 598         # Update the user.
 599         self._put(
 600             '/api/rgw/user/test01$teuth-test-user',
 601             params={'display_name': 'new name'})
 602         self.assertStatus(200)
 603         data = self.jsonBody()
 604         self._assert_user_data(data)
 605         self.assertEqual(data['display_name'], 'new name')
 606
 607         # Delete the user.
 608         self._delete('/api/rgw/user/test01$teuth-test-user')
 609         self.assertStatus(204)
 610         self.get_rgw_user('test01$teuth-test-user')
 611         self.assertStatus(500)
 612         resp = self.jsonBody()
 613         self.assertIn('detail', resp)
 614         self.assertIn('failed request with status code 404', resp['detail'])
 615         self.assertIn('"Code":"NoSuchUser"', resp['detail'])
 616         self.assertIn('"HostId"', resp['detail'])
 617         self.assertIn('"RequestId"', resp['detail'])
 618
 619
 620 class RgwUserCapabilityTest(RgwTestCase):
 621
 622     AUTH_ROLES = ['rgw-manager']
 623
 624     @classmethod
 625     def setUpClass(cls):
 626         cls.create_test_user = True
 627         super(RgwUserCapabilityTest, cls).setUpClass()
 628
 629     def test_set(self):
 630         self._post(
 631             '/api/rgw/user/teuth-test-user/capability',
 632             params={
 633                 'type': 'usage',
 634                 'perm': 'read'
 635             })
 636         self.assertStatus(201)
 637         data = self.jsonBody()
 638         self.assertEqual(len(data), 1)
 639         data = data[0]
 640         self.assertEqual(data['type'], 'usage')
 641         self.assertEqual(data['perm'], 'read')
 642
 643         # Get the user data to validate the capabilities.
 644         data = self.get_rgw_user('teuth-test-user')
 645         self.assertStatus(200)
 646         self.assertGreaterEqual(len(data['caps']), 1)
 647         self.assertEqual(data['caps'][0]['type'], 'usage')
 648         self.assertEqual(data['caps'][0]['perm'], 'read')
 649
 650     def test_delete(self):
 651         self._delete(
 652             '/api/rgw/user/teuth-test-user/capability',
 653             params={
 654                 'type': 'metadata',
 655                 'perm': 'write'
 656             })
 657         self.assertStatus(204)
 658
 659         # Get the user data to validate the capabilities.
 660         data = self.get_rgw_user('teuth-test-user')
 661         self.assertStatus(200)
 662         self.assertEqual(len(data['caps']), 0)
 663
 664
 665 class RgwUserKeyTest(RgwTestCase):
 666
 667     AUTH_ROLES = ['rgw-manager']
 668
 669     @classmethod
 670     def setUpClass(cls):
 671         cls.create_test_user = True
 672         super(RgwUserKeyTest, cls).setUpClass()
 673
 674     def test_create_s3(self):
 675         self._post(
 676             '/api/rgw/user/teuth-test-user/key',
 677             params={
 678                 'key_type': 's3',
 679                 'generate_key': 'false',
 680                 'access_key': 'abc987',
 681                 'secret_key': 'aaabbbccc'
 682             })
 683         data = self.jsonBody()
 684         self.assertStatus(201)
 685         self.assertGreaterEqual(len(data), 3)
 686         key = self.find_object_in_list('access_key', 'abc987', data)
 687         self.assertIsInstance(key, object)
 688         self.assertEqual(key['secret_key'], 'aaabbbccc')
 689
 690     def test_create_swift(self):
 691         self._post(
 692             '/api/rgw/user/teuth-test-user/key',
 693             params={
 694                 'key_type': 'swift',
 695                 'subuser': 'teuth-test-subuser',
 696                 'generate_key': 'false',
 697                 'secret_key': 'xxxyyyzzz'
 698             })
 699         data = self.jsonBody()
 700         self.assertStatus(201)
 701         self.assertGreaterEqual(len(data), 2)
 702         key = self.find_object_in_list('secret_key', 'xxxyyyzzz', data)
 703         self.assertIsInstance(key, object)
 704
 705     def test_delete_s3(self):
 706         self._delete(
 707             '/api/rgw/user/teuth-test-user/key',
 708             params={
 709                 'key_type': 's3',
 710                 'access_key': 'xyz123'
 711             })
 712         self.assertStatus(204)
 713
 714     def test_delete_swift(self):
 715         self._delete(
 716             '/api/rgw/user/teuth-test-user/key',
 717             params={
 718                 'key_type': 'swift',
 719                 'subuser': 'teuth-test-user:teuth-test-subuser2'
 720             })
 721         self.assertStatus(204)
 722
 723
 724 class RgwUserQuotaTest(RgwTestCase):
 725
 726     AUTH_ROLES = ['rgw-manager']
 727
 728     @classmethod
 729     def setUpClass(cls):
 730         cls.create_test_user = True
 731         super(RgwUserQuotaTest, cls).setUpClass()
 732
 733     def _assert_quota(self, data):
 734         self.assertIn('user_quota', data)
 735         self.assertIn('max_objects', data['user_quota'])
 736         self.assertIn('enabled', data['user_quota'])
 737         self.assertIn('max_size_kb', data['user_quota'])
 738         self.assertIn('max_size', data['user_quota'])
 739         self.assertIn('bucket_quota', data)
 740         self.assertIn('max_objects', data['bucket_quota'])
 741         self.assertIn('enabled', data['bucket_quota'])
 742         self.assertIn('max_size_kb', data['bucket_quota'])
 743         self.assertIn('max_size', data['bucket_quota'])
 744
 745     def test_get_quota(self):
 746         data = self._get('/api/rgw/user/teuth-test-user/quota')
 747         self.assertStatus(200)
 748         self._assert_quota(data)
 749
 750     def test_set_user_quota(self):
 751         self._put(
 752             '/api/rgw/user/teuth-test-user/quota',
 753             params={
 754                 'quota_type': 'user',
 755                 'enabled': 'true',
 756                 'max_size_kb': 2048,
 757                 'max_objects': 101
 758             })
 759         self.assertStatus(200)
 760
 761         data = self._get('/api/rgw/user/teuth-test-user/quota')
 762         self.assertStatus(200)
 763         self._assert_quota(data)
 764         self.assertEqual(data['user_quota']['max_objects'], 101)
 765         self.assertTrue(data['user_quota']['enabled'])
 766         self.assertEqual(data['user_quota']['max_size_kb'], 2048)
 767
 768     def test_set_bucket_quota(self):
 769         self._put(
 770             '/api/rgw/user/teuth-test-user/quota',
 771             params={
 772                 'quota_type': 'bucket',
 773                 'enabled': 'false',
 774                 'max_size_kb': 4096,
 775                 'max_objects': 2000
 776             })
 777         self.assertStatus(200)
 778
 779         data = self._get('/api/rgw/user/teuth-test-user/quota')
 780         self.assertStatus(200)
 781         self._assert_quota(data)
 782         self.assertEqual(data['bucket_quota']['max_objects'], 2000)
 783         self.assertFalse(data['bucket_quota']['enabled'])
 784         self.assertEqual(data['bucket_quota']['max_size_kb'], 4096)
 785
 786
 787 class RgwUserSubuserTest(RgwTestCase):
 788
 789     AUTH_ROLES = ['rgw-manager']
 790
 791     @classmethod
 792     def setUpClass(cls):
 793         cls.create_test_user = True
 794         super(RgwUserSubuserTest, cls).setUpClass()
 795
 796     def test_create_swift(self):
 797         self._post(
 798             '/api/rgw/user/teuth-test-user/subuser',
 799             params={
 800                 'subuser': 'tux',
 801                 'access': 'readwrite',
 802                 'key_type': 'swift'
 803             })
 804         self.assertStatus(201)
 805         data = self.jsonBody()
 806         subuser = self.find_object_in_list('id', 'teuth-test-user:tux', data)
 807         self.assertIsInstance(subuser, object)
 808         self.assertEqual(subuser['permissions'], 'read-write')
 809
 810         # Get the user data to validate the keys.
 811         data = self.get_rgw_user('teuth-test-user')
 812         self.assertStatus(200)
 813         key = self.find_object_in_list('user', 'teuth-test-user:tux',
 814                                        data['swift_keys'])
 815         self.assertIsInstance(key, object)
 816
 817     def test_create_s3(self):
 818         self._post(
 819             '/api/rgw/user/teuth-test-user/subuser',
 820             params={
 821                 'subuser': 'hugo',
 822                 'access': 'write',
 823                 'generate_secret': 'false',
 824                 'access_key': 'yyy',
 825                 'secret_key': 'xxx'
 826             })
 827         self.assertStatus(201)
 828         data = self.jsonBody()
 829         subuser = self.find_object_in_list('id', 'teuth-test-user:hugo', data)
 830         self.assertIsInstance(subuser, object)
 831         self.assertEqual(subuser['permissions'], 'write')
 832
 833         # Get the user data to validate the keys.
 834         data = self.get_rgw_user('teuth-test-user')
 835         self.assertStatus(200)
 836         key = self.find_object_in_list('user', 'teuth-test-user:hugo',
 837                                        data['keys'])
 838         self.assertIsInstance(key, object)
 839         self.assertEqual(key['secret_key'], 'xxx')
 840
 841     def test_delete_w_purge(self):
 842         self._delete(
 843             '/api/rgw/user/teuth-test-user/subuser/teuth-test-subuser2')
 844         self.assertStatus(204)
 845
 846         # Get the user data to check that the keys don't exist anymore.
 847         data = self.get_rgw_user('teuth-test-user')
 848         self.assertStatus(200)
 849         key = self.find_object_in_list(
 850             'user', 'teuth-test-user:teuth-test-subuser2', data['swift_keys'])
 851         self.assertIsNone(key)
 852
 853     def test_delete_wo_purge(self):
 854         self._delete(
 855             '/api/rgw/user/teuth-test-user/subuser/teuth-test-subuser',
 856             params={'purge_keys': 'false'})
 857         self.assertStatus(204)
 858
 859         # Get the user data to check whether they keys still exist.
 860         data = self.get_rgw_user('teuth-test-user')
 861         self.assertStatus(200)
 862         key = self.find_object_in_list(
 863             'user', 'teuth-test-user:teuth-test-subuser', data['keys'])
 864         self.assertIsInstance(key, object)