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 #
  39
  40
  41 def parse_nested_dict(d, key=""):
  42     """
  43     parse the nested dictionary and return values of
  44     given key of function parameter only
  45     """
  46     nested_elements = []
  47     for k, v in d.items():
  48         if isinstance(v, dict):
  49             sub_dict = parse_nested_dict(v, key)
  50             nested_elements.extend(sub_dict)
  51         else:
  52             if key:
  53                 if k == key:
  54                     nested_elements.append(v)
  55             else:
  56                 nested_elements.append(v)
  57
  58     return nested_elements
  59
  60
  61 def find_all_get_param_in_yml(yml):
  62     """
  63     Recursively find all referenced parameters in a parsed yaml body
  64     and return a list of parameters
  65     """
  66     os_pseudo_parameters = ["OS::stack_name", "OS::stack_id", "OS::project_id"]
  67
  68     if not hasattr(yml, "items"):
  69         return []
  70     params = []
  71     for k, v in yml.items():
  72         if k == "get_param" and v not in os_pseudo_parameters:
  73             if isinstance(v, list) and not isinstance(v[0], dict):
  74                 params.append(v[0])
  75             elif not isinstance(v, dict) and isinstance(v, str):
  76                 params.append(v)
  77             for item in v if isinstance(v, list) else [v]:
  78                 if isinstance(item, dict):
  79                     params.extend(find_all_get_param_in_yml(item))
  80             continue
  81         elif k == "list_join":
  82             for item in v if isinstance(v, list) else [v]:
  83                 if isinstance(item, list):
  84                     for d in item:
  85                         params.extend(find_all_get_param_in_yml(d))
  86             continue
  87         if isinstance(v, dict):
  88             params.extend(find_all_get_param_in_yml(v))
  89         elif isinstance(v, list):
  90             for d in v:
  91                 params.extend(find_all_get_param_in_yml(d))
  92     return params
  93
  94
  95 def find_all_get_resource_in_yml(yml):
  96     """
  97     Recursively find all referenced resources
  98     in a parsed yaml body and return a list of resource ids
  99     """
 100     if not hasattr(yml, "items"):
 101         return []
 102     resources = []
 103     for k, v in yml.items():
 104         if k == "get_resource":
 105             if isinstance(v, list):
 106                 resources.append(v[0])
 107             else:
 108                 resources.append(v)
 109             continue
 110         if isinstance(v, dict):
 111             resources.extend(find_all_get_resource_in_yml(v))
 112         elif isinstance(v, list):
 113             for d in v:
 114                 resources.extend(find_all_get_resource_in_yml(d))
 115     return resources
 116
 117
 118 def find_all_get_file_in_yml(yml):
 119     """
 120     Recursively find all get_file in a parsed yaml body
 121     and return the list of referenced files/urls
 122     """
 123     if not hasattr(yml, "items"):
 124         return []
 125     resources = []
 126     for k, v in yml.items():
 127         if k == "get_file":
 128             if isinstance(v, list):
 129                 resources.append(v[0])
 130             else:
 131                 resources.append(v)
 132             continue
 133         if isinstance(v, dict):
 134             resources.extend(find_all_get_file_in_yml(v))
 135         elif isinstance(v, list):
 136             for d in v:
 137                 resources.extend(find_all_get_file_in_yml(d))
 138     return resources
 139
 140
 141 def find_all_get_resource_in_resource(resource):
 142     """
 143     Recursively find all referenced resources
 144     in a heat resource and return a list of resource ids
 145     """
 146     if not hasattr(resource, "items"):
 147         return []
 148
 149     resources = []
 150     for k, v in resource.items():
 151         if k == "get_resource":
 152             if isinstance(v, list):
 153                 resources.append(v[0])
 154             else:
 155                 resources.append(v)
 156             continue
 157         if isinstance(v, dict):
 158             resources.extend(find_all_get_resource_in_resource(v))
 159         elif isinstance(v, list):
 160             for d in v:
 161                 resources.extend(find_all_get_resource_in_resource(d))
 162     return resources
 163
 164
 165 def get_associated_resources_per_resource(resources):
 166     """
 167     Recursively find all referenced resources for each resource
 168     in a list of resource ids
 169     """
 170     if not hasattr(resources, "items"):
 171         return None
 172
 173     resources_dict = {}
 174     resources_dict["resources"] = {}
 175     ref_resources = []
 176
 177     for res_key, res_value in resources.items():
 178         get_resources = []
 179
 180         for k, v in res_value:
 181             if k == "get_resource" and isinstance(v, dict):
 182                 get_resources = find_all_get_resource_in_resource(v)
 183
 184         # if resources found, add to dict
 185         if get_resources:
 186             ref_resources.extend(get_resources)
 187             resources_dict["resources"][res_key] = {
 188                 "res_value": res_value,
 189                 "get_resources": get_resources,
 190             }
 191
 192     resources_dict["ref_resources"] = set(ref_resources)
 193
 194     return resources_dict
 195
 196
 197 def flatten(items):
 198     """
 199     flatten items from any nested iterable
 200     """
 201
 202     merged_list = []
 203     for item in items:
 204         if isinstance(item, list):
 205             sub_list = flatten(item)
 206             merged_list.extend(sub_list)
 207         else:
 208             merged_list.append(item)
 209     return merged_list