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

97 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-03 16:41 +0000

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 collections.abc import Callable 

11from typing import Any, TypeVar, cast 

12 

13import msgspec 

14 

15from .characteristics.utils.ieee11073_parser import IEEE11073Parser 

16from .constants import ( 

17 ABSOLUTE_ZERO_CELSIUS, 

18 EXTENDED_PERCENTAGE_MAX, 

19 HEART_RATE_MAX, 

20 HEART_RATE_MIN, 

21 MAX_CONCENTRATION_PPM, 

22 MAX_POWER_WATTS, 

23 MAX_TEMPERATURE_CELSIUS, 

24 PERCENTAGE_MAX, 

25) 

26from .exceptions import DataValidationError, TypeMismatchError, ValueRangeError 

27 

28T = TypeVar("T") 

29 

30 

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

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

33 

34 field_name: str 

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

36 min_value: int | float | None = None 

37 max_value: int | float | None = None 

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

39 error_message: str | None = None 

40 

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

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

43 # Type check 

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

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

46 

47 # Range check 

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

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

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

51 numeric_value = cast(int | float, value) 

52 try: 

53 in_range = min_val <= numeric_value <= max_val 

54 except TypeError as exc: 

55 raise DataValidationError( 

56 self.field_name, 

57 value, 

58 "Value cannot be ordered for range validation", 

59 ) from exc 

60 

61 if not in_range: 

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

63 

64 # Custom validation 

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

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

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

68 

69 

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

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

72 

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

74 

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

76 """Add a validation rule.""" 

77 self.rules[rule.field_name] = rule 

78 

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

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

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

82 if field_name in self.rules: 

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

84 

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

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

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

88 if hasattr(obj, field_name): 

89 value = getattr(obj, field_name) 

90 rule.validate(value) 

91 

92 

93class CommonValidators: 

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

95 

96 @staticmethod 

97 def is_positive(value: float) -> bool: 

98 """Check if value is positive.""" 

99 return value > 0 

100 

101 @staticmethod 

102 def is_non_negative(value: float) -> bool: 

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

104 return value >= 0 

105 

106 @staticmethod 

107 def is_valid_percentage(value: float) -> bool: 

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

109 return 0 <= value <= PERCENTAGE_MAX 

110 

111 @staticmethod 

112 def is_valid_extended_percentage(value: float) -> bool: 

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

114 return 0 <= value <= EXTENDED_PERCENTAGE_MAX 

115 

116 @staticmethod 

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

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

119 return ABSOLUTE_ZERO_CELSIUS <= value <= MAX_TEMPERATURE_CELSIUS 

120 

121 @staticmethod 

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

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

124 return 0 <= value <= MAX_CONCENTRATION_PPM 

125 

126 @staticmethod 

127 def is_valid_power(value: float) -> bool: 

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

129 return 0 <= value <= MAX_POWER_WATTS 

130 

131 @staticmethod 

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

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

134 return HEART_RATE_MIN <= value <= HEART_RATE_MAX # Reasonable human heart rate range 

135 

136 @staticmethod 

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

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

139 return 0 <= value <= PERCENTAGE_MAX 

140 

141 @staticmethod 

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

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

144 special_values = { 

145 IEEE11073Parser.SFLOAT_NAN, 

146 IEEE11073Parser.SFLOAT_NRES, 

147 IEEE11073Parser.SFLOAT_POSITIVE_INFINITY, 

148 IEEE11073Parser.SFLOAT_NEGATIVE_INFINITY, 

149 } 

150 return value in special_values 

151 

152 

153# Pre-configured validators for common use cases 

154HEART_RATE_VALIDATOR = StrictValidator() 

155HEART_RATE_VALIDATOR.add_rule( 

156 ValidationRule( 

157 field_name="heart_rate", 

158 expected_type=int, 

159 min_value=30, 

160 max_value=300, 

161 custom_validator=CommonValidators.is_valid_heart_rate, 

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

163 ) 

164) 

165 

166BATTERY_VALIDATOR = StrictValidator() 

167BATTERY_VALIDATOR.add_rule( 

168 ValidationRule( 

169 field_name="battery_level", 

170 expected_type=int, 

171 min_value=0, 

172 max_value=PERCENTAGE_MAX, 

173 custom_validator=CommonValidators.is_valid_battery_level, 

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

175 ) 

176) 

177 

178TEMPERATURE_VALIDATOR = StrictValidator() 

179TEMPERATURE_VALIDATOR.add_rule( 

180 ValidationRule( 

181 field_name="temperature", 

182 expected_type=float, 

183 min_value=ABSOLUTE_ZERO_CELSIUS, 

184 max_value=MAX_TEMPERATURE_CELSIUS, 

185 custom_validator=CommonValidators.is_physical_temperature, 

186 error_message="Temperature must be physically reasonable", 

187 ) 

188) 

189 

190 

191def create_range_validator( 

192 field_name: str, 

193 expected_type: type, 

194 min_value: float | None = None, 

195 max_value: float | None = None, 

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

197) -> StrictValidator: 

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

199 validator = StrictValidator() 

200 validator.add_rule( 

201 ValidationRule( 

202 field_name=field_name, 

203 expected_type=expected_type, 

204 min_value=min_value, 

205 max_value=max_value, 

206 custom_validator=custom_validator, 

207 ) 

208 ) 

209 return validator 

210 

211 

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

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

214 if measurement_type == "heart_rate": 

215 HEART_RATE_VALIDATOR.validate_dict(data) 

216 elif measurement_type == "battery": 

217 BATTERY_VALIDATOR.validate_dict(data) 

218 elif measurement_type == "temperature": 

219 TEMPERATURE_VALIDATOR.validate_dict(data) 

220 else: 

221 # Generic validation for unknown types 

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

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

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

225 

226 return data