Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/hypothesis/internal/reflection.py: 71%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

273 statements  

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 

11"""This file can approximately be considered the collection of hypothesis going 

12to really unreasonable lengths to produce pretty output.""" 

13 

14import ast 

15import hashlib 

16import inspect 

17import re 

18import textwrap 

19import types 

20import warnings 

21from collections.abc import Callable, Sequence 

22from functools import partial, wraps 

23from inspect import Parameter, Signature 

24from io import StringIO 

25from keyword import iskeyword 

26from random import _inst as global_random_instance 

27from tokenize import COMMENT, generate_tokens, untokenize 

28from types import EllipsisType, ModuleType 

29from typing import TYPE_CHECKING, Any, TypeVar, Union 

30from unittest.mock import _patch as PatchType 

31 

32from hypothesis.errors import HypothesisWarning 

33from hypothesis.internal import lambda_sources 

34from hypothesis.internal.compat import is_typed_named_tuple 

35from hypothesis.utils.conventions import not_set 

36from hypothesis.vendor.pretty import pretty 

37 

38if TYPE_CHECKING: 

39 from hypothesis.strategies._internal.strategies import SearchStrategy 

40 

41T = TypeVar("T") 

42 

43 

44def is_mock(obj: object) -> bool: 

45 """Determine if the given argument is a mock type.""" 

46 

47 # We want to be able to detect these when dealing with various test 

48 # args. As they are sneaky and can look like almost anything else, 

49 # we'll check this by looking for an attribute with a name that it's really 

50 # unlikely to implement accidentally, and that anyone who implements it 

51 # deliberately should know what they're doing. This is more robust than 

52 # looking for types. 

53 return hasattr(obj, "hypothesis_internal_is_this_a_mock_check") 

54 

55 

56def _clean_source(src: str) -> bytes: 

57 """Return the source code as bytes, without decorators or comments. 

58 

59 Because this is part of our database key, we reduce the cache invalidation 

60 rate by ignoring decorators, comments, trailing whitespace, and empty lines. 

61 We can't just use the (dumped) AST directly because it changes between Python 

62 versions (e.g. ast.Constant) 

63 """ 

64 # Get the (one-indexed) line number of the function definition, and drop preceding 

65 # lines - i.e. any decorators, so that adding `@example()`s keeps the same key. 

66 try: 

67 funcdef = ast.parse(src).body[0] 

68 src = "".join(src.splitlines(keepends=True)[funcdef.lineno - 1 :]) 

69 except Exception: 

70 pass 

71 # Remove blank lines and use the tokenize module to strip out comments, 

72 # so that those can be changed without changing the database key. 

73 try: 

74 src = untokenize( 

75 t for t in generate_tokens(StringIO(src).readline) if t.type != COMMENT 

76 ) 

77 except Exception: 

78 pass 

79 # Finally, remove any trailing whitespace and empty lines as a last cleanup. 

80 return "\n".join(x.rstrip() for x in src.splitlines() if x.rstrip()).encode() 

81 

82 

83def function_digest(function: Any) -> bytes: 

84 """Returns a string that is stable across multiple invocations across 

85 multiple processes and is prone to changing significantly in response to 

86 minor changes to the function. 

87 

88 No guarantee of uniqueness though it usually will be. Digest collisions 

89 lead to unfortunate but not fatal problems during database replay. 

90 """ 

91 hasher = hashlib.sha384() 

92 try: 

93 src = inspect.getsource(function) 

94 except (OSError, TypeError): 

95 # If we can't actually get the source code, try for the name as a fallback. 

96 # NOTE: We might want to change this to always adding function.__qualname__, 

97 # to differentiate f.x. two classes having the same function implementation 

98 # with class-dependent behaviour. 

99 try: 

100 hasher.update(function.__name__.encode()) 

101 except AttributeError: 

102 pass 

103 else: 

104 hasher.update(_clean_source(src)) 

105 try: 

106 # This is additional to the source code because it can include the effects 

107 # of decorators, or of post-hoc assignment to the .__signature__ attribute. 

108 hasher.update(repr(get_signature(function)).encode()) 

109 except Exception: 

110 pass 

111 try: 

112 # We set this in order to distinguish e.g. @pytest.mark.parametrize cases. 

113 hasher.update(function._hypothesis_internal_add_digest) 

114 except AttributeError: 

115 pass 

116 return hasher.digest() 

117 

118 

119def check_signature(sig: Signature) -> None: # pragma: no cover # 3.10 only 

120 # Backport from Python 3.11; see https://github.com/python/cpython/pull/92065 

