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

92 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-03-18 11:17 +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 

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 if not min_val <= value <= max_val: 

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

53 

54 # Custom validation 

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

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

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

58 

59 

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

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

62 

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

64 

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

66 """Add a validation rule.""" 

67 self.rules[rule.field_name] = rule 

68 

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

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

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

72 if field_name in self.rules: 

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

74 

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

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

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

78 if hasattr(obj, field_name): 

79 value = getattr(obj, field_name) 

80 rule.validate(value) 

81 

82 

83class CommonValidators: 

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

85 

86 @staticmethod 

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

88 """Check if value is positive.""" 

89 return value > 0 

90 

91 @staticmethod 

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

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

94 return value >= 0 

95 

96 @staticmethod 

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

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

99 return 0 <= value <= PERCENTAGE_MAX 

100 

101 @staticmethod 

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

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

104 return 0 <= value <= EXTENDED_PERCENTAGE_MAX 

105 

106 @staticmethod 

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

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

109 return ABSOLUTE_ZERO_CELSIUS <= value <= MAX_TEMPERATURE_CELSIUS 

110 

111 @staticmethod 

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

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

114 return 0 <= value <= MAX_CONCENTRATION_PPM 

115 

116 @staticmethod 

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

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

119 return 0 <= value <= MAX_POWER_WATTS 

120 

121 @staticmethod 

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

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

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

125 

126 @staticmethod 

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

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

129 return 0 <= value <= PERCENTAGE_MAX 

130 

131 @staticmethod 

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

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

134 special_values = { 

135 IEEE11073Parser.SFLOAT_NAN, 

136 IEEE11073Parser.SFLOAT_NRES, 

137 IEEE11073Parser.SFLOAT_POSITIVE_INFINITY, 

138 IEEE11073Parser.SFLOAT_NEGATIVE_INFINITY, 

139 } 

140 return value in special_values 

141 

142 

143# Pre-configured validators for common use cases 

144HEART_RATE_VALIDATOR = StrictValidator() 

145HEART_RATE_VALIDATOR.add_rule( 

146 ValidationRule( 

147 field_name="heart_rate", 

148 expected_type=int, 

149 min_value=30, 

150 max_value=300, 

151 custom_validator=CommonValidators.is_valid_heart_rate, 

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

153 ) 

154) 

155 

156BATTERY_VALIDATOR = StrictValidator() 

157BATTERY_VALIDATOR.add_rule( 

158 ValidationRule( 

159 field_name="battery_level", 

160 expected_type=int, 

161 min_value=0, 

162 max_value=PERCENTAGE_MAX, 

163 custom_validator=CommonValidators.is_valid_battery_level, 

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

165 ) 

166) 

167 

168TEMPERATURE_VALIDATOR = StrictValidator() 

169TEMPERATURE_VALIDATOR.add_rule( 

170 ValidationRule( 

171 field_name="temperature", 

172 expected_type=float, 

173 min_value=ABSOLUTE_ZERO_CELSIUS, 

174 max_value=MAX_TEMPERATURE_CELSIUS, 

175 custom_validator=CommonValidators.is_physical_temperature, 

176 error_message="Temperature must be physically reasonable", 

177 ) 

178) 

179 

180 

181def create_range_validator( 

182 field_name: str, 

183 expected_type: type, 

184 min_value: float | None = None, 

185 max_value: float | None = None, 

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

187) -> StrictValidator: 

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

189 validator = StrictValidator() 

190 validator.add_rule( 

191 ValidationRule( 

192 field_name=field_name, 

193 expected_type=expected_type, 

194 min_value=min_value, 

195 max_value=max_value, 

196 custom_validator=custom_validator, 

197 ) 

198 ) 

199 return validator 

200 

201 

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

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

204 if measurement_type == "heart_rate": 

205 HEART_RATE_VALIDATOR.validate_dict(data) 

206 elif measurement_type == "battery": 

207 BATTERY_VALIDATOR.validate_dict(data) 

208 elif measurement_type == "temperature": 

209 TEMPERATURE_VALIDATOR.validate_dict(data) 

210 else: 

211 # Generic validation for unknown types 

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

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

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

215 

216 return data