Coverage for plugins/module_utils/utils.py : 19%

# -*- coding: utf-8 -*- 

# Copyright: (c) 2021, XLAB Steampunk <steampunk@xlab.si> 

# 

# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) 

from __future__ import absolute_import, division, print_function 

__metaclass__ = type 

def filter_dict(input, *field_names): 

output = {} 

for field_name in field_names: 

if field_name in input: 

value = input[field_name] 

if value is not None: 

output[field_name] = value 

return output 

def is_superset(superset, candidate): 

for k, v in candidate.items(): 

if k not in superset or superset[k] != v: 

return False 

return True 

def get_choices(module, mapping_field, default_payload_fields_mapping): 

if mapping_field not in module.params: 

return default_payload_fields_mapping 

if module.params[mapping_field] is None: 

return default_payload_fields_mapping 

overrides = module.params[mapping_field] 

clone = {} 

for key, item in default_payload_fields_mapping.items(): 

override = overrides.get(key) 

clone[key] = override if override else dict(item) 

return clone 

def get_mapper( 

module, 

mapping_field, 

default_payload_fields_mapping, 

sysparm_display_value="false", 

): 

choices = get_choices(module, mapping_field, default_payload_fields_mapping) 

mapper = PayloadMapper(choices, module.warn, sysparm_display_value) 

return mapper 

class PayloadMapper: 

def __init__( 

self, mapping, unknown_value_handler=None, sysparm_display_value="false" 

): 

# Convert 

# dict(a=[("s1", "a1"), ("s2", "a2")], b=[("s3", "a3")]) 

# to 

# _to_ansible == dict(a=dict(s1="a1", s2="a2"), b=dict(s3="a3")) 

# _to_snow == dict(a=dict(a1="s1", a2="s2"), b=dict(a3="s3")) 

# 

# This allows efficient transformation of our internal values to ServiceNow 

# equivalents. 

self._to_ansible = {} 

self._to_snow = {} 

self.unknown_value_handler = unknown_value_handler 

self.sysparm_display_value = sysparm_display_value 

for key, value_map in mapping.items(): 

if isinstance(value_map, dict): 

self._to_ansible[key] = value_map 

self._to_snow[key] = dict( 

(ansible_val, snow_val) 

for snow_val, ansible_val in value_map.items() 

) 

else: 

self._to_ansible[key] = dict(value_map) 

self._to_snow[key] = dict( 

(ansible_val, snow_val) for snow_val, ansible_val in value_map 

) 

def _map_key(self, key, val, mapping): 

if val in mapping[key]: 

return mapping[key][val] 

if self.unknown_value_handler: 

self.unknown_value_handler( 

"Encountered unknown value {0} while mapping field {1}.".format( 

val, key 

) 

) 

return val 

def _transform(self, mapping, data): 

result = {} 

for k, v in data.items(): 

if v is not None and k in mapping: 

result[k] = self._map_key(k, v, mapping) 

else: 

result[k] = v 

return result 

def to_ansible(self, snow_data): 

# If sysparm_display_value is set to "true" or "all", no need to transform as field display values are already 

# returned. 

if self.sysparm_display_value != "false": # is equal to either "all" or "true" 

return snow_data 

return self._transform(self._to_ansible, snow_data) 

def to_snow(self, ansible_data): 

# If sysparm_display_value is set to "true" or "all", no need to transform as field display values are already 

# returned. 

if self.sysparm_display_value != "false": # is equal to either "all" or "true" 

return ansible_data 

return self._transform(self._to_snow, ansible_data)