How to use the testslide.__init__.AggregatedExceptions function in TestSlide

To help you get started, we’ve selected a few TestSlide examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github facebookincubator / TestSlide / testslide / __init__.py View on Github external
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):
github facebookincubator / TestSlide / testslide / __init__.py View on Github external
def append_exception(self, exception):
        if isinstance(exception, AggregatedExceptions):
            self.exceptions.extend(exception.exceptions)
        else:
            self.exceptions.append(exception)
github facebookincubator / TestSlide / testslide / __init__.py View on Github external
def __init__(self):
        super(AggregatedExceptions, self).__init__()
        self.exceptions = []
github facebookincubator / TestSlide / testslide / __init__.py View on Github external
def __init__(self):
        super(_TestSlideTestResult, self).__init__()
        self.aggregated_exceptions = AggregatedExceptions()
github facebookincubator / TestSlide / testslide / __init__.py View on Github external
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)
github facebookincubator / TestSlide / testslide / __init__.py View on Github external
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()