Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/hypothesis/database.py: 27%
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/.
11import abc
12import errno
13import json
14import os
15import struct
16import sys
17import tempfile
18import warnings
19import weakref
20from collections.abc import Callable, Iterable
21from datetime import datetime, timedelta, timezone
22from functools import lru_cache
23from hashlib import sha384
24from os import PathLike, getenv
25from pathlib import Path, PurePath
26from queue import Queue
27from threading import Thread
28from typing import (
29 TYPE_CHECKING,
30 Any,
31 ClassVar,
32 Literal,
33 TypeAlias,
34 cast,
35)
36from zipfile import BadZipFile, ZipFile
38from hypothesis.configuration import StorageDirectory, storage_directory
39from hypothesis.errors import HypothesisException, HypothesisWarning
40from hypothesis.internal.conjecture.choice import ChoiceT
41from hypothesis.utils.conventions import UniqueIdentifier, not_set
42from hypothesis.utils.deprecation import note_deprecation
44__all__ = [
45 "DirectoryBasedExampleDatabase",
46 "ExampleDatabase",
47 "GitHubArtifactDatabase",
48 "InMemoryExampleDatabase",
49 "MultiplexedDatabase",
50 "ReadOnlyDatabase",
51]
53if TYPE_CHECKING:
54 from watchdog.observers.api import BaseObserver
56StrPathT: TypeAlias = str | PathLike[str]
57SaveDataT: TypeAlias = tuple[bytes, bytes] # key, value
58DeleteDataT: TypeAlias = tuple[bytes, bytes | None] # key, value
59ListenerEventT: TypeAlias = (
60 tuple[Literal["save"], SaveDataT] | tuple[Literal["delete"], DeleteDataT]
61)
62ListenerT: TypeAlias = Callable[[ListenerEventT], Any]
65def _usable_dir(path: StrPathT) -> bool:
66 """
67 Returns True if the desired path can be used as database path because
68 either the directory exists and can be used, or its root directory can
69 be used and we can make the directory as needed.
70 """
71 path = Path(path)
72 try:
73 while not path.exists():
74 # Loop terminates because the root dir ('/' on unix) always exists.
75 path = path.parent
76 return path.is_dir() and os.access(path, os.R_OK | os.W_OK | os.X_OK)
77 except PermissionError: # pragma: no cover
78 # path.exists() returns False on 3.14+ instead of raising. See
79 # https://docs.python.org/3.14/library/pathlib.html#querying-file-type-and-status
80 return False
83def _db_for_path(
84 path: StrPathT | UniqueIdentifier | Literal[":memory:"] | None = None,
85) -> "ExampleDatabase":
86 if path is not_set:
87 if os.getenv("HYPOTHESIS_DATABASE_FILE") is not None: # pragma: no cover
88 raise HypothesisException(
89 "The $HYPOTHESIS_DATABASE_FILE environment variable no longer has any "
90 "effect. Configure your database location via a settings profile instead.\n"
91 "https://hypothesis.readthedocs.io/en/latest/settings.html#settings-profiles"
92 )
94 storage_dir = storage_directory("examples", intent_to_write=False)
95 if not _usable_dir(storage_dir.path): # pragma: no cover
96 warnings.warn(
97 "The database setting is not configured, and the default "
98 "location is unusable - falling back to an in-memory "
99 f"database for this session. path={storage_dir.path!r}",
100 HypothesisWarning,
101 stacklevel=3,
102 )
103 return InMemoryExampleDatabase()
104 return _StorageDirectoryDatabase(storage_dir)
105 if path in (None, ":memory:"):
106 return InMemoryExampleDatabase()
107 path = cast(StrPathT, path)
108 return DirectoryBasedExampleDatabase(path)
111class _EDMeta(abc.ABCMeta):
112 def __call__(self, *args: Any, **kwargs: Any) -> "ExampleDatabase":
113 if self is ExampleDatabase:
114 note_deprecation(
115 "Creating a database using the abstract ExampleDatabase() class "
116 "is deprecated. Prefer using a concrete subclass, like "
117 "InMemoryExampleDatabase() or DirectoryBasedExampleDatabase(path). "
118 'In particular, the special string ExampleDatabase(":memory:") '
119 "should be replaced by InMemoryExampleDatabase().",
120 since="2025-04-07",
121 has_codemod=False,
122 )
123 return _db_for_path(*args, **kwargs)
124 return super().__call__(*args, **kwargs)
127# This __call__ method is picked up by Sphinx as the signature of all ExampleDatabase
128# subclasses, which is accurate, reasonable, and unhelpful. Fortunately Sphinx
129# maintains a list of metaclass-call-methods to ignore, and while they would prefer
130# not to maintain it upstream (https://github.com/sphinx-doc/sphinx/pull/8262) we
131# can insert ourselves here.
132#
133# This code only runs if Sphinx has already been imported; and it would live in our
134# docs/conf.py except that we would also like it to work for anyone documenting
135# downstream ExampleDatabase subclasses too.
136#
137# We avoid type-checking this block due to this combination facts:
138# * our check-types-api CI job runs under 3.14
139# * tools.txt therefore pins to a newer version of sphinx which uses 3.12+ `type`
140# syntax
141# * in test_mypy.py, mypy sees this block, sees sphinx is installed, tries parsing
142# sphinx code, and errors
143#
144# Putting `and not TYPE_CHECKING` here is just a convenience for our testing setup
145# (because we don't split mypy tests by running CI version, eg), not for runtime
146# behavior.
147if "sphinx" in sys.modules and not TYPE_CHECKING: # pragma: no cover
148 try:
149 import sphinx.ext.autodoc
151 signature = "hypothesis.database._EDMeta.__call__"
153 # _METACLASS_CALL_BLACKLIST moved in newer sphinx versions
154 try:
155 import sphinx.ext.autodoc._dynamic._signatures as _module
156 except ImportError:
157 _module = sphinx.ext.autodoc
159 # _METACLASS_CALL_BLACKLIST is a frozenset in later sphinx versions
160 if isinstance(_module._METACLASS_CALL_BLACKLIST, frozenset):
161 _module._METACLASS_CALL_BLACKLIST = _module._METACLASS_CALL_BLACKLIST | {
162 signature
163 }
164 else:
165 _module._METACLASS_CALL_BLACKLIST.append(signature)
166 except Exception:
167 pass
170class ExampleDatabase(metaclass=_EDMeta):
171 """
172 A Hypothesis database, for use in |settings.database|.
174 Hypothesis automatically saves failures to the database set in
175 |settings.database|. The next time the test is run, Hypothesis will replay
176 any failures from the database in |settings.database| for that test (in
177 |Phase.reuse|).
179 The database is best thought of as a cache that you never need to invalidate.
180 Entries may be transparently dropped when upgrading your Hypothesis version
181 or changing your test. Do not rely on the database for correctness; to ensure
182 Hypothesis always tries an input, use |@example|.
184 A Hypothesis database is a simple mapping of bytes to sets of bytes. Hypothesis
185 provides several concrete database subclasses. To write your own database class,
186 see :doc:`/how-to/custom-database`.
188 Change listening
189 ----------------
191 An optional extension to |ExampleDatabase| is change listening. On databases
192 which support change listening, calling |ExampleDatabase.add_listener| adds
193 a function as a change listener, which will be called whenever a value is
194 added, deleted, or moved inside the database. See |ExampleDatabase.add_listener|
195 for details.
197 All databases in Hypothesis support change listening. Custom database classes
198 are not required to support change listening, though they will not be compatible
199 with features that require change listening until they do so.
201 .. note::
203 While no Hypothesis features currently require change listening, change
204 listening is required by `HypoFuzz <https://hypofuzz.com/>`_.
206 Database methods
207 ----------------
209 Required methods:
211 * |ExampleDatabase.save|
212 * |ExampleDatabase.fetch|
213 * |ExampleDatabase.delete|
215 Optional methods:
217 * |ExampleDatabase.move|
219 Change listening methods:
221 * |ExampleDatabase.add_listener|
222 * |ExampleDatabase.remove_listener|
223 * |ExampleDatabase.clear_listeners|
224 * |ExampleDatabase._start_listening|
225 * |ExampleDatabase._stop_listening|
226 * |ExampleDatabase._broadcast_change|
227 """
229 def __init__(self) -> None:
230 self._listeners: list[ListenerT] = []
232 @abc.abstractmethod
233 def save(self, key: bytes, value: bytes) -> None:
234 """Save ``value`` under ``key``.
236 If ``value`` is already present in ``key``, silently do nothing.
237 """
238 raise NotImplementedError(f"{type(self).__name__}.save")
240 @abc.abstractmethod
241 def fetch(self, key: bytes) -> Iterable[bytes]:
242 """Return an iterable over all values matching this key."""
243 raise NotImplementedError(f"{type(self).__name__}.fetch")
245 @abc.abstractmethod
246 def delete(self, key: bytes, value: bytes) -> None:
247 """Remove ``value`` from ``key``.
249 If ``value`` is not present in ``key``, silently do nothing.
250 """
251 raise NotImplementedError(f"{type(self).__name__}.delete")
253 def move(self, src: bytes, dest: bytes, value: bytes) -> None:
254 """
255 Move ``value`` from key ``src`` to key ``dest``.
257 Equivalent to ``delete(src, value)`` followed by ``save(src, value)``,
258 but may have a more efficient implementation.
260 Note that ``value`` will be inserted at ``dest`` regardless of whether
261 it is currently present at ``src``.
262 """
263 if src == dest:
264 self.save(src, value)
265 return
266 self.delete(src, value)
267 self.save(dest, value)
269 def add_listener(self, f: ListenerT, /) -> None:
270 """
271 Add a change listener. ``f`` will be called whenever a value is saved,
272 deleted, or moved in the database.
274 ``f`` can be called with two different event values:
276 * ``("save", (key, value))``
277 * ``("delete", (key, value))``
279 where ``key`` and ``value`` are both ``bytes``.
281 There is no ``move`` event. Instead, a move is broadcasted as a
282 ``delete`` event followed by a ``save`` event.
284 For the ``delete`` event, ``value`` may be ``None``. This might occur if
285 the database knows that a deletion has occurred in ``key``, but does not
286 know what value was deleted.
287 """
288 had_listeners = bool(self._listeners)
289 self._listeners.append(f)
290 if not had_listeners:
291 self._start_listening()
293 def remove_listener(self, f: ListenerT, /) -> None:
294 """
295 Removes ``f`` from the list of change listeners.
297 If ``f`` is not in the list of change listeners, silently do nothing.
298 """
299 if f not in self._listeners:
300 return
301 self._listeners.remove(f)
302 if not self._listeners:
303 self._stop_listening()
305 def clear_listeners(self) -> None:
306 """Remove all change listeners."""
307 had_listeners = bool(self._listeners)
308 self._listeners.clear()
309 if had_listeners:
310 self._stop_listening()
312 def _broadcast_change(self, event: ListenerEventT) -> None:
313 """
314 Called when a value has been either added to or deleted from a key in
315 the underlying database store. The possible values for ``event`` are:
317 * ``("save", (key, value))``
318 * ``("delete", (key, value))``
320 ``value`` may be ``None`` for the ``delete`` event, indicating we know
321 that some value was deleted under this key, but not its exact value.
323 Note that you should not assume your instance is the only reference to
324 the underlying database store. For example, if two instances of
325 |DirectoryBasedExampleDatabase| reference the same directory,
326 _broadcast_change should be called whenever a file is added or removed
327 from the directory, even if that database was not responsible for
328 changing the file.
329 """
330 for listener in self._listeners:
331 listener(event)
333 def _start_listening(self) -> None:
334 """
335 Called when the database adds a change listener, and did not previously
336 have any change listeners. Intended to allow databases to wait to start
337 expensive listening operations until necessary.
339 ``_start_listening`` and ``_stop_listening`` are guaranteed to alternate,
340 so you do not need to handle the case of multiple consecutive
341 ``_start_listening`` calls without an intermediate ``_stop_listening``
342 call.
343 """
344 warnings.warn(
345 f"{self.__class__} does not support listening for changes",
346 HypothesisWarning,
347 stacklevel=4,
348 )
350 def _stop_listening(self) -> None:
351 """
352 Called whenever no change listeners remain on the database.
354 ``_stop_listening`` and ``_start_listening`` are guaranteed to alternate,
355 so you do not need to handle the case of multiple consecutive
356 ``_stop_listening`` calls without an intermediate ``_start_listening``
357 call.
358 """
359 warnings.warn(
360 f"{self.__class__} does not support stopping listening for changes",
361 HypothesisWarning,
362 stacklevel=4,
363 )
366class InMemoryExampleDatabase(ExampleDatabase):
367 """A non-persistent example database, implemented in terms of an in-memory
368 dictionary.
370 This can be useful if you call a test function several times in a single
371 session, or for testing other database implementations, but because it
372 does not persist between runs we do not recommend it for general use.
373 """
375 def __init__(self) -> None:
376 super().__init__()
377 self.data: dict[bytes, set[bytes]] = {}
379 def __repr__(self) -> str:
380 return f"InMemoryExampleDatabase({self.data!r})"
382 def __eq__(self, other: object) -> bool:
383 return isinstance(other, InMemoryExampleDatabase) and self.data is other.data
385 def fetch(self, key: bytes) -> Iterable[bytes]:
386 yield from self.data.get(key, ())
388 def save(self, key: bytes, value: bytes) -> None:
389 value = bytes(value)
390 values = self.data.setdefault(key, set())
391 changed = value not in values
392 values.add(value)
394 if changed:
395 self._broadcast_change(("save", (key, value)))
397 def delete(self, key: bytes, value: bytes) -> None:
398 value = bytes(value)
399 values = self.data.get(key, set())
400 changed = value in values
401 values.discard(value)
403 if changed:
404 self._broadcast_change(("delete", (key, value)))
406 def _start_listening(self) -> None:
407 # declare compatibility with the listener api, but do the actual
408 # implementation in .delete and .save, since we know we are the only
409 # writer to .data.
410 pass
412 def _stop_listening(self) -> None:
413 pass
416def _hash(key: bytes) -> str:
417 return sha384(key).hexdigest()[:16]
420class DirectoryBasedExampleDatabase(ExampleDatabase):
421 """Use a directory to store Hypothesis examples as files.
423 Each test corresponds to a directory, and each example to a file within that
424 directory. While the contents are fairly opaque, a
425 |DirectoryBasedExampleDatabase| can be shared by checking the directory
426 into version control, for example with the following ``.gitignore``::
428 # Ignore files cached by Hypothesis...
429 .hypothesis/*
430 # except for the examples directory
431 !.hypothesis/examples/
433 Note however that this only makes sense if you also pin to an exact version of
434 Hypothesis, and we would usually recommend implementing a shared database with
435 a network datastore - see |ExampleDatabase|, and the |MultiplexedDatabase| helper.
436 """
438 # we keep a database entry of the full values of all the database keys.
439 # currently only used for inverse mapping of hash -> key in change listening.
440 _metakeys_name: ClassVar[bytes] = b".hypothesis-keys"
441 _metakeys_hash: ClassVar[str] = _hash(_metakeys_name)
443 def __init__(self, path: StrPathT) -> None:
444 super().__init__()
445 self.path = Path(path)
446 self.keypaths: dict[bytes, Path] = {}
447 self._observer: BaseObserver | None = None
448 self._ensure_directory_exists_called = False
450 def _ensure_directory_exists(self) -> None:
451 # disk hits are expensive: early-return for performance
452 if self._ensure_directory_exists_called:
453 return
455 self.path.mkdir(exist_ok=True, parents=True)
456 self._ensure_directory_exists_called = True
458 def __repr__(self) -> str:
459 return f"DirectoryBasedExampleDatabase({self.path!r})"
461 def __eq__(self, other: object) -> bool:
462 return (
463 isinstance(other, DirectoryBasedExampleDatabase) and self.path == other.path
464 )
466 def _key_path(self, key: bytes) -> Path:
467 try:
468 return self.keypaths[key]
469 except KeyError:
470 pass
471 self.keypaths[key] = self.path / _hash(key)
472 return self.keypaths[key]
474 def _value_path(self, key: bytes, value: bytes) -> Path:
475 return self._key_path(key) / _hash(value)
477 def fetch(self, key: bytes) -> Iterable[bytes]:
478 kp = self._key_path(key)
479 if not kp.is_dir():
480 return
482 try:
483 for path in os.listdir(kp):
484 try:
485 yield (kp / path).read_bytes()
486 except OSError:
487 pass
488 except OSError: # pragma: no cover
489 # the `kp` directory might have been deleted in the meantime
490 pass
492 def save(self, key: bytes, value: bytes) -> None:
493 key_path = self._key_path(key)
494 if key_path.name != self._metakeys_hash:
495 # add this key to our meta entry of all keys - taking care to avoid
496 # infinite recursion.
497 self.save(self._metakeys_name, key)
499 # Note: we attempt to create the dir in question now. We
500 # already checked for permissions, but there can still be other issues,
501 # e.g. the disk is full, or permissions might have been changed.
502 try:
503 self._ensure_directory_exists()
504 key_path.mkdir(exist_ok=True, parents=True)
505 path = self._value_path(key, value)
506 if not path.exists():
507 # to mimic an atomic write, create and write in a temporary
508 # directory, and only move to the final path after. This avoids
509 # any intermediate state where the file is created (and empty)
510 # but not yet written to.
511 fd, tmpname = tempfile.mkstemp()
512 tmppath = Path(tmpname)
513 os.write(fd, value)
514 os.close(fd)
515 try:
516 tmppath.rename(path)
517 except OSError as err: # pragma: no cover
518 if err.errno == errno.EXDEV:
519 # Can't rename across filesystem boundaries, see e.g.
520 # https://github.com/HypothesisWorks/hypothesis/issues/4335
521 try:
522 path.write_bytes(tmppath.read_bytes())
523 except OSError:
524 pass
525 tmppath.unlink()
526 assert not tmppath.exists()
527 except OSError: # pragma: no cover
528 pass
530 def move(self, src: bytes, dest: bytes, value: bytes) -> None:
531 if src == dest:
532 self.save(src, value)
533 return
535 src_path = self._value_path(src, value)
536 dest_path = self._value_path(dest, value)
537 # if the dest key path does not exist, os.renames will create it for us,
538 # and we will never track its creation in the meta keys entry. Do so now.
539 if not self._key_path(dest).exists():
540 self.save(self._metakeys_name, dest)
542 try:
543 os.renames(src_path, dest_path)
544 except OSError:
545 self.delete(src, value)
546 self.save(dest, value)
548 def delete(self, key: bytes, value: bytes) -> None:
549 try:
550 self._value_path(key, value).unlink()
551 except OSError:
552 return
554 # try deleting the key dir, which will only succeed if the dir is empty
555 # (i.e. ``value`` was the last value in this key).
556 try:
557 self._key_path(key).rmdir()
558 except OSError:
559 pass
560 else:
561 # if the deletion succeeded, also delete this key entry from metakeys.
562 # (if this key happens to be the metakey itself, this deletion will
563 # fail; that's ok and faster than checking for this rare case.)
564 self.delete(self._metakeys_name, key)
566 def _start_listening(self) -> None:
567 try:
568 from watchdog.events import (
569 DirCreatedEvent,
570 DirDeletedEvent,
571 DirMovedEvent,
572 FileCreatedEvent,
573 FileDeletedEvent,
574 FileMovedEvent,
575 FileSystemEventHandler,
576 )
577 from watchdog.observers import Observer
578 except ImportError:
579 warnings.warn(
580 f"listening for changes in a {self.__class__.__name__} "
581 "requires the watchdog library. To install, run "
582 "`pip install hypothesis[watchdog]`",
583 HypothesisWarning,
584 stacklevel=4,
585 )
586 return
588 hash_to_key = {_hash(key): key for key in self.fetch(self._metakeys_name)}
589 _metakeys_hash = self._metakeys_hash
590 _broadcast_change = self._broadcast_change
592 class Handler(
593 FileSystemEventHandler
594 ): # pragma: no cover # skipped in test_database.py for now
595 def on_created(_self, event: FileCreatedEvent | DirCreatedEvent) -> None:
596 # we only registered for the file creation event
597 assert not isinstance(event, DirCreatedEvent)
598 # watchdog events are only bytes if we passed a byte path to
599 # .schedule
600 assert isinstance(event.src_path, str)
602 value_path = Path(event.src_path)
603 # the parent dir represents the key, and its name is the key hash
604 key_hash = value_path.parent.name
606 if key_hash == _metakeys_hash:
607 try:
608 hash_to_key[value_path.name] = value_path.read_bytes()
609 except OSError: # pragma: no cover
610 # this might occur if all the values in a key have been
611 # deleted and DirectoryBasedExampleDatabase removes its
612 # metakeys entry (which is `value_path` here`).
613 pass
614 return
616 key = hash_to_key.get(key_hash)
617 if key is None: # pragma: no cover
618 # we didn't recognize this key. This shouldn't ever happen,
619 # but some race condition trickery might cause this.
620 return
622 try:
623 value = value_path.read_bytes()
624 except OSError: # pragma: no cover
625 return
627 _broadcast_change(("save", (key, value)))
629 def on_deleted(self, event: FileDeletedEvent | DirDeletedEvent) -> None:
630 assert not isinstance(event, DirDeletedEvent)
631 assert isinstance(event.src_path, str)
633 value_path = Path(event.src_path)
634 key = hash_to_key.get(value_path.parent.name)
635 if key is None: # pragma: no cover
636 return
638 _broadcast_change(("delete", (key, None)))
640 def on_moved(self, event: FileMovedEvent | DirMovedEvent) -> None:
641 assert not isinstance(event, DirMovedEvent)
642 assert isinstance(event.src_path, str)
643 assert isinstance(event.dest_path, str)
645 src_path = Path(event.src_path)
646 dest_path = Path(event.dest_path)
647 k1 = hash_to_key.get(src_path.parent.name)
648 k2 = hash_to_key.get(dest_path.parent.name)
650 if k1 is None or k2 is None: # pragma: no cover
651 return
653 try:
654 value = dest_path.read_bytes()
655 except OSError: # pragma: no cover
656 return
658 _broadcast_change(("delete", (k1, value)))
659 _broadcast_change(("save", (k2, value)))
661 # If we add a listener to a DirectoryBasedExampleDatabase whose database
662 # directory doesn't yet exist, the watchdog observer will not fire any
663 # events, even after the directory gets created.
664 #
665 # Ensure the directory exists before starting the observer.
666 self._ensure_directory_exists()
667 self._observer = Observer()
668 self._observer.schedule(
669 Handler(),
670 # remove type: ignore when released
671 # https://github.com/gorakhargosh/watchdog/pull/1096
672 self.path, # type: ignore
673 recursive=True,
674 event_filter=[FileCreatedEvent, FileDeletedEvent, FileMovedEvent],
675 )
676 self._observer.start()
678 def _stop_listening(self) -> None:
679 assert self._observer is not None
680 self._observer.stop()
681 self._observer.join()
682 self._observer = None
685class _StorageDirectoryDatabase(DirectoryBasedExampleDatabase):
686 # A DirectoryBasedExampleDatabase which is located at the same directory as the storage
687 # directory. This lets our database logic interact with our logic for writing .gitignore
688 # files to the storage directory.
689 #
690 # The reason why we need this class is because the first interaction we have
691 # with .hypothesis might be writing a file to .hypothesis/examples, and
692 # DirectoryBasedExampleDatabase.save would otherwise create .hypothesis without
693 # performing our .gitignore logic.
695 def __init__(self, storage_dir: StorageDirectory) -> None:
696 super().__init__(storage_dir.path)
697 self._storage_dir = storage_dir
699 def _ensure_directory_exists(self) -> None:
700 if self._ensure_directory_exists_called:
701 return
703 self._storage_dir.create_if_missing()
704 self._ensure_directory_exists_called = True
707class ReadOnlyDatabase(ExampleDatabase):
708 """A wrapper to make the given database read-only.
710 The implementation passes through ``fetch``, and turns ``save``, ``delete``, and
711 ``move`` into silent no-ops.
713 Note that this disables Hypothesis' automatic discarding of stale examples.
714 It is designed to allow local machines to access a shared database (e.g. from CI
715 servers), without propagating changes back from a local or in-development branch.
716 """
718 def __init__(self, db: ExampleDatabase) -> None:
719 super().__init__()
720 assert isinstance(db, ExampleDatabase)
721 self._wrapped = db
723 def __repr__(self) -> str:
724 return f"ReadOnlyDatabase({self._wrapped!r})"
726 def __eq__(self, other: object) -> bool:
727 return isinstance(other, ReadOnlyDatabase) and self._wrapped == other._wrapped
729 def fetch(self, key: bytes) -> Iterable[bytes]:
730 yield from self._wrapped.fetch(key)
732 def save(self, key: bytes, value: bytes) -> None:
733 pass
735 def delete(self, key: bytes, value: bytes) -> None:
736 pass
738 def _start_listening(self) -> None:
739 # we're read only, so there are no changes to broadcast.
740 pass
742 def _stop_listening(self) -> None:
743 pass
746class MultiplexedDatabase(ExampleDatabase):
747 """A wrapper around multiple databases.
749 Each ``save``, ``fetch``, ``move``, or ``delete`` operation will be run against
750 all of the wrapped databases. ``fetch`` does not yield duplicate values, even
751 if the same value is present in two or more of the wrapped databases.
753 This combines well with a :class:`ReadOnlyDatabase`, as follows:
755 .. code-block:: python
757 local = DirectoryBasedExampleDatabase("/tmp/hypothesis/examples/")
758 shared = CustomNetworkDatabase()
760 settings.register_profile("ci", database=shared)
761 settings.register_profile(
762 "dev", database=MultiplexedDatabase(local, ReadOnlyDatabase(shared))
763 )
764 settings.load_profile("ci" if os.environ.get("CI") else "dev")
766 So your CI system or fuzzing runs can populate a central shared database;
767 while local runs on development machines can reproduce any failures from CI
768 but will only cache their own failures locally and cannot remove examples
769 from the shared database.
770 """
772 def __init__(self, *dbs: ExampleDatabase) -> None:
773 super().__init__()
774 assert all(isinstance(db, ExampleDatabase) for db in dbs)
775 self._wrapped = dbs
777 def __repr__(self) -> str:
778 return "MultiplexedDatabase({})".format(", ".join(map(repr, self._wrapped)))
780 def __eq__(self, other: object) -> bool:
781 return (
782 isinstance(other, MultiplexedDatabase) and self._wrapped == other._wrapped
783 )
785 def fetch(self, key: bytes) -> Iterable[bytes]:
786 seen = set()
787 for db in self._wrapped:
788 for value in db.fetch(key):
789 if value not in seen:
790 yield value
791 seen.add(value)
793 def save(self, key: bytes, value: bytes) -> None:
794 for db in self._wrapped:
795 db.save(key, value)
797 def delete(self, key: bytes, value: bytes) -> None:
798 for db in self._wrapped:
799 db.delete(key, value)
801 def move(self, src: bytes, dest: bytes, value: bytes) -> None:
802 for db in self._wrapped:
803 db.move(src, dest, value)
805 def _start_listening(self) -> None:
806 for db in self._wrapped:
807 db.add_listener(self._broadcast_change)
809 def _stop_listening(self) -> None:
810 for db in self._wrapped:
811 db.remove_listener(self._broadcast_change)
814class GitHubArtifactDatabase(ExampleDatabase):
815 """
816 A file-based database loaded from a `GitHub Actions <https://docs.github.com/en/actions>`_ artifact.
818 You can use this for sharing example databases between CI runs and developers, allowing
819 the latter to get read-only access to the former. This is particularly useful for
820 continuous fuzzing (i.e. with `HypoFuzz <https://hypofuzz.com/>`_),
821 where the CI system can help find new failing test cases through fuzzing,
822 and developers can reproduce them locally without any manual effort.
824 .. note::
825 You must provide ``GITHUB_TOKEN`` as an environment variable. In CI, Github Actions provides
826 this automatically, but it needs to be set manually for local usage. In a developer machine,
827 this would usually be a `Personal Access Token <https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens>`_.
828 If the repository is private, it's necessary for the token to have ``repo`` scope
829 in the case of a classic token, or ``actions:read`` in the case of a fine-grained token.
832 In most cases, this will be used
833 through the :class:`~hypothesis.database.MultiplexedDatabase`,
834 by combining a local directory-based database with this one. For example:
836 .. code-block:: python
838 local = DirectoryBasedExampleDatabase(".hypothesis/examples")
839 shared = ReadOnlyDatabase(GitHubArtifactDatabase("user", "repo"))
841 settings.register_profile("ci", database=local)
842 settings.register_profile("dev", database=MultiplexedDatabase(local, shared))
843 # We don't want to use the shared database in CI, only to populate its local one.
844 # which the workflow should then upload as an artifact.
845 settings.load_profile("ci" if os.environ.get("CI") else "dev")
847 .. note::
848 Because this database is read-only, you always need to wrap it with the
849 :class:`ReadOnlyDatabase`.
851 A setup like this can be paired with a GitHub Actions workflow including
852 something like the following:
854 .. code-block:: yaml
856 - name: Download example database
857 uses: dawidd6/action-download-artifact@v9
858 with:
859 name: hypothesis-example-db
860 path: .hypothesis/examples
861 if_no_artifact_found: warn
862 workflow_conclusion: completed
864 - name: Run tests
865 run: pytest
867 - name: Upload example database
868 uses: actions/upload-artifact@v3
869 if: always()
870 with:
871 name: hypothesis-example-db
872 path: .hypothesis/examples
874 In this workflow, we use `dawidd6/action-download-artifact <https://github.com/dawidd6/action-download-artifact>`_
875 to download the latest artifact given that the official `actions/download-artifact <https://github.com/actions/download-artifact>`_
876 does not support downloading artifacts from previous workflow runs.
878 The database automatically implements a simple file-based cache with a default expiration period
879 of 1 day. You can adjust this through the ``cache_timeout`` property.
881 For mono-repo support, you can provide a unique ``artifact_name`` (e.g. ``hypofuzz-example-db-frontend``).
882 """
884 def __init__(
885 self,
886 owner: str,
887 repo: str,
888 artifact_name: str = "hypothesis-example-db",
889 cache_timeout: timedelta = timedelta(days=1),
890 path: StrPathT | None = None,
891 ):
892 super().__init__()
893 self.owner = owner
894 self.repo = repo
895 self.artifact_name = artifact_name
896 self.cache_timeout = cache_timeout
898 # Get the GitHub token from the environment
899 # It's unnecessary to use a token if the repo is public
900 self.token: str | None = getenv("GITHUB_TOKEN")
902 self._storage_dir: StorageDirectory | None = None
903 if path is None:
904 self._storage_dir = storage_directory(
905 f"github-artifacts/{self.artifact_name}/"
906 )
907 self.path = self._storage_dir.path
908 else:
909 self.path = Path(path)
911 # We don't want to initialize the cache until we need to
912 self._initialized: bool = False
913 self._disabled: bool = False
915 # This is the path to the artifact in usage
916 # .hypothesis/github-artifacts/<artifact-name>/<modified_isoformat>.zip
917 self._artifact: Path | None = None
918 # This caches the artifact structure
919 self._access_cache: dict[PurePath, set[PurePath]] | None = None
921 # Message to display if user doesn't wrap around ReadOnlyDatabase
922 self._read_only_message = (
923 "This database is read-only. "
924 "Please wrap this class with ReadOnlyDatabase"
925 "i.e. ReadOnlyDatabase(GitHubArtifactDatabase(...))."
926 )
928 def __repr__(self) -> str:
929 return (
930 f"GitHubArtifactDatabase(owner={self.owner!r}, "
931 f"repo={self.repo!r}, artifact_name={self.artifact_name!r})"
932 )
934 def __eq__(self, other: object) -> bool:
935 return (
936 isinstance(other, GitHubArtifactDatabase)
937 and self.owner == other.owner
938 and self.repo == other.repo
939 and self.artifact_name == other.artifact_name
940 and self.path == other.path
941 )
943 def _prepare_for_io(self) -> None:
944 assert self._artifact is not None, "Artifact not loaded."
946 if self._initialized: # pragma: no cover
947 return
949 # Test that the artifact is valid
950 try:
951 with ZipFile(self._artifact) as f:
952 if f.testzip(): # pragma: no cover
953 raise BadZipFile
955 # Turns out that testzip() doesn't work quite well
956 # doing the cache initialization here instead
957 # will give us more coverage of the artifact.
959 # Cache the files inside each keypath
960 self._access_cache = {}
961 with ZipFile(self._artifact) as zf:
962 namelist = zf.namelist()
963 # Iterate over files in the artifact
964 for filename in namelist:
965 fileinfo = zf.getinfo(filename)
966 if fileinfo.is_dir():
967 self._access_cache.setdefault(PurePath(filename), set())
968 else:
969 # Get the keypath from the filename
970 keypath = PurePath(filename).parent
971 # Add the file to the keypath
972 self._access_cache.setdefault(keypath, set()).add(
973 PurePath(filename)
974 )
975 except BadZipFile:
976 warnings.warn(
977 "The downloaded artifact from GitHub is invalid. "
978 "This could be because the artifact was corrupted, "
979 "or because the artifact was not created by Hypothesis. ",
980 HypothesisWarning,
981 stacklevel=3,
982 )
983 self._disabled = True
985 self._initialized = True
987 def _initialize_db(self) -> None:
988 # Trigger warning that we suppressed earlier by intent_to_write=False
989 storage_directory(self.path.name)
990 # Create the cache directory if it doesn't exist
991 if self._storage_dir is not None: # pragma: no cover
992 self._storage_dir.create_if_missing()
993 else:
994 self.path.mkdir(exist_ok=True, parents=True)
996 # Get all artifacts
997 cached_artifacts = sorted(
998 self.path.glob("*.zip"),
999 key=lambda a: datetime.fromisoformat(a.stem.replace("_", ":")),
1000 )
1002 # Remove all but the latest artifact
1003 for artifact in cached_artifacts[:-1]:
1004 artifact.unlink()
1006 try:
1007 found_artifact = cached_artifacts[-1]
1008 except IndexError:
1009 found_artifact = None
1011 # Check if the latest artifact is a cache hit
1012 if found_artifact is not None and (
1013 datetime.now(timezone.utc)
1014 - datetime.fromisoformat(found_artifact.stem.replace("_", ":"))
1015 < self.cache_timeout
1016 ):
1017 self._artifact = found_artifact
1018 else:
1019 # Download the latest artifact from GitHub
1020 new_artifact = self._fetch_artifact()
1022 if new_artifact:
1023 if found_artifact is not None:
1024 found_artifact.unlink()
1025 self._artifact = new_artifact
1026 elif found_artifact is not None:
1027 warnings.warn(
1028 "Using an expired artifact as a fallback for the database: "
1029 f"{found_artifact}",
1030 HypothesisWarning,
1031 stacklevel=2,
1032 )
1033 self._artifact = found_artifact
1034 else:
1035 warnings.warn(
1036 "Couldn't acquire a new or existing artifact. Disabling database.",
1037 HypothesisWarning,
1038 stacklevel=2,
1039 )
1040 self._disabled = True
1041 return
1043 self._prepare_for_io()
1045 def _get_bytes(self, url: str) -> bytes | None: # pragma: no cover
1046 # Deferred imports: urllib.request pulls in http, email, and ssl, which
1047 # is a noticeable fraction of `import hypothesis`, and only this rarely
1048 # used database needs them.
1049 from urllib.error import HTTPError, URLError
1050 from urllib.request import Request, urlopen
1052 request = Request(
1053 url,
1054 headers={
1055 "Accept": "application/vnd.github+json",
1056 "X-GitHub-Api-Version": "2022-11-28 ",
1057 },
1058 )
1059 # see https://github.com/HypothesisWorks/hypothesis/pull/4791
1060 request.add_unredirected_header("Authorization", f"Bearer {self.token}")
1061 warning_message = None
1062 response_bytes: bytes | None = None
1063 try:
1064 with urlopen(request) as response:
1065 response_bytes = response.read()
1066 except HTTPError as e:
1067 if e.code == 401:
1068 warning_message = (
1069 "Authorization failed when trying to download artifact from GitHub. "
1070 "Check that you have a valid GITHUB_TOKEN set in your environment."
1071 )
1072 else:
1073 warning_message = (
1074 "Could not get the latest artifact from GitHub. "
1075 "This could be because the repository "
1076 "or artifact does not exist. "
1077 )
1078 # see https://github.com/python/cpython/issues/128734
1079 e.close()
1080 except URLError:
1081 warning_message = "Could not connect to GitHub to get the latest artifact. "
1082 except TimeoutError:
1083 warning_message = (
1084 "Could not connect to GitHub to get the latest artifact "
1085 "(connection timed out)."
1086 )
1088 if warning_message is not None:
1089 warnings.warn(warning_message, HypothesisWarning, stacklevel=4)
1090 return None
1092 return response_bytes
1094 def _fetch_artifact(self) -> Path | None: # pragma: no cover
1095 # Get the list of artifacts from GitHub
1096 url = f"https://api.github.com/repos/{self.owner}/{self.repo}/actions/artifacts"
1097 response_bytes = self._get_bytes(url)
1098 if response_bytes is None:
1099 return None
1101 artifacts = json.loads(response_bytes)["artifacts"]
1102 artifacts = [a for a in artifacts if a["name"] == self.artifact_name]
1104 if not artifacts:
1105 return None
1107 # Get the latest artifact from the list
1108 artifact = max(artifacts, key=lambda a: a["created_at"])
1109 url = artifact["archive_download_url"]
1111 # Download the artifact
1112 artifact_bytes = self._get_bytes(url)
1113 if artifact_bytes is None:
1114 return None
1116 # Save the artifact to the cache
1117 # We replace ":" with "_" to ensure the filenames are compatible
1118 # with Windows filesystems
1119 timestamp = datetime.now(timezone.utc).isoformat().replace(":", "_")
1120 artifact_path = self.path / f"{timestamp}.zip"
1121 try:
1122 artifact_path.write_bytes(artifact_bytes)
1123 except OSError:
1124 warnings.warn(
1125 "Could not save the latest artifact from GitHub. ",
1126 HypothesisWarning,
1127 stacklevel=3,
1128 )
1129 return None
1131 return artifact_path
1133 @staticmethod
1134 @lru_cache
1135 def _key_path(key: bytes) -> PurePath:
1136 return PurePath(_hash(key) + "/")
1138 def fetch(self, key: bytes) -> Iterable[bytes]:
1139 if self._disabled:
1140 return
1142 if not self._initialized:
1143 self._initialize_db()
1144 if self._disabled:
1145 return
1147 assert self._artifact is not None
1148 assert self._access_cache is not None
1150 kp = self._key_path(key)
1152 with ZipFile(self._artifact) as zf:
1153 # Get all the files in the kp from the cache
1154 filenames = self._access_cache.get(kp, ())
1155 for filename in filenames:
1156 with zf.open(filename.as_posix()) as f:
1157 yield f.read()
1159 # Read-only interface
1160 def save(self, key: bytes, value: bytes) -> None:
1161 raise RuntimeError(self._read_only_message)
1163 def move(self, src: bytes, dest: bytes, value: bytes) -> None:
1164 raise RuntimeError(self._read_only_message)
1166 def delete(self, key: bytes, value: bytes) -> None:
1167 raise RuntimeError(self._read_only_message)
1170class BackgroundWriteDatabase(ExampleDatabase):
1171 """A wrapper which defers writes on the given database to a background thread.
1173 Calls to :meth:`~hypothesis.database.ExampleDatabase.fetch` wait for any
1174 enqueued writes to finish before fetching from the database.
1175 """
1177 def __init__(self, db: ExampleDatabase) -> None:
1178 super().__init__()
1179 self._db = db
1180 self._queue: Queue[tuple[str, tuple[bytes, ...]]] = Queue()
1181 self._thread: Thread | None = None
1183 def _ensure_thread(self):
1184 if self._thread is None:
1185 self._thread = Thread(target=self._worker, daemon=True)
1186 self._thread.start()
1187 # avoid an unbounded timeout during gc. 0.1 should be plenty for most
1188 # use cases.
1189 weakref.finalize(self, self._join, 0.1)
1191 def __repr__(self) -> str:
1192 return f"BackgroundWriteDatabase({self._db!r})"
1194 def __eq__(self, other: object) -> bool:
1195 return isinstance(other, BackgroundWriteDatabase) and self._db == other._db
1197 def _worker(self) -> None:
1198 while True:
1199 method, args = self._queue.get()
1200 getattr(self._db, method)(*args)
1201 self._queue.task_done()
1203 def _join(self, timeout: float | None = None) -> None:
1204 # copy of Queue.join with a timeout. https://bugs.python.org/issue9634
1205 with self._queue.all_tasks_done:
1206 while self._queue.unfinished_tasks:
1207 self._queue.all_tasks_done.wait(timeout)
1209 def fetch(self, key: bytes) -> Iterable[bytes]:
1210 self._join()
1211 return self._db.fetch(key)
1213 def save(self, key: bytes, value: bytes) -> None:
1214 self._ensure_thread()
1215 self._queue.put(("save", (key, value)))
1217 def delete(self, key: bytes, value: bytes) -> None:
1218 self._ensure_thread()
1219 self._queue.put(("delete", (key, value)))
1221 def move(self, src: bytes, dest: bytes, value: bytes) -> None:
1222 self._ensure_thread()
1223 self._queue.put(("move", (src, dest, value)))
1225 def _start_listening(self) -> None:
1226 self._db.add_listener(self._broadcast_change)
1228 def _stop_listening(self) -> None:
1229 self._db.remove_listener(self._broadcast_change)
1232def _pack_uleb128(value: int) -> bytes:
1233 """
1234 Serialize an integer into variable-length bytes. For each byte, the first 7
1235 bits represent (part of) the integer, while the last bit indicates whether the
1236 integer continues into the next byte.
1238 https://en.wikipedia.org/wiki/LEB128
1239 """
1240 parts = bytearray()
1241 assert value >= 0
1242 while True:
1243 # chop off 7 bits
1244 byte = value & ((1 << 7) - 1)
1245 value >>= 7
1246 # set the continuation bit if we have more left
1247 if value:
1248 byte |= 1 << 7
1250 parts.append(byte)
1251 if not value:
1252 break
1253 return bytes(parts)
1256def _unpack_uleb128(buffer: bytes) -> tuple[int, int]:
1257 """
1258 Inverts _pack_uleb128, and also returns the index at which at which we stopped
1259 reading.
1260 """
1261 value = 0
1262 for i, byte in enumerate(buffer):
1263 n = byte & ((1 << 7) - 1)
1264 value |= n << (i * 7)
1266 if not byte >> 7:
1267 break
1268 return (i + 1, value)
1271def choices_to_bytes(choices: Iterable[ChoiceT], /) -> bytes:
1272 """Serialize a list of choices to a bytestring. Inverts choices_from_bytes."""
1273 # We use a custom serialization format for this, which might seem crazy - but our
1274 # data is a flat sequence of elements, and standard tools like protobuf or msgpack
1275 # don't deal well with e.g. nonstandard bit-pattern-NaNs, or invalid-utf8 unicode.
1276 #
1277 # We simply encode each element with a metadata byte, if needed a uint16 size, and
1278 # then the payload bytes. For booleans, the payload is inlined into the metadata.
1279 parts = []
1280 for choice in choices:
1281 if isinstance(choice, bool):
1282 # `000_0000v` - tag zero, low bit payload.
1283 parts.append(b"\1" if choice else b"\0")
1284 continue
1286 # `tag_ssss [uint16 size?] [payload]`
1287 if isinstance(choice, float):
1288 tag = 1 << 5
1289 choice = struct.pack("!d", choice)
1290 elif isinstance(choice, int):
1291 tag = 2 << 5
1292 choice = choice.to_bytes(1 + choice.bit_length() // 8, "big", signed=True)
1293 elif isinstance(choice, bytes):
1294 tag = 3 << 5
1295 else:
1296 assert isinstance(choice, str)
1297 tag = 4 << 5
1298 choice = choice.encode(errors="surrogatepass")
1300 size = len(choice)
1301 if size < 0b11111:
1302 parts.append((tag | size).to_bytes(1, "big"))
1303 else:
1304 parts.append((tag | 0b11111).to_bytes(1, "big"))
1305 parts.append(_pack_uleb128(size))
1306 parts.append(choice)
1308 return b"".join(parts)
1311def _choices_from_bytes(buffer: bytes, /) -> tuple[ChoiceT, ...]:
1312 # See above for an explanation of the format.
1313 parts: list[ChoiceT] = []
1314 idx = 0
1315 while idx < len(buffer):
1316 tag = buffer[idx] >> 5
1317 size = buffer[idx] & 0b11111
1318 idx += 1
1320 if tag == 0:
1321 parts.append(bool(size))
1322 continue
1323 if size == 0b11111:
1324 offset, size = _unpack_uleb128(buffer[idx:])
1325 idx += offset
1326 chunk = buffer[idx : idx + size]
1327 idx += size
1329 if tag == 1:
1330 assert size == 8, "expected float64"
1331 parts.extend(struct.unpack("!d", chunk))
1332 elif tag == 2:
1333 parts.append(int.from_bytes(chunk, "big", signed=True))
1334 elif tag == 3:
1335 parts.append(chunk)
1336 else:
1337 assert tag == 4
1338 parts.append(chunk.decode(errors="surrogatepass"))
1339 return tuple(parts)
1342def choices_from_bytes(buffer: bytes, /) -> tuple[ChoiceT, ...] | None:
1343 """
1344 Deserialize a bytestring to a tuple of choices. Inverts choices_to_bytes.
1346 Returns None if the given bytestring is not a valid serialization of choice
1347 sequences.
1348 """
1349 try:
1350 return _choices_from_bytes(buffer)
1351 except Exception:
1352 # deserialization error, eg because our format changed or someone put junk
1353 # data in the db.
1354 return None