Coverage for chalicelib/core/alerts/modules/helpers.py: 0%

1import decimal 

2import logging 

3 

4import schemas 

5from chalicelib.utils.TimeUTC import TimeUTC 

6 

7logger = logging.getLogger(__name__) 

8# This is the frequency of execution for each threshold 

9TimeInterval = { 

10 15: 3, 

11 30: 5, 

12 60: 10, 

13 120: 20, 

14 240: 30, 

15 1440: 60, 

16} 

17 

18 

19def __format_value(x): 

20 if x % 1 == 0: 

21 x = int(x) 

22 else: 

23 x = round(x, 2) 

24 return f"{x:,}" 

25 

26 

27def can_check(a) -> bool: 

28 now = TimeUTC.now() 

29 

30 repetitionBase = a["options"]["currentPeriod"] \ 

31 if a["detectionMethod"] == schemas.AlertDetectionMethod.CHANGE \ 

32 and a["options"]["currentPeriod"] > a["options"]["previousPeriod"] \ 

33 else a["options"]["previousPeriod"] 

34 

35 if TimeInterval.get(repetitionBase) is None: 

36 logger.error(f"repetitionBase: {repetitionBase} NOT FOUND") 

37 return False 

38 

39 return (a["options"]["renotifyInterval"] <= 0 or 

40 a["options"].get("lastNotification") is None or 

41 a["options"]["lastNotification"] <= 0 or 

42 ((now - a["options"]["lastNotification"]) > a["options"]["renotifyInterval"] * 60 * 1000)) \ 

43 and ((now - a["createdAt"]) % (TimeInterval[repetitionBase] * 60 * 1000)) < 60 * 1000 

44 

45 

46def generate_notification(alert, result): 

47 left = __format_value(result['value']) 

48 right = __format_value(alert['query']['right']) 

49 return { 

50 "alertId": alert["alertId"], 

51 "tenantId": alert["tenantId"], 

52 "title": alert["name"], 

53 "description": f"{alert['seriesName']} = {left} ({alert['query']['operator']} {right}).", 

54 "buttonText": "Check metrics for more details", 

55 "buttonUrl": f"/{alert['projectId']}/metrics", 

56 "imageUrl": None, 

57 "projectId": alert["projectId"], 

58 "projectName": alert["projectName"], 

59 "options": {"source": "ALERT", "sourceId": alert["alertId"], 

60 "sourceMeta": alert["detectionMethod"], 

61 "message": alert["options"]["message"], "projectId": alert["projectId"], 

62 "data": {"title": alert["name"], 

63 "limitValue": alert["query"]["right"], 

64 "actualValue": float(result["value"]) \ 

65 if isinstance(result["value"], decimal.Decimal) \ 

66 else result["value"], 

67 "operator": alert["query"]["operator"], 

68 "trigger": alert["query"]["left"], 

69 "alertId": alert["alertId"], 

70 "detectionMethod": alert["detectionMethod"], 

71 "currentPeriod": alert["options"]["currentPeriod"], 

72 "previousPeriod": alert["options"]["previousPeriod"], 

73 "createdAt": TimeUTC.now()}}, 

74 }