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
36T = TypeVar("T")
39def replace_all(
40 ls: Sequence[T],
41 replacements: Iterable[tuple[int, int, Sequence[T]]],
42) -> list[T]:
43 """Substitute multiple replacement values into a list.
45 Replacements is a list of (start, end, value) triples.
46 """
48 result: list[T] = []
49 prev = 0
50 offset = 0
51 for u, v, r in replacements:
52 result.extend(ls[prev:u])
53 result.extend(r)
54 prev = v
55 offset += len(r) - (v - u)
56 result.extend(ls[prev:])
57 assert len(result) == len(ls) + offset
58 return result
61class IntList(Sequence[int]):
62 """Class for storing a list of non-negative integers compactly.
64 We store them as the smallest size integer array we can get
65 away with. When we try to add an integer that is too large,
66 we upgrade the array to the smallest word size needed to store
67 the new value."""
69 ARRAY_CODES: ClassVar[list[str]] = ["B", "H", "I", "L", "Q", "O"]
70 NEXT_ARRAY_CODE: ClassVar[dict[str, str]] = dict(itertools.pairwise(ARRAY_CODES))
72 __slots__ = ("__underlying",)
74 def __init__(self, values: Sequence[int] = ()):
75 for code in self.ARRAY_CODES:
76 try:
77 underlying = self._array_or_list(code, values)
78 break
79 except OverflowError:
80 pass
81 else: # pragma: no cover
82 raise AssertionError(f"Could not create storage for {values!r}")
83 if isinstance(underlying, list):
84 for v in underlying:
85 if not isinstance(v, int) or v < 0:
86 raise ValueError(f"Could not create IntList for {values!r}")
87 self.__underlying: list[int] | ArrayType[int] = underlying
89 @classmethod
90 def of_length(cls, n: int) -> "IntList":
91 return cls(array.array("B", [0]) * n)
93 @staticmethod
94 def _array_or_list(
95 code: str, contents: Iterable[int]
96 ) -> Union[list[int], "ArrayType[int]"]:
97 if code == "O":
98 return list(contents)
99 return array.array(code, contents)
101 def count(self, value: int) -> int:
102 return self.__underlying.count(value)
104 def __repr__(self) -> str:
105 return f"IntList({list(self.__underlying)!r})"
107 def __len__(self) -> int:
108 return len(self.__underlying)
110 @overload
111 def __getitem__(self, i: int) -> int: ... # pragma: no cover
113 @overload
114 def __getitem__(
115 self, i: slice
116 ) -> "list[int] | ArrayType[int]": ... # pragma: no cover
118 def __getitem__(self, i: int | slice) -> "int | list[int] | ArrayType[int]":
119 return self.__underlying[i]
121 def __delitem__(self, i: int | slice) -> None:
122 del self.__underlying[i]
124 def insert(self, i: int, v: int) -> None:
125 self.__underlying.insert(i, v)
127 def __iter__(self) -> Iterator[int]:
128 return iter(self.__underlying)
130 def __eq__(self, other: object) -> bool:
131 if self is other:
132 return True
133 if not isinstance(other, IntList):
134 return NotImplemented
135 return self.__underlying == other.__underlying
137 def __ne__(self, other: object) -> bool:
138 if self is other:
139 return False
140 if not isinstance(other, IntList):
141 return NotImplemented
142 return self.__underlying != other.__underlying
144 def append(self, n: int) -> None:
145 # try the fast path of appending n first. If this overflows, use the
146 # __setitem__ path, which will upgrade the underlying array.
147 try:
148 self.__underlying.append(n)
149 except OverflowError:
150 i = len(self.__underlying)
151 self.__underlying.append(0)
152 self[i] = n
154 def __setitem__(self, i: int, n: int) -> None:
155 while True:
156 try:
157 self.__underlying[i] = n
158 return
159 except OverflowError:
160 assert n > 0
161 self.__upgrade()
163 def extend(self, ls: Iterable[int]) -> None:
164 for n in ls:
165 self.append(n)
167 def __upgrade(self) -> None:
168 assert isinstance(self.__underlying, array.array)
169 code = self.NEXT_ARRAY_CODE[self.__underlying.typecode]
170 self.__underlying = self._array_or_list(code, self.__underlying)
173def binary_search(lo: int, hi: int, f: Callable[[int], bool]) -> int:
174 """Binary searches in [lo , hi) to find
175 n such that f(n) == f(lo) but f(n + 1) != f(lo).
176 It is implicitly assumed and will not be checked
177 that f(hi) != f(lo).
178 """
180 reference = f(lo)
182 while lo + 1 < hi:
183 mid = (lo + hi) // 2
184 if f(mid) == reference:
185 lo = mid
186 else:
187 hi = mid
188 return lo
191class LazySequenceCopy(Generic[T]):
192 """A "copy" of a sequence that works by inserting a mask in front
193 of the underlying sequence, so that you can mutate it without changing
194 the underlying sequence. Effectively behaves as if you could do list(x)
195 in O(1) time. The full list API is not supported yet but there's no reason
196 in principle it couldn't be."""
198 def __init__(self, values: Sequence[T]):
199 self.__values = values
200 self.__len = len(values)
201 self.__mask: dict[int, T] | None = None
202 # Indices passed to pop(), in pop order, each in the coordinates that
203 # were current at the time of that pop. See __underlying_index.
204 self.__popped_at: list[int] | None = None
206 def __len__(self) -> int:
207 if self.__popped_at is None:
208 return self.__len
209 return self.__len - len(self.__popped_at)
211 def pop(self, i: int = -1) -> T:
212 if len(self) == 0:
213 raise IndexError("Cannot pop from empty list")
214 i, u = self.__underlying_index(i)
216 v = None
217 if self.__mask is not None:
218 v = self.__mask.pop(u, None)
219 if v is None:
220 v = self.__values[u]
222 if self.__popped_at is None:
223 self.__popped_at = []
224 self.__popped_at.append(i)
225 return v
227 def swap(self, i: int, j: int) -> None:
228 """Swap the elements ls[i], ls[j]."""
229 if i == j:
230 return
231 self[i], self[j] = self[j], self[i]
233 def __getitem__(self, i: int) -> T:
234 _, i = self.__underlying_index(i)
236 default = self.__values[i]
237 if self.__mask is None:
238 return default
239 else:
240 return self.__mask.get(i, default)
242 def __setitem__(self, i: int, v: T) -> None:
243 _, i = self.__underlying_index(i)
244 if self.__mask is None:
245 self.__mask = {}
246 self.__mask[i] = v
248 def __underlying_index(self, i: int) -> tuple[int, int]:
249 # given an index i in the popped representation of the list, compute
250 # its corresponding index u in the underlying list. given
251 # l = [1, 4, 2, 10, 188]
252 # l.pop(3)
253 # l.pop(1)
254 # assert l == [1, 2, 188]
255 #
256 # we want l[i] == self.__values[u], and return both the normalized
257 # (non-negative, bounds-checked) i and u.
258 #
259 # A pop at index p maps each later index x in the popped coordinates
260 # to x + (p <= x) in the coordinates current at the time of that pop,
261 # so applying this from the most recent pop back to the oldest turns i
262 # into an index into the underlying sequence. We add the comparison
263 # as an integer rather than branching on it so that, under
264 # symbolic-execution backends, symbolic indices pass through here
265 # without forcing any solver queries.
266 n = len(self)
267 if i < -n or i >= n:
268 raise IndexError(f"Index {i} out of range [0, {n})")
269 if i < 0:
270 i += n
271 assert 0 <= i < n
273 u = i
274 if self.__popped_at is not None:
275 assert len(self.__popped_at) <= len(self.__values)
276 for p in reversed(self.__popped_at):
277 u += p <= u
278 return i, u
280 # even though we have len + getitem, mypyc requires iter.
281 def __iter__(self) -> Iterable[T]:
282 for i in range(len(self)):
283 yield self[i]
286def stack_depth_of_caller() -> int:
287 """Get stack size for caller's frame.
289 From https://stackoverflow.com/a/47956089/9297601 , this is a simple
290 but much faster alternative to `len(inspect.stack(0))`. We use it
291 with get/set recursionlimit to make stack overflows non-flaky; see
292 https://github.com/HypothesisWorks/hypothesis/issues/2494 for details.
293 """
294 frame = sys._getframe(2)
295 size = 1
296 while frame:
297 frame = frame.f_back
298 size += 1
299 return size
302class StackframeLimiter:
303 # StackframeLimiter is used to make the recursion limit warning issued via
304 # ensure_free_stackframes thread-safe. We track the known values we have
305 # passed to sys.setrecursionlimit in _known_limits, and only issue a warning
306 # if sys.getrecursionlimit is not in _known_limits.
307 #
308 # This will always be an under-approximation of when we would ideally issue
309 # this warning, since a non-hypothesis caller could coincidentaly set the
310 # recursion limit to one of our known limits. Currently, StackframeLimiter
311 # resets _known_limits whenever all of the ensure_free_stackframes contexts
312 # have exited. We could increase the power of the warning by tracking a
313 # refcount for each limit, and removing it as soon as the refcount hits zero.
314 # I didn't think this extra complexity is worth the minor power increase for
315 # what is already only a "nice to have" warning.
317 def __init__(self):
318 self._active_contexts = 0
319 self._known_limits: set[int] = set()
320 self._original_limit: int | None = None
322 def _setrecursionlimit(self, new_limit: int, *, check: bool = True) -> None:
323 if (
324 check
325 and (current_limit := sys.getrecursionlimit()) not in self._known_limits
326 ):
327 warnings.warn(
328 "The recursion limit will not be reset, since it was changed "
329 f"during test execution (from {self._original_limit} to {current_limit}).",
330 HypothesisWarning,
331 stacklevel=4,
332 )
333 return
335 self._known_limits.add(new_limit)
336 sys.setrecursionlimit(new_limit)
338 def enter_context(self, new_limit: int, *, current_limit: int) -> None:
339 if self._active_contexts == 0:
340 # this is the first context on the stack. Record the true original
341 # limit, to restore later.
342 assert self._original_limit is None
343 self._original_limit = current_limit
344 self._known_limits.add(self._original_limit)
346 self._active_contexts += 1
347 self._setrecursionlimit(new_limit)
349 def exit_context(self, new_limit: int, *, check: bool = True) -> None:
350 assert self._active_contexts > 0
351 self._active_contexts -= 1
353 if self._active_contexts == 0:
354 # this is the last context to exit. Restore the true original
355 # limit and clear our known limits.
356 original_limit = self._original_limit
357 assert original_limit is not None
358 try:
359 self._setrecursionlimit(original_limit, check=check)
360 finally:
361 self._original_limit = None
362 # we want to clear the known limits, but preserve the limit
363 # we just set it to as known.
364 self._known_limits = {original_limit}
365 else:
366 self._setrecursionlimit(new_limit, check=check)
369_stackframe_limiter = StackframeLimiter()
370_stackframe_limiter_lock = Lock()
373class ensure_free_stackframes:
374 """Context manager that ensures there are at least N free stackframes (for
375 a reasonable value of N).
376 """
378 def __enter__(self) -> None:
379 cur_depth = stack_depth_of_caller()
380 with _stackframe_limiter_lock:
381 self.old_limit = sys.getrecursionlimit()
382 # The default CPython recursionlimit is 1000, but pytest seems to bump
383 # it to 3000 during test execution. Let's make it something reasonable:
384 self.new_limit = cur_depth + 2000
385 # Because we add to the recursion limit, to be good citizens we also
386 # add a check for unbounded recursion. The default limit is typically
387 # 1000/3000, so this can only ever trigger if something really strange
388 # is happening and it's hard to imagine an
389 # intentionally-deeply-recursive use of this code.
390 assert cur_depth <= 1000, (
391 f"Hypothesis would usually add {self.new_limit - self.old_limit} to "
392 f"the stack depth of {self.old_limit} here, but we are already much "
393 "deeper than expected. Aborting now, to avoid extending the stack "
394 "limit in an infinite loop..."
395 )
396 try:
397 _stackframe_limiter.enter_context(
398 self.new_limit, current_limit=self.old_limit
399 )
400 except Exception:
401 # if the stackframe limiter raises a HypothesisWarning (under eg
402 # -Werror), __exit__ is not called, since we errored in __enter__.
403 # Preserve the state of the stackframe limiter by exiting, and
404 # avoid showing a duplicate warning with check=False.
405 _stackframe_limiter.exit_context(self.old_limit, check=False)
406 raise
408 def __exit__(self, *args, **kwargs):
409 with _stackframe_limiter_lock:
410 _stackframe_limiter.exit_context(self.old_limit)
413def find_integer(f: Callable[[int], bool]) -> int:
414 """Finds a (hopefully large) integer such that f(n) is True and f(n + 1) is
415 False.
417 f(0) is assumed to be True and will not be checked.
418 """
419 # We first do a linear scan over the small numbers and only start to do
420 # anything intelligent if f(4) is true. This is because it's very hard to
421 # win big when the result is small. If the result is 0 and we try 2 first
422 # then we've done twice as much work as we needed to!
423 for i in range(1, 5):
424 if not f(i):
425 return i - 1
427 # We now know that f(4) is true. We want to find some number for which
428 # f(n) is *not* true.
429 # lo is the largest number for which we know that f(lo) is true.
430 lo = 4
432 # Exponential probe upwards until we find some value hi such that f(hi)
433 # is not true. Subsequently we maintain the invariant that hi is the
434 # smallest number for which we know that f(hi) is not true.
435 hi = 5
436 while f(hi):
437 lo = hi
438 hi *= 2
440 # Now binary search until lo + 1 = hi. At that point we have f(lo) and not
441 # f(lo + 1), as desired..
442 while lo + 1 < hi:
443 mid = (lo + hi) // 2
444 if f(mid):
445 lo = mid
446 else:
447 hi = mid
448 return lo
451_gc_initialized = False
452_gc_start: float = 0
453_gc_cumulative_time: float = 0
455# Since gc_callback potentially runs in test context, and perf_counter
456# might be monkeypatched, we store a reference to the real one.
457_perf_counter = time.perf_counter
460def gc_cumulative_time() -> float:
461 global _gc_initialized
463 # I don't believe we need a lock for the _gc_cumulative_time increment here,
464 # since afaik each gc callback is only executed once when the garbage collector
465 # runs, by the thread which initiated the gc.
467 if not _gc_initialized:
468 if hasattr(gc, "callbacks"):
469 # CPython
470 def gc_callback(
471 phase: Literal["start", "stop"], info: dict[str, int]
472 ) -> None:
473 global _gc_start, _gc_cumulative_time
474 try:
475 now = _perf_counter()
476 if phase == "start":
477 _gc_start = now
478 elif phase == "stop" and _gc_start > 0:
479 _gc_cumulative_time += now - _gc_start # pragma: no cover # ??
480 except RecursionError: # pragma: no cover
481 # Avoid flakiness via UnraisableException, which is caught and
482 # warned by pytest. The actual callback (this function) is
483 # validated to never trigger a RecursionError itself when
484 # when called by gc.collect.
485 # Anyway, we should hit the same error on "start"
486 # and "stop", but to ensure we don't get out of sync we just
487 # signal that there is no matching start.
488 _gc_start = 0
489 return
491 gc.callbacks.insert(0, gc_callback)
492 elif hasattr(gc, "hooks"): # pragma: no cover # pypy only
493 # PyPy
494 def hook(stats: Any) -> None:
495 global _gc_cumulative_time
496 try:
497 _gc_cumulative_time += stats.duration
498 except RecursionError:
499 pass
501 if gc.hooks.on_gc_minor is None:
502 gc.hooks.on_gc_minor = hook
503 if gc.hooks.on_gc_collect_step is None:
504 gc.hooks.on_gc_collect_step = hook
506 _gc_initialized = True
508 return _gc_cumulative_time
511def startswith(l1: Sequence[T], l2: Sequence[T]) -> bool:
512 if len(l1) < len(l2):
513 return False
514 return all(v1 == v2 for v1, v2 in zip(l1[: len(l2)], l2, strict=False))
517def endswith(l1: Sequence[T], l2: Sequence[T]) -> bool:
518 if len(l1) < len(l2):
519 return False
520 return all(v1 == v2 for v1, v2 in zip(l1[-len(l2) :], l2, strict=False))
523def bits_to_bytes(n: int) -> int:
524 """The number of bytes required to represent an n-bit number.
525 Equivalent to (n + 7) // 8, but slightly faster. This really is
526 called enough times that that matters."""
527 return (n + 7) >> 3