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 json
12import os
13import sys
14import threading
15from collections.abc import Callable
16from contextlib import contextmanager
17from typing import TypeVar
18
19from hypothesis.internal.reflection import proxies
20
21"""
22This module implements a custom coverage system that records conditions and
23then validates that every condition has been seen to be both True and False
24during the execution of our tests.
25
26The only thing we use it for at present is our argument validation functions,
27where we assert that every validation function has been seen to both pass and
28fail in the course of testing.
29
30When not running with a magic environment variable set, this module disables
31itself and has essentially no overhead.
32"""
33
34Func = TypeVar("Func", bound=Callable)
35pretty_file_name_cache: dict[str, str] = {}
36
37
38def pretty_file_name(f: str) -> str:
39 try:
40 return pretty_file_name_cache[f]
41 except KeyError:
42 pass
43
44 parts = f.split(os.path.sep)
45 if "hypothesis" in parts: # pragma: no branch
46 parts = parts[-parts[::-1].index("hypothesis") :]
47 result = os.path.sep.join(parts)
48 pretty_file_name_cache[f] = result
49 return result
50
51
52IN_COVERAGE_TESTS = os.getenv("HYPOTHESIS_INTERNAL_COVERAGE") == "true"
53# Thread-local, so that tests which run Hypothesis in several threads at once
54# don't interleave their descriptions into nonsensical branch names.
55_description_stack = threading.local()
56
57
58def _get_description_stack() -> list[str]:
59 try:
60 return _description_stack.value
61 except AttributeError:
62 _description_stack.value = []
63 return _description_stack.value
64
65
66if IN_COVERAGE_TESTS:
67 # By this point, "branch-check" should have already been deleted by the
68 # tox config. We can't delete it here because of #1718.
69
70 written: set[tuple[str, bool]] = set()
71
72 def record_branch(name, value):
73 key = (name, value)
74 if key in written:
75 return
76 written.add(key)
77 with open(f"branch-check-{os.getpid()}", mode="a", encoding="utf-8") as log:
78 log.write(json.dumps({"name": name, "value": value}) + "\n")
79
80 @contextmanager
81 def check_block(name, depth):
82 # We add an extra two callers to the stack: One for the contextmanager
83 # function, one for our actual caller, so we want to go two extra
84 # stack frames up.
85 caller = sys._getframe(depth + 2)
86 fname = pretty_file_name(caller.f_code.co_filename)
87 local_description = f"{name} at {fname}:{caller.f_lineno}"
88 description_stack = _get_description_stack()
89 try:
90 description_stack.append(local_description)
91 description = " in ".join(reversed(description_stack)) + " passed"
92 yield
93 record_branch(description, True)
94 except BaseException:
95 record_branch(description, False)
96 raise
97 finally:
98 description_stack.pop()
99
100 @contextmanager
101 def check(name):
102 with check_block(name, 2):
103 yield
104
105 def check_function(f: Func) -> Func:
106 @proxies(f)
107 def accept(*args, **kwargs):
108 # depth of 2 because of the proxy function calling us.
109 with check_block(f.__name__, 2):
110 return f(*args, **kwargs)
111
112 return accept
113
114else:
115
116 def check_function(f: Func) -> Func:
117 return f
118
119 @contextmanager
120 def check(name):
121 yield