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         if isinstance(v, dict):
  83             params.extend(find_all_get_param_in_yml(v))
  84         elif isinstance(v, list):
  85             for d in v:
  86                 params.extend(find_all_get_param_in_yml(d))
  87     return params
  88
  89
  90 def find_all_get_resource_in_yml(yml):
  91     '''
  92     Recursively find all referenced resources
  93     in a parsed yaml body and return a list of resource ids
  94     '''
  95     if not hasattr(yml, 'items'):
  96         return []
  97     resources = []
  98     for k, v in yml.items():
  99         if k == 'get_resource':
 100             if isinstance(v, list):
 101                 resources.append(v[0])
 102             else:
 103                 resources.append(v)
 104             continue
 105         if isinstance(v, dict):
 106             resources.extend(find_all_get_resource_in_yml(v))
 107         elif isinstance(v, list):
 108             for d in v:
 109                 resources.extend(find_all_get_resource_in_yml(d))
 110     return resources
 111
 112
 113 def find_all_get_file_in_yml(yml):
 114     '''
 115     Recursively find all get_file in a parsed yaml body
 116     and return the list of referenced files/urls
 117     '''
 118     if not hasattr(yml, 'items'):
 119         return []
 120     resources = []
 121     for k, v in yml.items():
 122         if k == 'get_file':
 123             if isinstance(v, list):
 124                 resources.append(v[0])
 125             else:
 126                 resources.append(v)
 127             continue
 128         if isinstance(v, dict):
 129             resources.extend(find_all_get_file_in_yml(v))
 130         elif isinstance(v, list):
 131             for d in v:
 132                 resources.extend(find_all_get_file_in_yml(d))
 133     return resources
 134
 135
 136 def find_all_get_resource_in_resource(resource):
 137     '''
 138     Recursively find all referenced resources
 139     in a heat resource and return a list of resource ids
 140     '''
 141     if not hasattr(resource, 'items'):
 142         return []
 143
 144     resources = []
 145     for k, v in resource.items():
 146         if k == 'get_resource':
 147             if isinstance(v, list):
 148                 resources.append(v[0])
 149             else:
 150                 resources.append(v)
 151             continue
 152         if isinstance(v, dict):
 153             resources.extend(
 154                 find_all_get_resource_in_resource(v))
 155         elif isinstance(v, list):
 156             for d in v:
 157                 resources.extend(
 158                     find_all_get_resource_in_resource(d))
 159     return resources
 160
 161
 162 def get_associated_resources_per_resource(resources):
 163     '''
 164     Recursively find all referenced resources for each resource
 165     in a list of resource ids
 166     '''
 167     if not hasattr(resources, 'items'):
 168         return None
 169
 170     resources_dict = {}
 171     resources_dict["resources"] = {}
 172     ref_resources = []
 173
 174     for res_key, res_value in resources.items():
 175         get_resources = []
 176
 177         for k, v in res_value:
 178             if k == 'get_resource' and\
 179                isinstance(v, dict):
 180                 get_resources = find_all_get_resource_in_resource(v)
 181
 182         # if resources found, add to dict
 183         if get_resources:
 184             ref_resources.extend(get_resources)
 185             resources_dict["resources"][res_key] = {
 186                 "res_value": res_value,
 187                 "get_resources": get_resources,
 188             }
 189
 190     resources_dict["ref_resources"] = set(ref_resources)
 191
 192     return resources_dict
 193
 194
 195 def flatten(items):
 196     '''
 197     flatten items from any nested iterable
 198     '''
 199
 200     merged_list = []
 201     for item in items:
 202         if isinstance(item, list):
 203             sub_list = flatten(item)
 204             merged_list.extend(sub_list)
 205         else:
 206             merged_list.append(item)
 207     return merged_list