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

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

29 statements  

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 Callable, Dict, List, 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) -> dict: 

46 if data.interesting_origin: 

47 status_reason = str(data.interesting_origin) 

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

49 status_reason = "exceeded size of current best example" 

50 else: 

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

52 

53 return { 

54 "type": "test_case", 

55 "run_start": start_timestamp, 

56 "property": test_name_or_nodeid, 

57 "status": { 

58 Status.OVERRUN: "gave_up", 

59 Status.INVALID: "gave_up", 

60 Status.VALID: "passed", 

61 Status.INTERESTING: "failed", 

62 }[data.status], 

63 "status_reason": status_reason, 

64 "representation": string_repr, 

65 "arguments": arguments or {}, 

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

67 "features": { 

68 **{ 

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

70 }, 

71 **data.events, 

72 }, 

73 "timing": timing, 

74 "metadata": { 

75 "traceback": getattr(data.extra_information, "_expected_traceback", None), 

76 "predicates": data._observability_predicates, 

77 **_system_metadata(), 

78 }, 

79 "coverage": coverage, 

80 } 

81 

82 

83_WROTE_TO = set() 

84 

85 

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

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

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

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

90 _WROTE_TO.add(fname) 

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

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

93 

94 

95_imported_at = time.time() 

96 

97 

98@lru_cache 

99def _system_metadata(): 

100 return { 

101 "sys.argv": sys.argv, 

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

103 "imported_at": _imported_at, 

104 } 

105 

106 

107OBSERVABILITY_COLLECT_COVERAGE = ( 

108 "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY_NOCOVER" not in os.environ 

109) 

110if OBSERVABILITY_COLLECT_COVERAGE is False and sys.version_info[:2] >= ( 

111 3, 

112 12, 

113): # pragma: no cover 

114 warnings.warn( 

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

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

117 HypothesisWarning, 

118 stacklevel=2, 

119 ) 

120if ( 

121 "HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY" in os.environ 

122 or OBSERVABILITY_COLLECT_COVERAGE is False 

123): # pragma: no cover 

124 TESTCASE_CALLBACKS.append(_deliver_to_file) 

125 

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

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

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

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

130 f.unlink(missing_ok=True)