Coverage for src / bluetooth_sig / gatt / validation.py: 82%

1"""Enhanced validation utilities for strict type checking and data validation. 

2 

3This module provides additional validation capabilities beyond the basic 

4utils, focusing on strict type safety and comprehensive data integrity 

5checks. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import Any, Callable, TypeVar 

11 

12import msgspec 

13 

14from .characteristics.utils.ieee11073_parser import IEEE11073Parser 

15from .constants import ( 

16 ABSOLUTE_ZERO_CELSIUS, 

17 MAX_CONCENTRATION_PPM, 

18 MAX_POWER_WATTS, 

19 MAX_TEMPERATURE_CELSIUS, 

20 PERCENTAGE_MAX, 

21) 

22from .exceptions import DataValidationError, TypeMismatchError, ValueRangeError 

23 

24T = TypeVar("T") 

25 

26 

27class ValidationRule(msgspec.Struct, kw_only=True): 

28 """Represents a validation rule with optional custom validator.""" 

29 

30 field_name: str 

31 expected_type: type | tuple[type, ...] # Allow tuple of types for isinstance 

32 min_value: int | float | None = None 

33 max_value: int | float | None = None 

34 custom_validator: Callable[[Any], bool] | None = None 

35 error_message: str | None = None 

36 

37 def validate(self, value: Any) -> None: # noqa: ANN401 # Validates values of various types 

38 """Apply this validation rule to a value.""" 

39 # Type check 

40 if not isinstance(value, self.expected_type): 

41 raise TypeMismatchError(self.field_name, value, self.expected_type) 

42 

43 # Range check 

44 if self.min_value is not None or self.max_value is not None: 

45 min_val = self.min_value if self.min_value is not None else float("-inf") 

46 max_val = self.max_value if self.max_value is not None else float("inf") 

47 if not min_val <= value <= max_val: 

48 raise ValueRangeError(self.field_name, value, min_val, max_val) 

49 

50 # Custom validation 

51 if self.custom_validator: 

52 # Note: custom_validator is type-hinted as Callable, but pylint doesn't recognize this 

53 if not self.custom_validator(value): # pylint: disable=not-callable 

54 message = self.error_message or f"Custom validation failed for {self.field_name}" 

55 raise DataValidationError(self.field_name, value, message) 

56 

57 

58class StrictValidator(msgspec.Struct, kw_only=True): 

59 """Strict validation engine for complex data structures.""" 

60 

61 rules: dict[str, ValidationRule] = msgspec.field(default_factory=dict) 

62 

63 def add_rule(self, rule: ValidationRule) -> None: 

64 """Add a validation rule.""" 

65 self.rules[rule.field_name] = rule 

66 

67 def validate_dict(self, data: dict[str, Any]) -> None: 

68 """Validate a dictionary against all rules.""" 

69 for field_name, value in data.items(): 

70 if field_name in self.rules: 

71 self.rules[field_name].validate(value) 

72 

73 def validate_object(self, obj: Any) -> None: # noqa: ANN401 # Validates objects of any type 

74 """Validate an object's attributes against all rules.""" 

75 for field_name, rule in self.rules.items(): 

76 if hasattr(obj, field_name): 

77 value = getattr(obj, field_name) 

78 rule.validate(value) 

79 

80 

81class CommonValidators: 

82 """Collection of commonly used validation functions.""" 

83 

84 @staticmethod 

85 def is_positive(value: int | float) -> bool: 

86 """Check if value is positive.""" 

87 return value > 0 

88 

89 @staticmethod 

90 def is_non_negative(value: int | float) -> bool: 

91 """Check if value is non-negative.""" 

92 return value >= 0 

93 

94 @staticmethod 

95 def is_valid_percentage(value: int | float) -> bool: 

96 """Check if value is a valid percentage (0-100).""" 

97 return 0 <= value <= PERCENTAGE_MAX 

98 

99 @staticmethod 

100 def is_valid_extended_percentage(value: int | float) -> bool: 

101 """Check if value is a valid extended percentage (0-200).""" 

102 return 0 <= value <= 200 

103 

104 @staticmethod 

105 def is_physical_temperature(value: float) -> bool: 

106 """Check if temperature is physically reasonable.""" 

