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

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 

11from collections.abc import Callable, Sequence 

12from inspect import signature 

13from typing import Any 

14from weakref import WeakKeyDictionary 

15 

16from hypothesis.configuration import check_sideeffect_during_initialization 

17from hypothesis.internal.conjecture.choice import ChoiceT 

18from hypothesis.internal.conjecture.data import ConjectureData 

19from hypothesis.internal.reflection import ( 

20 convert_keyword_arguments, 

21 convert_positional_arguments, 

22 get_pretty_function_description, 

23 repr_call, 

24) 

25from hypothesis.strategies._internal.deferred import DeferredStrategy 

26from hypothesis.strategies._internal.strategies import ( 

27 Ex, 

28 RecurT, 

29 SearchStrategy, 

30 _filter_location_override, 

31 current_filter_call_site, 

32) 

33from hypothesis.utils.threading import ThreadLocal 

34 

35threadlocal = ThreadLocal(unwrap_depth=int, unwrap_cache=WeakKeyDictionary) 

36 

37 

38def unwrap_strategies(s): 

39 # optimization 

40 if not isinstance(s, (LazyStrategy, DeferredStrategy)): 

41 return s 

42 

43 # A fully-unwrapped LazyStrategy never changes, so we memoize that on the 

44 # instance; the threadlocal cache below only handles cycles mid-unwrap. 

45 if (result := s.__dict__.get("_unwrapped")) is not None: 

46 return result 

47 

48 try: 

49 return threadlocal.unwrap_cache[s] 

50 except KeyError: 

51 pass 

52 

53 threadlocal.unwrap_cache[s] = s 

54 threadlocal.unwrap_depth += 1 

55 

56 try: 

57 result = unwrap_strategies(s.wrapped_strategy) 

58 threadlocal.unwrap_cache[s] = result 

59 

60 try: 

61 assert result.force_has_reusable_values == s.force_has_reusable_values 

62 except AttributeError: 

63 pass 

64 

65 try: 

66 result.force_has_reusable_values = s.force_has_reusable_values 

67 except AttributeError: 

68 pass 

69 

70 if threadlocal.unwrap_depth == 1 and isinstance(s, LazyStrategy): 

71 s._unwrapped = result 

72 return result 

73 finally: 

74 threadlocal.unwrap_depth -= 1 

75 if threadlocal.unwrap_depth <= 0: 

76 threadlocal.unwrap_cache.clear() 

77 assert threadlocal.unwrap_depth >= 0 

78 

79 

80class LazyStrategy(SearchStrategy[Ex]): 

81 """A strategy which is defined purely by conversion to and from another 

82 strategy. 

83 

84 Its parameter and distribution come from that other strategy. 

85 """ 

86 

87 def __init__( 

88 self, 

89 function: Callable[..., SearchStrategy[Ex]], 

90 args: Sequence[object], 

91 kwargs: dict[str, object], 

92 *, 

93 # (name, function, location of the .filter()/.map() call, if known) 

94 transforms: tuple[tuple[str, Callable[..., Any], str | None], ...] = (), 

95 force_repr: str | None = None, 

96 ): 

97 super().__init__() 

98 self.__wrapped_strategy: SearchStrategy[Ex] | None = None 

99 self.__representation: str | None = force_repr 

100 self.function = function 

101 self.__args = args 

102 self.__kwargs = kwargs 

103 self._transformations = transforms 

104 

105 def calc_is_empty(self, recur: RecurT) -> bool: 

106 return recur(self.wrapped_strategy) 

107 

108 def calc_has_reusable_values(self, recur: RecurT) -> bool: 

109 return recur(self.wrapped_strategy) 

110 

111 def calc_is_cacheable(self, recur: RecurT) -> bool: 

112 for source in (self.__args, self.__kwargs.values()): 

113 for v in source: 

114 if isinstance(v, SearchStrategy) and not v.is_cacheable: 

115 return False 

116 return True 

117 

118 def calc_label(self) -> int: 

119 return self.wrapped_strategy.label 

120 

121 @property 

122 def wrapped_strategy(self) -> SearchStrategy[Ex]: 

123 if self.__wrapped_strategy is None: 

124 check_sideeffect_during_initialization("lazy evaluation of {!r}", self) 

125 

126 unwrapped_args = tuple(unwrap_strategies(s) for s in self.__args) 

127 unwrapped_kwargs = { 

128 k: unwrap_strategies(v) for k, v in self.__kwargs.items() 

129 } 

130 

131 base = self.function(*self.__args, **self.__kwargs) 

132 if unwrapped_args == self.__args and unwrapped_kwargs == self.__kwargs: 

133 _wrapped_strategy = base 

134 else: 

135 _wrapped_strategy = self.function(*unwrapped_args, **unwrapped_kwargs) 

136 for method, fn, location in self._transformations: 

137 # Carry the location of the original .filter() call through to 

138 # the underlying strategy's .filter(), for reporting. 

139 with _filter_location_override.with_value(location): 

140 _wrapped_strategy = getattr(_wrapped_strategy, method)(fn) 

141 self.__wrapped_strategy = _wrapped_strategy 

142 assert self.__wrapped_strategy is not None 

143 return self.__wrapped_strategy 

144 

145 def __with_transform(self, method, fn, *, location=None): 

146 repr_ = self.__representation 

147 if repr_: 

148 repr_ = f"{repr_}.{method}({get_pretty_function_description(fn)})" 

149 return LazyStrategy( 

150 self.function, 

151 self.__args, 

152 self.__kwargs, 

153 transforms=(*self._transformations, (method, fn, location)), 

154 force_repr=repr_, 

155 ) 

156 

157 def map(self, pack): 

158 return self.__with_transform("map", pack) 

159 

160 def filter(self, condition): 

161 return self.__with_transform( 

162 "filter", condition, location=current_filter_call_site() 

163 ) 

164 

165 def do_validate(self) -> None: 

166 w = self.wrapped_strategy 

167 assert isinstance(w, SearchStrategy), f"{self!r} returned non-strategy {w!r}" 

168 w.validate() 

169 

170 def __repr__(self) -> str: 

171 if self.__representation is None: 

172 sig = signature(self.function) 

173 pos = [p for p in sig.parameters.values() if "POSITIONAL" in p.kind.name] 

174 if len(pos) > 1 or any(p.default is not sig.empty for p in pos): 

175 _args, _kwargs = convert_positional_arguments( 

176 self.function, self.__args, self.__kwargs 

177 ) 

178 else: 

179 _args, _kwargs = convert_keyword_arguments( 

180 self.function, self.__args, self.__kwargs 

181 ) 

182 kwargs_for_repr = { 

183 k: v 

184 for k, v in _kwargs.items() 

185 if k not in sig.parameters or v is not sig.parameters[k].default 

186 } 

187 self.__representation = repr_call( 

188 self.function, _args, kwargs_for_repr, reorder=False 

189 ) + "".join( 

190 f".{method}({get_pretty_function_description(fn)})" 

191 for method, fn, _ in self._transformations 

192 ) 

193 return self.__representation 

194 

195 def do_draw(self, data: ConjectureData) -> Ex: 

196 return data.draw(self.wrapped_strategy) 

197 

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

199 return self.wrapped_strategy._invert(value)