projects / ceph.git / blame
| Commit | Line | Data |
|---|---|---|
| 11fdf7f2 TL |
1 | # -*- coding: utf-8 -*- |
| 2 | from __future__ import absolute_import | |
| 3 | ||
| 4 | import logging | |
| 5 | import urllib | |
| 6 | import six | |
| 7 | ||
| 8 | from .helper import DashboardTestCase, JObj, JList, JLeaf | |
| 9 | ||
| 10 | logger = logging.getLogger(__name__) | |
| 11 | ||
| 12 | ||
| 13 | class RgwTestCase(DashboardTestCase): | |
| 14 | ||
| 15 | maxDiff = None | |
| 16 | create_test_user = False | |
| 17 | ||
| 18 | AUTH_ROLES = ['rgw-manager'] | |
| 19 | ||
| 20 | @classmethod | |
| 21 | def setUpClass(cls): | |
| 22 | super(RgwTestCase, cls).setUpClass() | |
| 23 | # Create the administrator account. | |
| 24 | cls._radosgw_admin_cmd([ | |
| 25 | 'user', 'create', '--uid', 'admin', '--display-name', 'admin', | |
| 26 | '--system', '--access-key', 'admin', '--secret', 'admin' | |
| 27 | ]) | |
| 28 | # Update the dashboard configuration. | |
| 29 | cls._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', 'admin']) | |
| 30 | cls._ceph_cmd(['dashboard', 'set-rgw-api-access-key', 'admin']) | |
| 31 | # Create a test user? | |
| 32 | if cls.create_test_user: | |
| 33 | cls._radosgw_admin_cmd([ | |
| 34 | 'user', 'create', '--uid', 'teuth-test-user', '--display-name', | |
| 35 | 'teuth-test-user' | |
| 36 | ]) | |
| 37 | cls._radosgw_admin_cmd([ | |
| 38 | 'caps', 'add', '--uid', 'teuth-test-user', '--caps', | |
| 39 | 'metadata=write' | |
| 40 | ]) | |
| 41 | cls._radosgw_admin_cmd([ | |
| 42 | 'subuser', 'create', '--uid', 'teuth-test-user', '--subuser', | |
| 43 | 'teuth-test-subuser', '--access', 'full', '--key-type', 's3', | |
| 44 | '--access-key', 'xyz123' | |
| 45 | ]) | |
| 46 | cls._radosgw_admin_cmd([ | |
| 47 | 'subuser', 'create', '--uid', 'teuth-test-user', '--subuser', | |
| 48 | 'teuth-test-subuser2', '--access', 'full', '--key-type', | |
| 49 | 'swift' | |
| 50 | ]) | |
| 51 | ||
| 52 | @classmethod | |
| 53 | def tearDownClass(cls): | |
| 54 | if cls.create_test_user: | |
| 55 | cls._radosgw_admin_cmd(['user', 'rm', '--uid=teuth-test-user']) | |
| 56 | super(RgwTestCase, cls).tearDownClass() | |
| 57 | ||
| 58 | def setUp(self): | |
| 59 | super(RgwTestCase, self).setUp() | |
| 60 | ||
| 61 | def get_rgw_user(self, uid): | |
| 62 | return self._get('/api/rgw/user/{}'.format(uid)) | |
| 63 | ||
| 64 | ||
| 65 | class RgwApiCredentialsTest(RgwTestCase): | |
| 66 | ||
| 67 | AUTH_ROLES = ['rgw-manager'] | |
| 68 | ||
| 69 | def setUp(self): | |
| 70 | # Restart the Dashboard module to ensure that the connection to the | |
| 71 | # RGW Admin Ops API is re-established with the new credentials. | |
| 72 | self.logout() | |
| 73 | self._ceph_cmd(['mgr', 'module', 'disable', 'dashboard']) | |
| 74 | self._ceph_cmd(['mgr', 'module', 'enable', 'dashboard', '--force']) | |
| 75 | # Set the default credentials. | |
| 76 | self._ceph_cmd(['dashboard', 'set-rgw-api-user-id', '']) | |
| 77 | self._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', 'admin']) | |
| 78 | self._ceph_cmd(['dashboard', 'set-rgw-api-access-key', 'admin']) | |
| 79 | super(RgwApiCredentialsTest, self).setUp() | |
| 80 | ||
| 81 | def test_no_access_secret_key(self): | |
| 82 | self._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', '']) | |
| 83 | self._ceph_cmd(['dashboard', 'set-rgw-api-access-key', '']) | |
| 84 | resp = self._get('/api/rgw/user') | |
| 85 | self.assertStatus(500) | |
| 86 | self.assertIn('detail', resp) | |
| 87 | self.assertIn('component', resp) | |
| 88 | self.assertIn('No RGW credentials found', resp['detail']) | |
| 89 | self.assertEqual(resp['component'], 'rgw') | |
| 90 | ||
| 91 | def test_success(self): | |
| 92 | data = self._get('/api/rgw/status') | |
| 93 | self.assertStatus(200) | |
| 94 | self.assertIn('available', data) | |
| 95 | self.assertIn('message', data) | |
| 96 | self.assertTrue(data['available']) | |
| 97 | ||
| 98 | def test_invalid_user_id(self): | |
| 99 | self._ceph_cmd(['dashboard', 'set-rgw-api-user-id', 'xyz']) | |
| 100 | data = self._get('/api/rgw/status') | |
| 101 | self.assertStatus(200) | |
| 102 | self.assertIn('available', data) | |
| 103 | self.assertIn('message', data) | |
| 104 | self.assertFalse(data['available']) | |
| 105 | self.assertIn('The user "xyz" is unknown to the Object Gateway.', | |
| 106 | data['message']) | |
| 107 | ||
| 108 | ||
| 109 | class RgwBucketTest(RgwTestCase): | |
| 110 | ||
| 111 | AUTH_ROLES = ['rgw-manager'] | |
| 112 | ||
| 113 | @classmethod | |
| 114 | def setUpClass(cls): | |
| 115 | cls.create_test_user = True | |
| 116 | super(RgwBucketTest, cls).setUpClass() | |
| eafe8130 | 117 | # Create tenanted users. |
| 11fdf7f2 TL |
118 | cls._radosgw_admin_cmd([ |
| 119 | 'user', 'create', '--tenant', 'testx', '--uid', 'teuth-test-user', | |
| 120 | '--display-name', 'tenanted teuth-test-user' | |
| 121 | ]) | |
| eafe8130 TL |
122 | cls._radosgw_admin_cmd([ |
| 123 | 'user', 'create', '--tenant', 'testx', '--uid', 'teuth-test-user2', | |
| 124 | '--display-name', 'tenanted teuth-test-user 2' | |
| 125 | ]) | |
| 11fdf7f2 TL |
126 | |
| 127 | @classmethod | |
| 128 | def tearDownClass(cls): | |
| 129 | cls._radosgw_admin_cmd( | |
| 130 | ['user', 'rm', '--tenant', 'testx', '--uid=teuth-test-user']) | |
| eafe8130 TL |
131 | cls._radosgw_admin_cmd( |
| 132 | ['user', 'rm', '--tenant', 'testx', '--uid=teuth-test-user2']) | |
| 11fdf7f2 TL |
133 | super(RgwBucketTest, cls).tearDownClass() |
| 134 | ||
| 135 | def test_all(self): | |
| 136 | # Create a new bucket. | |
| 137 | self._post( | |
| 138 | '/api/rgw/bucket', | |
| 139 | params={ | |
| 140 | 'bucket': 'teuth-test-bucket', | |
| 141 | 'uid': 'admin' | |
| 142 | }) | |
| 143 | self.assertStatus(201) | |
| 144 | data = self.jsonBody() | |
| 145 | self.assertSchema(data, JObj(sub_elems={ | |
| 146 | 'bucket_info': JObj(sub_elems={ | |
| 147 | 'bucket': JObj(allow_unknown=True, sub_elems={ | |
| 148 | 'name': JLeaf(str), | |
| 149 | 'bucket_id': JLeaf(str), | |
| 150 | 'tenant': JLeaf(str) | |
| 151 | }), | |
| 152 | 'quota': JObj(sub_elems={}, allow_unknown=True), | |
| 153 | 'creation_time': JLeaf(str) | |
| 154 | }, allow_unknown=True) | |
| 155 | }, allow_unknown=True)) | |
| 156 | data = data['bucket_info']['bucket'] | |
| 157 | self.assertEqual(data['name'], 'teuth-test-bucket') | |
| 158 | self.assertEqual(data['tenant'], '') | |
| 159 | ||
| 160 | # List all buckets. | |
| 161 | data = self._get('/api/rgw/bucket') | |
| 162 | self.assertStatus(200) | |
| 163 | self.assertEqual(len(data), 1) | |
| 164 | self.assertIn('teuth-test-bucket', data) | |
| 165 | ||
| 166 | # Get the bucket. | |
| 167 | data = self._get('/api/rgw/bucket/teuth-test-bucket') | |
| 168 | self.assertStatus(200) | |
| 169 | self.assertSchema(data, JObj(sub_elems={ | |
| 170 | 'id': JLeaf(str), | |
| 171 | 'bid': JLeaf(str), | |
| 172 | 'tenant': JLeaf(str), | |
| 173 | 'bucket': JLeaf(str), | |
| 174 | 'bucket_quota': JObj(sub_elems={}, allow_unknown=True), | |
| 175 | 'owner': JLeaf(str) | |
| 176 | }, allow_unknown=True)) | |
| 177 | self.assertEqual(data['bucket'], 'teuth-test-bucket') | |
| 178 | self.assertEqual(data['owner'], 'admin') | |
| 179 | ||
| 180 | # Update the bucket. | |
| 181 | self._put( | |
| 182 | '/api/rgw/bucket/teuth-test-bucket', | |
| 183 | params={ | |
| 184 | 'bucket_id': data['id'], | |
| 185 | 'uid': 'teuth-test-user' | |
| 186 | }) | |
| 187 | self.assertStatus(200) | |
| 188 | data = self._get('/api/rgw/bucket/teuth-test-bucket') | |
| 189 | self.assertStatus(200) | |
| 190 | self.assertSchema(data, JObj(sub_elems={ | |
| 191 | 'owner': JLeaf(str), | |
| 192 | 'bid': JLeaf(str), | |
| 193 | 'tenant': JLeaf(str) | |
| 194 | }, allow_unknown=True)) | |
| 195 | self.assertEqual(data['owner'], 'teuth-test-user') | |
| 196 | ||
| 197 | # Delete the bucket. | |
| 198 | self._delete('/api/rgw/bucket/teuth-test-bucket') | |
| 199 | self.assertStatus(204) | |
| 200 | data = self._get('/api/rgw/bucket') | |
| 201 | self.assertStatus(200) | |
| 202 | self.assertEqual(len(data), 0) | |
| 203 | ||
| 204 | def test_create_get_update_delete_w_tenant(self): | |
| 205 | # Create a new bucket. The tenant of the user is used when | |
| 206 | # the bucket is created. | |
| 207 | self._post( | |
| 208 | '/api/rgw/bucket', | |
| 209 | params={ | |
| 210 | 'bucket': 'teuth-test-bucket', | |
| 211 | 'uid': 'testx$teuth-test-user' | |
| 212 | }) | |
| 213 | self.assertStatus(201) | |
| 214 | # It's not possible to validate the result because there | |
| 215 | # IS NO result object returned by the RGW Admin OPS API | |
| 216 | # when a tenanted bucket is created. | |
| 217 | data = self.jsonBody() | |
| 218 | self.assertIsNone(data) | |
| 219 | ||
| 220 | # List all buckets. | |
| 221 | data = self._get('/api/rgw/bucket') | |
| 222 | self.assertStatus(200) | |
| 223 | self.assertEqual(len(data), 1) | |
| 224 | self.assertIn('testx/teuth-test-bucket', data) | |
| 225 | ||
| 226 | # Get the bucket. | |
| 227 | data = self._get('/api/rgw/bucket/{}'.format( | |
| 228 | urllib.quote_plus('testx/teuth-test-bucket'))) | |
| 229 | self.assertStatus(200) | |
| 230 | self.assertSchema(data, JObj(sub_elems={ | |
| 231 | 'owner': JLeaf(str), | |
| 232 | 'bucket': JLeaf(str), | |
| 233 | 'tenant': JLeaf(str), | |
| 234 | 'bid': JLeaf(str) | |
| 235 | }, allow_unknown=True)) | |
| 236 | self.assertEqual(data['owner'], 'testx$teuth-test-user') | |
| 237 | self.assertEqual(data['bucket'], 'teuth-test-bucket') | |
| 238 | self.assertEqual(data['tenant'], 'testx') | |
| 239 | self.assertEqual(data['bid'], 'testx/teuth-test-bucket') | |
| 240 | ||
| eafe8130 TL |
241 | # Update bucket: different user from same tenant. |
| 242 | self._put( | |
| 243 | '/api/rgw/bucket/{}'.format( | |
| 244 | urllib.quote_plus('testx/teuth-test-bucket')), | |
| 245 | params={ | |
| 246 | 'bucket_id': data['id'], | |
| 247 | 'uid': 'testx$teuth-test-user2' | |
| 248 | }) | |
| 249 | self.assertStatus(200) | |
| 250 | data = self._get('/api/rgw/bucket/{}'.format( | |
| 251 | urllib.quote_plus('testx/teuth-test-bucket'))) | |
| 252 | self.assertStatus(200) | |
| 253 | self.assertIn('owner', data) | |
| 254 | self.assertEqual(data['owner'], 'testx$teuth-test-user2') | |
| 255 | ||
| 256 | # Update bucket: different user from empty tenant. | |
| 11fdf7f2 TL |
257 | self._put( |
| 258 | '/api/rgw/bucket/{}'.format( | |
| 259 | urllib.quote_plus('testx/teuth-test-bucket')), | |
| 260 | params={ | |
| 261 | 'bucket_id': data['id'], | |
| 262 | 'uid': 'admin' | |
| 263 | }) | |
| 264 | self.assertStatus(200) | |
| 265 | data = self._get('/api/rgw/bucket/{}'.format( | |
| 266 | urllib.quote_plus('testx/teuth-test-bucket'))) | |
| 267 | self.assertStatus(200) | |
| 268 | self.assertIn('owner', data) | |
| 269 | self.assertEqual(data['owner'], 'admin') | |
| 270 | ||
| 271 | # Delete the bucket. | |
| 272 | self._delete('/api/rgw/bucket/{}'.format( | |
| 273 | urllib.quote_plus('testx/teuth-test-bucket'))) | |
| 274 | self.assertStatus(204) | |
| 275 | data = self._get('/api/rgw/bucket') | |
| 276 | self.assertStatus(200) | |
| 277 | self.assertEqual(len(data), 0) | |
| 278 | ||
| 279 | ||
| 280 | class RgwDaemonTest(DashboardTestCase): | |
| 281 | ||
| 282 | AUTH_ROLES = ['rgw-manager'] | |
| 283 | ||
| 284 | @DashboardTestCase.RunAs('test', 'test', [{ | |
| 285 | 'rgw': ['create', 'update', 'delete'] | |
| 286 | }]) | |
| 287 | def test_read_access_permissions(self): | |
| 288 | self._get('/api/rgw/daemon') | |
| 289 | self.assertStatus(403) | |
| 290 | self._get('/api/rgw/daemon/id') | |
| 291 | self.assertStatus(403) | |
| 292 | ||
| 293 | def test_list(self): | |
| 294 | data = self._get('/api/rgw/daemon') | |
| 295 | self.assertStatus(200) | |
| 296 | self.assertEqual(len(data), 1) | |
| 297 | data = data[0] | |
| 298 | self.assertIn('id', data) | |
| 299 | self.assertIn('version', data) | |
| 300 | self.assertIn('server_hostname', data) | |
| 301 | ||
| 302 | def test_get(self): | |
| 303 | data = self._get('/api/rgw/daemon') | |
| 304 | self.assertStatus(200) | |
| 305 | ||
| 306 | data = self._get('/api/rgw/daemon/{}'.format(data[0]['id'])) | |
| 307 | self.assertStatus(200) | |
| 308 | self.assertIn('rgw_metadata', data) | |
| 309 | self.assertIn('rgw_id', data) | |
| 310 | self.assertIn('rgw_status', data) | |
| 311 | self.assertTrue(data['rgw_metadata']) | |
| 312 | ||
| 313 | def test_status(self): | |
| 314 | self._radosgw_admin_cmd([ | |
| 315 | 'user', 'create', '--uid=admin', '--display-name=admin', | |
| 316 | '--system', '--access-key=admin', '--secret=admin' | |
| 317 | ]) | |
| 318 | self._ceph_cmd(['dashboard', 'set-rgw-api-user-id', 'admin']) | |
| 319 | self._ceph_cmd(['dashboard', 'set-rgw-api-secret-key', 'admin']) | |
| 320 | self._ceph_cmd(['dashboard', 'set-rgw-api-access-key', 'admin']) | |
| 321 | ||
| 322 | data = self._get('/api/rgw/status') | |
| 323 | self.assertStatus(200) | |
| 324 | self.assertIn('available', data) | |
| 325 | self.assertIn('message', data) | |
| 326 | self.assertTrue(data['available']) | |
| 327 | ||
| 328 | ||
| 329 | class RgwUserTest(RgwTestCase): | |
| 330 | ||
| 331 | AUTH_ROLES = ['rgw-manager'] | |
| 332 | ||
| 333 | @classmethod | |
| 334 | def setUpClass(cls): | |
| 335 | super(RgwUserTest, cls).setUpClass() | |
| 336 | ||
| 337 | def _assert_user_data(self, data): | |
| 338 | self.assertSchema(data, JObj(sub_elems={ | |
| 339 | 'caps': JList(JObj(sub_elems={}, allow_unknown=True)), | |
| 340 | 'display_name': JLeaf(str), | |
| 341 | 'email': JLeaf(str), | |
| 342 | 'keys': JList(JObj(sub_elems={}, allow_unknown=True)), | |
| 343 | 'max_buckets': JLeaf(int), | |
| 344 | 'subusers': JList(JLeaf(str)), | |
| 345 | 'suspended': JLeaf(int), | |
| 346 | 'swift_keys': JList(JObj(sub_elems={}, allow_unknown=True)), | |
| 347 | 'tenant': JLeaf(str), | |
| 348 | 'user_id': JLeaf(str), | |
| 349 | 'uid': JLeaf(str) | |
| 350 | }, allow_unknown=True)) | |
| 351 | self.assertGreaterEqual(len(data['keys']), 1) | |
| 352 | ||
| 353 | def test_get(self): | |
| 354 | data = self.get_rgw_user('admin') | |
| 355 | self.assertStatus(200) | |
| 356 | self._assert_user_data(data) | |
| 357 | self.assertEqual(data['user_id'], 'admin') | |
| 358 | ||
| 359 | def test_list(self): | |
| 360 | data = self._get('/api/rgw/user') | |
| 361 | self.assertStatus(200) | |
| 362 | self.assertGreaterEqual(len(data), 1) | |
| 363 | self.assertIn('admin', data) | |
| 364 | ||
| 365 | def test_create_get_update_delete(self): | |
| 366 | # Create a new user. | |
| 367 | self._post('/api/rgw/user', params={ | |
| 368 | 'uid': 'teuth-test-user', | |
| 369 | 'display_name': 'display name' | |
| 370 | }) | |
| 371 | self.assertStatus(201) | |
| 372 | data = self.jsonBody() | |
| 373 | self._assert_user_data(data) | |
| 374 | self.assertEqual(data['user_id'], 'teuth-test-user') | |
| 375 | self.assertEqual(data['display_name'], 'display name') | |
| 376 | ||
| 377 | # Get the user. | |
| 378 | data = self.get_rgw_user('teuth-test-user') | |
| 379 | self.assertStatus(200) | |
| 380 | self._assert_user_data(data) | |
| 381 | self.assertEqual(data['tenant'], '') | |
| 382 | self.assertEqual(data['user_id'], 'teuth-test-user') | |
| 383 | self.assertEqual(data['uid'], 'teuth-test-user') | |
| 384 | ||
| 385 | # Update the user. | |
| 386 | self._put( | |
| 387 | '/api/rgw/user/teuth-test-user', | |
| 388 | params={'display_name': 'new name'}) | |
| 389 | self.assertStatus(200) | |
| 390 | data = self.jsonBody() | |
| 391 | self._assert_user_data(data) | |
| 392 | self.assertEqual(data['display_name'], 'new name') | |
| 393 | ||
| 394 | # Delete the user. | |
| 395 | self._delete('/api/rgw/user/teuth-test-user') | |
| 396 | self.assertStatus(204) | |
| 397 | self.get_rgw_user('teuth-test-user') | |
| 398 | self.assertStatus(500) | |
| 399 | resp = self.jsonBody() | |
| 400 | self.assertIn('detail', resp) | |
| 401 | self.assertIn('failed request with status code 404', resp['detail']) | |
| 402 | self.assertIn('"Code":"NoSuchUser"', resp['detail']) | |
| 403 | self.assertIn('"HostId"', resp['detail']) | |
| 404 | self.assertIn('"RequestId"', resp['detail']) | |
| 405 | ||
| 406 | def test_create_get_update_delete_w_tenant(self): | |
| 407 | # Create a new user. | |
| 408 | self._post( | |
| 409 | '/api/rgw/user', | |
| 410 | params={ | |
| 411 | 'uid': 'test01$teuth-test-user', | |
| 412 | 'display_name': 'display name' | |
| 413 | }) | |
| 414 | self.assertStatus(201) | |
| 415 | data = self.jsonBody() | |
| 416 | self._assert_user_data(data) | |
| 417 | self.assertEqual(data['user_id'], 'teuth-test-user') | |
| 418 | self.assertEqual(data['display_name'], 'display name') | |
| 419 | ||
| 420 | # Get the user. | |
| 421 | data = self.get_rgw_user('test01$teuth-test-user') | |
| 422 | self.assertStatus(200) | |
| 423 | self._assert_user_data(data) | |
| 424 | self.assertEqual(data['tenant'], 'test01') | |
| 425 | self.assertEqual(data['user_id'], 'teuth-test-user') | |
| 426 | self.assertEqual(data['uid'], 'test01$teuth-test-user') | |
| 427 | ||
| 428 | # Update the user. | |
| 429 | self._put( | |
| 430 | '/api/rgw/user/test01$teuth-test-user', | |
| 431 | params={'display_name': 'new name'}) | |
| 432 | self.assertStatus(200) | |
| 433 | data = self.jsonBody() | |
| 434 | self._assert_user_data(data) | |
| 435 | self.assertEqual(data['display_name'], 'new name') | |
| 436 | ||
| 437 | # Delete the user. | |
| 438 | self._delete('/api/rgw/user/test01$teuth-test-user') | |
| 439 | self.assertStatus(204) | |
| 440 | self.get_rgw_user('test01$teuth-test-user') | |
| 441 | self.assertStatus(500) | |
| 442 | resp = self.jsonBody() | |
| 443 | self.assertIn('detail', resp) | |
| 444 | self.assertIn('failed request with status code 404', resp['detail']) | |
| 445 | self.assertIn('"Code":"NoSuchUser"', resp['detail']) | |
| 446 | self.assertIn('"HostId"', resp['detail']) | |
| 447 | self.assertIn('"RequestId"', resp['detail']) | |
| 448 | ||
| 449 | ||
| 450 | class RgwUserCapabilityTest(RgwTestCase): | |
| 451 | ||
| 452 | AUTH_ROLES = ['rgw-manager'] | |
| 453 | ||
| 454 | @classmethod | |
| 455 | def setUpClass(cls): | |
| 456 | cls.create_test_user = True | |
| 457 | super(RgwUserCapabilityTest, cls).setUpClass() | |
| 458 | ||
| 459 | def test_set(self): | |
| 460 | self._post( | |
| 461 | '/api/rgw/user/teuth-test-user/capability', | |
| 462 | params={ | |
| 463 | 'type': 'usage', | |
| 464 | 'perm': 'read' | |
| 465 | }) | |
| 466 | self.assertStatus(201) | |
| 467 | data = self.jsonBody() | |
| 468 | self.assertEqual(len(data), 1) | |
| 469 | data = data[0] | |
| 470 | self.assertEqual(data['type'], 'usage') | |
| 471 | self.assertEqual(data['perm'], 'read') | |
| 472 | ||
| 473 | # Get the user data to validate the capabilities. | |
| 474 | data = self.get_rgw_user('teuth-test-user') | |
| 475 | self.assertStatus(200) | |
| 476 | self.assertGreaterEqual(len(data['caps']), 1) | |
| 477 | self.assertEqual(data['caps'][0]['type'], 'usage') | |
| 478 | self.assertEqual(data['caps'][0]['perm'], 'read') | |
| 479 | ||
| 480 | def test_delete(self): | |
| 481 | self._delete( | |
| 482 | '/api/rgw/user/teuth-test-user/capability', | |
| 483 | params={ | |
| 484 | 'type': 'metadata', | |
| 485 | 'perm': 'write' | |
| 486 | }) | |
| 487 | self.assertStatus(204) | |
| 488 | ||
| 489 | # Get the user data to validate the capabilities. | |
| 490 | data = self.get_rgw_user('teuth-test-user') | |
| 491 | self.assertStatus(200) | |
| 492 | self.assertEqual(len(data['caps']), 0) | |
| 493 | ||
| 494 | ||
| 495 | class RgwUserKeyTest(RgwTestCase): | |
| 496 | ||
| 497 | AUTH_ROLES = ['rgw-manager'] | |
| 498 | ||
| 499 | @classmethod | |
| 500 | def setUpClass(cls): | |
| 501 | cls.create_test_user = True | |
| 502 | super(RgwUserKeyTest, cls).setUpClass() | |
| 503 | ||
| 504 | def test_create_s3(self): | |
| 505 | self._post( | |
| 506 | '/api/rgw/user/teuth-test-user/key', | |
| 507 | params={ | |
| 508 | 'key_type': 's3', | |
| 509 | 'generate_key': 'false', | |
| 510 | 'access_key': 'abc987', | |
| 511 | 'secret_key': 'aaabbbccc' | |
| 512 | }) | |
| 513 | data = self.jsonBody() | |
| 514 | self.assertStatus(201) | |
| 515 | self.assertGreaterEqual(len(data), 3) | |
| 516 | key = self.find_object_in_list('access_key', 'abc987', data) | |
| 517 | self.assertIsInstance(key, object) | |
| 518 | self.assertEqual(key['secret_key'], 'aaabbbccc') | |
| 519 | ||
| 520 | def test_create_swift(self): | |
| 521 | self._post( | |
| 522 | '/api/rgw/user/teuth-test-user/key', | |
| 523 | params={ | |
| 524 | 'key_type': 'swift', | |
| 525 | 'subuser': 'teuth-test-subuser', | |
| 526 | 'generate_key': 'false', | |
| 527 | 'secret_key': 'xxxyyyzzz' | |
| 528 | }) | |
| 529 | data = self.jsonBody() | |
| 530 | self.assertStatus(201) | |
| 531 | self.assertGreaterEqual(len(data), 2) | |
| 532 | key = self.find_object_in_list('secret_key', 'xxxyyyzzz', data) | |
| 533 | self.assertIsInstance(key, object) | |
| 534 | ||
| 535 | def test_delete_s3(self): | |
| 536 | self._delete( | |
| 537 | '/api/rgw/user/teuth-test-user/key', | |
| 538 | params={ | |
| 539 | 'key_type': 's3', | |
| 540 | 'access_key': 'xyz123' | |
| 541 | }) | |
| 542 | self.assertStatus(204) | |
| 543 | ||
| 544 | def test_delete_swift(self): | |
| 545 | self._delete( | |
| 546 | '/api/rgw/user/teuth-test-user/key', | |
| 547 | params={ | |
| 548 | 'key_type': 'swift', | |
| 549 | 'subuser': 'teuth-test-user:teuth-test-subuser2' | |
| 550 | }) | |
| 551 | self.assertStatus(204) | |
| 552 | ||
| 553 | ||
| 554 | class RgwUserQuotaTest(RgwTestCase): | |
| 555 | ||
| 556 | AUTH_ROLES = ['rgw-manager'] | |
| 557 | ||
| 558 | @classmethod | |
| 559 | def setUpClass(cls): | |
| 560 | cls.create_test_user = True | |
| 561 | super(RgwUserQuotaTest, cls).setUpClass() | |
| 562 | ||
| 563 | def _assert_quota(self, data): | |
| 564 | self.assertIn('user_quota', data) | |
| 565 | self.assertIn('max_objects', data['user_quota']) | |
| 566 | self.assertIn('enabled', data['user_quota']) | |
| 567 | self.assertIn('max_size_kb', data['user_quota']) | |
| 568 | self.assertIn('max_size', data['user_quota']) | |
| 569 | self.assertIn('bucket_quota', data) | |
| 570 | self.assertIn('max_objects', data['bucket_quota']) | |
| 571 | self.assertIn('enabled', data['bucket_quota']) | |
| 572 | self.assertIn('max_size_kb', data['bucket_quota']) | |
| 573 | self.assertIn('max_size', data['bucket_quota']) | |
| 574 | ||
| 575 | def test_get_quota(self): | |
| 576 | data = self._get('/api/rgw/user/teuth-test-user/quota') | |
| 577 | self.assertStatus(200) | |
| 578 | self._assert_quota(data) | |
| 579 | ||
| 580 | def test_set_user_quota(self): | |
| 581 | self._put( | |
| 582 | '/api/rgw/user/teuth-test-user/quota', | |
| 583 | params={ | |
| 584 | 'quota_type': 'user', | |
| 585 | 'enabled': 'true', | |
| 586 | 'max_size_kb': 2048, | |
| 587 | 'max_objects': 101 | |
| 588 | }) | |
| 589 | self.assertStatus(200) | |
| 590 | ||
| 591 | data = self._get('/api/rgw/user/teuth-test-user/quota') | |
| 592 | self.assertStatus(200) | |
| 593 | self._assert_quota(data) | |
| 594 | self.assertEqual(data['user_quota']['max_objects'], 101) | |
| 595 | self.assertTrue(data['user_quota']['enabled']) | |
| 596 | self.assertEqual(data['user_quota']['max_size_kb'], 2048) | |
| 597 | ||
| 598 | def test_set_bucket_quota(self): | |
| 599 | self._put( | |
| 600 | '/api/rgw/user/teuth-test-user/quota', | |
| 601 | params={ | |
| 602 | 'quota_type': 'bucket', | |
| 603 | 'enabled': 'false', | |
| 604 | 'max_size_kb': 4096, | |
| 605 | 'max_objects': 2000 | |
| 606 | }) | |
| 607 | self.assertStatus(200) | |
| 608 | ||
| 609 | data = self._get('/api/rgw/user/teuth-test-user/quota') | |
| 610 | self.assertStatus(200) | |
| 611 | self._assert_quota(data) | |
| 612 | self.assertEqual(data['bucket_quota']['max_objects'], 2000) | |
| 613 | self.assertFalse(data['bucket_quota']['enabled']) | |
| 614 | self.assertEqual(data['bucket_quota']['max_size_kb'], 4096) | |
| 615 | ||
| 616 | ||
| 617 | class RgwUserSubuserTest(RgwTestCase): | |
| 618 | ||
| 619 | AUTH_ROLES = ['rgw-manager'] | |
| 620 | ||
| 621 | @classmethod | |
| 622 | def setUpClass(cls): | |
| 623 | cls.create_test_user = True | |
| 624 | super(RgwUserSubuserTest, cls).setUpClass() | |
| 625 | ||
| 626 | def test_create_swift(self): | |
| 627 | self._post( | |
| 628 | '/api/rgw/user/teuth-test-user/subuser', | |
| 629 | params={ | |
| 630 | 'subuser': 'tux', | |
| 631 | 'access': 'readwrite', | |
| 632 | 'key_type': 'swift' | |
| 633 | }) | |
| 634 | self.assertStatus(201) | |
| 635 | data = self.jsonBody() | |
| 636 | subuser = self.find_object_in_list('id', 'teuth-test-user:tux', data) | |
| 637 | self.assertIsInstance(subuser, object) | |
| 638 | self.assertEqual(subuser['permissions'], 'read-write') | |
| 639 | ||
| 640 | # Get the user data to validate the keys. | |
| 641 | data = self.get_rgw_user('teuth-test-user') | |
| 642 | self.assertStatus(200) | |
| 643 | key = self.find_object_in_list('user', 'teuth-test-user:tux', | |
| 644 | data['swift_keys']) | |
| 645 | self.assertIsInstance(key, object) | |
| 646 | ||
| 647 | def test_create_s3(self): | |
| 648 | self._post( | |
| 649 | '/api/rgw/user/teuth-test-user/subuser', | |
| 650 | params={ | |
| 651 | 'subuser': 'hugo', | |
| 652 | 'access': 'write', | |
| 653 | 'generate_secret': 'false', | |
| 654 | 'access_key': 'yyy', | |
| 655 | 'secret_key': 'xxx' | |
| 656 | }) | |
| 657 | self.assertStatus(201) | |
| 658 | data = self.jsonBody() | |
| 659 | subuser = self.find_object_in_list('id', 'teuth-test-user:hugo', data) | |
| 660 | self.assertIsInstance(subuser, object) | |
| 661 | self.assertEqual(subuser['permissions'], 'write') | |
| 662 | ||
| 663 | # Get the user data to validate the keys. | |
| 664 | data = self.get_rgw_user('teuth-test-user') | |
| 665 | self.assertStatus(200) | |
| 666 | key = self.find_object_in_list('user', 'teuth-test-user:hugo', | |
| 667 | data['keys']) | |
| 668 | self.assertIsInstance(key, object) | |
| 669 | self.assertEqual(key['secret_key'], 'xxx') | |
| 670 | ||
| 671 | def test_delete_w_purge(self): | |
| 672 | self._delete( | |
| 673 | '/api/rgw/user/teuth-test-user/subuser/teuth-test-subuser2') | |
| 674 | self.assertStatus(204) | |
| 675 | ||
| 676 | # Get the user data to check that the keys don't exist anymore. | |
| 677 | data = self.get_rgw_user('teuth-test-user') | |
| 678 | self.assertStatus(200) | |
| 679 | key = self.find_object_in_list( | |
| 680 | 'user', 'teuth-test-user:teuth-test-subuser2', data['swift_keys']) | |
| 681 | self.assertIsNone(key) | |
| 682 | ||
| 683 | def test_delete_wo_purge(self): | |
| 684 | self._delete( | |
| 685 | '/api/rgw/user/teuth-test-user/subuser/teuth-test-subuser', | |
| 686 | params={'purge_keys': 'false'}) | |
| 687 | self.assertStatus(204) | |
| 688 | ||
| 689 | # Get the user data to check whether they keys still exist. | |
| 690 | data = self.get_rgw_user('teuth-test-user') | |
| 691 | self.assertStatus(200) | |
| 692 | key = self.find_object_in_list( | |
| 693 | 'user', 'teuth-test-user:teuth-test-subuser', data['keys']) | |
| 694 | self.assertIsInstance(key, object) |