107 return ABSOLUTE_ZERO_CELSIUS <= value <= MAX_TEMPERATURE_CELSIUS 

108 

109 @staticmethod 

110 def is_valid_concentration(value: float) -> bool: 

111 """Check if concentration is in valid range.""" 

112 return 0 <= value <= MAX_CONCENTRATION_PPM 

113 

114 @staticmethod 

115 def is_valid_power(value: int | float) -> bool: 

116 """Check if power value is reasonable.""" 

117 return 0 <= value <= MAX_POWER_WATTS 

118 

119 @staticmethod 

120 def is_valid_heart_rate(value: int) -> bool: 

121 """Check if heart rate is in human range.""" 

122 return 30 <= value <= 300 # Reasonable human heart rate range 

123 

124 @staticmethod 

125 def is_valid_battery_level(value: int) -> bool: 

126 """Check if battery level is valid percentage.""" 

127 return 0 <= value <= PERCENTAGE_MAX 

128 

129 @staticmethod 

130 def is_ieee11073_special_value(value: int) -> bool: 

131 """Check if value is a valid IEEE 11073 special value.""" 

132 special_values = { 

133 IEEE11073Parser.SFLOAT_NAN, 

134 IEEE11073Parser.SFLOAT_NRES, 

135 IEEE11073Parser.SFLOAT_POSITIVE_INFINITY, 

136 IEEE11073Parser.SFLOAT_NEGATIVE_INFINITY, 

137 } 

138 return value in special_values 

139 

140 

141# Pre-configured validators for common use cases 

142HEART_RATE_VALIDATOR = StrictValidator() 

143HEART_RATE_VALIDATOR.add_rule( 

144 ValidationRule( 

145 field_name="heart_rate", 

146 expected_type=int, 

147 min_value=30, 

148 max_value=300, 

149 custom_validator=CommonValidators.is_valid_heart_rate, 

150 error_message="Heart rate must be between 30-300 bpm", 

151 ) 

152) 

153 

154BATTERY_VALIDATOR = StrictValidator() 

155BATTERY_VALIDATOR.add_rule( 

156 ValidationRule( 

157 field_name="battery_level", 

158 expected_type=int, 

159 min_value=0, 

160 max_value=PERCENTAGE_MAX, 

161 custom_validator=CommonValidators.is_valid_battery_level, 

162 error_message="Battery level must be 0-100%", 

163 ) 

164) 

165 

166TEMPERATURE_VALIDATOR = StrictValidator() 

167TEMPERATURE_VALIDATOR.add_rule( 

168 ValidationRule( 

169 field_name="temperature", 

170 expected_type=float, 

171 min_value=ABSOLUTE_ZERO_CELSIUS, 

172 max_value=MAX_TEMPERATURE_CELSIUS, 

173 custom_validator=CommonValidators.is_physical_temperature, 

174 error_message="Temperature must be physically reasonable", 

175 ) 

176) 

177 

178 

179def create_range_validator( 

180 field_name: str, 

181 expected_type: type, 

182 min_value: int | float | None = None, 

183 max_value: int | float | None = None, 

184 custom_validator: Callable[[Any], bool] | None = None, 

185) -> StrictValidator: 

186 """Factory function to create a validator for a specific range.""" 

187 validator = StrictValidator() 

188 validator.add_rule( 

189 ValidationRule( 

190 field_name=field_name, 

191 expected_type=expected_type, 

192 min_value=min_value, 

193 max_value=max_value, 

194 custom_validator=custom_validator, 

195 ) 

196 ) 

197 return validator 

198 

199 

200def validate_measurement_data(data: dict[str, Any], measurement_type: str) -> dict[str, Any]: 

201 """Validate measurement data based on type and return validated data.""" 

202 if measurement_type == "heart_rate": 

203 HEART_RATE_VALIDATOR.validate_dict(data) 

204 elif measurement_type == "battery": 

205 BATTERY_VALIDATOR.validate_dict(data) 

206 elif measurement_type == "temperature": 

207 TEMPERATURE_VALIDATOR.validate_dict(data) 

208 else: 

209 # Generic validation for unknown types 

210 for key, value in data.items(): 

211 if isinstance(value, (int, float)) and value < 0: 

212 raise ValueRangeError(key, value, 0, float("inf")) 

213 

214 return data