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