Coverage for kube_notify/utils/selectors.py: 100%

1def check_selector( 

2 resource_name: str, 

3 resources: list[dict], 

4 resource_type: str, 

5 labels: dict, 

6 namespace: str, 

7 involved_object_kind: str, 

8 reason: str, 

9) -> bool: 

10 for resource in resources: 

11 if resource["name"] == resource_name: 

12 result = True 

13 if "selector" in resource: 

14 selectors = resource.get("selector", {}) 

15 result = ( 

16 resource_type in selectors.get("types", [resource_type]) 

17 and ( 

18 selectors.get("labels") is None 

19 or any( 

20 ( 

21 label.get("key"), 

22 label.get("value", labels.get(label.get("key"))), 

23 ) 

24 in labels.items() 

25 for label in selectors.get("labels", []) 

26 ) 

27 ) 

28 and ( 

29 selectors.get("involvedObjectKind") is None 

30 or involved_object_kind in selectors.get("involvedObjectKind") 

31 ) 

32 and ( 

33 selectors.get("namespaces") is None 

34 or namespace in selectors.get("namespaces") 

35 ) 

36 and ( 

37 selectors.get("reasons") is None 

38 or reason in selectors.get("reasons") 

39 ) 

40 ) 

41 elif "excludeSelector" in resource: 

42 selectors = resource.get("excludeSelector", {}) 

43 result = not ( 

44 resource_type in selectors.get("types", []) 

45 or any( 

46 ( 

47 label.get("key"), 

48 label.get("value", labels.get(label.get("key"))), 

49 ) 

50 in labels.items() 

51 for label in selectors.get("labels", []) 

52 ) 

53 or involved_object_kind in selectors.get("involvedObjectKind", []) 

54 or namespace in selectors.get("namespaces", []) 

55 or reason in selectors.get("reasons", []) 

56 ) 

57 if result: 

58 return result 

59 return False