Coverage for postrfp/web/suxint/validate.py: 100%

1import inspect 

2import logging 

3 

4from pydantic import BaseModel 

5from webob.response import Response 

6 

7from .extractors import PathArg 

8from postrfp.shared.exceptions import RoutingError 

9from postrfp.shared.types import SuxType 

10 

11 

12log = logging.getLogger(__name__) 

13 

14 

15def validate_handlers( 

16 sux_instance: SuxType, strict: bool = True, check_return_types: bool = True 

17) -> None: 

18 """ 

19 Validate all handlers for missing adaptors, return types, and duplicate names. 

20 

21 Args: 

22 strict: If True, raise RoutingError if any validation error is found. 

23 check_return_types: If True, also validate handler return types. 

24 

25 Raises: 

26 RoutingError: If any errors are found and strict mode is enabled. 

27 """ 

28 from typing import get_origin, Annotated, get_args 

29 

30 def _unwrap_annotated(type_): 

31 if get_origin(type_) is Annotated: 

32 return get_args(type_)[0] 

33 return type_ 

34 

35 errors = [] 

36 warnings = [] 

37 seen_handler_names = set() 

38 

39 for handler in sux_instance.iter_handlers(): 

40 # Check for duplicate handler names 

41 if handler.name in seen_handler_names: 

42 errors.append(f"Duplicate handler name detected: '{handler.name}'.") 

43 else: 

44 seen_handler_names.add(handler.name) 

45 

46 # Validate required arguments 

47 for param_name in handler.required_arguments: 

48 adaptor = getattr(sux_instance.adaptor_module, param_name, None) 

49 if adaptor is None: 

50 adaptor_module_name = ( 

51 sux_instance.adaptor_module.__name__ 

52 if sux_instance.adaptor_module 

53 else "Unknown" 

54 ) 

55 errors.append( 

56 f"Adaptor for required parameter '{param_name}' not found in adaptor module {adaptor_module_name} " 

57 f"for handler '{handler.name}'." 

58 ) 

59 elif isinstance(adaptor, PathArg): 

60 if f"{{{adaptor.doc_name}}}" not in handler.path: 

61 errors.append( 

62 f"PathArg for '{param_name}' not found as '{{{adaptor.doc_name}}}' in handler path '{handler.path}'." 

63 ) 

64 

65 # Validate optional arguments 

66 for param_name in handler.optional_arguments: 

67 adaptor = getattr(sux_instance.adaptor_module, param_name, None) 

68 if adaptor is None: 

69 warnings.append( 

70 f"Adaptor for optional parameter '{param_name}' " 

71 f"in handler '{handler.name}' not found in adaptor module {sux_instance.adaptor_module}." 

72 ) 

73 

74 # Validate return types 

75 if check_return_types: 

76 retval = handler.return_value 

77 if retval is not None: 

78 # First, unwrap Annotated to get the underlying type 

79 unwrapped_type = _unwrap_annotated(retval) 

80 # Then, get the origin (e.g., list from list[int]) or the type itself 

81 origin_or_base = get_origin(unwrapped_type) or unwrapped_type 

82 

83 # If the origin/base isn't a class (e.g. TypeVar, NewType, typing.Union args), flag it 

84 if not inspect.isclass(origin_or_base): 

85 errors.append( 

86 f"Handler '{handler.name}' return type '{retval}' resolves to non-class '{origin_or_base}'." 

87 ) 

88 # Perform the subclass check against allowed types 

89 elif not issubclass(origin_or_base, (BaseModel, dict, list, Response)): 

90 errors.append( 

91 f"Handler '{handler.name}' return type '{origin_or_base.__name__}' " 

92 f"is not an instance or subclass of BaseModel, dict, list or Response." 

93 ) 

94 

95 if warnings: 

96 log.warning(f"Validation Warnings: \n\n {'\n'.join(warnings)}") 

97 

98 if errors: 

99 errstring = "\n ".join(errors) 

100 log.error("Validation Errors:\n {errstring}") 

101 if strict: 

102 raise RoutingError( 

103 f"Validation Errors during initialisation: \n\n {errstring}" 

104 )