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

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 try: 

44 return threadlocal.unwrap_cache[s] 

45 except KeyError: 

46 pass 

47 

48 threadlocal.unwrap_cache[s] = s 

49 threadlocal.unwrap_depth += 1 

50 

51 try: 

52 result = unwrap_strategies(s.wrapped_strategy) 

53 threadlocal.unwrap_cache[s] = result 

54 

55 try: 

56 assert result.force_has_reusable_values == s.force_has_reusable_values 

57 except AttributeError: 

58 pass 

59 

60 try: 

61 result.force_has_reusable_values = s.force_has_reusable_values 

62 except AttributeError: 

63 pass 

64 

65 return result 

66 finally: 

67 threadlocal.unwrap_depth -= 1 

68 if threadlocal.unwrap_depth <= 0: 

69 threadlocal.unwrap_cache.clear() 

70 assert threadlocal.unwrap_depth >= 0 

71 

72 

73class LazyStrategy(SearchStrategy[Ex]): 

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

75 strategy. 

76 

77 Its parameter and distribution come from that other strategy. 

78 """ 

79 

80 def __init__( 

81 self, 

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

83 args: Sequence[object], 

84 kwargs: dict[str, object], 

85 *, 

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

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

88 force_repr: str | None = None, 

89 ): 

90 super().__init__() 

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

92 self.__representation: str | None = force_repr 

93 self.function = function 

94 self.__args = args 

95 self.__kwargs = kwargs 

96 self._transformations = transforms 

97 

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

99 return recur(self.wrapped_strategy) 

100 

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

102 return recur(self.wrapped_strategy) 

103 

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

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

106 for v in source: 

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

108 return False 

109 return True 

110 

111 def calc_label(self) -> int: 

112 return self.wrapped_strategy.label 

113 

114 @property 

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

116 if self.__wrapped_strategy is None: 

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

118 

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

120 unwrapped_kwargs = { 

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

122 } 

123 

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

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

126 _wrapped_strategy = base 

127 else: 

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

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

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

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

132 with _filter_location_override.with_value(location): 

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

134 self.__wrapped_strategy = _wrapped_strategy 

135 assert self.__wrapped_strategy is not None 

136 return self.__wrapped_strategy 

137 

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

139 repr_ = self.__representation 

140 if repr_: 

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

142 return LazyStrategy( 

143 self.function, 

144 self.__args, 

145 self.__kwargs, 

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

147 force_repr=repr_, 

148 ) 

149 

150 def map(self, pack): 

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

152 

153 def filter(self, condition): 

154 return self.__with_transform( 

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

156 ) 

157 

158 def do_validate(self) -> None: 

159 w = self.wrapped_strategy 

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

161 w.validate() 

162 

163 def __repr__(self) -> str: 

164 if self.__representation is None: 

165 sig = signature(self.function) 

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

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

168 _args, _kwargs = convert_positional_arguments( 

169 self.function, self.__args, self.__kwargs 

170 ) 

171 else: 

172 _args, _kwargs = convert_keyword_arguments( 

173 self.function, self.__args, self.__kwargs 

174 ) 

175 kwargs_for_repr = { 

176 k: v 

177 for k, v in _kwargs.items() 

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

179 } 

180 self.__representation = repr_call( 

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

182 ) + "".join( 

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

184 for method, fn, _ in self._transformations 

185 ) 

186 return self.__representation 

187 

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

189 return data.draw(self.wrapped_strategy) 

190 

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

192 return self.wrapped_strategy._invert(value)