ice_validator/tests/utils/nested_iterables.py

   1 # -*- coding: utf8 -*-
   2 # ============LICENSE_START=======================================================
   3 # org.onap.vvp/validation-scripts
   4 # ===================================================================
   5 # Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6 # ===================================================================
   7 #
   8 # Unless otherwise specified, all software contained herein is licensed
   9 # under the Apache License, Version 2.0 (the “License”);
  10 # you may not use this software except in compliance with the License.
  11 # You may obtain a copy of the License at
  12 #
  13 #             http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS,
  17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18 # See the License for the specific language governing permissions and
  19 # limitations under the License.
  20 #
  21 #
  22 #
  23 # Unless otherwise specified, all documentation contained herein is licensed
  24 # under the Creative Commons License, Attribution 4.0 Intl. (the “License”);
  25 # you may not use this documentation except in compliance with the License.
  26 # You may obtain a copy of the License at
  27 #
  28 #             https://creativecommons.org/licenses/by/4.0/
  29 #
  30 # Unless required by applicable law or agreed to in writing, documentation
  31 # distributed under the License is distributed on an "AS IS" BASIS,
  32 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  33 # See the License for the specific language governing permissions and
  34 # limitations under the License.
  35 #
  36 # ============LICENSE_END============================================
  37 #
  38 # ECOMP is a trademark and service mark of AT&T Intellectual Property.
  39 #
  40
  41
  42 def parse_nested_dict(d, key=""):
  43     '''
  44     parse the nested dictionary and return values of
  45     given key of function parameter only
  46     '''
  47     nested_elements = []
  48     for k, v in d.items():
  49         if isinstance(v, dict):
  50             sub_dict = parse_nested_dict(v, key)
  51             nested_elements.extend(sub_dict)
  52         else:
  53             if key:
  54                 if k == key:
  55                     nested_elements.append(v)
  56             else:
  57                 nested_elements.append(v)
  58
  59     return nested_elements
  60
  61
  62 def find_all_get_param_in_yml(yml):
  63     '''
  64     Recursively find all referenced parameters in a parsed yaml body
  65     and return a list of parameters
  66     '''
  67     os_pseudo_parameters = ['OS::stack_name',
  68                             'OS::stack_id',
  69                             'OS::project_id']
  70
  71     if not hasattr(yml, 'items'):
  72         return []
  73     params = []
  74     for k, v in yml.items():
  75         if k == 'get_param' and v not in os_pseudo_parameters:
  76             for item in (v if isinstance(v, list) else [v]):
  77                 if isinstance(item, dict):
  78                     params.extend(find_all_get_param_in_yml(item))
  79                 elif isinstance(item, str):
  80                     params.append(item)
  81             continue
  82         elif k == 'list_join':
  83             for item in (v if isinstance(v, list) else [v]):
  84                 if isinstance(item, list):
  85                     for d in item:
  86                         params.extend(find_all_get_param_in_yml(d))
  87             continue
  88         if isinstance(v, dict):
  89             params.extend(find_all_get_param_in_yml(v))
  90         elif isinstance(v, list):
  91             for d in v:
  92                 params.extend(find_all_get_param_in_yml(d))
  93     return params
  94
  95
  96 def find_all_get_resource_in_yml(yml):
  97     '''
  98     Recursively find all referenced resources
  99     in a parsed yaml body and return a list of resource ids
 100     '''
 101     if not hasattr(yml, 'items'):
 102         return []
 103     resources = []
 104     for k, v in yml.items():
 105         if k == 'get_resource':
 106             if isinstance(v, list):
 107                 resources.append(v[0])
 108             else:
 109                 resources.append(v)
 110             continue
 111         if isinstance(v, dict):
 112             resources.extend(find_all_get_resource_in_yml(v))
 113         elif isinstance(v, list):
 114             for d in v:
 115                 resources.extend(find_all_get_resource_in_yml(d))
 116     return resources
 117
 118
 119 def find_all_get_file_in_yml(yml):
 120     '''
 121     Recursively find all get_file in a parsed yaml body
 122     and return the list of referenced files/urls
 123     '''
 124     if not hasattr(yml, 'items'):
 125         return []
 126     resources = []
 127     for k, v in yml.items():
 128         if k == 'get_file':
 129             if isinstance(v, list):
 130                 resources.append(v[0])
 131             else:
 132                 resources.append(v)
 133             continue
 134         if isinstance(v, dict):
 135             resources.extend(find_all_get_file_in_yml(v))
 136         elif isinstance(v, list):
 137             for d in v:
 138                 resources.extend(find_all_get_file_in_yml(d))
 139     return resources
 140
 141
 142 def find_all_get_resource_in_resource(resource):
 143     '''
 144     Recursively find all referenced resources
 145     in a heat resource and return a list of resource ids
 146     '''
 147     if not hasattr(resource, 'items'):
 148         return []
 149
 150     resources = []
 151     for k, v in resource.items():
 152         if k == 'get_resource':
 153             if isinstance(v, list):
 154                 resources.append(v[0])
 155             else:
 156                 resources.append(v)
 157             continue
 158         if isinstance(v, dict):
 159             resources.extend(
 160                 find_all_get_resource_in_resource(v))
 161         elif isinstance(v, list):
 162             for d in v:
 163                 resources.extend(
 164                     find_all_get_resource_in_resource(d))
 165     return resources
 166
 167
 168 def get_associated_resources_per_resource(resources):
 169     '''
 170     Recursively find all referenced resources for each resource
 171     in a list of resource ids
 172     '''
 173     if not hasattr(resources, 'items'):
 174         return None
 175
 176     resources_dict = {}
 177     resources_dict["resources"] = {}
 178     ref_resources = []
 179
 180     for res_key, res_value in resources.items():
 181         get_resources = []
 182
 183         for k, v in res_value:
 184             if k == 'get_resource' and\
 185                isinstance(v, dict):
 186                 get_resources = find_all_get_resource_in_resource(v)
 187
 188         # if resources found, add to dict
 189         if get_resources:
 190             ref_resources.extend(get_resources)
 191             resources_dict["resources"][res_key] = {
 192                 "res_value": res_value,
 193                 "get_resources": get_resources,
 194             }
 195
 196     resources_dict["ref_resources"] = set(ref_resources)
 197
 198     return resources_dict
 199
 200
 201 def flatten(items):
 202     '''
 203     flatten items from any nested iterable
 204     '''
 205
 206     merged_list = []
 207     for item in items:
 208         if isinstance(item, list):
 209             sub_list = flatten(item)
 210             merged_list.extend(sub_list)
 211         else:
 212             merged_list.append(item)
 213     return merged_list