Coverage for human_requests/pytest_plugin/__init__.py: 86%

133 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-05-28 00:39 +0000

1from __future__ import annotations 

2 

3from collections.abc import Generator 

4from typing import Any 

5 

6import pytest 

7from _pytest.reports import TestReport 

8 

9try: 

10 from _pytest.subtests import SubtestReport 

11except ImportError: # pragma: no cover - pytest without built-in subtests support 

12 SubtestReport = None # type: ignore[assignment, misc] 

13 

14from ..autotest_report import AutotestCrash 

15from ._config import get_start_class_path, register_ini_options 

16from ._constants import AUTOTEST_TEST_NAME 

17from ._runtime import _autotest_anyio_runner, run_autotest_tree_anyio, run_autotest_tree_sync 

18 

19 

20class _AutotestRunnerReport(TestReport): 

21 @property 

22 def count_towards_summary(self) -> bool: 

23 return False 

24 

25 

26def pytest_addoption(parser: pytest.Parser) -> None: 

27 register_ini_options(parser) 

28 

29 

30def pytest_collection_modifyitems( 

31 session: pytest.Session, 

32 config: pytest.Config, 

33 items: list[pytest.Item], 

34) -> None: 

35 if not get_start_class_path(config): 

36 return 

37 

38 callobj = run_autotest_tree_sync 

39 if _has_anyio_plugin(config): 

40 callobj = run_autotest_tree_anyio 

41 

42 runner_parent = _pick_runner_parent(session=session, items=items) 

43 runner = pytest.Function.from_parent( 

44 parent=runner_parent, 

45 name=AUTOTEST_TEST_NAME, 

46 callobj=callobj, 

47 ) 

48 setattr(runner, "_human_requests_autotest_runner", True) 

49 items.append(runner) 

50 terminalreporter = config.pluginmanager.get_plugin("terminalreporter") 

51 if terminalreporter is not None and hasattr(terminalreporter, "_numcollected"): 

52 terminalreporter._numcollected += 1 

53 

54 

55@pytest.hookimpl(hookwrapper=True, tryfirst=True) 

56def pytest_runtest_makereport( 

57 item: pytest.Item, 

58 call: pytest.CallInfo[object], 

59) -> Generator[None, Any, None]: 

60 outcome = yield 

61 report = outcome.get_result() 

62 if getattr(item, "_human_requests_autotest_runner", False): 

63 report.__class__ = _AutotestRunnerReport 

64 if call.when != "call" or call.excinfo is None: 

65 return 

66 

67 error = call.excinfo.value 

68 if isinstance(error, AutotestCrash): 

69 report.longrepr = error.to_longrepr() 

70 

71 

72@pytest.hookimpl(hookwrapper=True, tryfirst=True) 

73def pytest_report_teststatus( 

74 report: pytest.TestReport, 

75 config: pytest.Config, 

76) -> Generator[None, Any, None]: 

77 outcome = yield 

78 result = outcome.get_result() 

79 if not isinstance(result, tuple) or len(result) != 3: 

80 return 

81 if SubtestReport is not None and isinstance(report, SubtestReport): 

82 if AUTOTEST_TEST_NAME not in report.nodeid: 

83 return 

84 category, _letter, word = result 

85 if report.passed: 

86 outcome.force_result((category, ".", word)) 

87 elif report.failed: 

88 outcome.force_result((category, "f", word)) 

89 return 

90 

91 if isinstance(report, _AutotestRunnerReport) and report.when == "call": 

92 if report.failed and isinstance(report.longrepr, str): 

93 if "failed subtest" in report.longrepr: 

94 status = _resolve_runner_teststatus(config) 

95 if status is not None: 

96 outcome.force_result(status) 

97 return 

98 

99 

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

101 del exitstatus 

102 terminalreporter = session.config.pluginmanager.get_plugin("terminalreporter") 

103 if terminalreporter is None: 

104 return 

105 

106 _maybe_suppress_failure_section(terminalreporter) 

107 

108 

109@pytest.hookimpl(tryfirst=True) 