121 for p in sig.parameters.values(): 

122 if iskeyword(p.name) and p.kind is not p.POSITIONAL_ONLY: 

123 raise ValueError( 

124 f"Signature {sig!r} contains a parameter named {p.name!r}, " 

125 f"but this is a SyntaxError because `{p.name}` is a keyword. " 

126 "You, or a library you use, must have manually created an " 

127 "invalid signature - this will be an error in Python 3.11+" 

128 ) 

129 

130 

131def get_signature( 

132 target: Any, *, follow_wrapped: bool = True, eval_str: bool = False 

133) -> Signature: 

134 # Special case for use of `@unittest.mock.patch` decorator, mimicking the 

135 # behaviour of getfullargspec instead of reporting unusable arguments. 

136 patches = getattr(target, "patchings", None) 

137 if isinstance(patches, list) and all(isinstance(p, PatchType) for p in patches): 

138 return Signature( 

139 [ 

140 Parameter("args", Parameter.VAR_POSITIONAL), 

141 Parameter("keywargs", Parameter.VAR_KEYWORD), 

142 ] 

143 ) 

144 

145 if isinstance(getattr(target, "__signature__", None), Signature): 

146 # This special case covers unusual codegen like Pydantic models 

147 sig = target.__signature__ 

148 check_signature(sig) 

149 # And *this* much more complicated block ignores the `self` argument 

150 # if that's been (incorrectly) included in the custom signature. 

151 if sig.parameters and (inspect.isclass(target) or inspect.ismethod(target)): 

152 selfy = next(iter(sig.parameters.values())) 

153 if ( 

154 selfy.name == "self" 

155 and selfy.default is Parameter.empty 

156 and selfy.kind.name.startswith("POSITIONAL_") 

157 ): 

158 return sig.replace( 

159 parameters=[v for k, v in sig.parameters.items() if k != "self"] 

160 ) 

161 return sig 

162 sig = inspect.signature(target, follow_wrapped=follow_wrapped, eval_str=eval_str) 

163 check_signature(sig) 

164 return sig 

165 

166 

167def arg_is_required(param: Parameter) -> bool: 

168 return param.default is Parameter.empty and param.kind in ( 

169 Parameter.POSITIONAL_OR_KEYWORD, 

170 Parameter.KEYWORD_ONLY, 

171 ) 

172 

173 

174def required_args( 

175 target: Callable[..., Any], 

176 args: tuple["SearchStrategy[Any]", ...] = (), 

177 kwargs: dict[str, Union["SearchStrategy[Any]", EllipsisType]] | None = None, 

178) -> set[str]: 

179 """Return a set of names of required args to target that were not supplied 

180 in args or kwargs. 

181 

182 This is used in builds() to determine which arguments to attempt to 

183 fill from type hints. target may be any callable (including classes 

184 and bound methods). args and kwargs should be as they are passed to 

185 builds() - that is, a tuple of values and a dict of names: values. 

186 """ 

187 kwargs = {} if kwargs is None else kwargs 

188 # We start with a workaround for NamedTuples, which don't have nice inits 

189 if inspect.isclass(target) and is_typed_named_tuple(target): 

190 fields = target._fields # type: ignore 

191 provided = set(kwargs) | set(fields[: len(args)]) 

192 return set(fields) - provided 

193 # Then we try to do the right thing with inspect.signature 

194 try: 

195 sig = get_signature(target) 

196 except (ValueError, TypeError): 

197 return set() 

198 return { 

199 name 

200 for name, param in list(sig.parameters.items())[len(args) :] 

201 if arg_is_required(param) and name not in kwargs 

202 } 

203 

204 

205def convert_keyword_arguments( 

206 function: Any, args: Sequence[object], kwargs: dict[str, object] 

207) -> tuple[tuple[object, ...], dict[str, object]]: 

208 """Returns a pair of a tuple and a dictionary which would be equivalent 

209 passed as positional and keyword args to the function. Unless function has 

210 kwonlyargs or **kwargs the dictionary will always be empty. 

211 """ 

212 sig = inspect.signature(function, follow_wrapped=False) 

213 bound = sig.bind(*args, **kwargs) 

214 return bound.args, bound.kwargs 

215 

216 

217def convert_positional_arguments( 

218 function: Any, args: Sequence[object], kwargs: dict[str, object] 

219) -> tuple[tuple[object, ...], dict[str, object]]: 

220 """Return a tuple (new_args, new_kwargs) where all possible arguments have 

221 been moved to kwargs. 

222 

223 new_args will only be non-empty if function has pos-only args or *args. 

224 """ 

