Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/hypothesis/strategies/_internal/recursive.py: 84%

1# This file is part of Hypothesis, which may be found at 

2# https://github.com/HypothesisWorks/hypothesis/ 

3# 

4# Copyright the Hypothesis Authors. 

5# Individual contributors are listed in AUTHORS.rst and the git log. 

6# 

7# This Source Code Form is subject to the terms of the Mozilla Public License, 

8# v. 2.0. If a copy of the MPL was not distributed with this file, You can 

9# obtain one at https://mozilla.org/MPL/2.0/. 

10 

11import threading 

12import warnings 

13from collections.abc import Callable, Generator 

14from contextlib import contextmanager 

15from typing import Any, TypeVar 

16 

17from hypothesis.errors import HypothesisWarning, InvalidArgument 

18from hypothesis.internal.conjecture.choice import ChoiceT 

19from hypothesis.internal.conjecture.data import ConjectureData 

20from hypothesis.internal.reflection import ( 

21 get_pretty_function_description, 

22 is_first_param_referenced_in_function, 

23 is_identity_function, 

24) 

25from hypothesis.internal.validation import check_type 

26from hypothesis.strategies._internal.strategies import ( 

27 OneOfStrategy, 

28 SearchStrategy, 

29 check_strategy, 

30) 

31from hypothesis.utils.deprecation import note_deprecation 

32 

33T = TypeVar("T") 

34 

35 

36class LimitReached(BaseException): 

37 pass 

38 

39 

40class LimitedStrategy(SearchStrategy[T]): 

41 def __init__(self, strategy: SearchStrategy[T]): 

42 super().__init__() 

43 self.base_strategy = strategy 

44 self._threadlocal = threading.local() 

45 

46 @property 

47 def marker(self) -> int: 

48 return getattr(self._threadlocal, "marker", 0) 

49 

50 @marker.setter 

51 def marker(self, value: int) -> None: 

52 self._threadlocal.marker = value 

53 

54 @property 

55 def currently_capped(self) -> bool: 

56 return getattr(self._threadlocal, "currently_capped", False) 

57 

58 @currently_capped.setter 

59 def currently_capped(self, value: bool) -> None: 

60 self._threadlocal.currently_capped = value 

61 

62 def __repr__(self) -> str: 

63 return f"LimitedStrategy({self.base_strategy!r})" 

64 

65 def do_validate(self) -> None: 

66 self.base_strategy.validate() 

67 

68 def do_draw(self, data: ConjectureData) -> T: 

69 assert self.currently_capped 

70 if self.marker <= 0: 

71 raise LimitReached 

72 self.marker -= 1 

73 return data.draw(self.base_strategy) 

74 

75 def _invert(self, value: Any) -> tuple[ChoiceT, ...]: 

76 # The marker is not part of the choice sequence, so at the choice 

77 # level a LimitedStrategy draw is exactly a base_strategy draw. If a 

78 # replay of the inversion exhausts the marker it raises LimitReached 

79 # partway through the choices, and the caller's replay-verification 

80 # rejects the attempt - as _invert's best-effort contract permits. 

81 return self.base_strategy._invert(value) 

82 

83 @contextmanager 

84 def capped(self, max_templates: int) -> Generator[None, None, None]: 

85 try: 

86 was_capped = self.currently_capped 

87 self.currently_capped = True 

88 self.marker = max_templates 

89 yield 

90 finally: 

91 self.currently_capped = was_capped 

92 

93 

94class RecursiveStrategy(SearchStrategy): 

95 def __init__( 

96 self, 

97 base: SearchStrategy, 

98 extend: Callable[[SearchStrategy], SearchStrategy], 

99 min_leaves: int | None, 

100 max_leaves: int, 

101 ): 

102 super().__init__() 

103 self.min_leaves = min_leaves 

104 self.max_leaves = max_leaves 

105 self.base = base 

106 self.limited_base = LimitedStrategy(base) 

107 self.extend = extend 

108 

