Coverage for human_requests/pytest_plugin/_runtime.py: 90%

1from __future__ import annotations 

2 

3import asyncio 

4from collections.abc import Coroutine 

5from typing import Any, Callable, TypeVar 

6 

7import pytest 

8 

9from ..autotest import execute_autotests, execute_autotests_with_subtests 

10from ._config import ( 

11 get_trace_limit, 

12 get_truncation_context_lines, 

13 get_typecheck_mode, 

14 resolve_runtime_dependencies, 

15) 

16 

17T = TypeVar("T") 

18 

19 

20def run_autotest_tree_sync(request: pytest.FixtureRequest) -> None: 

21 api, schemashot = resolve_runtime_dependencies(request) 

22 typecheck_mode = get_typecheck_mode(request.config) 

23 trace_limit = get_trace_limit(request.config) 

24 truncation_context_lines = get_truncation_context_lines(request.config) 

25 subtests = _resolve_subtests_fixture(request) 

26 case_status_recorder = _resolve_case_status_recorder(request) 

27 success_recorder = _resolve_success_recorder(request) 

28 executed_count = run_coroutine( 

29 _execute_autotests_async( 

30 api=api, 

31 schemashot=schemashot, 

32 typecheck_mode=typecheck_mode, 

33 trace_limit=trace_limit, 

34 truncation_context_lines=truncation_context_lines, 

35 subtests=subtests, 

36 case_status_recorder=case_status_recorder, 

37 success_recorder=success_recorder, 

38 ) 

39 ) 

40 if executed_count == 0: 

41 pytest.skip("No methods marked with @autotest were found in the api tree.") 

42 

43 

44@pytest.mark.usefixtures("_autotest_anyio_runner") 

45def run_autotest_tree_anyio(request: pytest.FixtureRequest) -> None: 

46 runner = request.getfixturevalue("_autotest_anyio_runner") 

47 api, schemashot = resolve_runtime_dependencies(request) 

48 typecheck_mode = get_typecheck_mode(request.config) 

49 trace_limit = get_trace_limit(request.config) 

50 truncation_context_lines = get_truncation_context_lines(request.config) 

51 subtests = _resolve_subtests_fixture(request) 

52 case_status_recorder = _resolve_case_status_recorder(request) 

53 success_recorder = _resolve_success_recorder(request) 

54 executed_count = runner.run_test( 

55 _execute_autotests_async, 

56 { 

57 "api": api, 

58 "schemashot": schemashot, 

59 "typecheck_mode": typecheck_mode, 

60 "trace_limit": trace_limit, 

61 "truncation_context_lines": truncation_context_lines, 

62 "subtests": subtests, 

63 "case_status_recorder": case_status_recorder, 

64 "success_recorder": success_recorder, 

65 }, 

66 ) 

67 if executed_count == 0: 

68 pytest.skip("No methods marked with @autotest were found in the api tree.") 

69 

70 

71@pytest.fixture 

72def _autotest_anyio_runner(anyio_backend: Any) -> Any: 

73 from anyio.pytest_plugin import extract_backend_and_options, get_runner 

74 

75 backend_name, backend_options = extract_backend_and_options(anyio_backend) 

76 with get_runner(backend_name, backend_options) as runner: 

77 yield runner 

78 

79 

80async def _execute_autotests_async( 

81 api: object, 

82 schemashot: Any, 

83 typecheck_mode: str, 

84 trace_limit: int, 

85 truncation_context_lines: int, 

86 subtests: Any | None = None, 

87 case_status_recorder: Callable[[str, str], None] | None = None, 

88 success_recorder: Any | None = None, 

89) -> int: 

90 if subtests is not None: 

91 return await execute_autotests_with_subtests( 

92 api=api, 

93 schemashot=schemashot, 

94 subtests=subtests, 

95 typecheck_mode=typecheck_mode, 

96 trace_limit=trace_limit, 

97 truncation_context_lines=truncation_context_lines, 

98 case_status_recorder=case_status_recorder, 

99 success_recorder=success_recorder, 

100 ) 

101 return await execute_autotests( 

102 api=api, 

103 schemashot=schemashot, 

104 typecheck_mode=typecheck_mode, 

105 trace_limit=trace_limit, 

106 truncation_context_lines=truncation_context_lines, 

107 case_status_recorder=case_status_recorder, 

108 success_recorder=success_recorder, 

109 ) 

110 

111 

112def _resolve_subtests_fixture(request: pytest.FixtureRequest) -> Any | None: 

113 if not request.config.pluginmanager.has_plugin("subtests"): 

114 return None 

115 try: 

116 return request.getfixturevalue("subtests") 

117 except pytest.FixtureLookupError: 

118 return None 

119 

120 

121def _resolve_success_recorder(request: pytest.FixtureRequest) -> Callable[[str], None]: 

122 config = request.config 

123 setattr(config, "_human_requests_autotest_success_labels", []) 

124 

125 def _record_success(label: str) -> None: 

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

127 labels.append(label) 

128 

129 return _record_success 

130 

131 

132def _resolve_case_status_recorder(request: pytest.FixtureRequest) -> Callable[[str, str], None]: 

133 config = request.config 

134 setattr(config, "_human_requests_autotest_case_records", []) 

135 

136 def _record_status(label: str, status: str) -> None: 

137 records = getattr(config, "_human_requests_autotest_case_records", []) 

138 records.append((label, status)) 

139 

140 return _record_status 

141 

142 

143def run_coroutine(coro: Coroutine[Any, Any, T]) -> T: 

144 try: 

145 asyncio.get_running_loop() 

146 except RuntimeError: 

147 return asyncio.run(coro) 

148 raise RuntimeError( 

149 "Autotest plugin is running inside an active event loop. " 

150 "Run it from a synchronous pytest context." 

151 )