def setUpClass(cls):
cls.create_test_user = True
super(RgwBucketTest, cls).setUpClass()
- # Create a tenanted user.
+ # Create tenanted users.
cls._radosgw_admin_cmd([
'user', 'create', '--tenant', 'testx', '--uid', 'teuth-test-user',
'--display-name', 'tenanted teuth-test-user'
])
+ cls._radosgw_admin_cmd([
+ 'user', 'create', '--tenant', 'testx', '--uid', 'teuth-test-user2',
+ '--display-name', 'tenanted teuth-test-user 2'
+ ])
@classmethod
def tearDownClass(cls):
cls._radosgw_admin_cmd(
['user', 'rm', '--tenant', 'testx', '--uid=teuth-test-user'])
+ cls._radosgw_admin_cmd(
+ ['user', 'rm', '--tenant', 'testx', '--uid=teuth-test-user2'])
super(RgwBucketTest, cls).tearDownClass()
def test_all(self):
self.assertEqual(data['tenant'], 'testx')
self.assertEqual(data['bid'], 'testx/teuth-test-bucket')
- # Update the bucket.
+ # Update bucket: different user from same tenant.
+ self._put(
+ '/api/rgw/bucket/{}'.format(
+ urllib.quote_plus('testx/teuth-test-bucket')),
+ params={
+ 'bucket_id': data['id'],
+ 'uid': 'testx$teuth-test-user2'
+ })
+ self.assertStatus(200)
+ data = self._get('/api/rgw/bucket/{}'.format(
+ urllib.quote_plus('testx/teuth-test-bucket')))
+ self.assertStatus(200)
+ self.assertIn('owner', data)
+ self.assertEqual(data['owner'], 'testx$teuth-test-user2')
+
+ # Update bucket: different user from empty tenant.
self._put(
'/api/rgw/bucket/{}'.format(
urllib.quote_plus('testx/teuth-test-bucket')),
if bucket['tenant'] else bucket['bucket']
return bucket
+ @staticmethod
+ def strip_tenant_from_bucket_name(bucket_name, uid):
+ # type (str, str) => str
+ """
+ When linking a bucket to a new user belonging to same tenant
+ as the previous owner, tenant must be removed from the bucket name.
+ >>> RgwBucket.strip_tenant_from_bucket_name('tenant/bucket-name', 'tenant$user1')
+ 'bucket-name'
+ >>> RgwBucket.strip_tenant_from_bucket_name('tenant/bucket-name', 'tenant2$user2')
+ 'tenant/bucket-name'
+ >>> RgwBucket.strip_tenant_from_bucket_name('bucket-name', 'user1')
+ 'bucket-name'
+ """
+ bucket_tenant = bucket_name[:bucket_name.find('/')] if bucket_name.find('/') >= 0 else None
+ uid_tenant = uid[:uid.find('$')] if uid.find('$') >= 0 else None
+ if bucket_tenant and uid_tenant and bucket_tenant == uid_tenant:
+ return bucket_name[bucket_name.find('/') + 1:]
+
+ return bucket_name
+
def list(self):
return self.proxy('GET', 'bucket')
def set(self, bucket, bucket_id, uid):
result = self.proxy('PUT', 'bucket', {
- 'bucket': bucket,
+ 'bucket': RgwBucket.strip_tenant_from_bucket_name(bucket, uid),
'bucket-id': bucket_id,
'uid': uid
}, json_response=False)