Coverage for /home/crpier/Projects/snektest/snektest/decorators.py: 47%

1import asyncio 

2from collections.abc import Callable 

3from concurrent.futures import Future 

4from functools import wraps 

5from inspect import Parameter, Signature, currentframe, iscoroutinefunction 

6from typing import Any, Literal, Protocol, TypeVar, cast, overload 

7 

8from hypothesis import given 

9 

10from snektest.annotations import ( 

11 AsyncFixture, 

12 AsyncSessionFixture, 

13 Coroutine, 

14 Fixture, 

15 SessionFixture, 

16) 

17from snektest.fixtures import ( 

18 is_session_fixture, 

19 load_function_fixture, 

20 load_session_fixture, 

21 register_session_fixture_from_namespace, 

22) 

23from snektest.models import Param 

24from snektest.utils import get_code_from_generator, mark_test_function 

25 

26_given = cast("Any", given) 

27 

28T_co = TypeVar("T_co", covariant=True) 

29 

30 

31class SearchStrategy(Protocol[T_co]): 

32 def example(self) -> T_co: ... 

33 

34 

35type Marker = Literal["fast", "medium", "slow"] 

36"""Markers for test functions. 

37 

38Markers describe the resources a test may use, 

39not how long it is expected to take. 

40 

41`fast` means the test runs entirely in memory, 

42without IO, threads, or subprocesses. 

43`medium` means the test may use local IO or threads, 

44but not network IO or subprocesses. 

45`slow` means the test may use network IO, subprocesses, 

46or other expensive external resources. 

47""" 

48 

49 

50VALID_MARKERS: tuple[Marker, ...] = ("slow", "medium", "fast") 

51 

52 

53def _normalize_marker_entry(entry: object) -> str: 

54 if entry in VALID_MARKERS: 

55 return entry 

56 msg = "Markers must be Marker literals" 

57 raise TypeError(msg) 

58 

59 

60def _normalize_markers(mark: object | None) -> tuple[str, ...]: 

61 if mark is None: 

62 return () 

63 if mark in VALID_MARKERS: 

64 return (_normalize_marker_entry(mark),) 

65 msg = "Markers must be a single Marker literal" 

66 raise TypeError(msg) 

67 

68 

69@overload 

70def test( 

71 *params: list[Param[Any]], 

72 mark: Marker | None = None, 

73) -> Callable[ 

74 [Callable[[*tuple[Any, ...]], Coroutine[None] | None]], 

75 Callable[[*tuple[Any, ...]], Coroutine[None] | None], 

76]: ... 

77 

78 

79@overload 

80def test( 

81 *params: list[Param[Any]], 

82 mark: object | None = None, 

83) -> Callable[ 

84 [Callable[[*tuple[Any, ...]], Coroutine[None] | None]], 

85 Callable[[*tuple[Any, ...]], Coroutine[None] | None], 

86]: ... 

87 

88 

89def test( 

90 *params: list[Param[Any]], 

91 mark: object | None = None, 

92) -> Callable[ 

93 [Callable[[*tuple[Any, ...]], Coroutine[None] | None]], 

94 Callable[[*tuple[Any, ...]], Coroutine[None] | None], 

95]: 

96 """Mark a function as a test function with an optional built-in marker.""" 

97 

98 markers = _normalize_markers(mark) 

99 

100 def decorator( 

101 test_func: Callable[[*tuple[Any, ...]], Coroutine[None] | None], 

102 ) -> Callable[[*tuple[Any, ...]], Coroutine[None] | None]: 

103 mark_test_function(test_func, params, markers) 

104 return test_func 

105 

106 return decorator 

107 

108 

109def _maybe_apply_hypothesis_settings( 

110 source: Callable[..., object], 

111 target: Callable[..., object], 

112) -> Callable[..., object]: 

113 """Propagate `@hypothesis.settings` applied to `source` onto `target`.""" 

114 

115 settings_obj = getattr(source, "_hypothesis_internal_use_settings", None) 

116 if settings_obj is None: 

117 return target 

118 

119 settings_decorator = cast( 

120 "Callable[[Callable[..., object]], Callable[..., object]]", 

121 settings_obj, 

122 ) 

123 return settings_decorator(target) 

124 

125 

126def _run_hypothesis( 

127 wrapper: Callable[..., object], 

128 strategies: tuple[SearchStrategy[Any], ...], 

129 run_one_example: Callable[..., None], 

130) -> None: 

131 def hypothesis_runner(*strategy_values: Any) -> None: 

132 run_one_example(*strategy_values) 

133 

134 signature = Signature( 

135 parameters=[ 

136 Parameter(f"arg{i}", kind=Parameter.POSITIONAL_OR_KEYWORD) 

137 for i in range(len(strategies)) 

138 ] 

139 ) 

140 hypothesis_runner.__signature__ = signature # pyright: ignore[reportFunctionMemberAccess] 

141 

142 hypothesis_runner_wrapped = _given(*strategies)(hypothesis_runner) 