225 sig = inspect.signature(function, follow_wrapped=False) 

226 bound = sig.bind(*args, **kwargs) 

227 new_args = [] 

228 new_kwargs = dict(bound.arguments) 

229 for p in sig.parameters.values(): 

230 if p.name in new_kwargs: 

231 if p.kind is p.POSITIONAL_ONLY: 

232 new_args.append(new_kwargs.pop(p.name)) 

233 elif p.kind is p.VAR_POSITIONAL: 

234 new_args.extend(new_kwargs.pop(p.name)) 

235 elif p.kind is p.VAR_KEYWORD: 

236 assert set(new_kwargs[p.name]).isdisjoint(set(new_kwargs) - {p.name}) 

237 new_kwargs.update(new_kwargs.pop(p.name)) 

238 return tuple(new_args), new_kwargs 

239 

240 

241def ast_arguments_matches_signature(args: ast.arguments, sig: Signature) -> bool: 

242 expected: list[tuple[str, int]] = [] 

243 for node in args.posonlyargs: 

244 expected.append((node.arg, Parameter.POSITIONAL_ONLY)) 

245 for node in args.args: 

246 expected.append((node.arg, Parameter.POSITIONAL_OR_KEYWORD)) 

247 if args.vararg is not None: 

248 expected.append((args.vararg.arg, Parameter.VAR_POSITIONAL)) 

249 for node in args.kwonlyargs: 

250 expected.append((node.arg, Parameter.KEYWORD_ONLY)) 

251 if args.kwarg is not None: 

252 expected.append((args.kwarg.arg, Parameter.VAR_KEYWORD)) 

253 return expected == [(p.name, p.kind) for p in sig.parameters.values()] 

254 

255 

256def is_first_param_referenced_in_function(f: Any) -> bool: 

257 """Is the given name referenced within f?""" 

258 try: 

259 tree = ast.parse(textwrap.dedent(inspect.getsource(f))) 

260 except Exception: 

261 return True # Assume it's OK unless we know otherwise 

262 name = next(iter(get_signature(f).parameters)) 

263 return any( 

264 isinstance(node, ast.Name) 

265 and node.id == name 

266 and isinstance(node.ctx, ast.Load) 

267 for node in ast.walk(tree) 

268 ) 

269 

270 

271def get_pretty_function_description(f: object) -> str: 

272 # Anything which knows how to pretty-print itself knows better than we do - 

273 # e.g. the constant functions from st.functions(), which would otherwise 

274 # borrow the name of the function they imitate. We skip classes, whose 

275 # instance method would be called with the printer as `self`, and require 

276 # a callable so that pretty() can't bounce straight back to us. 

277 if not isinstance(f, type) and callable(getattr(f, "_repr_pretty_", None)): 

278 return pretty(f) 

279 if isinstance(f, partial): 

280 return pretty(f) 

281 if not hasattr(f, "__name__"): 

282 return repr(f) 

283 name = f.__name__ 

284 if name == "<lambda>": 

285 return lambda_sources.lambda_description(f) 

286 elif isinstance(f, (types.MethodType, types.BuiltinMethodType)): 

287 self = f.__self__ 

288 # Some objects, like `builtins.abs` are of BuiltinMethodType but have 

289 # their module as __self__. This might include c-extensions generally? 

290 if not (self is None or inspect.isclass(self) or inspect.ismodule(self)): 

291 if self is global_random_instance: 

292 return f"random.{name}" 

293 return f"{self!r}.{name}" 

294 elif isinstance(name, str) and getattr(dict, name, object()) is f: 

295 # special case for keys/values views in from_type() / ghostwriter output 

296 return f"dict.{name}" 

297 return name 

298 

299 

300def nicerepr(v: Any) -> str: 

301 if inspect.isfunction(v): 

302 return get_pretty_function_description(v) 

303 elif isinstance(v, type): 

304 return v.__name__ 

305 else: 

306 # With TypeVar T, show List[T] instead of TypeError on List[~T] 

307 return re.sub(r"(\[)~([A-Z][a-z]*\])", r"\g<1>\g<2>", pretty(v)) 

308 

309 

310def repr_call( 

311 f: Any, args: Sequence[object], kwargs: dict[str, object], *, reorder: bool = True 

312) -> str: 

313 # Note: for multi-line pretty-printing, see RepresentationPrinter.repr_call() 

314 if reorder: 

315 args, kwargs = convert_positional_arguments(f, args, kwargs) 

316 

317 bits = [nicerepr(x) for x in args] 

318 

319 for p in get_signature(f).parameters.values(): 

