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
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/.
11"""A module for miscellaneous useful bits and bobs that don't
12obviously belong anywhere else. If you spot a better home for
13anything that lives here, please move it."""
15import array
16import gc
17import itertools
18import sys
19import time
20import warnings
21from array import ArrayType
22from collections.abc import Callable, Iterable, Iterator, Sequence
23from threading import Lock
24from typing import (
25 Any,
26 ClassVar,
27 Generic,
28 Literal,
29 TypeVar,
30 Union,
31 overload,
32)
34from hypothesis.errors import HypothesisWarning
35from hypothesis.internal.floats import float_to_int
37T = TypeVar("T")
40def equal_values(a: Any, b: Any) -> bool:
41 """Like ==, but requires identical types (so True != 1 != 1.0) and
42 compares floats by bitpattern (so nan == nan and 0.0 != -0.0), applying
43 both rules elementwise through lists and tuples.
45 Values of any other type - dicts, sets, arbitrary objects - are compared
46 by plain == after the type check, so a NaN, signed zero, or bool/int
47 confusion nested inside them is missed: a NaN there compares unequal and
48 a nested True matches 1. Callers use this for best-effort,
49 replay-verified inversion, where such a miss costs a widening opportunity
50 - the replay produces a wrong or unequal value and is rejected - never
51 correctness.
52 """
53 if type(a) is not type(b):
54 return False
55 if isinstance(a, float):
56 # by bitpattern, matching choice_equal
57 return float_to_int(a) == float_to_int(b)
58 if isinstance(a, (list, tuple)):
59 return len(a) == len(b) and all(map(equal_values, a, b))
60 return bool(a == b)
63def replace_all(
64 ls: Sequence[T],
65 replacements: Iterable[tuple[int, int, Sequence[T]]],
66) -> list[T]:
67 """Substitute multiple replacement values into a list.
69 Replacements is a list of (start, end, value) triples.
70 """
72 result: list[T] = []
73 prev = 0
74 offset = 0
75 for u, v, r in replacements:
76 result.extend(ls[prev:u])
77 result.extend(r)
78 prev = v
79 offset += len(r) - (v - u)
80 result.extend(ls[prev:])
81 assert len(result) == len(ls) + offset
82 return result
85class IntList(Sequence[int]):
86 """Class for storing a list of non-negative integers compactly.
88 We store them as the smallest size integer array we can get
89 away with. When we try to add an integer that is too large,
90 we upgrade the array to the smallest word size needed to store
91 the new value."""
93 ARRAY_CODES: ClassVar[list[str]] = ["B", "H", "I", "L", "Q", "O"]
94 NEXT_ARRAY_CODE: ClassVar[dict[str, str]] = dict(itertools.pairwise(ARRAY_CODES))
96 __slots__ = ("__underlying",)
98 def __init__(self, values: Sequence[int] = ()):
99 for code in self.ARRAY_CODES:
100 try:
101 underlying = self._array_or_list(code, values)
102 break
103 except OverflowError:
104 pass
105 else: # pragma: no cover
106 raise AssertionError(f"Could not create storage for {values!r}")
107 if isinstance(underlying, list):
108 for v in underlying:
109 if not isinstance(v, int) or v < 0:
110 raise ValueError(f"Could not create IntList for {values!r}")
111 self.__underlying: list[int] | ArrayType[int] = underlying
113 @classmethod
114 def of_length(cls, n: int) -> "IntList":
115 return cls(array.array("B", [0]) * n)
117 @staticmethod
118 def _array_or_list(
119 code: str, contents: Iterable[int]
120 ) -> Union[list[int], "ArrayType[int]"]:
121 if code == "O":
122 return list(contents)
123 return array.array(code, contents)
125 def count(self, value: int) -> int:
126 return self.__underlying.count(value)
128 def __repr__(self) -> str:
129 return f"IntList({list(self.__underlying)!r})"
131 def __len__(self) -> int:
132 return len(self.__underlying)
134 @overload
135 def __getitem__(self, i: int) -> int: ...
137 @overload
138 def __getitem__(self, i: slice) -> "list[int] | ArrayType[int]": ...
140 def __getitem__(self, i: int | slice) -> "int | list[int] | ArrayType[int]":
141 return self.__underlying[i]
143 def __delitem__(self, i: int | slice) -> None:
144 del self.__underlying[i]
146 def insert(self, i: int, v: int) -> None:
147 self.__underlying.insert(i, v)
149 def __iter__(self) -> Iterator[int]:
150 return iter(self.__underlying)
152 def __eq__(self, other: object) -> bool:
153 if self is other:
154 return True
155 if not isinstance(other, IntList):
156 return NotImplemented
157 return self.__underlying == other.__underlying
159 def __ne__(self, other: object) -> bool:
160 if self is other:
161 return False
162 if not isinstance(other, IntList):
163 return NotImplemented
164 return self.__underlying != other.__underlying
166 def append(self, n: int) -> None:
167 # try the fast path of appending n first. If this overflows, use the
168 # __setitem__ path, which will upgrade the underlying array.
169 try:
170 self.__underlying.append(n)
171 except OverflowError:
172 i = len(self.__underlying)
173 self.__underlying.append(0)
174 self[i] = n
176 def __setitem__(self, i: int, n: int) -> None:
177 while True:
178 try:
179 self.__underlying[i] = n
180 return
181 except OverflowError:
182 assert n > 0
183 self.__upgrade()
185 def extend(self, ls: Iterable[int]) -> None:
186 for n in ls:
187 self.append(n)
189 def __upgrade(self) -> None:
190 assert isinstance(self.__underlying, array.array)
191 code = self.NEXT_ARRAY_CODE[self.__underlying.typecode]
192 self.__underlying = self._array_or_list(code, self.__underlying)
195def binary_search(lo: int, hi: int, f: Callable[[int], bool]) -> int:
196 """Binary searches in [lo , hi) to find
197 n such that f(n) == f(lo) but f(n + 1) != f(lo).
198 It is implicitly assumed and will not be checked
199 that f(hi) != f(lo).
200 """
202 reference = f(lo)
204 while lo + 1 < hi:
205 mid = (lo + hi) // 2
206 if f(mid) == reference:
207 lo = mid
208 else:
209 hi = mid
210 return lo
213class LazySequenceCopy(Generic[T]):
214 """A "copy" of a sequence that works by inserting a mask in front
215 of the underlying sequence, so that you can mutate it without changing
216 the underlying sequence. Effectively behaves as if you could do list(x)
217 in O(1) time. The full list API is not supported yet but there's no reason
218 in principle it couldn't be."""
220 def __init__(self, values: Sequence[T]):
221 self.__values = values
222 self.__len = len(values)
223 self.__mask: dict[int, T] | None = None
224 # Indices passed to pop(), in pop order, each in the coordinates that
225 # were current at the time of that pop. See __underlying_index.
226 self.__popped_at: list[int] | None = None
228 def __len__(self) -> int:
229 if self.__popped_at is None:
230 return self.__len
231 return self.__len - len(self.__popped_at)
233 def pop(self, i: int = -1) -> T:
234 if len(self) == 0:
235 raise IndexError("Cannot pop from empty list")
236 i, u = self.__underlying_index(i)
238 v = None
239 if self.__mask is not None:
240 v = self.__mask.pop(u, None)
241 if v is None:
242 v = self.__values[u]
244 if self.__popped_at is None:
245 self.__popped_at = []
246 self.__popped_at.append(i)
247 return v
249 def swap(self, i: int, j: int) -> None:
250 """Swap the elements ls[i], ls[j]."""
251 if i == j:
252 return
253 self[i], self[j] = self[j], self[i]
255 def __getitem__(self, i: int) -> T:
256 _, i = self.__underlying_index(i)
258 default = self.__values[i]
259 if self.__mask is None:
260 return default
261 else:
262 return self.__mask.get(i, default)
264 def __setitem__(self, i: int, v: T) -> None:
265 _, i = self.__underlying_index(i)
266 if self.__mask is None:
267 self.__mask = {}
268 self.__mask[i] = v
270 def __underlying_index(self, i: int) -> tuple[int, int]:
271 # given an index i in the popped representation of the list, compute
272 # its corresponding index u in the underlying list. given
273 # l = [1, 4, 2, 10, 188]
274 # l.pop(3)
275 # l.pop(1)
276 # assert l == [1, 2, 188]
277 #
278 # we want l[i] == self.__values[u], and return both the normalized
279 # (non-negative, bounds-checked) i and u.
280 #
281 # A pop at index p maps each later index x in the popped coordinates
282 # to x + (p <= x) in the coordinates current at the time of that pop,
283 # so applying this from the most recent pop back to the oldest turns i
284 # into an index into the underlying sequence. We add the comparison
285 # as an integer rather than branching on it so that, under
286 # symbolic-execution backends, symbolic indices pass through here
287 # without forcing any solver queries.
288 n = len(self)
289 if i < -n or i >= n:
290 raise IndexError(f"Index {i} out of range [0, {n})")
291 if i < 0:
292 i += n
293 assert 0 <= i < n
295 u = i
296 if self.__popped_at is not None:
297 assert len(self.__popped_at) <= len(self.__values)
298 for p in reversed(self.__popped_at):
299 u += p <= u
300 return i, u
302 # even though we have len + getitem, mypyc requires iter.
303 def __iter__(self) -> Iterable[T]:
304 for i in range(len(self)):
305 yield self[i]
308def stack_depth_of_caller() -> int:
309 """Get stack size for caller's frame.
311 From https://stackoverflow.com/a/47956089/9297601 , this is a simple
312 but much faster alternative to `len(inspect.stack(0))`. We use it
313 with get/set recursionlimit to make stack overflows non-flaky; see
314 https://github.com/HypothesisWorks/hypothesis/issues/2494 for details.
315 """
316 frame = sys._getframe(2)
317 size = 1
318 while frame:
319 frame = frame.f_back
320 size += 1
321 return size
324class StackframeLimiter:
325 # StackframeLimiter is used to make the recursion limit warning issued via
326 # ensure_free_stackframes thread-safe. We track the known values we have
327 # passed to sys.setrecursionlimit in _known_limits, and only issue a warning
328 # if sys.getrecursionlimit is not in _known_limits.
329 #
330 # This will always be an under-approximation of when we would ideally issue
331 # this warning, since a non-hypothesis caller could coincidentaly set the
332 # recursion limit to one of our known limits. Currently, StackframeLimiter
333 # resets _known_limits whenever all of the ensure_free_stackframes contexts
334 # have exited. We could increase the power of the warning by tracking a
335 # refcount for each limit, and removing it as soon as the refcount hits zero.
336 # I didn't think this extra complexity is worth the minor power increase for
337 # what is already only a "nice to have" warning.
339 def __init__(self):
340 self._active_contexts = 0
341 self._known_limits: set[int] = set()
342 self._original_limit: int | None = None
344 def _setrecursionlimit(self, new_limit: int, *, check: bool = True) -> None:
345 if (
346 check
347 and (current_limit := sys.getrecursionlimit()) not in self._known_limits
348 ):
349 warnings.warn(
350 "The recursion limit will not be reset, since it was changed "
351 f"during test execution (from {self._original_limit} to {current_limit}).",
352 HypothesisWarning,
353 stacklevel=4,
354 )
355 return
357 self._known_limits.add(new_limit)
358 sys.setrecursionlimit(new_limit)
360 def enter_context(self, new_limit: int, *, current_limit: int) -> None:
361 if self._active_contexts == 0:
362 # this is the first context on the stack. Record the true original
363 # limit, to restore later.
364 assert self._original_limit is None
365 self._original_limit = current_limit
366 self._known_limits.add(self._original_limit)
368 self._active_contexts += 1
369 self._setrecursionlimit(new_limit)
371 def exit_context(self, new_limit: int, *, check: bool = True) -> None:
372 assert self._active_contexts > 0
373 self._active_contexts -= 1
375 if self._active_contexts == 0:
376 # this is the last context to exit. Restore the true original
377 # limit and clear our known limits.
378 original_limit = self._original_limit
379 assert original_limit is not None
380 try:
381 self._setrecursionlimit(original_limit, check=check)
382 finally:
383 self._original_limit = None
384 # we want to clear the known limits, but preserve the limit
385 # we just set it to as known.
386 self._known_limits = {original_limit}
387 else:
388 self._setrecursionlimit(new_limit, check=check)
391_stackframe_limiter = StackframeLimiter()
392_stackframe_limiter_lock = Lock()
395class ensure_free_stackframes:
396 """Context manager that ensures there are at least N free stackframes (for
397 a reasonable value of N).
398 """
400 def __enter__(self) -> None:
401 cur_depth = stack_depth_of_caller()
402 with _stackframe_limiter_lock:
403 self.old_limit = sys.getrecursionlimit()
404 # The default CPython recursionlimit is 1000, but pytest seems to bump
405 # it to 3000 during test execution. Let's make it something reasonable:
406 self.new_limit = cur_depth + 2000
407 # Because we add to the recursion limit, to be good citizens we also
408 # add a check for unbounded recursion. The default limit is typically
409 # 1000/3000, so this can only ever trigger if something really strange
410 # is happening and it's hard to imagine an
411 # intentionally-deeply-recursive use of this code.
412 assert cur_depth <= 1000, (
413 f"Hypothesis would usually add {self.new_limit - self.old_limit} to "
414 f"the stack depth of {self.old_limit} here, but we are already much "
415 "deeper than expected. Aborting now, to avoid extending the stack "
416 "limit in an infinite loop..."
417 )
418 try:
419 _stackframe_limiter.enter_context(
420 self.new_limit, current_limit=self.old_limit
421 )
422 except Exception:
423 # if the stackframe limiter raises a HypothesisWarning (under eg
424 # -Werror), __exit__ is not called, since we errored in __enter__.
425 # Preserve the state of the stackframe limiter by exiting, and
426 # avoid showing a duplicate warning with check=False.
427 _stackframe_limiter.exit_context(self.old_limit, check=False)
428 raise
430 def __exit__(self, *args, **kwargs):
431 with _stackframe_limiter_lock:
432 _stackframe_limiter.exit_context(self.old_limit)
435def find_integer(f: Callable[[int], bool]) -> int:
436 """Finds a (hopefully large) integer such that f(n) is True and f(n + 1) is
437 False.
439 f(0) is assumed to be True and will not be checked.
440 """
441 # We first do a linear scan over the small numbers and only start to do
442 # anything intelligent if f(4) is true. This is because it's very hard to
443 # win big when the result is small. If the result is 0 and we try 2 first
444 # then we've done twice as much work as we needed to!
445 for i in range(1, 5):
446 if not f(i):
447 return i - 1
449 # We now know that f(4) is true. We want to find some number for which
450 # f(n) is *not* true.
451 # lo is the largest number for which we know that f(lo) is true.
452 lo = 4
454 # Exponential probe upwards until we find some value hi such that f(hi)
455 # is not true. Subsequently we maintain the invariant that hi is the
456 # smallest number for which we know that f(hi) is not true.
457 hi = 5
458 while f(hi):
459 lo = hi
460 hi *= 2
462 # Now binary search until lo + 1 = hi. At that point we have f(lo) and not
463 # f(lo + 1), as desired..
464 while lo + 1 < hi:
465 mid = (lo + hi) // 2
466 if f(mid):
467 lo = mid
468 else:
469 hi = mid
470 return lo
473_gc_initialized = False
474_gc_start: float = 0
475_gc_cumulative_time: float = 0
477# Since gc_callback potentially runs in test context, and perf_counter
478# might be monkeypatched, we store a reference to the real one.
479_perf_counter = time.perf_counter
482def gc_cumulative_time() -> float:
483 global _gc_initialized
485 # I don't believe we need a lock for the _gc_cumulative_time increment here,
486 # since afaik each gc callback is only executed once when the garbage collector
487 # runs, by the thread which initiated the gc.
489 if not _gc_initialized:
490 if hasattr(gc, "callbacks"):
491 # CPython
492 def gc_callback(
493 phase: Literal["start", "stop"], info: dict[str, int]
494 ) -> None:
495 global _gc_start, _gc_cumulative_time
496 try:
497 now = _perf_counter()
498 if phase == "start":
499 _gc_start = now
500 elif phase == "stop" and _gc_start > 0:
501 _gc_cumulative_time += now - _gc_start # pragma: no cover # ??
502 except RecursionError: # pragma: no cover
503 # Avoid flakiness via UnraisableException, which is caught and
504 # warned by pytest. The actual callback (this function) is
505 # validated to never trigger a RecursionError itself when
506 # when called by gc.collect.
507 # Anyway, we should hit the same error on "start"
508 # and "stop", but to ensure we don't get out of sync we just
509 # signal that there is no matching start.
510 _gc_start = 0
511 return
513 gc.callbacks.insert(0, gc_callback)
514 elif hasattr(gc, "hooks"): # pragma: no cover # pypy only
515 # PyPy
516 def hook(stats: Any) -> None:
517 global _gc_cumulative_time
518 try:
519 _gc_cumulative_time += stats.duration
520 except RecursionError:
521 pass
523 if gc.hooks.on_gc_minor is None:
524 gc.hooks.on_gc_minor = hook
525 if gc.hooks.on_gc_collect_step is None:
526 gc.hooks.on_gc_collect_step = hook
528 _gc_initialized = True
530 return _gc_cumulative_time
533def startswith(l1: Sequence[T], l2: Sequence[T]) -> bool:
534 if len(l1) < len(l2):
535 return False
536 return all(v1 == v2 for v1, v2 in zip(l1[: len(l2)], l2, strict=False))
539def endswith(l1: Sequence[T], l2: Sequence[T]) -> bool:
540 if len(l1) < len(l2):
541 return False
542 return all(v1 == v2 for v1, v2 in zip(l1[-len(l2) :], l2, strict=False))
545def bits_to_bytes(n: int) -> int:
546 """The number of bytes required to represent an n-bit number.
547 Equivalent to (n + 7) // 8, but slightly faster. This really is
548 called enough times that that matters."""
549 return (n + 7) >> 3