Coverage for pytest_steps/steps_generator.py : 77%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Authors: Sylvain MARIE <sylvain.marie@se.com>
2# + All contributors to <https://github.com/smarie/python-pytest-steps>
3#
4# License: 3-clause BSD, <https://github.com/smarie/python-pytest-steps/blob/master/LICENSE>
5try:
6 from collections.abc import Iterable as It
7except ImportError:
8 from collections import Iterable as It
10from makefun import add_signature_parameters, wraps
11from wrapt import ObjectProxy
13# try: # python 3.2+
14# from functools import lru_cache
15# except ImportError:
16# from functools32 import lru_cache
18try: # python 3.3+
19 from inspect import signature, Parameter
20except ImportError:
21 from funcsigs import signature, Parameter
23from inspect import isgeneratorfunction
25try: # python 3+
26 from typing import Iterable, Any, Union
27except ImportError:
28 pass
30import pytest
32from .common_mini_six import string_types, reraise
33from .steps_common import create_pytest_param_str_id, get_pytest_node_hash_id, get_scope
36class ExceptionHook(object):
37 """
38 A context manager used to register a hook for exceptions.
39 This hook (a method provided in constructor) will be called if an exception is caught.
40 The exception will then be raised as usual.
42 Example:
43 --------
44 >>>with ExceptionHook(print):
45 >>> assert False
47 The hook method ('print' in the above example) will be called with the type of exception, the exception value,
48 and the exception traceback, before the exception is actually raised.
49 """
50 def __init__(self, exc_handler):
51 """
52 Constructor.
54 :param exc_handler: the method that should be called if an exception is raised inside this context manager.
55 """
56 self.exc_handler = exc_handler
58 def __enter__(self):
59 pass
61 def __exit__(self, exc_type, exc_val, exc_tb):
62 if exc_type is not None:
63 self.exc_handler(exc_type, exc_val, exc_tb)
65 # return False so that the exception is always raised
66 return False
69class StepExecutionError(Exception):
70 """
71 Exception raised by a StepsMonitor when a step cannot be executed because the underlying generator function
72 returned a StopIteration.
73 """
74 def __init__(self, step_name):
75 self.step_name = step_name
76 Exception.__init__(self)
78 def __str__(self):
79 return "Error executing step '%s': could not reach the next `yield` statement (received `StopIteration`). " \
80 "This may be caused by use of a `return` statement instead of a `yield`, or by a missing `yield`" \
81 "" % self.step_name
84class StepYieldError(Exception):
85 """
86 Exception raised by a StepsMonitor when a step does not yield anything (None), or yields a string that is not the
87 correct step name, or yields an object that is not an `optional_step`
88 """
89 def __init__(self, step_name, received):
90 self.step_name = step_name
91 self.received = received
92 Exception.__init__(self)
94 def __str__(self):
95 return "Error collecting results from step '%s': received '%s' from the `yield` statement, which is different" \
96 " from the current step or step name. Please either use `yield`, `yield '%s'` or wrap your step with " \
97 "`with optional_step(...) as my_step:` and use `yield my_step`" \
98 % (self.step_name, self.received, self.step_name)
101class _OnePerStepFixtureProxy(ObjectProxy):
102 """
103 An object container for which one can change the inner instance.
104 By default wrapt.ObjectProxy does the job perfectly, so this object behaves
105 transparently like the fixture it wraps.
107 If for some reason you still wish to access the underlying fixture object, please rely on the public API
108 `get_underlying_fixture(obj)` rather than calling `obj.__wrapped__`.
110 We go one step further by also proxying the representation
111 """
112 def __repr__(self):
113 return repr(self.__wrapped__)
116def is_replacable_fixture_wrapper(obj):
117 """
118 Returns True when the object results from a function-scoped fixture that has been decorated with
119 @one_fixture_per_step.
121 In that case the fixture value of the first step is wrapped in a `_OnePerStepFixtureProxy`, so that we can inject
122 the other fixture values in it later. Indeed otherwise the fixture values for the other steps will never be injected
123 in the generator test function (because its args are provided only once at the first step).
125 :param obj:
126 :return:
127 """
128 return isinstance(obj, _OnePerStepFixtureProxy)
131def replace_fixture(rfw1, rfw2):
132 """
133 Replaces the contents of fixture obj1 with the ones from fixture obj2. This only works if both are replaceable
134 fixture wrappers
136 :param rfw1:
137 :param rfw2:
138 :return:
139 """
140 if is_replacable_fixture_wrapper(rfw1) and is_replacable_fixture_wrapper(rfw2): 140 ↛ 143line 140 didn't jump to line 143, because the condition on line 140 was never false
141 rfw1.__wrapped__ = rfw2.__wrapped__
142 else:
143 raise TypeError("both objects should come from the same fixture, decorated with @one_fixture_per_step")
146def get_underlying_fixture(rfw):
147 """
148 Returns the underlying fixture object inside this fixture wrapper, or returns the fixture itself in case it is
149 not a one_fixture_per_step fixture wrapper
151 :param rfw:
152 :return:
153 """
154 if is_replacable_fixture_wrapper(rfw): 154 ↛ 155line 154 didn't jump to line 155, because the condition on line 154 was never true
155 return rfw.__wrapped__
156 else:
157 return rfw
160def one_fixture_per_step(*args):
161 """
162 A decorator for a function-scoped fixture so that it works well with generator-mode test functions. You do not have
163 to use it in parametrizer mode, although it does not hurt.
165 By default if you do not use this decorator but use the fixture in a generator-mode test function, only the
166 fixture created for the first step will be injected in your test function, and all subsequent steps will see that
167 same instance.
169 Decorating your fixture with `@one_fixture_per_step` tells `@test_steps` to transparently replace the fixture object
170 instance by the one created for each step, before each step executes in your test function. This results in all
171 steps using different fixture instances, as expected.
173 It is recommended that you put this decorator as the second decorator, right after `@pytest.fixture`:
175 ```python
176 @pytest.fixture
177 @one_fixture_per_step
178 def my_cool_fixture():
179 return random()
180 ```
182 Note: When a fixture is decorated with `@one_fixture_per_step`, the object that is injected in your test function
183 is a transparent proxy of the fixture, so it behaves exactly like the fixture. If for some reason you want to get
184 the "true" inner wrapped object, you can do so using `get_underlying_fixture(my_fixture)`.
185 :return:
186 """
187 if len(args) == 1 and callable(args[0]): 187 ↛ 189line 187 didn't jump to line 189, because the condition on line 187 was never false
188 return one_fixture_per_step_decorate(args[0])
189 elif len(args) == 0:
190 return one_fixture_per_step_decorate
191 else:
192 raise ValueError("@one_fixture_per_step accepts no argument")
195one_per_step = one_fixture_per_step
196"""Deprecated alias for `@one_fixture_per_step`."""
199def one_fixture_per_step_decorate(fixture_fun):
200 """ Implementation of the @one_fixture_per_step decorator, for manual decoration"""
202 def _check_scope(request):
203 scope = get_scope(request)
204 if scope != 'function': 204 ↛ 206line 204 didn't jump to line 206, because the condition on line 204 was never true
205 # session- or module-scope
206 raise Exception("The `@one_fixture_per_step` decorator is only useful for function-scope fixtures. `%s`"
207 " seems to have scope='%s'. Consider removing `@one_fixture_per_step` or changing "
208 "the scope to 'function'." % (fixture_fun, scope))
210 # We will expose a new signature with additional arguments
211 orig_sig = signature(fixture_fun)
212 func_needs_request = 'request' in orig_sig.parameters
213 if not func_needs_request: 213 ↛ 216line 213 didn't jump to line 216, because the condition on line 213 was never false
214 new_sig = add_signature_parameters(orig_sig, first=Parameter('request', kind=Parameter.POSITIONAL_OR_KEYWORD))
215 else:
216 new_sig = orig_sig
218 if not isgeneratorfunction(fixture_fun): 218 ↛ 226line 218 didn't jump to line 226, because the condition on line 218 was never false
219 @wraps(fixture_fun, new_sig=new_sig)
220 def _steps_aware_decorated_function(*args, **kwargs):
221 request = kwargs['request'] if func_needs_request else kwargs.pop('request')
222 _check_scope(request)
223 res = fixture_fun(*args, **kwargs)
224 return _OnePerStepFixtureProxy(res)
225 else:
226 @wraps(fixture_fun, new_sig=new_sig)
227 def _steps_aware_decorated_function(*args, **kwargs):
228 request = kwargs['request'] if func_needs_request else kwargs.pop('request')
229 _check_scope(request)
230 gen = fixture_fun(*args, **kwargs)
231 res = next(gen)
232 yield _OnePerStepFixtureProxy(res)
233 next(gen)
235 return _steps_aware_decorated_function
238one_per_step_decorate = one_fixture_per_step_decorate
239"""Deprecated alias for `one_fixture_per_step_decorate`"""
242class StepsMonitor(object):
243 """
244 An object responsible to _monitor execution of a test function with steps.
245 The function should be a generator
246 """
247 def __init__(self, step_names, test_function, first_step_args, first_step_kwargs):
248 """
249 Constructor with declaration of all step names in advance,
250 as well as the test function to execute and the first step args and kwargs
252 Nothing will be executed here, the test function will only be called once to create the generator.
254 :param step_names:
255 :param test_function:
256 :param first_step_args:
257 :param first_step_kwargs:
258 """
259 self.steps = step_names
260 self.exceptions = dict()
262 # Remember objects that should be replaced in subsequent steps
263 # -- for positional arguments, store in a dict under key=position
264 self.replaceable_args = dict()
265 for i, a in enumerate(first_step_args): 265 ↛ 266line 265 didn't jump to line 266, because the loop on line 265 never started
266 if is_replacable_fixture_wrapper(a):
267 self.replaceable_args[i] = a
269 # -- for keyword arguments, store in a dict under key=name
270 self.replaceable_kwargs = dict()
271 for k, a in first_step_kwargs.items():
272 if is_replacable_fixture_wrapper(a):
273 self.replaceable_kwargs[k] = a
275 # create the generator
276 self.gen = test_function(*first_step_args, **first_step_kwargs)
278 def can_execute(self, step_name):
279 """
280 As of today a step can execute if there are no registered exceptions (in previous mandatory steps).
281 :param step_name:
282 :return:
283 """
284 return len(self.exceptions) == 0
286 def execute(self, step_name, args, kwargs):
287 """
288 Executes one iteration of the monitored generator.
290 :param step_name:
291 :return:
292 """
294 if self.can_execute(step_name):
295 # Replace all objects that should be replaced
296 for i, a in self.replaceable_args.items(): 296 ↛ 297line 296 didn't jump to line 297, because the loop on line 296 never started
297 replace_fixture(a, args[i])
298 for k, a in self.replaceable_kwargs.items():
299 replace_fixture(a, kwargs[k])
301 # Execute the step
302 with self._monitor(step_name):
303 try:
304 res = next(self.gen)
305 except StopIteration:
306 raise StepExecutionError(step_name)
308 # Manage exceptions in optional steps
309 if res is None:
310 # We now accept None yields
311 # raise StepYieldError(step_name, None)
312 pass
314 elif isinstance(res, string_types):
315 if res != step_name: 315 ↛ 316line 315 didn't jump to line 316, because the condition on line 315 was never true
316 raise StepYieldError(step_name, res)
318 elif isinstance(res, optional_step): 318 ↛ 346line 318 didn't jump to line 346, because the condition on line 318 was never false
319 # optional step: check if the execution went well
320 if res.exec_result is None: 320 ↛ 321line 320 didn't jump to line 321, because the condition on line 320 was never true
321 raise ValueError("Internal error: this should not happen")
323 elif isinstance(res.exec_result, OptionalStepException):
324 # An exception happened in the optional step. We can now raise it safely
325 # (raising it sooner would break the generator)
326 reraise(res.exec_result.exc_type, res.exec_result.exc_val, res.exec_result.tb)
328 elif isinstance(res.exec_result, _DependentTestsNotRunException): 328 ↛ 339line 328 didn't jump to line 339, because the condition on line 328 was never false
329 # This exception has been put here to declare that the optional step did not run because a
330 # dependency is missing. >> Skip or fail
331 # TODO add fail_instead_of_skip argument like in depends_on
332 should_fail = False
333 msg = "This test step '%s' depends on other steps, and the following have failed: %s" \
334 "" % (step_name, res.exec_result.dependency_names)
335 if should_fail: 335 ↛ 336line 335 didn't jump to line 336, because the condition on line 335 was never true
336 pytest.fail(msg)
337 else:
338 pytest.skip(msg)
339 elif res.exec_result is True:
340 # normal execution success
341 pass
342 else:
343 raise ValueError("Internal error: this should not happen")
344 else:
345 # the generator yielded a res that is not of an accepted type
346 raise StepYieldError(step_name, res)
347 else:
348 # A mandatory step failed before this one. The generator is broken, no need to even try >> Skip or fail
349 # TODO add fail_instead_of_skip argument like in depends_on
350 should_fail2 = False
351 failed_step = next(iter(self.exceptions.keys())) if len(self.exceptions) == 1 \
352 else list(self.exceptions.keys())
353 msg = "This test step '%s' is not run because non-optional previous step '%s' has failed" \
354 "" % (step_name, failed_step)
355 if should_fail2: 355 ↛ 356line 355 didn't jump to line 356, because the condition on line 355 was never true
356 pytest.fail(msg)
357 else:
358 pytest.skip(msg)
360 def _monitor(self, step_name):
361 """ returns a context manager that registers all captured exceptions in self, under given step name """
363 def handle_exception(exc_type, exc_val, exc_tb):
364 if exc_type in {_DependentTestsNotRunException, OptionalStepException}: 364 ↛ 366line 364 didn't jump to line 366, because the condition on line 364 was never true
365 # Do not register this exception
366 pass
367 else:
368 self.exceptions[step_name] = exc_type, exc_val, exc_tb
370 return ExceptionHook(handle_exception)
373class StepMonitorsContainer(dict):
374 """
375 A dictionary of step monitors
376 It contains all the StepsMonitor created for this function.
377 there will be one StepsMonitor created for each unique function call
378 """
380 def __init__(self, test_func, step_ids):
381 self.test_func = test_func
382 self.step_ids = step_ids
383 dict.__init__(self)
385 def get_execution_monitor(self, pytest_node, args, kwargs):
386 """
387 Returns the StepsMonitor in charge of monitoring execution of the provided pytest node. The same StepsMonitor
388 will be used to execute all steps of the generator function.
390 If there is no monitor yet (first function call with this combination of parameters), then one is created,
391 that will be used subsequently.
393 :param pytest_node:
394 :param args:
395 :param kwargs:
396 :return:
397 """
398 # Get the unique id that is shared between the steps of the same execution, by removing the step parameter
399 # Note: when the id was using not only param values but also fixture values we had to discard
400 # 'request' and maybe some fixtures here. But that's not the case anymore,simply discard the "test step" param
401 id_without_steps = get_pytest_node_hash_id(pytest_node, params_to_ignore=(GENERATOR_MODE_STEP_ARGNAME,))
403 if id_without_steps not in self:
404 # First time we call the function with this combination of parameters
405 # print("DEBUG - creating StepsMonitor for %s" % id_without_steps)
407 # create the monitor, in charge of managing the execution flow
408 self[id_without_steps] = StepsMonitor(self.step_ids, self.test_func, args, kwargs)
410 return self[id_without_steps]
413GENERATOR_MODE_STEP_ARGNAME = "________step_name_"
416def get_generator_decorator(steps # type: Iterable[Any]
417 ):
418 """
419 Subroutine of `test_steps` used to perform the test function parametrization when mode is 'generator'.
421 :param steps: a list of steps that the decorated function is supposed to perform. The decorated function should be
422 a generator, and should `yield` as many steps as declared in the decorator.
423 :return: a function decorator
424 """
425 def steps_decorator(test_func):
426 """
427 The test function decorator. When a function is decorated it
428 - checks that the function is a generator
429 - checks that the function signature does not contain our private name `GENERATOR_MODE_STEP_ARGNAME` "by
430 chance"
431 - wraps the function
432 :param test_func:
433 :return:
434 """
436 # ------VALIDATION -------
437 # Check if function is a generator
438 if not isgeneratorfunction(test_func): 438 ↛ 439line 438 didn't jump to line 439, because the condition on line 438 was never true
439 raise ValueError("Decorated function is not a generator. You either forgot to add `yield` statements in "
440 "its body, or you are using mode='generator' instead of mode='parametrizer' or 'auto'."
441 "See help(pytest_steps) for details")
443 # check that our name for the additional 'test step' parameter is valid (it does not exist in f signature)
444 f_sig = signature(test_func)
445 test_step_argname = GENERATOR_MODE_STEP_ARGNAME
446 if test_step_argname in f_sig.parameters: 446 ↛ 447line 446 didn't jump to line 447, because the condition on line 446 was never true
447 raise ValueError("Your test function relies on arg name %s that is needed by @test_steps in generator "
448 "mode" % test_step_argname)
450 # ------CORE -------
451 # Transform the steps into ids if needed
452 step_ids = [create_pytest_param_str_id(f) for f in steps]
454 # Create the container that will hold all execution monitors for this function
455 # TODO maybe have later a single 'monitor' instance at plugin level... like in pytest-benchmark
456 all_monitors = StepMonitorsContainer(test_func, step_ids)
458 # Create the function wrapper.
459 # We will expose a new signature with additional 'request' arguments if needed, and the test step
460 orig_sig = signature(test_func)
461 func_needs_request = 'request' in orig_sig.parameters
462 additional_params = (
463 (Parameter(test_step_argname, kind=Parameter.POSITIONAL_OR_KEYWORD), )
464 + ((Parameter('request', kind=Parameter.POSITIONAL_OR_KEYWORD), ) if not func_needs_request else ())
465 )
466 # add request parameter last, as first may be 'self'
467 new_sig = add_signature_parameters(orig_sig, last=additional_params)
469 # -- first create the logic
470 @wraps(test_func, new_sig=new_sig)
471 def wrapped_test_function(*args, **kwargs):
472 step_name = kwargs.pop(test_step_argname)
473 request = kwargs['request'] if func_needs_request else kwargs.pop('request')
474 if request is None:
475 # we are manually called outside of pytest. let's execute all steps at nce
476 if step_name is None:
477 # print("@test_steps - decorated function '%s' is being called manually. The `%s` parameter is set "
478 # "to None so all steps will be executed in order" % (f, test_step_argname))
479 step_names = step_ids
480 else:
481 # print("@test_steps - decorated function '%s' is being called manually. The `%s` parameter is set "
482 # "to %s so only these steps will be executed in order. Note that the order should be
483 # feasible"
484 # "" % (f, test_step_argname, step_name))
485 if not isinstance(step_name, (list, tuple)):
486 step_names = [create_pytest_param_str_id(step_name)]
487 else:
488 step_names = [create_pytest_param_str_id(f) for f in step_name]
489 steps_monitor = StepsMonitor(step_ids, test_func, args, kwargs)
490 for i, (step_name, ref_step_name) in enumerate(zip(step_names, step_ids)):
491 if step_name != ref_step_name:
492 raise ValueError("Incorrect sequence of steps provided for manual execution. Step #%s should"
493 " be named '%s', found '%s'" % (i+1, ref_step_name, step_name))
494 steps_monitor.execute(step_name, args, kwargs)
495 else:
496 # Retrieve or create the corresponding execution monitor
497 steps_monitor = all_monitors.get_execution_monitor(request.node, args, kwargs)
499 # execute the step
500 # print("DEBUG - executing step %s" % step_name)
501 steps_monitor.execute(step_name, args, kwargs)
503 # With this hack we will be ordered correctly by pytest https://github.com/pytest-dev/pytest/issues/4429
504 wrapped_test_function.place_as = test_func
506 # Parametrize the wrapper function with the test step ids
507 parametrizer = pytest.mark.parametrize(test_step_argname, step_ids, ids=str)
509 # finally apply parametrizer
510 parametrized_step_function_wrapper = parametrizer(wrapped_test_function)
511 return parametrized_step_function_wrapper
513 return steps_decorator
516# ----------- optional steps
518class _DependentTestsNotRunException(Exception):
519 """
520 An internal exception that is actually never raised: it is used by the optional_step context manager
521 """
523 def __init__(self, step_name, dependency_name):
524 self.step_name = step_name
525 self.dependency_names = [dependency_name]
528class OptionalStepException(Exception):
529 """ A wrapper for an exception caught during an optional step execution """
531 def __init__(self, exc_type, exc_val, tb):
532 self.exc_type = exc_type
533 self.exc_val = exc_val
534 self.tb = tb
537class optional_step(object):
538 """
539 A context manager that you can use in *generator* mode in order to declare a step as independent, so that next
540 steps in the generator can continue to execute even if this one fails.
542 When this context manager is used you should not forget to yield the context object ! Otherwise the test step will
543 be marked as successful even if it was not.
545 You can optionally declare dependencies using the `depends_on=` argument in the constructor. If so, you should use
546 the .should_execute() method if you wish your code block to be properly skipped.
548 ```python
549 from pytest_steps import test_steps, optional_step
551 @test_steps('step_a', 'step_b', 'step_c', 'step_d')
552 def test_suite_opt():
553 # Step A
554 assert not False
555 yield
557 # Step B
558 with optional_step('step_b') as step_b:
559 assert False
560 yield step_b
562 # Step C depends on step B
563 with optional_step('step_c', depends_on=step_b) as step_c:
564 if step_c.should_run():
565 assert True
566 yield step_c
568 # Step D
569 assert not False
570 yield
571 ```
572 """
574 def __init__(self,
575 step_name, # type: str
576 depends_on=None # type: Union[optional_step, Iterable[optional_step]]
577 ):
578 """
579 Creates the context manager for an optional step named `step_name` with optional dependencies on other
580 optional steps.
582 :param step_name: the name of this optional step. This name will be used in pytest failure/skip messages when
583 other steps depend on this one and are skipped/failed because this one was skipped/failed.
584 :param depends_on: an optional dependency or list of dependencies, that should all be optional steps created
585 with an `optional_step` context manager.
586 """
587 # default values
588 self.step_name = step_name
589 self.exec_result = None
590 self.depends_on = depends_on or []
592 # coerce depends_on to a list
593 if not isinstance(self.depends_on, It):
594 self.depends_on = [self.depends_on]
596 # dependencies should be optional steps too
597 for dependency in self.depends_on:
598 if not isinstance(dependency, optional_step): 598 ↛ 599line 598 didn't jump to line 599, because the condition on line 598 was never true
599 raise ValueError("depends_on should only contain optional_step instances")
601 def __str__(self):
602 return self.step_name
604 def __enter__(self):
605 # check that all dependencies have run
606 for dependency in self.depends_on:
607 if not dependency.ran_with_success(): 607 ↛ 606line 607 didn't jump to line 606, because the condition on line 607 was never false
608 if self.exec_result is None: 608 ↛ 611line 608 didn't jump to line 611, because the condition on line 608 was never false
609 self.exec_result = _DependentTestsNotRunException(self.step_name, dependency.step_name)
610 else:
611 self.exec_result.dependency_names.append(dependency.step_name)
613 # Unfortunately if we raise an exception here it will not be caught by the __exit__ method
614 # So there is absolutely no way to prevent the code block to execute,
615 # - even with an ExitStack (I tried !): indeed the ExitStack is nothing more than a context manager so if an
616 # error is raised during its __enter__ method, then its __exit__ stack will not be called
617 # - A PEP 377 was created for that but was rejected
618 # - A hack also exists but it interferes with the debugger so it is too complex
619 # See https://stackoverflow.com/questions/12594148/skipping-execution-of-with-block
621 return self
623 def should_run(self):
624 """
625 Return True if there are no exceptions, false otherwise
626 :return:
627 """
628 return self.exec_result is None
630 def __exit__(self, exc_type, exc_val, traceback):
631 # Store this step's execution result for later
632 if exc_type is None:
633 if self.exec_result is None: 633 ↛ 635, 633 ↛ 6382 missed branches: 1) line 633 didn't jump to line 635, because the condition on line 633 was never true, 2) line 633 didn't jump to line 638, because the condition on line 633 was never false
634 # Success !
635 self.exec_result = True
636 else:
637 # there was a dependency that had a failure, we can not forget it
638 pass
639 else:
640 # Failure: remember the exception
641 self.exec_result = OptionalStepException(exc_type, exc_val, traceback)
643 # We have to *cancel* the exception in the stack of the test function since it is a generator and we need to
644 # be able to execute subsequent steps
645 # see https://docs.python.org/3/reference/datamodel.html#object.__exit__
646 return True
648 def ran_with_success(self):
649 """
650 Return True if self.exec_result is True
651 :return:
652 """
653 return self.exec_result is True