]>
git.proxmox.com Git - ceph.git/blob - ceph/qa/tasks/mgr/dashboard/test_rgw.py
1 # -*- coding: utf-8 -*-
2 from __future__
import absolute_import
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
13 from .helper
import DashboardTestCase
, JObj
, JList
, JLeaf
15 logger
= logging
.getLogger(__name__
)
18 class RgwTestCase(DashboardTestCase
):
21 create_test_user
= False
23 AUTH_ROLES
= ['rgw-manager']
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'
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'])
38 if cls
.create_test_user
:
39 cls
._radosgw
_admin
_cmd
([
40 'user', 'create', '--uid', 'teuth-test-user', '--display-name',
43 cls
._radosgw
_admin
_cmd
([
44 'caps', 'add', '--uid', 'teuth-test-user', '--caps',
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'
52 cls
._radosgw
_admin
_cmd
([
53 'subuser', 'create', '--uid', 'teuth-test-user', '--subuser',
54 'teuth-test-subuser2', '--access', 'full', '--key-type',
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()
66 def get_rgw_user(self
, uid
):
67 return self
._get
('/api/rgw/user/{}'.format(uid
))
70 class RgwApiCredentialsTest(RgwTestCase
):
72 AUTH_ROLES
= ['rgw-manager']
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.
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()
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')
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'])
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.',
115 class RgwSiteTest(RgwTestCase
):
117 AUTH_ROLES
= ['rgw-manager']
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({
124 'placement_targets': JList(JObj({
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))
136 class RgwBucketTest(RgwTestCase
):
138 _mfa_token_serial
= '1'
139 _mfa_token_seed
= '23456723'
140 _mfa_token_time_step
= 2
142 AUTH_ROLES
= ['rgw-manager']
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'
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'
159 cls
._radosgw
_admin
_cmd
([
160 'user', 'create', '--tenant', 'testx2', '--uid', 'teuth-test-user2',
161 '--display-name', 'tenanted teuth-test-user 2'
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()
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
)
180 # Create a new bucket.
184 'bucket': 'teuth-test-bucket',
186 'zonegroup': 'default',
187 'placement_target': 'default-placement'
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
={
195 'bucket_id': JLeaf(str),
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'], '')
207 data
= self
._get
('/api/rgw/bucket')
208 self
.assertStatus(200)
209 self
.assertEqual(len(data
), 1)
210 self
.assertIn('teuth-test-bucket', data
)
213 data
= self
._get
('/api/rgw/bucket/teuth-test-bucket')
214 self
.assertStatus(200)
215 self
.assertSchema(data
, JObj(sub_elems
={
218 'tenant': JLeaf(str),
219 'bucket': JLeaf(str),
220 'bucket_quota': JObj(sub_elems
={}, allow_unknown
=True),
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')
228 # Update bucket: change owner, enable versioning.
230 '/api/rgw/bucket/teuth-test-bucket',
232 'bucket_id': data
['id'],
233 'uid': 'teuth-test-user',
234 'versioning_state': 'Enabled'
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
={
243 }, allow_unknown
=True))
244 self
.assertEqual(data
['owner'], 'teuth-test-user')
245 self
.assertEqual(data
['versioning'], 'Enabled')
247 # Update bucket: enable MFA Delete.
249 '/api/rgw/bucket/teuth-test-bucket',
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
()
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')
264 # Update bucket: disable versioning & MFA Delete.
265 time
.sleep(self
._mfa
_token
_time
_step
* 3) # Required to get new TOTP pin.
267 '/api/rgw/bucket/teuth-test-bucket',
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
()
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')
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)
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.
295 'bucket': 'teuth-test-bucket',
296 'uid': 'testx$teuth-test-user',
297 'zonegroup': 'default',
298 'placement_target': 'default-placement'
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
)
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
)
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
={
320 'bucket': JLeaf(str),
321 'tenant': 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
)
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')
335 # Update bucket: different user with different tenant, enable versioning.
337 '/api/rgw/bucket/{}'.format(
338 parse
.quote_plus('testx/teuth-test-bucket')),
340 'bucket_id': data
['id'],
341 'uid': 'testx2$teuth-test-user2',
342 'versioning_state': 'Enabled'
344 data
= _verify_tenant_bucket('teuth-test-bucket', 'testx2', 'teuth-test-user2')
345 self
.assertEqual(data
['versioning'], 'Enabled')
347 # Change owner to a non-tenanted user
349 '/api/rgw/bucket/{}'.format(
350 parse
.quote_plus('testx2/teuth-test-bucket')),
352 'bucket_id': data
['id'],
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')
365 # Change owner back to tenanted user, suspend versioning.
367 '/api/rgw/bucket/teuth-test-bucket',
369 'bucket_id': data
['id'],
370 'uid': 'testx$teuth-test-user',
371 'versioning_state': 'Suspended'
373 self
.assertStatus(200)
374 data
= _verify_tenant_bucket('teuth-test-bucket', 'testx', 'teuth-test-user')
375 self
.assertEqual(data
['versioning'], 'Suspended')
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)
385 def test_crud_w_locking(self
):
387 self
._post
('/api/rgw/bucket',
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'
398 self
.assertStatus(201)
400 data
= self
._get
('/api/rgw/bucket/teuth-test-bucket')
401 self
.assertStatus(200)
405 'lock_enabled': JLeaf(bool),
406 'lock_mode': JLeaf(str),
407 'lock_retention_period_days': JLeaf(int),
408 'lock_retention_period_years': JLeaf(int)
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)
416 self
._put
('/api/rgw/bucket/teuth-test-bucket',
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'
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)
432 self
._delete
('/api/rgw/bucket/teuth-test-bucket')
433 self
.assertStatus(204)
436 class RgwDaemonTest(RgwTestCase
):
438 AUTH_ROLES
= ['rgw-manager']
440 @DashboardTestCase.RunAs('test', 'test', [{
441 'rgw': ['create', 'update', 'delete']
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)
450 data
= self
._get
('/api/rgw/daemon')
451 self
.assertStatus(200)
452 self
.assertEqual(len(data
), 1)
454 self
.assertIn('id', data
)
455 self
.assertIn('version', data
)
456 self
.assertIn('server_hostname', data
)
459 data
= self
._get
('/api/rgw/daemon')
460 self
.assertStatus(200)
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'])
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'])
477 class RgwUserTest(RgwTestCase
):
479 AUTH_ROLES
= ['rgw-manager']
483 super(RgwUserTest
, cls
).setUpClass()
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),
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),
498 }, allow_unknown
=True))
499 self
.assertGreaterEqual(len(data
['keys']), 1)
502 data
= self
.get_rgw_user('admin')
503 self
.assertStatus(200)
504 self
._assert
_user
_data
(data
)
505 self
.assertEqual(data
['user_id'], 'admin')
508 data
= self
._get
('/api/rgw/user')
509 self
.assertStatus(200)
510 self
.assertGreaterEqual(len(data
), 1)
511 self
.assertIn('admin', data
)
513 def test_get_emails(self
):
514 data
= self
._get
('/api/rgw/user/get_emails')
515 self
.assertStatus(200)
516 self
.assertSchema(data
, JList(str))
518 def test_create_get_update_delete(self
):
520 self
._post
('/api/rgw/user', params
={
521 'uid': 'teuth-test-user',
522 'display_name': 'display name'
524 self
.assertStatus(201)
525 data
= self
.jsonBody()
526 self
._assert
_user
_data
(data
)
527 self
.assertEqual(data
['user_id'], 'teuth-test-user')
528 self
.assertEqual(data
['display_name'], 'display name')
531 data
= self
.get_rgw_user('teuth-test-user')
532 self
.assertStatus(200)
533 self
._assert
_user
_data
(data
)
534 self
.assertEqual(data
['tenant'], '')
535 self
.assertEqual(data
['user_id'], 'teuth-test-user')
536 self
.assertEqual(data
['uid'], 'teuth-test-user')
540 '/api/rgw/user/teuth-test-user',
541 params
={'display_name': 'new name'})
542 self
.assertStatus(200)
543 data
= self
.jsonBody()
544 self
._assert
_user
_data
(data
)
545 self
.assertEqual(data
['display_name'], 'new name')
548 self
._delete
('/api/rgw/user/teuth-test-user')
549 self
.assertStatus(204)
550 self
.get_rgw_user('teuth-test-user')
551 self
.assertStatus(500)
552 resp
= self
.jsonBody()
553 self
.assertIn('detail', resp
)
554 self
.assertIn('failed request with status code 404', resp
['detail'])
555 self
.assertIn('"Code":"NoSuchUser"', resp
['detail'])
556 self
.assertIn('"HostId"', resp
['detail'])
557 self
.assertIn('"RequestId"', resp
['detail'])
559 def test_create_get_update_delete_w_tenant(self
):
564 'uid': 'test01$teuth-test-user',
565 'display_name': 'display name'
567 self
.assertStatus(201)
568 data
= self
.jsonBody()
569 self
._assert
_user
_data
(data
)
570 self
.assertEqual(data
['user_id'], 'teuth-test-user')
571 self
.assertEqual(data
['display_name'], 'display name')
574 data
= self
.get_rgw_user('test01$teuth-test-user')
575 self
.assertStatus(200)
576 self
._assert
_user
_data
(data
)
577 self
.assertEqual(data
['tenant'], 'test01')
578 self
.assertEqual(data
['user_id'], 'teuth-test-user')
579 self
.assertEqual(data
['uid'], 'test01$teuth-test-user')
583 '/api/rgw/user/test01$teuth-test-user',
584 params
={'display_name': 'new name'})
585 self
.assertStatus(200)
586 data
= self
.jsonBody()
587 self
._assert
_user
_data
(data
)
588 self
.assertEqual(data
['display_name'], 'new name')
591 self
._delete
('/api/rgw/user/test01$teuth-test-user')
592 self
.assertStatus(204)
593 self
.get_rgw_user('test01$teuth-test-user')
594 self
.assertStatus(500)
595 resp
= self
.jsonBody()
596 self
.assertIn('detail', resp
)
597 self
.assertIn('failed request with status code 404', resp
['detail'])
598 self
.assertIn('"Code":"NoSuchUser"', resp
['detail'])
599 self
.assertIn('"HostId"', resp
['detail'])
600 self
.assertIn('"RequestId"', resp
['detail'])
603 class RgwUserCapabilityTest(RgwTestCase
):
605 AUTH_ROLES
= ['rgw-manager']
609 cls
.create_test_user
= True
610 super(RgwUserCapabilityTest
, cls
).setUpClass()
614 '/api/rgw/user/teuth-test-user/capability',
619 self
.assertStatus(201)
620 data
= self
.jsonBody()
621 self
.assertEqual(len(data
), 1)
623 self
.assertEqual(data
['type'], 'usage')
624 self
.assertEqual(data
['perm'], 'read')
626 # Get the user data to validate the capabilities.
627 data
= self
.get_rgw_user('teuth-test-user')
628 self
.assertStatus(200)
629 self
.assertGreaterEqual(len(data
['caps']), 1)
630 self
.assertEqual(data
['caps'][0]['type'], 'usage')
631 self
.assertEqual(data
['caps'][0]['perm'], 'read')
633 def test_delete(self
):
635 '/api/rgw/user/teuth-test-user/capability',
640 self
.assertStatus(204)
642 # Get the user data to validate the capabilities.
643 data
= self
.get_rgw_user('teuth-test-user')
644 self
.assertStatus(200)
645 self
.assertEqual(len(data
['caps']), 0)
648 class RgwUserKeyTest(RgwTestCase
):
650 AUTH_ROLES
= ['rgw-manager']
654 cls
.create_test_user
= True
655 super(RgwUserKeyTest
, cls
).setUpClass()
657 def test_create_s3(self
):
659 '/api/rgw/user/teuth-test-user/key',
662 'generate_key': 'false',
663 'access_key': 'abc987',
664 'secret_key': 'aaabbbccc'
666 data
= self
.jsonBody()
667 self
.assertStatus(201)
668 self
.assertGreaterEqual(len(data
), 3)
669 key
= self
.find_object_in_list('access_key', 'abc987', data
)
670 self
.assertIsInstance(key
, object)
671 self
.assertEqual(key
['secret_key'], 'aaabbbccc')
673 def test_create_swift(self
):
675 '/api/rgw/user/teuth-test-user/key',
678 'subuser': 'teuth-test-subuser',
679 'generate_key': 'false',
680 'secret_key': 'xxxyyyzzz'
682 data
= self
.jsonBody()
683 self
.assertStatus(201)
684 self
.assertGreaterEqual(len(data
), 2)
685 key
= self
.find_object_in_list('secret_key', 'xxxyyyzzz', data
)
686 self
.assertIsInstance(key
, object)
688 def test_delete_s3(self
):
690 '/api/rgw/user/teuth-test-user/key',
693 'access_key': 'xyz123'
695 self
.assertStatus(204)
697 def test_delete_swift(self
):
699 '/api/rgw/user/teuth-test-user/key',
702 'subuser': 'teuth-test-user:teuth-test-subuser2'
704 self
.assertStatus(204)
707 class RgwUserQuotaTest(RgwTestCase
):
709 AUTH_ROLES
= ['rgw-manager']
713 cls
.create_test_user
= True
714 super(RgwUserQuotaTest
, cls
).setUpClass()
716 def _assert_quota(self
, data
):
717 self
.assertIn('user_quota', data
)
718 self
.assertIn('max_objects', data
['user_quota'])
719 self
.assertIn('enabled', data
['user_quota'])
720 self
.assertIn('max_size_kb', data
['user_quota'])
721 self
.assertIn('max_size', data
['user_quota'])
722 self
.assertIn('bucket_quota', data
)
723 self
.assertIn('max_objects', data
['bucket_quota'])
724 self
.assertIn('enabled', data
['bucket_quota'])
725 self
.assertIn('max_size_kb', data
['bucket_quota'])
726 self
.assertIn('max_size', data
['bucket_quota'])
728 def test_get_quota(self
):
729 data
= self
._get
('/api/rgw/user/teuth-test-user/quota')
730 self
.assertStatus(200)
731 self
._assert
_quota
(data
)
733 def test_set_user_quota(self
):
735 '/api/rgw/user/teuth-test-user/quota',
737 'quota_type': 'user',
742 self
.assertStatus(200)
744 data
= self
._get
('/api/rgw/user/teuth-test-user/quota')
745 self
.assertStatus(200)
746 self
._assert
_quota
(data
)
747 self
.assertEqual(data
['user_quota']['max_objects'], 101)
748 self
.assertTrue(data
['user_quota']['enabled'])
749 self
.assertEqual(data
['user_quota']['max_size_kb'], 2048)
751 def test_set_bucket_quota(self
):
753 '/api/rgw/user/teuth-test-user/quota',
755 'quota_type': 'bucket',
760 self
.assertStatus(200)
762 data
= self
._get
('/api/rgw/user/teuth-test-user/quota')
763 self
.assertStatus(200)
764 self
._assert
_quota
(data
)
765 self
.assertEqual(data
['bucket_quota']['max_objects'], 2000)
766 self
.assertFalse(data
['bucket_quota']['enabled'])
767 self
.assertEqual(data
['bucket_quota']['max_size_kb'], 4096)
770 class RgwUserSubuserTest(RgwTestCase
):
772 AUTH_ROLES
= ['rgw-manager']
776 cls
.create_test_user
= True
777 super(RgwUserSubuserTest
, cls
).setUpClass()
779 def test_create_swift(self
):
781 '/api/rgw/user/teuth-test-user/subuser',
784 'access': 'readwrite',
787 self
.assertStatus(201)
788 data
= self
.jsonBody()
789 subuser
= self
.find_object_in_list('id', 'teuth-test-user:tux', data
)
790 self
.assertIsInstance(subuser
, object)
791 self
.assertEqual(subuser
['permissions'], 'read-write')
793 # Get the user data to validate the keys.
794 data
= self
.get_rgw_user('teuth-test-user')
795 self
.assertStatus(200)
796 key
= self
.find_object_in_list('user', 'teuth-test-user:tux',
798 self
.assertIsInstance(key
, object)
800 def test_create_s3(self
):
802 '/api/rgw/user/teuth-test-user/subuser',
806 'generate_secret': 'false',
810 self
.assertStatus(201)
811 data
= self
.jsonBody()
812 subuser
= self
.find_object_in_list('id', 'teuth-test-user:hugo', data
)
813 self
.assertIsInstance(subuser
, object)
814 self
.assertEqual(subuser
['permissions'], 'write')
816 # Get the user data to validate the keys.
817 data
= self
.get_rgw_user('teuth-test-user')
818 self
.assertStatus(200)
819 key
= self
.find_object_in_list('user', 'teuth-test-user:hugo',
821 self
.assertIsInstance(key
, object)
822 self
.assertEqual(key
['secret_key'], 'xxx')
824 def test_delete_w_purge(self
):
826 '/api/rgw/user/teuth-test-user/subuser/teuth-test-subuser2')
827 self
.assertStatus(204)
829 # Get the user data to check that the keys don't exist anymore.
830 data
= self
.get_rgw_user('teuth-test-user')
831 self
.assertStatus(200)
832 key
= self
.find_object_in_list(
833 'user', 'teuth-test-user:teuth-test-subuser2', data
['swift_keys'])
834 self
.assertIsNone(key
)
836 def test_delete_wo_purge(self
):
838 '/api/rgw/user/teuth-test-user/subuser/teuth-test-subuser',
839 params
={'purge_keys': 'false'})
840 self
.assertStatus(204)
842 # Get the user data to check whether they keys still exist.
843 data
= self
.get_rgw_user('teuth-test-user')
844 self
.assertStatus(200)
845 key
= self
.find_object_in_list(
846 'user', 'teuth-test-user:teuth-test-subuser', data
['keys'])
847 self
.assertIsInstance(key
, object)