320 if p.name in kwargs and not p.kind.name.startswith("VAR_"): 

321 bits.append(f"{p.name}={nicerepr(kwargs.pop(p.name))}") 

322 if kwargs: 

323 for a in sorted(kwargs): 

324 bits.append(f"{a}={nicerepr(kwargs[a])}") 

325 

326 rep = nicerepr(f) 

327 if rep.startswith("lambda") and ":" in rep: 

328 rep = f"({rep})" 

329 repr_len = len(rep) + sum(len(b) for b in bits) # approx 

330 if repr_len > 30000: 

331 warnings.warn( 

332 "Generating overly large repr. This is an expensive operation, and with " 

333 f"a length of {repr_len//1000} kB is unlikely to be useful. Use -Wignore " 

334 "to ignore the warning, or -Werror to get a traceback.", 

335 HypothesisWarning, 

336 stacklevel=2, 

337 ) 

338 return rep + "(" + ", ".join(bits) + ")" 

339 

340 

341def check_valid_identifier(identifier: str) -> None: 

342 if not identifier.isidentifier(): 

343 raise ValueError(f"{identifier!r} is not a valid python identifier") 

344 

345 

346eval_cache: dict[str, ModuleType] = {} 

347 

348 

349def source_exec_as_module(source: str) -> ModuleType: 

350 try: 

351 return eval_cache[source] 

352 except KeyError: 

353 pass 

354 

355 hexdigest = hashlib.sha384(source.encode()).hexdigest() 

356 result = ModuleType("hypothesis_temporary_module_" + hexdigest) 

357 assert isinstance(source, str) 

358 exec(source, result.__dict__) 

359 eval_cache[source] = result 

360 return result 

361 

362 

363COPY_SIGNATURE_SCRIPT = """ 

364from contextlib import aclosing, closing 

365 

366from hypothesis.utils.conventions import not_set 

367 

368def accept({funcname}): 

369 {def_prefix}def {name}{signature}: 

370 {body} 

371 return {name} 

372""".lstrip() 

373 

374 

375def get_varargs( 

376 sig: Signature, kind: int = Parameter.VAR_POSITIONAL 

377) -> Parameter | None: 

378 for p in sig.parameters.values(): 

379 if p.kind is kind: 

380 return p 

381 return None 

382 

383 

384def define_function_signature(name, docstring, signature): 

385 """A decorator which sets the name, signature and docstring of the function 

386 passed into it.""" 

387 if name == "<lambda>": 

388 name = "_lambda_" 

389 check_valid_identifier(name) 

390 for a in signature.parameters: 

391 check_valid_identifier(a) 

392 

393 used_names = {*signature.parameters, name} 

394 

395 newsig = signature.replace( 

396 parameters=[ 

397 p if p.default is signature.empty else p.replace(default=not_set) 

398 for p in ( 

399 p.replace(annotation=signature.empty) 

400 for p in signature.parameters.values() 

401 ) 

402 ], 

403 return_annotation=signature.empty, 

404 ) 

405 

406 pos_args = [ 

407 p 

408 for p in signature.parameters.values() 

409 if p.kind.name.startswith("POSITIONAL_") 

410 ] 

411 

412 def accept(f): 

413 fsig = inspect.signature(f, follow_wrapped=False) 

414 must_pass_as_kwargs = [] 

415 invocation_parts = [] 

416 for p in pos_args: 

417 if p.name not in fsig.parameters and get_varargs(fsig) is None: 

418 must_pass_as_kwargs.append(p.name) 

419 else: 

420 invocation_parts.append(p.name) 

421 if get_varargs(signature) is not None: 

422 invocation_parts.append("*" + get_varargs(signature).name) 

423 for k in must_pass_as_kwargs: 

424 invocation_parts.append(f"{k}={k}") 

425 for p in signature.parameters.values(): 

426 if p.kind is p.KEYWORD_ONLY: 

427 invocation_parts.append(f"{p.name}={p.name}") 

428 varkw = get_varargs(signature, kind=Parameter.VAR_KEYWORD) 

429 if varkw: 

430 invocation_parts.append("**" + varkw.name) 

431 

432 candidate_names = ["f"] + [f"f_{i}" for i in range(1, len(used_names) + 4)] 

433 free_names = [n for n in candidate_names if n not in used_names] 

434 funcname, gen, val = free_names[:3] 

435 

436 invocation = f"{funcname}({', '.join(invocation_parts)})" 

437 # Preserve the kind of the wrapped function, so that e.g. proxies for 

438 # async or generator functions are themselves async or generators - 

