tripleo-common/tripleo_common/utils/roles.py

126 lines
4.1 KiB
Python

# Copyright 2017 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import os
import shutil
import yaml
import six
from six.moves import cStringIO as StringIO
from tripleo_common.exception import NotFound
from tripleo_common.exception import RoleMetadataError
def get_roles_list_from_directory(directory):
"""Return array of roles in roles path"""
if not os.path.exists(directory):
raise ValueError("Invalid roles path specified: {}".format(
directory))
roles = []
for f in os.listdir(directory):
if f.endswith(".yaml"):
roles.append(f[:-5])
roles.sort()
return roles
def check_role_exists(available_roles, requested_roles):
"""Performs a check on the requested roles to ensure they exist
:param available_roles list of available role names
:param requested_roles list of requested role names
:exception NotFound if a role in the requested list is not available
"""
role_check = set(requested_roles) - set(available_roles)
if len(role_check) > 0:
msg = "Invalid roles requested: {}\nValid Roles:\n{}".format(
','.join(role_check), '\n'.join(available_roles)
)
raise NotFound(msg)
def generate_roles_data_from_directory(directory, roles, validate=True):
"""Generate a roles data file using roles from a local path
:param directory local filesystem path to the roles
:param roles ordered list of roles
:param validate validate the metadata format in the role yaml files
:returns string contents of the roles_data.yaml
"""
available_roles = get_roles_list_from_directory(directory)
check_role_exists(available_roles, roles)
output = StringIO()
header = ["#" * 79,
"# File generated by TripleO",
"#" * 79,
""]
output.write("\n".join(header))
for role in roles:
file_path = os.path.join(directory, "{}.yaml".format(role))
if validate:
validate_role_yaml(role_path=file_path)
with open(file_path, "r") as f:
shutil.copyfileobj(f, output)
return output.getvalue()
def validate_role_yaml(role_data=None, role_path=None):
"""Basic role yaml validation
:param role_data the role yaml in string form
:param role_path the path to the yaml file to validate.
:exception RoleMetadataError
:returns parsed role yaml object
"""
if role_data and role_path or (not role_data and not role_path):
raise ValueError('Either role_data OR role_path must be specified')
if role_path:
with open(role_path, 'r') as f:
role_data = f.read()
try:
role = yaml.safe_load(role_data)[0]
except yaml.YAMLError:
raise RoleMetadataError('Unable to parse role yaml')
schema = {
'name': {'type': six.string_types},
'CountDefault': {'type': int},
'HostnameFormatDefault': {'type': six.string_types},
'disable_constraints': {'type': bool},
'upgrade_batch_size': {'type': int},
'ServicesDefault': {'type': list},
'tags': {'type': list},
'description': {'type': six.string_types},
'networks': {'type': list},
'networks_skip_config': {'type': list},
}
if 'name' not in role:
raise RoleMetadataError('Role name is missing from the role')
# validate numeric metadata is numeric
for k in schema:
if k in role and not isinstance(role[k], schema[k]['type']):
msg = "Role '{}': {} is not of expected type {}".format(
role['name'], k, schema[k]['type'])
raise RoleMetadataError(msg)
return role