from keystoneauth1 import session
from cinderclient import client as cinderclient
from glanceclient import client as glanceclient
+from heatclient import client as heatclient
from novaclient import client as novaclient
from keystoneclient import client as keystoneclient
from neutronclient.neutron import client as neutronclient
logger = ft_logger.Logger("openstack_utils").getLogger()
DEFAULT_API_VERSION = '2'
+DEFAULT_HEAT_API_VERSION = '1'
# *********************************************
'OS_PROJECT_DOMAIN_NAME': 'project_domain_name',
'OS_PROJECT_NAME': 'project_name',
'OS_ENDPOINT_TYPE': 'endpoint_type',
- 'OS_REGION_NAME': 'region_name'
+ 'OS_REGION_NAME': 'region_name',
+ 'OS_CACERT': 'https_cacert'
}
return env_cred_dict
with open(rc_file, "r") as f:
for line in f:
var = line.rstrip('"\n').replace('export ', '').split("=")
- key = re.sub(r'^ *| *$', '', var[0])
- value = re.sub(r'^[" ]*|[ "]*$', '', "".join(var[1:]))
+ # The two next lines should be modified as soon as rc_file
+ # conforms with common rules. Be aware that it could induce
+ # issues if value starts with '
+ key = re.sub(r'^["\' ]*|[ \'"]*$', '', var[0])
+ value = re.sub(r'^["\' ]*|[ \'"]*$', '', "".join(var[1:]))
os.environ[key] = value
if region_name is not None:
cred_key = env_cred_dict.get('OS_REGION_NAME')
rally_conf[cred_key] = region_name
+
+ cacert = os.getenv('OS_CACERT')
+ if cacert is not None:
+ cred_key = env_cred_dict.get('OS_CACERT')
+ rally_conf[cred_key] = cacert
return rally_conf
def get_session(other_creds={}):
auth = get_session_auth(other_creds)
- return session.Session(auth=auth)
+ cacert = os.getenv('OS_CACERT')
+ if cacert is not None:
+ if not os.path.isfile(cacert):
+ raise Exception("The 'OS_CACERT' environment"
+ "variable is set to %s but the file"
+ "does not exist.", cacert)
+
+ return session.Session(auth=auth, verify=cacert)
# *********************************************
return glanceclient.Client(get_glance_client_version(), session=sess)
+def get_heat_client_version():
+ api_version = os.getenv('OS_ORCHESTRATION_API_VERSION')
+ if api_version is not None:
+ logger.info("OS_ORCHESTRATION_API_VERSION is set in env as '%s'",
+ api_version)
+ return api_version
+ return DEFAULT_HEAT_API_VERSION
+
+
+def get_heat_client(other_creds={}):
+ sess = get_session(other_creds)
+ return heatclient.Client(get_heat_client_version(), session=sess)
+
+
# *********************************************
# NOVA
# *********************************************
def create_secgroup_rule(neutron_client, sg_id, direction, protocol,
port_range_min=None, port_range_max=None):
- if port_range_min is None and port_range_max is None:
- json_body = {'security_group_rule': {'direction': direction,
- 'security_group_id': sg_id,
- 'protocol': protocol}}
- elif port_range_min is not None and port_range_max is not None:
- json_body = {'security_group_rule': {'direction': direction,
- 'security_group_id': sg_id,
- 'port_range_min': port_range_min,
- 'port_range_max': port_range_max,
- 'protocol': protocol}}
+ # We create a security group in 2 steps
+ # 1 - we check the format and set the json body accordingly
+ # 2 - we call neturon client to create the security group
+
+ # Format check
+ json_body = {'security_group_rule': {'direction': direction,
+ 'security_group_id': sg_id,
+ 'protocol': protocol}}
+ # parameters may be
+ # - both None => we do nothing
+ # - both Not None => we add them to the json description
+ # but one cannot be None is the other is not None
+ if (port_range_min is not None and port_range_max is not None):
+ # add port_range in json description
+ json_body['security_group_rule']['port_range_min'] = port_range_min
+ json_body['security_group_rule']['port_range_max'] = port_range_max
+ logger.debug("Security_group format set (port range included)")
else:
- logger.error("Error [create_secgroup_rule(neutron_client, '%s', '%s', "
- "'%s', '%s', '%s', '%s')]:" % (neutron_client,
- sg_id, direction,
- port_range_min,
- port_range_max,
- protocol),
- " Invalid values for port_range_min, port_range_max")
- return False
+ # either both port range are set to None => do nothing
+ # or one is set but not the other => log it and return False
+ if port_range_min is None and port_range_max is None:
+ logger.debug("Security_group format set (no port range mentioned)")
+ else:
+ logger.error("Bad security group format."
+ "One of the port range is not properly set:"
+ "range min: {},"
+ "range max: {}".format(port_range_min,
+ port_range_max))
+ return False
+
+ # Create security group using neutron client
try:
neutron_client.create_security_group_rule(json_body)
return True
- except Exception, e:
- logger.error("Error [create_secgroup_rule(neutron_client, '%s', '%s', "
- "'%s', '%s', '%s', '%s')]: %s" % (neutron_client,
- sg_id,
- direction,
- port_range_min,
- port_range_max,
- protocol, e))
+ except:
+ logger.exception("Impossible to create_security_group_rule,"
+ "security group rule probably already exists")
return False
+def get_security_group_rules(neutron_client, sg_id):
+ try:
+ security_rules = neutron_client.list_security_group_rules()[
+ 'security_group_rules']
+ security_rules = [rule for rule in security_rules
+ if rule["security_group_id"] == sg_id]
+ return security_rules
+ except Exception, e:
+ logger.error("Error [get_security_group_rules(neutron_client, sg_id)]:"
+ " %s" % e)
+ return None
+
+
+def check_security_group_rules(neutron_client, sg_id, direction, protocol,
+ port_min=None, port_max=None):
+ try:
+ security_rules = get_security_group_rules(neutron_client, sg_id)
+ security_rules = [rule for rule in security_rules
+ if (rule["direction"].lower() == direction
+ and rule["protocol"].lower() == protocol
+ and rule["port_range_min"] == port_min
+ and rule["port_range_max"] == port_max)]
+ if len(security_rules) == 0:
+ return True
+ else:
+ return False
+ except Exception, e:
+ logger.error("Error [check_security_group_rules("
+ " neutron_client, sg_id, direction,"
+ " protocol, port_min=None, port_max=None)]: "
+ "%s" % e)
+ return None
+
+
def create_security_group_full(neutron_client,
sg_name, sg_description):
sg_id = get_security_group_id(neutron_client, sg_name)
return None
+def get_or_create_tenant(keystone_client, tenant_name, tenant_description):
+ tenant_id = get_tenant_id(keystone_client, tenant_name)
+ if not tenant_id:
+ tenant_id = create_tenant(keystone_client, tenant_name,
+ tenant_description)
+
+ return tenant_id
+
+
def create_user(keystone_client, user_name, user_password,
user_email, tenant_id):
try:
return None
+def get_or_create_user(keystone_client, user_name, user_password,
+ tenant_id, user_email=None):
+ user_id = get_user_id(keystone_client, user_name)
+ if not user_id:
+ user_id = create_user(keystone_client, user_name, user_password,
+ user_email, tenant_id)
+ return user_id
+
+
def add_role_user(keystone_client, user_id, role_id, tenant_id):
try:
if is_keystone_v3():
logger.error("Error [delete_user(keystone_client, '%s')]: %s"
% (user_id, e))
return False
+
+
+# *********************************************
+# HEAT
+# *********************************************
+def get_resource(heat_client, stack_id, resource):
+ try:
+ resources = heat_client.resources.get(stack_id, resource)
+ return resources
+ except Exception, e:
+ logger.error("Error [get_resource]: %s" % e)
+ return None
Code Review / functest.git / blobdiff