Coverage for src/pytest_maxcov/plugin.py: 8%

1import os 

2import re 

3from collections import defaultdict, namedtuple 

4from enum import StrEnum 

5from time import time 

6from typing import Dict, List, Optional, Set, Tuple, Union 

7from warnings import warn 

8 

9import pytest 

10from coverage import CoverageData 

11 

12 

13def pytest_addoption(parser): 

14 group = parser.getgroup("coverage runtime minimisation") 

15 group.addoption( 

16 "--maxcov", 

17 action="store_true", 

18 default=False, 

19 help=( 

20 "Run the subset of tests provides maximum coverage with the minimum" 

21 "execution time. Requires a previous pytest run using --maxcov-record" 

22 ), 

23 ) 

24 group.addoption( 

25 "--maxcov-record", 

26 action="store_true", 

27 default=False, 

28 help="Record coverage and timing data for the --maxcov option", 

29 ) 

30 group.addoption( 

31 "--maxcov-threshold", 

32 type=float, 

33 default=100.0, 

34 metavar="PERCENT", 

35 help="Set the threshold for computing maximum coverage. Default: 100.0", 

36 ) 

37 

38 

39def pytest_configure(config): 

40 config.pluginmanager.register(MaxCovPlugin(config), "maxcov-plugin") 

41 

42 

43Subset = namedtuple("Subset", ["set", "cost", "context"]) 

44 

45 

46class CacheType(StrEnum): 

47 COST = "$" 

48 COVERAGE = "?" 

49 

50 

51class TestDuration: 

52 def __init__(self): 

53 self.start = None 

54 self.end = None 

55 

56 @property 

57 def delta(self) -> time: 

58 return self.end - self.start 

59 

60 

61class MaxCovPlugin: 

62 def __init__(self, config: pytest.Config): 

63 self.config = config 

64 self.durations = defaultdict(TestDuration) 

65 if not 0.0 < config.option.maxcov_threshold <= 100.0: 

66 raise pytest.UsageError("--maxcov-threshold must be >0.0 and <=100.0") 

67 

68 def pytest_collection(self, session: pytest.Session) -> None: 

69 if not self.config.pluginmanager.has_plugin("pytest_cov"): 

70 raise pytest.UsageError("pytest-maxcov uses pytest-cov which is not enabled") 

71 if self.config.option.no_cov: 

72 raise pytest.UsageError("pytest-maxcov uses coverage which is disabled with --no-cov") 

73 if self.config.option.maxcov_record and not self.config.option.cov_context: 

74 raise pytest.UsageError("pytest-maxcov requires --cov_context=test") 

75 

76 def pytest_collection_modifyitems( 

77 self, session: pytest.Session, config: pytest.Config, items: List[pytest.Item] 

78 ) -> None: 

79 if not config.option.maxcov: 

80 return 

81 

82 (costs, coverage) = self.read_cache() 

83 if len(coverage) == 0: 

84 warn(pytest.PytestWarning("No coverage data available: --maxcov aborted")) 

85 return 

86 

87 collected_nodeids = set([x.nodeid for x in items]) 

88 cached_nodeids = set(costs.keys()) 

89 if not collected_nodeids.issubset(cached_nodeids): 

90 missing_nodeids = ", ".join(collected_nodeids - cached_nodeids) 

91 raise pytest.UsageError(f"no recorded durations for nodes: {missing_nodeids}") 

92 

93 universe = set() 

94 subsets = [] 

95 for context, hits in coverage.items(): 

96 subset = set(hits.split("|")) 

97 universe |= subset 

98 subsets.append(Subset(set=subset, cost=costs[context], context=context)) 

99 

100 cover_subsets = set_cover(universe, subsets) 

101 maxcov_contexts = [x.context for x in cover_subsets] 

102 

103 skip_maxcov = pytest.mark.skip(reason="excluded by --maxcov") 

104 maxcov_subsets = set_cover(universe, subsets, config.option.maxcov_threshold / 100) 

105 maxcov_contexts = [x.context for x in maxcov_subsets] 

106 

107 for item in items: 

108 if item.nodeid not in maxcov_contexts: 

109 item.add_marker(skip_maxcov) 

110 

111 def pytest_runtest_logstart( 

112 self, nodeid: str, location: Tuple[str, Optional[int], str] 

113 ) -> None: 

