Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _sync_run_all_hooks_and_example(self, context_data, around_functions=None):
"""
***********************************************************************
***********************************************************************
WARNING
***********************************************************************
***********************************************************************
This function **MUST** be keep the exact same execution flow of
_real_async_run_all_hooks_and_example()!!!
"""
if around_functions is None:
around_functions = list(reversed(self.example.context.all_around_functions))
if not around_functions:
aggregated_exceptions = AggregatedExceptions()
with aggregated_exceptions.catch():
for before_code in self.example.context.all_before_functions:
if hasattr(before_code, "_memoize_before_code"):
self.formatter.dsl_memoize_before(
self.example, before_code._memoize_before_code
)
else:
self.formatter.dsl_before(self.example, before_code)
self._fail_if_coroutine_function(before_code, context_data)
self.formatter.dsl_example(self.example, self.example.code)
self._fail_if_coroutine_function(self.example.code, context_data)
after_functions = []
after_functions.extend(context_data._mock_callable_after_functions)
after_functions.extend(self.example.context.all_after_functions)
after_functions.extend(context_data._after_functions)
for after_code in reversed(after_functions):
def append_exception(self, exception):
if isinstance(exception, AggregatedExceptions):
self.exceptions.extend(exception.exceptions)
else:
self.exceptions.append(exception)
def __init__(self):
super(AggregatedExceptions, self).__init__()
self.exceptions = []
def __init__(self):
super(_TestSlideTestResult, self).__init__()
self.aggregated_exceptions = AggregatedExceptions()
def _init_sub_example(self):
self._sub_examples_agg_ex = AggregatedExceptions()
def real_assert_sub_examples(self):
if self._sub_examples_agg_ex.exceptions:
self._sub_examples_agg_ex.raise_correct_exception()
if self._example.is_async:
async def assert_sub_examples(self):
real_assert_sub_examples(self)
else:
def assert_sub_examples(self):
real_assert_sub_examples(self)
self.after(assert_sub_examples)
def logger_warning(msg, *args, **kwargs):
if re.compile("^Executing .+ took .+ seconds$").match(str(msg)):
msg = (
f"{msg}\n"
"During the execution of the async test a slow callback "
"that blocked the event loop was detected.\n"
"Tip: you can customize the detection threshold with:\n"
" asyncio.get_running_loop().slow_callback_duration = seconds"
)
caught_failures.append(SlowCallback(msg % args))
else:
original_logger_warning(msg, *args, **kwargs)
asyncio.log.logger.warning = logger_warning
aggregated_exceptions = AggregatedExceptions()
try:
with aggregated_exceptions.catch():
yield
finally:
warnings.showwarning = original_showwarning
asyncio.log.logger.warning = original_logger_warning
for failure in caught_failures:
with aggregated_exceptions.catch():
raise failure
aggregated_exceptions.raise_correct_exception()