Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@wraps(fixture_func, new_sig=new_sig)
def wrapped_fixture_func(*args, **kwargs):
if not is_used_request(kwargs['request']):
yield NOT_USED
else:
args, kwargs = _get_arguments(*args, **kwargs)
for res in fixture_func(*args, **kwargs):
yield res
@wraps(fixture_fun, new_sig=new_sig)
def _steps_aware_decorated_function(*args, **kwargs):
request = kwargs['request'] if func_needs_request else kwargs.pop('request')
id_without_steps = _init_and_check(request)
try:
# already available: this is a subsequent step.
yield ref_dct[id_without_steps]
except KeyError:
# not yet cached, this is probably the first step
gen = fixture_fun(*args, **kwargs)
res = next(gen)
ref_dct[id_without_steps] = res
yield res
# TODO this teardown hook should actually be executed after all steps...
next(gen)
@wraps(test_func, new_sig=new_sig)
def wrapped_test_func(*args, **kwargs): # noqa
if kwargs.get(fixture_union_name, None) is NOT_USED:
# TODO why this ? it is probably useless: this fixture
# is private and will never end up in another union
return NOT_USED
else:
replace_paramfixture_with_values(kwargs)
return test_func(*args, **kwargs)
@wraps(fixture_func, new_sig=new_sig)
def wrapped_fixture_func(*_args, **_kwargs):
if not is_used_request(_kwargs['request']):
yield NOT_USED
else:
_args, _kwargs = _map_arguments(*_args, **_kwargs)
for res in fixture_func(*_args, **_kwargs):
yield res
@wraps(fixture_func, new_sig=new_sig)
def wrapped_fixture_func(*args, **kwargs):
request = kwargs['request'] if func_needs_request else kwargs.pop('request')
if is_used_request(request):
return fixture_func(*args, **kwargs)
else:
return NOT_USED
@wraps(test_func, new_sig=new_sig)
def wrapped_test_function(*args, **kwargs):
"""Executes the current step only if its dependencies are correct, and registers its execution result"""
request = kwargs['request'] if func_needs_request else kwargs.pop('request')
if request is None:
# manual call (maybe for pre-loading?), no dependency management, ability to execute several steps
_execute_manually(test_func, s, test_step_argname, step_ids, steps, args, kwargs)
else:
# (a) retrieve the "current step" function
current_step_fun = get_fixture_or_param_value(request, test_step_argname)
# Get the unique id that is shared between the steps of the same execution
# Note: when the id was using not only param values but also fixture values we had to discard
# steps_data_holder_name and 'request'. But that's not the case anymore, simply discard "test step"
test_id_without_steps = get_pytest_node_hash_id(request.node, params_to_ignore={test_step_argname})
# Make sure that it has a field to store its execution success
@wraps(fixture_fun, new_sig=new_sig)
def _steps_aware_decorated_function(*args, **kwargs):
request = kwargs['request'] if func_needs_request else kwargs.pop('request')
_check_scope(request)
gen = fixture_fun(*args, **kwargs)
res = next(gen)
yield _OnePerStepFixtureProxy(res)
next(gen)
@wraps(fixture_func, new_sig=new_sig)
def wrapped_fixture_func(*_args, **_kwargs):
if not is_used_request(_kwargs['request']):
return NOT_USED
else:
_args, _kwargs = _map_arguments(*_args, **_kwargs)
return fixture_func(*_args, **_kwargs)
@makefun.wraps(func)
def validated_func(*args, **kwargs):
"""
Args:
contains an argument named 't' for failure interval
"""
args_dict = _get_args_dict(func, args, kwargs)
if args_dict["t"] <= 0:
raise InvalidInterval("Failure interval must be greater than 0!")
return func(*args, **kwargs)