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