neutron/tests/tempest/services/identity/v3/json/identity_client.py

   1 # Copyright 2013 OpenStack Foundation
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 from oslo_serialization import jsonutils as json
  17 from six.moves.urllib import parse
  18
  19 from neutron.tests.tempest.common import service_client
  20
  21
  22 class IdentityV3ClientJSON(service_client.ServiceClient):
  23     api_version = "v3"
  24
  25     def create_user(self, user_name, password=None, project_id=None,
  26                     email=None, domain_id='default', **kwargs):
  27         """Creates a user."""
  28         en = kwargs.get('enabled', True)
  29         description = kwargs.get('description', None)
  30         default_project_id = kwargs.get('default_project_id')
  31         post_body = {
  32             'project_id': project_id,
  33             'default_project_id': default_project_id,
  34             'description': description,
  35             'domain_id': domain_id,
  36             'email': email,
  37             'enabled': en,
  38             'name': user_name,
  39             'password': password
  40         }
  41         post_body = json.dumps({'user': post_body})
  42         resp, body = self.post('users', post_body)
  43         self.expected_success(201, resp.status)
  44         body = json.loads(body)
  45         return service_client.ResponseBody(resp, body['user'])
  46
  47     def update_user(self, user_id, name, **kwargs):
  48         """Updates a user."""
  49         body = self.get_user(user_id)
  50         email = kwargs.get('email', body['email'])
  51         en = kwargs.get('enabled', body['enabled'])
  52         project_id = kwargs.get('project_id', body['project_id'])
  53         if 'default_project_id' in body.keys():
  54             default_project_id = kwargs.get('default_project_id',
  55                                             body['default_project_id'])
  56         else:
  57             default_project_id = kwargs.get('default_project_id')
  58         description = kwargs.get('description', body['description'])
  59         domain_id = kwargs.get('domain_id', body['domain_id'])
  60         post_body = {
  61             'name': name,
  62             'email': email,
  63             'enabled': en,
  64             'project_id': project_id,
  65             'default_project_id': default_project_id,
  66             'id': user_id,
  67             'domain_id': domain_id,
  68             'description': description
  69         }
  70         post_body = json.dumps({'user': post_body})
  71         resp, body = self.patch('users/%s' % user_id, post_body)
  72         self.expected_success(200, resp.status)
  73         body = json.loads(body)
  74         return service_client.ResponseBody(resp, body['user'])
  75
  76     def update_user_password(self, user_id, password, original_password):
  77         """Updates a user password."""
  78         update_user = {
  79             'password': password,
  80             'original_password': original_password
  81         }
  82         update_user = json.dumps({'user': update_user})
  83         resp, _ = self.post('users/%s/password' % user_id, update_user)
  84         self.expected_success(204, resp.status)
  85         return service_client.ResponseBody(resp)
  86
  87     def list_user_projects(self, user_id):
  88         """Lists the projects on which a user has roles assigned."""
  89         resp, body = self.get('users/%s/projects' % user_id)
  90         self.expected_success(200, resp.status)
  91         body = json.loads(body)
  92         return service_client.ResponseBodyList(resp, body['projects'])
  93
  94     def get_users(self, params=None):
  95         """Get the list of users."""
  96         url = 'users'
  97         if params:
  98             url += '?%s' % parse.urlencode(params)
  99         resp, body = self.get(url)
 100         self.expected_success(200, resp.status)
 101         body = json.loads(body)
 102         return service_client.ResponseBodyList(resp, body['users'])
 103
 104     def get_user(self, user_id):
 105         """GET a user."""
 106         resp, body = self.get("users/%s" % user_id)
 107         self.expected_success(200, resp.status)
 108         body = json.loads(body)
 109         return service_client.ResponseBody(resp, body['user'])
 110
 111     def delete_user(self, user_id):
 112         """Deletes a User."""
 113         resp, body = self.delete("users/%s" % user_id)
 114         self.expected_success(204, resp.status)
 115         return service_client.ResponseBody(resp, body)
 116
 117     def create_project(self, name, **kwargs):
 118         """Creates a project."""
 119         description = kwargs.get('description', None)
 120         en = kwargs.get('enabled', True)
 121         domain_id = kwargs.get('domain_id', 'default')
 122         post_body = {
 123             'description': description,
 124             'domain_id': domain_id,
 125             'enabled': en,
 126             'name': name
 127         }
 128         post_body = json.dumps({'project': post_body})
 129         resp, body = self.post('projects', post_body)
 130         self.expected_success(201, resp.status)
 131         body = json.loads(body)
 132         return service_client.ResponseBody(resp, body['project'])
 133
 134     def list_projects(self, params=None):
 135         url = "projects"
 136         if params:
 137             url += '?%s' % parse.urlencode(params)
 138         resp, body = self.get(url)
 139         self.expected_success(200, resp.status)
 140         body = json.loads(body)
 141         return service_client.ResponseBodyList(resp, body['projects'])
 142
 143     def update_project(self, project_id, **kwargs):
 144         body = self.get_project(project_id)
 145         name = kwargs.get('name', body['name'])
 146         desc = kwargs.get('description', body['description'])
 147         en = kwargs.get('enabled', body['enabled'])
 148         domain_id = kwargs.get('domain_id', body['domain_id'])
 149         post_body = {
 150             'id': project_id,
 151             'name': name,
 152             'description': desc,
 153             'enabled': en,
 154             'domain_id': domain_id,
 155         }
 156         post_body = json.dumps({'project': post_body})
 157         resp, body = self.patch('projects/%s' % project_id, post_body)
 158         self.expected_success(200, resp.status)
 159         body = json.loads(body)
 160         return service_client.ResponseBody(resp, body['project'])
 161
 162     def get_project(self, project_id):
 163         """GET a Project."""
 164         resp, body = self.get("projects/%s" % project_id)
 165         self.expected_success(200, resp.status)
 166         body = json.loads(body)
 167         return service_client.ResponseBody(resp, body['project'])
 168
 169     def delete_project(self, project_id):
 170         """Delete a project."""
 171         resp, body = self.delete('projects/%s' % str(project_id))
 172         self.expected_success(204, resp.status)
 173         return service_client.ResponseBody(resp, body)
 174
 175     def create_role(self, name):
 176         """Create a Role."""
 177         post_body = {
 178             'name': name
 179         }
 180         post_body = json.dumps({'role': post_body})
 181         resp, body = self.post('roles', post_body)
 182         self.expected_success(201, resp.status)
 183         body = json.loads(body)
 184         return service_client.ResponseBody(resp, body['role'])
 185
 186     def get_role(self, role_id):
 187         """GET a Role."""
 188         resp, body = self.get('roles/%s' % str(role_id))
 189         self.expected_success(200, resp.status)
 190         body = json.loads(body)
 191         return service_client.ResponseBody(resp, body['role'])
 192
 193     def list_roles(self):
 194         """Get the list of Roles."""
 195         resp, body = self.get("roles")
 196         self.expected_success(200, resp.status)
 197         body = json.loads(body)
 198         return service_client.ResponseBodyList(resp, body['roles'])
 199
 200     def update_role(self, name, role_id):
 201         """Create a Role."""
 202         post_body = {
 203             'name': name
 204         }
 205         post_body = json.dumps({'role': post_body})
 206         resp, body = self.patch('roles/%s' % str(role_id), post_body)
 207         self.expected_success(200, resp.status)
 208         body = json.loads(body)
 209         return service_client.ResponseBody(resp, body['role'])
 210
 211     def delete_role(self, role_id):
 212         """Delete a role."""
 213         resp, body = self.delete('roles/%s' % str(role_id))
 214         self.expected_success(204, resp.status)
 215         return service_client.ResponseBody(resp, body)
 216
 217     def assign_user_role(self, project_id, user_id, role_id):
 218         """Add roles to a user on a project."""
 219         resp, body = self.put('projects/%s/users/%s/roles/%s' %
 220                               (project_id, user_id, role_id), None)
 221         self.expected_success(204, resp.status)
 222         return service_client.ResponseBody(resp, body)
 223
 224     def create_domain(self, name, **kwargs):
 225         """Creates a domain."""
 226         description = kwargs.get('description', None)
 227         en = kwargs.get('enabled', True)
 228         post_body = {
 229             'description': description,
 230             'enabled': en,
 231             'name': name
 232         }
 233         post_body = json.dumps({'domain': post_body})
 234         resp, body = self.post('domains', post_body)
 235         self.expected_success(201, resp.status)
 236         body = json.loads(body)
 237         return service_client.ResponseBody(resp, body['domain'])
 238
 239     def delete_domain(self, domain_id):
 240         """Delete a domain."""
 241         resp, body = self.delete('domains/%s' % str(domain_id))
 242         self.expected_success(204, resp.status)
 243         return service_client.ResponseBody(resp, body)
 244
 245     def list_domains(self):
 246         """List Domains."""
 247         resp, body = self.get('domains')
 248         self.expected_success(200, resp.status)
 249         body = json.loads(body)
 250         return service_client.ResponseBodyList(resp, body['domains'])
 251
 252     def update_domain(self, domain_id, **kwargs):
 253         """Updates a domain."""
 254         body = self.get_domain(domain_id)
 255         description = kwargs.get('description', body['description'])
 256         en = kwargs.get('enabled', body['enabled'])
 257         name = kwargs.get('name', body['name'])
 258         post_body = {
 259             'description': description,
 260             'enabled': en,
 261             'name': name
 262         }
 263         post_body = json.dumps({'domain': post_body})
 264         resp, body = self.patch('domains/%s' % domain_id, post_body)
 265         self.expected_success(200, resp.status)
 266         body = json.loads(body)
 267         return service_client.ResponseBody(resp, body['domain'])
 268
 269     def get_domain(self, domain_id):
 270         """Get Domain details."""
 271         resp, body = self.get('domains/%s' % domain_id)
 272         self.expected_success(200, resp.status)
 273         body = json.loads(body)
 274         return service_client.ResponseBody(resp, body['domain'])
 275
 276     def get_token(self, resp_token):
 277         """Get token details."""
 278         headers = {'X-Subject-Token': resp_token}
 279         resp, body = self.get("auth/tokens", headers=headers)
 280         self.expected_success(200, resp.status)
 281         body = json.loads(body)
 282         return service_client.ResponseBody(resp, body['token'])
 283
 284     def delete_token(self, resp_token):
 285         """Deletes token."""
 286         headers = {'X-Subject-Token': resp_token}
 287         resp, body = self.delete("auth/tokens", headers=headers)
 288         self.expected_success(204, resp.status)
 289         return service_client.ResponseBody(resp, body)
 290
 291     def create_group(self, name, **kwargs):
 292         """Creates a group."""
 293         description = kwargs.get('description', None)
 294         domain_id = kwargs.get('domain_id', 'default')
 295         project_id = kwargs.get('project_id', None)
 296         post_body = {
 297             'description': description,
 298             'domain_id': domain_id,
 299             'project_id': project_id,
 300             'name': name
 301         }
 302         post_body = json.dumps({'group': post_body})
 303         resp, body = self.post('groups', post_body)
 304         self.expected_success(201, resp.status)
 305         body = json.loads(body)
 306         return service_client.ResponseBody(resp, body['group'])
 307
 308     def get_group(self, group_id):
 309         """Get group details."""
 310         resp, body = self.get('groups/%s' % group_id)
 311         self.expected_success(200, resp.status)
 312         body = json.loads(body)
 313         return service_client.ResponseBody(resp, body['group'])
 314
 315     def list_groups(self):
 316         """Lists the groups."""
 317         resp, body = self.get('groups')
 318         self.expected_success(200, resp.status)
 319         body = json.loads(body)
 320         return service_client.ResponseBodyList(resp, body['groups'])
 321
 322     def update_group(self, group_id, **kwargs):
 323         """Updates a group."""
 324         body = self.get_group(group_id)
 325         name = kwargs.get('name', body['name'])
 326         description = kwargs.get('description', body['description'])
 327         post_body = {
 328             'name': name,
 329             'description': description
 330         }
 331         post_body = json.dumps({'group': post_body})
 332         resp, body = self.patch('groups/%s' % group_id, post_body)
 333         self.expected_success(200, resp.status)
 334         body = json.loads(body)
 335         return service_client.ResponseBody(resp, body['group'])
 336
 337     def delete_group(self, group_id):
 338         """Delete a group."""
 339         resp, body = self.delete('groups/%s' % str(group_id))
 340         self.expected_success(204, resp.status)
 341         return service_client.ResponseBody(resp, body)
 342
 343     def add_group_user(self, group_id, user_id):
 344         """Add user into group."""
 345         resp, body = self.put('groups/%s/users/%s' % (group_id, user_id),
 346                               None)
 347         self.expected_success(204, resp.status)
 348         return service_client.ResponseBody(resp, body)
 349
 350     def list_group_users(self, group_id):
 351         """List users in group."""
 352         resp, body = self.get('groups/%s/users' % group_id)
 353         self.expected_success(200, resp.status)
 354         body = json.loads(body)
 355         return service_client.ResponseBodyList(resp, body['users'])
 356
 357     def list_user_groups(self, user_id):
 358         """Lists groups which a user belongs to."""
 359         resp, body = self.get('users/%s/groups' % user_id)
 360         self.expected_success(200, resp.status)
 361         body = json.loads(body)
 362         return service_client.ResponseBodyList(resp, body['groups'])
 363
 364     def delete_group_user(self, group_id, user_id):
 365         """Delete user in group."""
 366         resp, body = self.delete('groups/%s/users/%s' % (group_id, user_id))
 367         self.expected_success(204, resp.status)
 368         return service_client.ResponseBody(resp, body)
 369
 370     def assign_user_role_on_project(self, project_id, user_id, role_id):
 371         """Add roles to a user on a project."""
 372         resp, body = self.put('projects/%s/users/%s/roles/%s' %
 373                               (project_id, user_id, role_id), None)
 374         self.expected_success(204, resp.status)
 375         return service_client.ResponseBody(resp, body)
 376
 377     def assign_user_role_on_domain(self, domain_id, user_id, role_id):
 378         """Add roles to a user on a domain."""
 379         resp, body = self.put('domains/%s/users/%s/roles/%s' %
 380                               (domain_id, user_id, role_id), None)
 381         self.expected_success(204, resp.status)
 382         return service_client.ResponseBody(resp, body)
 383
 384     def list_user_roles_on_project(self, project_id, user_id):
 385         """list roles of a user on a project."""
 386         resp, body = self.get('projects/%s/users/%s/roles' %
 387                               (project_id, user_id))
 388         self.expected_success(200, resp.status)
 389         body = json.loads(body)
 390         return service_client.ResponseBodyList(resp, body['roles'])
 391
 392     def list_user_roles_on_domain(self, domain_id, user_id):
 393         """list roles of a user on a domain."""
 394         resp, body = self.get('domains/%s/users/%s/roles' %
 395                               (domain_id, user_id))
 396         self.expected_success(200, resp.status)
 397         body = json.loads(body)
 398         return service_client.ResponseBodyList(resp, body['roles'])
 399
 400     def revoke_role_from_user_on_project(self, project_id, user_id, role_id):
 401         """Delete role of a user on a project."""
 402         resp, body = self.delete('projects/%s/users/%s/roles/%s' %
 403                                  (project_id, user_id, role_id))
 404         self.expected_success(204, resp.status)
 405         return service_client.ResponseBody(resp, body)
 406
 407     def revoke_role_from_user_on_domain(self, domain_id, user_id, role_id):
 408         """Delete role of a user on a domain."""
 409         resp, body = self.delete('domains/%s/users/%s/roles/%s' %
 410                                  (domain_id, user_id, role_id))
 411         self.expected_success(204, resp.status)
 412         return service_client.ResponseBody(resp, body)
 413
 414     def assign_group_role_on_project(self, project_id, group_id, role_id):
 415         """Add roles to a user on a project."""
 416         resp, body = self.put('projects/%s/groups/%s/roles/%s' %
 417                               (project_id, group_id, role_id), None)
 418         self.expected_success(204, resp.status)
 419         return service_client.ResponseBody(resp, body)
 420
 421     def assign_group_role_on_domain(self, domain_id, group_id, role_id):
 422         """Add roles to a user on a domain."""
 423         resp, body = self.put('domains/%s/groups/%s/roles/%s' %
 424                               (domain_id, group_id, role_id), None)
 425         self.expected_success(204, resp.status)
 426         return service_client.ResponseBody(resp, body)
 427
 428     def list_group_roles_on_project(self, project_id, group_id):
 429         """list roles of a user on a project."""
 430         resp, body = self.get('projects/%s/groups/%s/roles' %
 431                               (project_id, group_id))
 432         self.expected_success(200, resp.status)
 433         body = json.loads(body)
 434         return service_client.ResponseBodyList(resp, body['roles'])
 435
 436     def list_group_roles_on_domain(self, domain_id, group_id):
 437         """list roles of a user on a domain."""
 438         resp, body = self.get('domains/%s/groups/%s/roles' %
 439                               (domain_id, group_id))
 440         self.expected_success(200, resp.status)
 441         body = json.loads(body)
 442         return service_client.ResponseBodyList(resp, body['roles'])
 443
 444     def revoke_role_from_group_on_project(self, project_id, group_id, role_id):
 445         """Delete role of a user on a project."""
 446         resp, body = self.delete('projects/%s/groups/%s/roles/%s' %
 447                                  (project_id, group_id, role_id))
 448         self.expected_success(204, resp.status)
 449         return service_client.ResponseBody(resp, body)
 450
 451     def revoke_role_from_group_on_domain(self, domain_id, group_id, role_id):
 452         """Delete role of a user on a domain."""
 453         resp, body = self.delete('domains/%s/groups/%s/roles/%s' %
 454                                  (domain_id, group_id, role_id))
 455         self.expected_success(204, resp.status)
 456         return service_client.ResponseBody(resp, body)
 457
 458     def create_trust(self, trustor_user_id, trustee_user_id, project_id,
 459                      role_names, impersonation, expires_at):
 460         """Creates a trust."""
 461         roles = [{'name': n} for n in role_names]
 462         post_body = {
 463             'trustor_user_id': trustor_user_id,
 464             'trustee_user_id': trustee_user_id,
 465             'project_id': project_id,
 466             'impersonation': impersonation,
 467             'roles': roles,
 468             'expires_at': expires_at
 469         }
 470         post_body = json.dumps({'trust': post_body})
 471         resp, body = self.post('OS-TRUST/trusts', post_body)
 472         self.expected_success(201, resp.status)
 473         body = json.loads(body)
 474         return service_client.ResponseBody(resp, body['trust'])
 475
 476     def delete_trust(self, trust_id):
 477         """Deletes a trust."""
 478         resp, body = self.delete("OS-TRUST/trusts/%s" % trust_id)
 479         self.expected_success(204, resp.status)
 480         return service_client.ResponseBody(resp, body)
 481
 482     def get_trusts(self, trustor_user_id=None, trustee_user_id=None):
 483         """GET trusts."""
 484         if trustor_user_id:
 485             resp, body = self.get("OS-TRUST/trusts?trustor_user_id=%s"
 486                                   % trustor_user_id)
 487         elif trustee_user_id:
 488             resp, body = self.get("OS-TRUST/trusts?trustee_user_id=%s"
 489                                   % trustee_user_id)
 490         else:
 491             resp, body = self.get("OS-TRUST/trusts")
 492         self.expected_success(200, resp.status)
 493         body = json.loads(body)
 494         return service_client.ResponseBodyList(resp, body['trusts'])
 495
 496     def get_trust(self, trust_id):
 497         """GET trust."""
 498         resp, body = self.get("OS-TRUST/trusts/%s" % trust_id)
 499         self.expected_success(200, resp.status)
 500         body = json.loads(body)
 501         return service_client.ResponseBody(resp, body['trust'])
 502
 503     def get_trust_roles(self, trust_id):
 504         """GET roles delegated by a trust."""
 505         resp, body = self.get("OS-TRUST/trusts/%s/roles" % trust_id)
 506         self.expected_success(200, resp.status)
 507         body = json.loads(body)
 508         return service_client.ResponseBodyList(resp, body['roles'])
 509
 510     def get_trust_role(self, trust_id, role_id):
 511         """GET role delegated by a trust."""
 512         resp, body = self.get("OS-TRUST/trusts/%s/roles/%s"
 513                               % (trust_id, role_id))
 514         self.expected_success(200, resp.status)
 515         body = json.loads(body)
 516         return service_client.ResponseBody(resp, body['role'])
 517
 518     def check_trust_role(self, trust_id, role_id):
 519         """HEAD Check if role is delegated by a trust."""
 520         resp, body = self.head("OS-TRUST/trusts/%s/roles/%s"
 521                                % (trust_id, role_id))
 522         self.expected_success(200, resp.status)
 523         return service_client.ResponseBody(resp, body)