Coverage for python/pyairflowtester/scoring.py: 98%

1""" 

2Scoring and risk assessment module. 

3""" 

4 

5from typing import Any, Dict, List 

6 

7from pyairflowtester.models import SEVERITY_WEIGHTS 

8 

9 

10class Scorer: 

11 """Risk scoring engine.""" 

12 

13 def __init__(self): 

14 """Initialize scorer.""" 

15 self.severity_weights = SEVERITY_WEIGHTS 

16 

17 def calculate_risk_score(self, violations: List[Dict[str, Any]]) -> float: 

18 """ 

19 Calculate risk score from violations (0-100). 

20 

21 Args: 

22 violations: List of violations 

23 

24 Returns: 

25 Risk score (0-100) 

26 """ 

27 if not violations: 

28 return 100.0 

29 

30 total_severity = sum( 

31 self.severity_weights.get(v.get("severity", "info"), 0) for v in violations 

32 ) 

33 

34 avg_severity = total_severity / len(violations) 

35 violation_count_factor = min(len(violations) / 10.0, 1.0) 

36 

37 base_risk = (1.0 - avg_severity) * 100.0 

38 adjusted_risk = base_risk * (1.0 + violation_count_factor) 

39 

40 return min(adjusted_risk, 100.0) 

41 

42 def aggregate_by_severity(self, violations: List[Dict[str, Any]]) -> Dict[str, int]: 

43 """ 

44 Aggregate violations by severity. 

45 

46 Args: 

47 violations: List of violations 

48 

49 Returns: 

50 Aggregation by severity 

51 """ 

52 aggregated = { 

53 "critical": 0, 

54 "high": 0, 

55 "medium": 0, 

56 "low": 0, 

57 "info": 0, 

58 } 

59 

60 for v in violations: 

61 severity = v.get("severity", "info") 

62 if severity in aggregated: 62 ↛ 60line 62 didn't jump to line 60 because the condition on line 62 was always true

63 aggregated[severity] += 1 

64 

65 return aggregated 

66 

67 def categorize_risk(self, score: float) -> str: 

68 """ 

69 Categorize risk level. 

70 

71 Args: 

72 score: Risk score (0-100) 

73 

74 Returns: 

75 Risk level 

76 """ 

77 if score <= 25.0: 

78 return "low" 

79 elif score <= 50.0: 

80 return "medium" 

81 elif score <= 75.0: 

82 return "high" 

83 else: 

84 return "critical" 

85 

86 def calculate_health_score( 

87 self, 

88 reliability: float, 

89 performance: float, 

90 maintainability: float, 

91 ) -> float: 

92 """ 

93 Calculate health score based on multiple dimensions. 

94 

95 Args: 

96 reliability: Reliability score (0-100) 

97 performance: Performance score (0-100) 

98 maintainability: Maintainability score (0-100) 

99 

100 Returns: 

101 Health score (0-100) 

102 """ 

103 # Weighted average: reliability 50%, performance 30%, maintainability 20% 

104 score = reliability * 0.5 + performance * 0.3 + maintainability * 0.2 

105 return min(max(score, 0.0), 100.0) 

106 

107 def calculate_trend(self, current_score: float, previous_score: float) -> str: 

108 """ 

109 Calculate trend. 

110 

111 Args: 

112 current_score: Current score 

113 previous_score: Previous score 

114 

115 Returns: 

116 Trend (improving/stable/degrading) 

117 """ 

118 diff = current_score - previous_score 

119 if diff > 5.0: 

120 return "improving" 

121 elif diff < -5.0: 

122 return "degrading" 

123 else: 

124 return "stable" 

125 

126 def filter_by_severity( 

127 self, 

128 violations: List[Dict[str, Any]], 

129 min_severity: str, 

130 ) -> List[Dict[str, Any]]: 

131 """ 

132 Filter violations by minimum severity. 

133 

134 Args: 

135 violations: List of violations 

136 min_severity: Minimum severity level 

137 

138 Returns: 

139 Filtered violations 

140 """ 

141 min_weight = self.severity_weights.get(min_severity, 0) 

142 

143 return [ 

144 v 

145 for v in violations 

146 if self.severity_weights.get(v.get("severity", "info"), 0) >= min_weight 

147 ]