def get_netloc(url):
"""Return the netloc from a URL.
If the input value is not a value URL the method will raise an Ansible
filter exception.
:param url: the URL to parse
:type url: ``str``
:returns: ``str``
"""
try:
netloc = urlparse(url).netloc
except Exception as exp:
raise errors.AnsibleFilterError(
'Failed to return the netloc of: "%s"' % str(exp)
)
else:
return netloc
python类AnsibleFilterError()的实例源码
def get_netorigin(url):
"""Return the netloc from a URL.
If the input value is not a value URL the method will raise an Ansible
filter exception.
:param url: the URL to parse
:type url: ``str``
:returns: ``str``
"""
try:
parsed_url = urlparse(url)
netloc = parsed_url.netloc
scheme = parsed_url.scheme
except Exception as exp:
raise errors.AnsibleFilterError(
'Failed to return the netorigin of: "%s"' % str(exp)
)
else:
return '%s://%s' % (scheme, netloc)
def extract_by_prefix(values, prefix):
"""
This function takes a list as 'values' parameter and extract a first value from the list which contains prefix.
The prefix is defined in parameter 'prefix'
"""
if values is None:
raise errors.AnsibleFilterError('Values is not provided')
if prefix is None:
raise errors.AnsibleFilterError('Prefix is not provided')
if not isinstance(values, list):
raise errors.AnsibleFilterError('Wrong type for values')
if not isinstance(prefix, str):
raise errors.AnsibleFilterError('Wrong type for the prefix')
if len(values) == 0:
raise errors.AnsibleFilterError('Empty list. Nothing to extract')
for v in values:
if v.startswith(prefix):
return v
raise errors.AnsibleFilterError('Value not found')
def filter_by_prefix(values, prefix):
"""
This function takes a list as 'values' parameter and filters out all list values which contain prefix.
The prefix is defined in parameter 'prefix'
"""
if values is None:
raise errors.AnsibleFilterError('Values is not provided')
if prefix is None:
raise errors.AnsibleFilterError('Prefix is not provided')
if not isinstance(values, list):
raise errors.AnsibleFilterError('Wrong type for values')
if not isinstance(prefix, str):
raise errors.AnsibleFilterError('Wrong type for the prefix')
return filter(lambda x: x.startswith(prefix), values)
def log(value, base):
"""
This function returns the logarithm of 'value' to the given base 'base'
"""
if value is None:
raise errors.AnsibleFilterError('value is not provided')
if base is None:
raise errors.AnsibleFilterError('base is not provided')
if not isinstance(value, int):
raise errors.AnsibleFilterError('Wrong type for value')
if not isinstance(base, int):
raise errors.AnsibleFilterError('Wrong type for base')
return math.log(value, base)
def _net_interface_type(context, name, inventory_hostname):
"""Return a string describing the network interface type.
Possible types include 'ether', 'bridge', 'bond'.
"""
bridge_ports = net_bridge_ports(context, name, inventory_hostname)
bond_slaves = net_bond_slaves(context, name, inventory_hostname)
if bridge_ports is not None and bond_slaves is not None:
raise errors.AnsibleFilterError(
"Network %s on host %s has both bridge ports and bond slaves "
"defined" %
(name,
_get_hostvar(context, 'inventory_hostname', inventory_hostname)))
if bridge_ports is None and bond_slaves is None:
return 'ether'
if bridge_ports is not None:
return 'bridge'
if bond_slaves is not None:
return 'bond'
def previous_nth_usable(value, offset):
try:
vtype = ipaddr(value, 'type')
if vtype == 'address':
v = ipaddr(value, 'cidr')
elif vtype == 'network':
v = ipaddr(value, 'subnet')
v = netaddr.IPNetwork(v)
except:
return False
if type(offset) != int:
raise errors.AnsibleFilterError('Must pass in an interger')
if v.size > 1:
first_usable, last_usable = _first_last(v)
nth_ip = int(netaddr.IPAddress(int(v.ip) - offset))
if nth_ip >= first_usable and nth_ip <= last_usable:
return str(netaddr.IPAddress(int(v.ip) - offset))
def rand(environment, end, start=None, step=None, seed=None):
if seed is None:
r = SystemRandom()
else:
r = Random(seed)
if isinstance(end, integer_types):
if not start:
start = 0
if not step:
step = 1
return r.randrange(start, end, step)
elif hasattr(end, '__iter__'):
if start or step:
raise errors.AnsibleFilterError('start and step can only be used with integer values')
return r.choice(end)
else:
raise errors.AnsibleFilterError('random can only be used on sequences and integers')
def changed(result):
''' Test if task result yields changed '''
if not isinstance(result, MutableMapping):
raise errors.AnsibleFilterError("|changed expects a dictionary")
if 'changed' not in result:
changed = False
if (
'results' in result and # some modules return a 'results' key
isinstance(result['results'], MutableSequence) and
isinstance(result['results'][0], MutableMapping)
):
for res in result['results']:
if res.get('changed', False):
changed = True
break
else:
changed = result.get('changed', False)
return changed
def get_attr(data, attribute=None):
""" This looks up dictionary attributes of the form a.b.c and returns
the value.
If the key isn't present, None is returned.
Ex: data = {'a': {'b': {'c': 5}}}
attribute = "a.b.c"
returns 5
"""
if not attribute:
raise errors.AnsibleFilterError("|failed expects attribute to be set")
ptr = data
for attr in attribute.split('.'):
if attr in ptr:
ptr = ptr[attr]
else:
ptr = None
break
return ptr
def oo_select_keys_from_list(data, keys):
""" This returns a list, which contains the value portions for the keys
Ex: data = { 'a':1, 'b':2, 'c':3 }
keys = ['a', 'c']
returns [1, 3]
"""
if not isinstance(data, list):
raise errors.AnsibleFilterError("|failed expects to filter on a list")
if not isinstance(keys, list):
raise errors.AnsibleFilterError("|failed expects first param is a list")
# Gather up the values for the list of keys passed in
retval = [FilterModule.oo_select_keys(item, keys) for item in data]
return FilterModule.oo_flatten(retval)
def oo_select_keys(data, keys):
""" This returns a list, which contains the value portions for the keys
Ex: data = { 'a':1, 'b':2, 'c':3 }
keys = ['a', 'c']
returns [1, 3]
"""
if not isinstance(data, Mapping):
raise errors.AnsibleFilterError("|failed expects to filter on a dict or object")
if not isinstance(keys, list):
raise errors.AnsibleFilterError("|failed expects first param is a list")
# Gather up the values for the list of keys passed in
retval = [data[key] for key in keys if key in data]
return retval
def oo_filter_list(data, filter_attr=None):
""" This returns a list, which contains all items where filter_attr
evaluates to true
Ex: data = [ { a: 1, b: True },
{ a: 3, b: False },
{ a: 5, b: True } ]
filter_attr = 'b'
returns [ { a: 1, b: True },
{ a: 5, b: True } ]
"""
if not isinstance(data, list):
raise errors.AnsibleFilterError("|failed expects to filter on a list")
if not isinstance(filter_attr, basestring):
raise errors.AnsibleFilterError("|failed expects filter_attr is a str or unicode")
# Gather up the values for the list of keys passed in
return [x for x in data if filter_attr in x and x[filter_attr]]
def oo_31_rpm_rename_conversion(rpms, openshift_version=None):
""" Filters a list of 3.0 rpms and return the corresponding 3.1 rpms
names with proper version (if provided)
If 3.1 rpms are passed in they will only be augmented with the
correct version. This is important for hosts that are running both
Masters and Nodes.
"""
if not isinstance(rpms, list):
raise errors.AnsibleFilterError("failed expects to filter on a list")
if openshift_version is not None and not isinstance(openshift_version, basestring):
raise errors.AnsibleFilterError("failed expects openshift_version to be a string")
rpms_31 = []
for rpm in rpms:
if not 'atomic' in rpm:
rpm = rpm.replace("openshift", "atomic-openshift")
if openshift_version:
rpm = rpm + openshift_version
rpms_31.append(rpm)
return rpms_31
def oo_pods_match_component(pods, deployment_type, component):
""" Filters a list of Pods and returns the ones matching the deployment_type and component
"""
if not isinstance(pods, list):
raise errors.AnsibleFilterError("failed expects to filter on a list")
if not isinstance(deployment_type, basestring):
raise errors.AnsibleFilterError("failed expects deployment_type to be a string")
if not isinstance(component, basestring):
raise errors.AnsibleFilterError("failed expects component to be a string")
image_prefix = 'openshift/origin-'
if deployment_type in ['enterprise', 'online', 'openshift-enterprise']:
image_prefix = 'openshift3/ose-'
elif deployment_type == 'atomic-enterprise':
image_prefix = 'aep3_beta/aep-'
matching_pods = []
image_regex = image_prefix + component + r'.*'
for pod in pods:
for container in pod['spec']['containers']:
if re.search(image_regex, container['image']):
matching_pods.append(pod)
break # stop here, don't add a pod more than once
return matching_pods
def oo_image_tag_to_rpm_version(version, include_dash=False):
""" Convert an image tag string to an RPM version if necessary
Empty strings and strings that are already in rpm version format
are ignored. Also remove non semantic version components.
Ex. v3.2.0.10 -> -3.2.0.10
v1.2.0-rc1 -> -1.2.0
"""
if not isinstance(version, basestring):
raise errors.AnsibleFilterError("|failed expects a string or unicode")
if version.startswith("v"):
version = version[1:]
# Strip release from requested version, we no longer support this.
version = version.split('-')[0]
if include_dash and version and not version.startswith("-"):
version = "-" + version
return version
def validate_idp_list(idp_list, openshift_version, deployment_type):
''' validates a list of idps '''
login_providers = [x.name for x in idp_list if x.login]
multiple_logins_unsupported = False
if len(login_providers) > 1:
if deployment_type in ['enterprise', 'online', 'atomic-enterprise', 'openshift-enterprise']:
if LooseVersion(openshift_version) < LooseVersion('3.2'):
multiple_logins_unsupported = True
if deployment_type in ['origin']:
if LooseVersion(openshift_version) < LooseVersion('1.2'):
multiple_logins_unsupported = True
if multiple_logins_unsupported:
raise errors.AnsibleFilterError("|failed multiple providers are "
"not allowed for login. login "
"providers: {0}".format(', '.join(login_providers)))
names = [x.name for x in idp_list]
if len(set(names)) != len(names):
raise errors.AnsibleFilterError("|failed more than one provider configured with the same name")
for idp in idp_list:
idp.validate()
def validate(self):
''' validate this idp instance '''
IdentityProviderBase.validate(self)
if not isinstance(self.provider['attributes'], dict):
raise errors.AnsibleFilterError("|failed attributes for provider "
"{0} must be a dictionary".format(self.__class__.__name__))
attrs = ['id', 'email', 'name', 'preferredUsername']
for attr in attrs:
if attr in self.provider['attributes'] and not isinstance(self.provider['attributes'][attr], list):
raise errors.AnsibleFilterError("|failed {0} attribute for "
"provider {1} must be a list".format(attr, self.__class__.__name__))
unknown_attrs = set(self.provider['attributes'].keys()) - set(attrs)
if len(unknown_attrs) > 0:
raise errors.AnsibleFilterError("|failed provider {0} has unknown "
"attributes: {1}".format(self.__class__.__name__, ', '.join(unknown_attrs)))
def translate_idps(idps, api_version, openshift_version, deployment_type):
''' Translates a list of dictionaries into a valid identityProviders config '''
idp_list = []
if not isinstance(idps, list):
raise errors.AnsibleFilterError("|failed expects to filter on a list of identity providers")
for idp in idps:
if not isinstance(idp, dict):
raise errors.AnsibleFilterError("|failed identity providers must be a list of dictionaries")
cur_module = sys.modules[__name__]
idp_class = getattr(cur_module, idp['kind'], None)
idp_inst = idp_class(api_version, idp) if idp_class is not None else IdentityProviderBase(api_version, idp)
idp_inst.set_provider_items()
idp_list.append(idp_inst)
IdentityProviderBase.validate_idp_list(idp_list, openshift_version, deployment_type)
return yaml.safe_dump([idp.to_dict() for idp in idp_list], default_flow_style=False)
def validate_pcs_cluster(data, masters=None):
''' Validates output from "pcs status", ensuring that each master
provided is online.
Ex: data = ('...',
'PCSD Status:',
'master1.example.com: Online',
'master2.example.com: Online',
'master3.example.com: Online',
'...')
masters = ['master1.example.com',
'master2.example.com',
'master3.example.com']
returns True
'''
if not issubclass(type(data), basestring):
raise errors.AnsibleFilterError("|failed expects data is a string or unicode")
if not issubclass(type(masters), list):
raise errors.AnsibleFilterError("|failed expects masters is a list")
valid = True
for master in masters:
if "{0}: Online".format(master) not in data:
valid = False
return valid
def oo_htpasswd_users_from_file(file_contents):
''' return a dictionary of htpasswd users from htpasswd file contents '''
htpasswd_entries = {}
if not isinstance(file_contents, basestring):
raise errors.AnsibleFilterError("failed, expects to filter on a string")
for line in file_contents.splitlines():
user = None
passwd = None
if len(line) == 0:
continue
if ':' in line:
user, passwd = line.split(':', 1)
if user is None or len(user) == 0 or passwd is None or len(passwd) == 0:
error_msg = "failed, expects each line to be a colon separated string representing the user and passwd"
raise errors.AnsibleFilterError(error_msg)
htpasswd_entries[user] = passwd
return htpasswd_entries
def get_attr(data, attribute=None):
""" This looks up dictionary attributes of the form a.b.c and returns
the value.
If the key isn't present, None is returned.
Ex: data = {'a': {'b': {'c': 5}}}
attribute = "a.b.c"
returns 5
"""
if not attribute:
raise errors.AnsibleFilterError("|failed expects attribute to be set")
ptr = data
for attr in attribute.split('.'):
if attr in ptr:
ptr = ptr[attr]
else:
ptr = None
break
return ptr
def oo_select_keys(data, keys):
""" This returns a list, which contains the value portions for the keys
Ex: data = { 'a':1, 'b':2, 'c':3 }
keys = ['a', 'c']
returns [1, 3]
"""
if not isinstance(data, Mapping):
raise errors.AnsibleFilterError("|failed expects to filter on a dict or object")
if not isinstance(keys, list):
raise errors.AnsibleFilterError("|failed expects first param is a list")
# Gather up the values for the list of keys passed in
retval = [data[key] for key in keys if key in data]
return retval
def oo_ami_selector(data, image_name):
""" This takes a list of amis and an image name and attempts to return
the latest ami.
"""
if not isinstance(data, list):
raise errors.AnsibleFilterError("|failed expects first param is a list")
if not data:
return None
else:
if image_name is None or not image_name.endswith('_*'):
ami = sorted(data, key=itemgetter('name'), reverse=True)[0]
return ami['ami_id']
else:
ami_info = [(ami, ami['name'].split('_')[-1]) for ami in data]
ami = sorted(ami_info, key=itemgetter(1), reverse=True)[0][0]
return ami['ami_id']
def oo_filter_list(data, filter_attr=None):
""" This returns a list, which contains all items where filter_attr
evaluates to true
Ex: data = [ { a: 1, b: True },
{ a: 3, b: False },
{ a: 5, b: True } ]
filter_attr = 'b'
returns [ { a: 1, b: True },
{ a: 5, b: True } ]
"""
if not isinstance(data, list):
raise errors.AnsibleFilterError("|failed expects to filter on a list")
if not isinstance(filter_attr, basestring):
raise errors.AnsibleFilterError("|failed expects filter_attr is a str or unicode")
# Gather up the values for the list of keys passed in
return [x for x in data if filter_attr in x and x[filter_attr]]
def oo_openshift_env(hostvars):
''' Return facts which begin with "openshift_" and translate
legacy facts to their openshift_env counterparts.
Ex: hostvars = {'openshift_fact': 42,
'theyre_taking_the_hobbits_to': 'isengard'}
returns = {'openshift_fact': 42}
'''
if not issubclass(type(hostvars), dict):
raise errors.AnsibleFilterError("|failed expects hostvars is a dict")
facts = {}
regex = re.compile('^openshift_.*')
for key in hostvars:
if regex.match(key):
facts[key] = hostvars[key]
migrations = {'openshift_router_selector': 'openshift_hosted_router_selector',
'openshift_registry_selector': 'openshift_hosted_registry_selector'}
for old_fact, new_fact in migrations.iteritems():
if old_fact in facts and new_fact not in facts:
facts[new_fact] = facts[old_fact]
return facts
def oo_31_rpm_rename_conversion(rpms, openshift_version=None):
""" Filters a list of 3.0 rpms and return the corresponding 3.1 rpms
names with proper version (if provided)
If 3.1 rpms are passed in they will only be augmented with the
correct version. This is important for hosts that are running both
Masters and Nodes.
"""
if not isinstance(rpms, list):
raise errors.AnsibleFilterError("failed expects to filter on a list")
if openshift_version is not None and not isinstance(openshift_version, basestring):
raise errors.AnsibleFilterError("failed expects openshift_version to be a string")
rpms_31 = []
for rpm in rpms:
if not 'atomic' in rpm:
rpm = rpm.replace("openshift", "atomic-openshift")
if openshift_version:
rpm = rpm + openshift_version
rpms_31.append(rpm)
return rpms_31
def oo_pods_match_component(pods, deployment_type, component):
""" Filters a list of Pods and returns the ones matching the deployment_type and component
"""
if not isinstance(pods, list):
raise errors.AnsibleFilterError("failed expects to filter on a list")
if not isinstance(deployment_type, basestring):
raise errors.AnsibleFilterError("failed expects deployment_type to be a string")
if not isinstance(component, basestring):
raise errors.AnsibleFilterError("failed expects component to be a string")
image_prefix = 'openshift/origin-'
if deployment_type in ['enterprise', 'online', 'openshift-enterprise']:
image_prefix = 'openshift3/ose-'
elif deployment_type == 'atomic-enterprise':
image_prefix = 'aep3_beta/aep-'
matching_pods = []
image_regex = image_prefix + component + r'.*'
for pod in pods:
for container in pod['spec']['containers']:
if re.search(image_regex, container['image']):
matching_pods.append(pod)
break # stop here, don't add a pod more than once
return matching_pods
def oo_image_tag_to_rpm_version(version, include_dash=False):
""" Convert an image tag string to an RPM version if necessary
Empty strings and strings that are already in rpm version format
are ignored. Also remove non semantic version components.
Ex. v3.2.0.10 -> -3.2.0.10
v1.2.0-rc1 -> -1.2.0
"""
if not isinstance(version, basestring):
raise errors.AnsibleFilterError("|failed expects a string or unicode")
if version.startswith("v"):
version = version[1:]
# Strip release from requested version, we no longer support this.
version = version.split('-')[0]
if include_dash and version and not version.startswith("-"):
version = "-" + version
return version
def validate(self):
''' validate this idp instance '''
IdentityProviderBase.validate(self)
if not isinstance(self.provider['attributes'], dict):
raise errors.AnsibleFilterError("|failed attributes for provider "
"{0} must be a dictionary".format(self.__class__.__name__))
attrs = ['id', 'email', 'name', 'preferredUsername']
for attr in attrs:
if attr in self.provider['attributes'] and not isinstance(self.provider['attributes'][attr], list):
raise errors.AnsibleFilterError("|failed {0} attribute for "
"provider {1} must be a list".format(attr, self.__class__.__name__))
unknown_attrs = set(self.provider['attributes'].keys()) - set(attrs)
if len(unknown_attrs) > 0:
raise errors.AnsibleFilterError("|failed provider {0} has unknown "
"attributes: {1}".format(self.__class__.__name__, ', '.join(unknown_attrs)))