109 strategies = [self.limited_base, self.extend(self.limited_base)] 

110 while 2 ** (len(strategies) - 1) <= max_leaves: 

111 strategies.append(extend(OneOfStrategy(tuple(strategies)))) 

112 # If min_leaves > 1, we can never draw from base directly 

113 if min_leaves is not None and min_leaves > 1: 

114 strategies = strategies[1:] 

115 self.strategy = OneOfStrategy(strategies) 

116 

117 def __repr__(self) -> str: 

118 if not hasattr(self, "_cached_repr"): 

119 self._cached_repr = ( 

120 f"recursive({self.base!r}, " 

121 f"{get_pretty_function_description(self.extend)}, " 

122 f"min_leaves={self.min_leaves}, max_leaves={self.max_leaves})" 

123 ) 

124 return self._cached_repr 

125 

126 def do_validate(self) -> None: 

127 check_strategy(self.base, "base") 

128 extended = self.extend(self.limited_base) 

129 check_strategy(extended, f"extend({self.limited_base!r})") 

130 self.limited_base.validate() 

131 extended.validate() 

132 

133 if is_identity_function(self.extend): 

134 warnings.warn( 

135 "extend=lambda x: x is a no-op; you probably want to use a " 

136 "different extend function, or just use the base strategy directly.", 

137 HypothesisWarning, 

138 stacklevel=5, 

139 ) 

140 

141 if not is_first_param_referenced_in_function(self.extend): 

142 msg = ( 

143 f"extend={get_pretty_function_description(self.extend)} doesn't use " 

144 "it's argument, and thus can't actually recurse!" 

145 ) 

146 if self.min_leaves is None: 

147 note_deprecation( 

148 msg, 

149 since="2026-01-12", 

150 has_codemod=False, 

151 stacklevel=1, 

152 ) 

153 else: 

154 raise InvalidArgument(msg) 

155 

156 if self.min_leaves is not None: 

157 check_type(int, self.min_leaves, "min_leaves") 

158 check_type(int, self.max_leaves, "max_leaves") 

159 if self.min_leaves is not None and self.min_leaves <= 0: 

160 raise InvalidArgument( 

161 f"min_leaves={self.min_leaves!r} must be greater than zero" 

162 ) 

163 if self.max_leaves <= 0: 

164 raise InvalidArgument( 

165 f"max_leaves={self.max_leaves!r} must be greater than zero" 

166 ) 

167 if (self.min_leaves or 1) > self.max_leaves: 

168 raise InvalidArgument( 

169 f"min_leaves={self.min_leaves!r} must be less than or equal to " 

170 f"max_leaves={self.max_leaves!r}" 

171 ) 

172 

173 def do_draw(self, data: ConjectureData) -> Any: 

174 min_leaves_retries = 0 

175 while True: 

176 try: 

177 with self.limited_base.capped(self.max_leaves): 

178 result = data.draw(self.strategy) 

179 leaves_drawn = self.max_leaves - self.limited_base.marker 

180 if self.min_leaves and leaves_drawn < self.min_leaves: 

181 data.events[ 

182 f"Draw for {self!r} had fewer than " 

183 f"min_leaves={self.min_leaves} and had to be retried" 

184 ] = "" 

185 min_leaves_retries += 1 

186 if min_leaves_retries < 5: 

187 continue 

188 data.mark_invalid(f"min_leaves={self.min_leaves} unsatisfied") 

189 return result 

190 except LimitReached: 

191 data.events[ 

192 f"Draw for {self!r} exceeded " 

193 f"max_leaves={self.max_leaves} and had to be retried" 

194 ] = "" 

195 

196 def _invert(self, value: Any) -> tuple[ChoiceT, ...]: 

197 # do_draw makes exactly one draw from self.strategy per attempt; its 

198 # retry paths (LimitReached, min_leaves unsatisfied) only replay under 

199 # a bad inversion, which the caller's replay-verification rejects. 

200 return self.strategy._invert(value)