Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/c7n/element.py: 78%

1# Copyright 2020 Cloud Custodian Authors. 

2# Copyright The Cloud Custodian Authors. 

3# SPDX-License-Identifier: Apache-2.0 

4 

5from c7n import deprecated 

6from c7n.executor import ThreadPoolExecutor 

7from c7n.utils import jmespath_search 

8 

9 

10class Element: 

11 """Parent base class for filters and actions. 

12 """ 

13 

14 permissions = () 

15 metrics = () 

16 

17 executor_factory = ThreadPoolExecutor 

18 

19 schema = {'type': 'object'} 

20 # schema aliases get hoisted into a jsonschema definition 

21 # location, and then referenced inline. 

22 schema_alias = None 

23 

24 def get_permissions(self): 

25 return self.permissions 

26 

27 def validate(self): 

28 """Validate the current element's configuration. 

29 

30 Should raise a validation error if there are any configuration issues. 

31 

32 This method will always be called prior to element execution/process() method 

33 being called and thus can act as a point of lazy initialization. 

34 """ 

35 

36 def filter_resources(self, resources, key_expr, allowed_values=()): 

37 # many filters implementing a resource state transition only allow 

38 # a given set of starting states, this method will filter resources 

39 # and issue a warning log, as implicit filtering in filters means 

40 # our policy metrics are off, and they should be added as policy 

41 # filters. 

42 resource_count = len(resources) 

43 search_expr = key_expr 

44 if not search_expr.startswith('[].'): 

45 search_expr = '[].' + key_expr 

46 results = [r for value, r in zip( 

47 jmespath_search(search_expr, resources), resources) 

48 if value in allowed_values] 

49 if resource_count != len(results): 

50 self.log.warning( 

51 "%s implicitly filtered %d of %d resources key:%s on %s", 

52 self.type, len(results), resource_count, key_expr, 

53 (', '.join(map(str, allowed_values)))) 

54 return results 

55 

56 def get_deprecations(self): 

57 """Return any matching deprecations for the policy fields itself.""" 

58 return deprecated.check_deprecations(self, self.type + ":")