tests/user.py

   1 ##############################################################################
   2 # Copyright (c) 2017 ZTE Corporation and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9 import os
  10
  11 from oslo_config import cfg
  12
  13 from identity_auth import get_session
  14 from os_clients import keystone_client
  15 from os_clients import nova_client
  16
  17
  18 OPTS = [
  19     cfg.StrOpt('doctor_user',
  20                default='doctor',
  21                help='the name of test user',
  22                required=True),
  23     cfg.StrOpt('doctor_passwd',
  24                default='doctor',
  25                help='the password of test user',
  26                required=True),
  27     cfg.StrOpt('doctor_project',
  28                default='doctor',
  29                help='the name of test project',
  30                required=True),
  31     cfg.StrOpt('doctor_role',
  32                default='_member_',
  33                help='the role of test user',
  34                required=True),
  35     cfg.IntOpt('quota_instances',
  36                default=os.environ.get('VM_COUNT', 1),
  37                help='the quota of instances in test user',
  38                required=True),
  39     cfg.IntOpt('quota_cores',
  40                default=os.environ.get('VM_COUNT', 1),
  41                help='the quota of cores in test user',
  42                required=True),
  43 ]
  44
  45
  46 class User(object):
  47
  48     def __init__(self, conf, log):
  49         self.conf = conf
  50         self.log = log
  51         self.keystone = \
  52             keystone_client(get_session())
  53         self.nova = \
  54             nova_client(conf.nova_version, get_session())
  55         self.users = {}
  56         self.projects = {}
  57         self.roles = {}
  58         self.roles_for_user = {}
  59         self.roles_for_admin = {}
  60
  61     def create(self):
  62         """create test user, project and etc"""
  63         self.log.info('user create start......')
  64
  65         self._create_project()
  66         self._create_user()
  67         self._create_role()
  68         self._add_user_role_in_project(is_admin=False)
  69         self._add_user_role_in_project(is_admin=True)
  70
  71         self.log.info('user create end......')
  72
  73     def _create_project(self):
  74         """create test project"""
  75         self.projects = {project.name: project
  76                     for project in self.keystone.tenants.list()}
  77         if self.conf.doctor_project not in self.projects:
  78             test_project = \
  79                 self.keystone.tenants.create(self.conf.doctor_project)
  80             self.projects[test_project.name] = test_project
  81
  82     def _create_user(self):
  83         """create test user"""
  84         project = self.projects.get(self.conf.doctor_project)
  85         self.users = {user.name: user for user in self.keystone.users.list()}
  86         if self.conf.doctor_user not in self.users:
  87             test_user = self.keystone.users.create(
  88                 self.conf.doctor_user,
  89                 password=self.conf.doctor_passwd,
  90                 tenant_id=project.id)
  91             self.users[test_user.name] = test_user
  92
  93     def _create_role(self):
  94         """create test role"""
  95         self.roles = {role.name: role for role in self.keystone.roles.list()}
  96         if self.conf.doctor_role not in self.roles:
  97             test_role = self.keystone.roles.create(self.conf.doctor_role)
  98             self.roles[test_role.name] = test_role
  99
 100     def _add_user_role_in_project(self, is_admin=False):
 101         """add test user with test role in test project"""
 102         project = self.projects.get(self.conf.doctor_project)
 103
 104         user_name = 'admin' if is_admin else self.conf.doctor_user
 105         user = self.users.get(user_name)
 106
 107         role_name = 'admin' if is_admin else self.conf.doctor_role
 108         role = self.roles.get(role_name)
 109
 110         roles_for_user = self.roles_for_admin \
 111             if is_admin else self.roles_for_user
 112
 113         roles_for_user = \
 114             {role.name: role for role in
 115              self.keystone.roles.roles_for_user(user, tenant=project)}
 116         if role_name not in roles_for_user:
 117             self.keystone.roles.add_user_role(user, role, tenant=project)
 118             roles_for_user[role_name] = role
 119
 120     def delete(self):
 121         """delete the test user, project and role"""
 122         self.log.info('user delete start......')
 123
 124         project = self.projects.get(self.conf.doctor_project)
 125         user = self.users.get(self.conf.doctor_user)
 126         role = self.roles.get(self.conf.doctor_role)
 127
 128         if project:
 129             if 'admin' in self.roles_for_admin:
 130                 self.keystone.roles.remove_user_role(
 131                     self.users['admin'],
 132                     self.roles['admin'],
 133                     tenant=project)
 134
 135             if user:
 136                 if role and self.conf.doctor_role in self.roles_for_user:
 137                     self.keystone.roles.remove_user_role(
 138                         user, role, tenant=project)
 139                     self.keystone.roles.delete(role)
 140                 self.keystone.users.delete(user)
 141
 142             self.keystone.tenants.delete(project)
 143         self.log.info('user delete end......')
 144
 145     def update_quota(self):
 146         self.log.info('user quota update start......')
 147         project = self.projects.get(self.conf.doctor_project)
 148         user = self.users.get(self.conf.doctor_user)
 149
 150         if project and user:
 151             self.quota = self.nova.quotas.get(project.id,
 152                                               user_id=user.id)
 153             if self.conf.quota_instances > self.quota.instances:
 154                 self.nova.quotas.update(project.id,
 155                                         instances=self.conf.quota_instances,
 156                                         user_id=user.id)
 157             if self.conf.quota_cores > self.quota.cores:
 158                 self.nova.quotas.update(project.id,
 159                                         cores=self.conf.quota_cores,
 160                                         user_id=user.id)
 161             self.log.info('user quota update end......')
 162         else:
 163             raise Exception('No project or role for update quota')