110def pytest_terminal_summary(terminalreporter: pytest.TerminalReporter) -> None: 

111 config = terminalreporter.config 

112 labels = getattr(config, "_human_requests_autotest_success_labels", []) 

113 if labels: 

114 passed_reports = terminalreporter.stats.setdefault("passed", []) 

115 for index, label in enumerate(labels, start=1): 

116 nodeid = f"{AUTOTEST_TEST_NAME}[success-{index}:{label}]" 

117 passed_reports.append( 

118 TestReport( 

119 nodeid=nodeid, 

120 location=("", None, label), 

121 keywords={}, 

122 outcome="passed", 

123 longrepr=None, 

124 when="call", 

125 ) 

126 ) 

127 

128 _maybe_suppress_failure_section(terminalreporter) 

129 

130 

131def _has_anyio_plugin(config: pytest.Config) -> bool: 

132 return bool(config.pluginmanager.has_plugin("anyio")) 

133 

134 

135def _pick_runner_parent(session: pytest.Session, items: list[pytest.Item]) -> pytest.Collector: 

136 for item in items: 

137 if isinstance(item, pytest.Function): 

138 parent = item.parent 

139 if isinstance(parent, pytest.Collector): 

140 return parent 

141 return session 

142 

143 

144def _resolve_runner_teststatus( 

145 config: pytest.Config, 

146) -> tuple[str, str, str | tuple[str, dict[str, bool]]] | None: 

147 records = _get_case_records(config) 

148 if not records: 

149 return None 

150 

151 statuses = [status for _label, status in records] 

152 passed = any(status == "passed" for status in statuses) 

153 failed = any(status == "failed" for status in statuses) 

154 if passed and failed: 

155 return "failed", "M", ("mixed", {"yellow": True}) 

156 if failed: 

157 return "failed", "F", ("failed", {"red": True}) 

158 return "passed", ".", ("passed", {"green": True}) 

159 

160 

161def _maybe_suppress_failure_section(terminalreporter: pytest.TerminalReporter) -> None: 

162 if getattr(terminalreporter, "_human_requests_autotest_skip_failures", False): 

163 return 

164 

165 failed_reports = terminalreporter.stats.get("failed", []) 

166 if not failed_reports: 

167 return 

168 

169 if not any(_is_autotest_runner_report(report) for report in failed_reports): 

170 return 

171 

172 setattr(terminalreporter, "_human_requests_autotest_skip_failures", True) 

173 original_summary_failures = terminalreporter.summary_failures 

174 

175 def _summary_failures_without_autotest_runner() -> None: 

176 reports = terminalreporter.stats.get("failed", []) 

177 filtered_reports = [report for report in reports if not _is_autotest_runner_report(report)] 

178 if len(filtered_reports) == len(reports): 

179 return original_summary_failures() 

180 

181 terminalreporter.stats["failed"] = filtered_reports 

182 try: 

183 return original_summary_failures() 

184 finally: 

185 terminalreporter.stats["failed"] = reports 

186 

187 setattr(terminalreporter, "summary_failures", _summary_failures_without_autotest_runner) 

188 

189 

190def _is_autotest_runner_report(report: TestReport) -> bool: 

191 return isinstance(report, _AutotestRunnerReport) 

192 

193 

194def _get_case_records(config: pytest.Config) -> list[tuple[str, str]]: 

195 records = getattr(config, "_human_requests_autotest_case_records", None) 

196 if records: 

197 return list(records) 

198 

199 legacy_statuses = getattr(config, "_human_requests_autotest_case_statuses", None) 

200 if not legacy_statuses: 

201 return [] 

202 

203 if legacy_statuses and isinstance(legacy_statuses[0], tuple): 

204 return list(legacy_statuses) 

205 

206 return [(str(index), str(status)) for index, status in enumerate(legacy_statuses, start=1)] 

207 

208 

209__all__ = [ 

210 "pytest_addoption", 

211 "pytest_collection_modifyitems", 

212 "pytest_runtest_makereport", 

213 "pytest_terminal_summary", 

214 "_autotest_anyio_runner", 

215]