Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.10/site-packages/hypothesis/internal/observability.py: 67%

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"""Observability tools to spit out analysis-ready tables, one row per test case.""" 

12 

13import json 

14import os 

15import sys 

16import time 

17import warnings 

18from datetime import date, timedelta 

19from functools import lru_cache 

20from typing import Any, Callable, Optional 

21 

22from hypothesis.configuration import storage_directory 

23from hypothesis.errors import HypothesisWarning 

24from hypothesis.internal.conjecture.data import ConjectureData, Status 

25 

26TESTCASE_CALLBACKS: list[Callable[[dict], None]] = [] 

27 

28 

29def deliver_json_blob(value: dict) -> None: 

30 for callback in TESTCASE_CALLBACKS: 

31 callback(value) 

32 

33 

34def make_testcase( 

35 *, 

36 start_timestamp: float, 

37 test_name_or_nodeid: str, 

38 data: ConjectureData, 

39 how_generated: str, 

40 string_repr: str = "<unknown>", 

41 arguments: Optional[dict] = None, 

42 timing: dict[str, float], 

43 coverage: Optional[dict[str, list[int]]] = None, 

44 phase: Optional[str] = None, 

45 backend_metadata: Optional[dict[str, Any]] = None, 

46) -> dict: 

47 from hypothesis.core import reproduction_decorator 

48 

49 if data.interesting_origin: 

50 status_reason = str(data.interesting_origin) 

51 elif phase == "shrink" and data.status == Status.OVERRUN: 

52 status_reason = "exceeded size of current best example" 

53 else: 

54 status_reason = str(data.events.pop("invalid because", "")) 

55 

56 return { 

57 "type": "test_case", 

58 "run_start": start_timestamp, 

59 "property": test_name_or_nodeid, 

60 "status": { 

61 Status.OVERRUN: "gave_up", 

62 Status.INVALID: "gave_up", 

63 Status.VALID: "passed", 

64 Status.INTERESTING: "failed", 

65 }[data.status], 

66 "status_reason": status_reason, 

67 "representation": string_repr, 

68 "arguments": { 

69 k.removeprefix("generate:"): v for k, v in (arguments or {}).items() 

70 }, 

71 "how_generated": how_generated, # iid, mutation, etc. 

72 "features": { 

73 **{ 

74 f"target:{k}".strip(":"): v for k, v in data.target_observations.items() 

75 }, 

76 **data.events, 

77 }, 

78 "timing": timing, 

79 "metadata": { 

80 "traceback": data.expected_traceback, 

81 "reproduction_decorator": ( 

82 reproduction_decorator(data.choices) 

83 if data.status is Status.INTERESTING 

84 else None 

85 ), 

86 "predicates": dict(data._observability_predicates), 

87 "backend": backend_metadata or {}, 

88 **_system_metadata(), 

89 }, 

90 "coverage": coverage, 

91 } 

92 

93 

94_WROTE_TO = set() 

95 

96 

97def _deliver_to_file(value): # pragma: no cover 

98 kind = "testcases" if value["type"] == "test_case" else "info" 

99 fname = storage_directory("observed", f"{date.today().isoformat()}_{kind}.jsonl") 

100 fname.parent.mkdir(exist_ok=True, parents=True) 

101 _WROTE_TO.add(fname) 

102 with fname.open(mode="a") as f: 

103 f.write(json.dumps(value) + "\n") 

104 

105 

106_imported_at = time.time() 

107 

108 

109@lru_cache 

110def _system_metadata(): 

111 return { 

112 "sys.argv": sys.argv, 

113 "os.getpid()": os.getpid(), 

114 "imported_at": _imported_at, 

115 } 

116 

117 

118OBSERVABILITY_COLLECT_COVERAGE = ( 

119 "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY_NOCOVER" not in os.environ 

120) 

121if OBSERVABILITY_COLLECT_COVERAGE is False and ( 

122 sys.version_info[:2] >= (3, 12) 

123): # pragma: no cover 

124 warnings.warn( 

125 "Coverage data collection should be quite fast in Python 3.12 or later " 

126 "so there should be no need to turn coverage reporting off.", 

127 HypothesisWarning, 

128 stacklevel=2, 

129 ) 

130if "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY" in os.environ or ( 

131 OBSERVABILITY_COLLECT_COVERAGE is False 

132): # pragma: no cover 

133 TESTCASE_CALLBACKS.append(_deliver_to_file) 

134 

135 # Remove files more than a week old, to cap the size on disk 

136 max_age = (date.today() - timedelta(days=8)).isoformat() 

137 for f in storage_directory("observed", intent_to_write=False).glob("*.jsonl"): 

138 if f.stem < max_age: # pragma: no branch 

139 f.unlink(missing_ok=True)