143 

144 runner = cast( 

145 "Callable[[], None]", 

146 _maybe_apply_hypothesis_settings(wrapper, hypothesis_runner_wrapped), 

147 ) 

148 runner() 

149 

150 

151def _run_async_example( 

152 loop: asyncio.AbstractEventLoop, 

153 test_func: Callable[..., Coroutine[None] | None], 

154 *, 

155 strategy_values: tuple[Any, ...], 

156 param_values: tuple[Any, ...], 

157) -> None: 

158 done: Future[None] = Future() 

159 

160 def schedule() -> None: 

161 try: 

162 res = cast( 

163 "Coroutine[None]", 

164 test_func(*strategy_values, *param_values), 

165 ) 

166 task: asyncio.Task[None] = loop.create_task(res) 

167 except Exception as exc: 

168 done.set_exception(exc) 

169 return 

170 

171 def on_done(task: asyncio.Task[None]) -> None: 

172 try: 

173 task.result() 

174 except Exception as exc: 

175 done.set_exception(exc) 

176 else: 

177 done.set_result(None) 

178 

179 task.add_done_callback(on_done) 

180 

181 _ = loop.call_soon_threadsafe(schedule) 

182 done.result() 

183 

184 

185def test_hypothesis( 

186 *strategies: SearchStrategy[Any], 

187 mark: Marker | None = None, 

188) -> Callable[ 

189 [Callable[..., Coroutine[None] | None]], 

190 Callable[..., Coroutine[None] | None], 

191]: 

192 """Mark a function as a property-based test using Hypothesis. 

193 

194 Strategies are positional and fill function arguments from left to right. 

195 Use `mark=` with a `Marker` literal to attach a built-in snektest marker. 

196 

197 Notes: 

198 - Hypothesis cannot directly run async functions; for `async def` tests we run 

199 the Hypothesis engine in a worker thread and schedule the async test body 

200 back onto the main event loop. 

201 - Apply `@hypothesis.settings(...)` above or below this decorator to adjust 

202 Hypothesis behavior. 

203 """ 

204 

205 if len(strategies) == 0: 

206 msg = "test_hypothesis() requires at least one strategy" 

207 raise ValueError(msg) 

208 

209 strategies_tuple = tuple(strategies) 

210 markers = _normalize_markers(mark) 

211 

212 def decorator( 

213 test_func: Callable[..., Coroutine[None] | None], 

214 ) -> Callable[..., Coroutine[None] | None]: 

215 if iscoroutinefunction(test_func): 

216 

217 @wraps(test_func) 

218 async def async_wrapper() -> None: 

219 loop = asyncio.get_running_loop() 

220 

221 def run_one_example(*strategy_values: Any) -> None: 

222 _run_async_example( 

223 loop, 

224 test_func, 

225 strategy_values=tuple(strategy_values), 

226 param_values=(), 

227 ) 

228 

229 def run_hypothesis() -> None: 

230 _run_hypothesis(async_wrapper, strategies_tuple, run_one_example) 

231 

232 await asyncio.to_thread(run_hypothesis) 

233 

234 mark_test_function(async_wrapper, (), markers) 

235 return async_wrapper 

236 

237 @wraps(test_func) 

238 def sync_wrapper() -> None: 

239 def run_one_example(*strategy_values: Any) -> None: 

240 _ = test_func(*strategy_values) 

241 

242 _run_hypothesis(sync_wrapper, strategies_tuple, run_one_example) 

243 

244 mark_test_function(sync_wrapper, (), markers) 

245 return sync_wrapper 

246 

247 return decorator 

248 

249 

250@overload 

251def load_fixture[R]( 

252 fixture_gen: Fixture[R] | SessionFixture[R], 

253) -> R: ... 

254 

255 

256@overload 

257def load_fixture[R]( 

258 fixture_gen: AsyncFixture[R] | AsyncSessionFixture[R], 

259) -> Coroutine[R]: ... 

260 

261 

262def load_fixture[R]( 

263 fixture_gen: Fixture[R] 

264 | AsyncFixture[R] 

265 | SessionFixture[R] 

266 | AsyncSessionFixture[R], 

267) -> Coroutine[R] | R: 

268 """Load a fixture from a generator. 

269 When loading a fixture, `snektest` takes care to handle tearing down the 

270 fixture after the test has finished.""" 

271 fixture_gen_code = get_code_from_generator(fixture_gen) 

272 frame = currentframe() 

273 caller_frame = frame.f_back if frame is not None else None 

274 if caller_frame is not None: 

275 register_session_fixture_from_namespace( 

276 fixture_gen_code, caller_frame.f_globals 

277 ) 

278 register_session_fixture_from_namespace(fixture_gen_code, caller_frame.f_locals) 

279 del frame 

280 del caller_frame 

281 

282 if is_session_fixture(fixture_gen_code): 

283 return load_session_fixture(fixture_gen) 

284 

285 return load_function_fixture(fixture_gen)