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