114 if self.config.option.maxcov_record: 

115 self.durations[nodeid].start = time() 

116 

117 def pytest_runtest_logfinish(self, nodeid, location: Tuple[str, Optional[int], str]) -> None: 

118 if self.config.option.maxcov_record: 

119 self.durations[nodeid].end = time() 

120 

121 def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int) -> None: 

122 if self.config.option.maxcov_record: 

123 self.write_cache() 

124 

125 def read_cache(self) -> Tuple[Dict[str, str], Dict[str, str]]: 

126 costs = {} 

127 coverage = {} 

128 lineno = 1 

129 try: 

130 fh = open(".maxcov", "r") 

131 except FileNotFoundError as e: 

132 raise pytest.UsageError(str(e)) from e 

133 

134 for line in fh.readlines(): 

135 if line.startswith(f"{CacheType.COST}:"): 

136 line_type = CacheType.COST 

137 elif line.startswith(f"{CacheType.COVERAGE}:"): 

138 line_type = CacheType.COVERAGE 

139 else: 

140 raise pytest.UsageError(f".maxcov is corrupted at line {lineno}") 

141 

142 try: 

143 (context, data) = line[2:].split("=") 

144 except ValueError as e: 

145 raise pytest.UsageError(f".maxcov is corrupted at line {lineno}") from e 

146 

147 if line_type == CacheType.COST: 

148 costs[context] = float(data) 

149 else: 

150 coverage[context] = data 

151 lineno += 1 

152 

153 return costs, coverage 

154 

155 def write_cache(self) -> None: 

156 coverage = coverage_data() 

157 try: 

158 fh = open(".maxcov", "w") 

159 except (FileNotFoundError, PermissionError) as e: 

160 raise pytest.UsageError(str(e)) from e 

161 

162 for nodeid, duration in self.durations.items(): 

163 print(f"{CacheType.COST}:{nodeid}={duration.delta}", file=fh) 

164 for context, hits in coverage.items(): 

165 print(f"{CacheType.COVERAGE}:{context}={hits}", file=fh) 

166 

167 

168def coverage_data() -> Dict[str, str]: 

169 """Read coverage data from disk and return a list of lines/arcs hit per coverage context""" 

170 if os.environ.get("COVERAGE_FILE"): 

171 data = CoverageData(os.environ.get("COVERAGE_FILE")) 

172 else: 

173 data = CoverageData(".coverage") 

174 

175 data.read() 

176 root_dir = os.path.dirname(data.base_filename()) + "/" 

177 coverage_data = {} 

178 for context in data.measured_contexts(): 

179 if not context: 

180 continue 

181 data.set_query_context(context) 

182 if data.has_arcs(): 

183 arcs = [ 

184 [f"{filename.replace(root_dir,'')}:{arc}" for arc in data.arcs(filename)] 

185 for filename in data.measured_files() 

186 ] 

187 hits = set(e for s in arcs for e in s) 

188 else: 

189 lines = [ 

190 [f"{filename.replace(root_dir,'')}:{line}" for line in data.lines(filename)] 

191 for filename in data.measured_files() 

192 ] 

193 hits = set(e for s in lines for e in s) 

194 context = re.sub(r"\|.*", "", context) 

195 coverage_data[context] = "|".join(hits) 

196 return coverage_data 

197 

198 

199def set_cover( 

200 universe: Set, subsets: List[Subset], confidence: Union[None, float] = None 

201) -> List[list]: 

202 all_elements = set(e for s in subsets for e in s.set) 

203 if universe != all_elements: 

204 raise RuntimeError("union of all sets doesn't match universe") 

205 

206 if confidence is not None and (confidence <= 0.0 or confidence > 1.0): 

207 raise RuntimeError("subset match confidence must be >0.0 and <=1.0") 

208 

209 covered = set() 

210 cover_subsets = [] 

211 while True: 

212 best_subset = max(subsets, key=lambda x: len(x.set - covered) / x.cost) 

213 cover_subsets.append(best_subset) 

214 covered |= best_subset.set 

215 

216 if confidence is None: 

217 if len(covered) == len(universe): 

218 break 

219 else: 

220 if len(covered) >= len(universe) * confidence: 

221 break 

222 

223 return cover_subsets