439 # closing the proxy also closes the underlying (async) generator. 

440 if inspect.iscoroutinefunction(f): 

441 def_prefix, body = "async ", f"return await {invocation}" 

442 elif inspect.isasyncgenfunction(f): 

443 def_prefix = "async " 

444 body = ( 

445 f"async with aclosing({invocation}) as {gen}:\n" 

446 f" async for {val} in {gen}:\n" 

447 f" yield {val}" 

448 ) 

449 elif inspect.isgeneratorfunction(f): 

450 def_prefix = "" 

451 body = ( 

452 f"with closing({invocation}) as {gen}:\n" 

453 f" return (yield from {gen})" 

454 ) 

455 else: 

456 def_prefix, body = "", f"return {invocation}" 

457 

458 source = COPY_SIGNATURE_SCRIPT.format( 

459 name=name, 

460 funcname=funcname, 

461 signature=str(newsig), 

462 def_prefix=def_prefix, 

463 body=body, 

464 ) 

465 result = source_exec_as_module(source).accept(f) 

466 result.__doc__ = docstring 

467 result.__defaults__ = tuple( 

468 p.default 

469 for p in signature.parameters.values() 

470 if p.default is not signature.empty and "POSITIONAL" in p.kind.name 

471 ) 

472 kwdefaults = { 

473 p.name: p.default 

474 for p in signature.parameters.values() 

475 if p.default is not signature.empty and p.kind is p.KEYWORD_ONLY 

476 } 

477 if kwdefaults: 

478 result.__kwdefaults__ = kwdefaults 

479 annotations = { 

480 p.name: p.annotation 

481 for p in signature.parameters.values() 

482 if p.annotation is not signature.empty 

483 } 

484 if signature.return_annotation is not signature.empty: 

485 annotations["return"] = signature.return_annotation 

486 if annotations: 

487 result.__annotations__ = annotations 

488 return result 

489 

490 return accept 

491 

492 

493def impersonate(target): 

494 """Decorator to update the attributes of a function so that to external 

495 introspectors it will appear to be the target function. 

496 

497 Note that this updates the function in place, it doesn't return a 

498 new one. 

499 """ 

500 

501 def accept(f): 

502 # Lie shamelessly about where this code comes from, to hide the hypothesis 

503 # internals from pytest, ipython, and other runtime introspection. 

504 f.__code__ = f.__code__.replace( 

505 co_filename=target.__code__.co_filename, 

506 co_firstlineno=target.__code__.co_firstlineno, 

507 ) 

508 f.__name__ = target.__name__ 

509 f.__module__ = target.__module__ 

510 f.__doc__ = target.__doc__ 

511 f.__globals__["__hypothesistracebackhide__"] = True 

512 # But leave an breadcrumb for _describe_lambda to follow, it's 

513 # just confused by the lies above 

514 f.__wrapped_target = target 

515 return f 

516 

517 return accept 

518 

519 

520def proxies(target: T) -> Callable[[Callable], T]: 

521 replace_sig = define_function_signature( 

522 target.__name__.replace("<lambda>", "_lambda_"), # type: ignore 

523 target.__doc__, 

524 get_signature(target, follow_wrapped=False), 

525 ) 

526 

527 def accept(proxy): 

528 return impersonate(target)(wraps(target)(replace_sig(proxy))) 

529 

530 return accept 

531 

532 

533def is_identity_function(f: Callable) -> bool: 

534 try: 

535 code = f.__code__ 

536 except AttributeError: 

537 try: 

538 f = f.__call__ # type: ignore 

539 code = f.__code__ 

540 except AttributeError: 

541 return False 

542 

543 # We only accept a single unbound argument. While it would be possible to 

544 # accept extra defaulted arguments, it would be pointless as they couldn't 

545 # be referenced at all in the code object (or the co_code check would fail). 

546 bound_args = int(inspect.ismethod(f)) 

547 if code.co_argcount != bound_args + 1 or code.co_kwonlyargcount > 0: 

548 return False 

549 

550 # We know that f accepts a single positional argument, now check that its 

551 # code object is simply "return first unbound argument". 

552 template = (lambda self, x: x) if bound_args else (lambda x: x) 

553 try: 

554 return code.co_code == template.__code__.co_code 

555 except AttributeError: # pragma: no cover # pypy only 

556 # In PyPy, some builtin functions have a code object ('builtin-code') 

557 # lacking co_code, perhaps because they are native-compiled and don't have 

558 # a corresponding bytecode. Regardless, since Python doesn't have any 

559 # builtin identity function it seems safe to say that this one isn't 

560 return False