sqlmesh.core.test.result

  1from __future__ import annotations
  2
  3import types
  4import typing as t
  5import unittest
  6
  7from sqlmesh.core.test.definition import ModelTest
  8
  9if t.TYPE_CHECKING:
 10    ErrorType = t.Union[
 11        t.Tuple[type[BaseException], BaseException, types.TracebackType],
 12        t.Tuple[None, None, None],
 13    ]
 14
 15
 16class ModelTextTestResult(unittest.TextTestResult):
 17    successes: t.List[unittest.TestCase]
 18
 19    def __init__(self, *args: t.Any, **kwargs: t.Any):
 20        self.console = kwargs.pop("console", None)
 21        super().__init__(*args, **kwargs)
 22        self.successes = []
 23        self.original_failures: t.List[t.Tuple[unittest.TestCase, ErrorType]] = []
 24        self.failure_tables: t.List[t.Tuple[t.Any, ...]] = []
 25        self.original_errors: t.List[t.Tuple[unittest.TestCase, ErrorType]] = []
 26        self.duration: t.Optional[float] = None
 27
 28    def addSubTest(
 29        self,
 30        test: unittest.TestCase,
 31        subtest: unittest.TestCase,
 32        err: t.Optional[ErrorType],
 33    ) -> None:
 34        """Called at the end of a subtest.
 35
 36        The traceback is suppressed because it is redundant and not useful.
 37
 38        Args:
 39            test: The test case.
 40            subtest: The subtest instance.
 41            err: A tuple of the form returned by sys.exc_info(), i.e., (type, value, traceback).
 42        """
 43        if err:
 44            exctype, value, tb = err
 45            err = (exctype, value, None)  # type: ignore
 46
 47            if err[0] and issubclass(err[0], test.failureException):
 48                self.addFailure(test, err)
 49            else:
 50                self.addError(test, err)
 51
 52    def _print_char(self, char: str) -> None:
 53        from sqlmesh.core.console import TerminalConsole
 54
 55        if isinstance(self.console, TerminalConsole):
 56            self.console._print(char, end="")
 57
 58    def addFailure(self, test: unittest.TestCase, err: ErrorType) -> None:
 59        """Called when the test case test signals a failure.
 60
 61        The traceback is suppressed because it is redundant and not useful.
 62
 63        Args:
 64            test: The test case.
 65            err: A tuple of the form returned by sys.exc_info(), i.e., (type, value, traceback).
 66        """
 67        exctype, value, _ = err
 68
 69        if value and value.args:
 70            exception_msg, rich_tables = value.args[:1], value.args[1:]
 71            value.args = exception_msg
 72
 73            if rich_tables:
 74                self.failure_tables.append(rich_tables)
 75
 76        self._print_char("F")
 77
 78        self.original_failures.append((test, err))
 79
 80        # Intentionally ignore the traceback to hide it from the user
 81        return super().addFailure(test, (exctype, value, None))  # type: ignore
 82
 83    def addError(self, test: unittest.TestCase, err: ErrorType) -> None:
 84        """Called when the test case test signals an error.
 85
 86        Args:
 87            test: The test case.
 88            err: A tuple of the form returned by sys.exc_info(), i.e., (type, value, traceback).
 89        """
 90        exctype, value, _ = err
 91        self.original_errors.append((test, err))
 92
 93        self._print_char("E")
 94
 95        # Intentionally ignore the traceback to hide it from the user
 96        return super().addError(test, (exctype, value, None))  # type: ignore
 97
 98    def addSuccess(self, test: unittest.TestCase) -> None:
 99        """Called when the test case test succeeds.
100
101        Args:
102            test: The test case
103        """
104        super().addSuccess(test)
105
106        self._print_char(".")
107
108        self.successes.append(test)
109
110    def merge(self, other: ModelTextTestResult) -> None:
111        if other.successes:
112            self.addSuccess(other.successes[0])
113        elif other.errors:
114            for error_test, error in other.original_errors:
115                self.addError(error_test, error)
116        elif other.failures:
117            for failure_test, failure in other.original_failures:
118                self.addFailure(failure_test, failure)
119
120            self.failure_tables.extend(other.failure_tables)
121        elif other.skipped:
122            skipped_args = other.skipped[0]
123            self.addSkip(skipped_args[0], skipped_args[1])
124
125        self.testsRun += other.testsRun
126
127    def get_fail_and_error_tests(self) -> t.List[ModelTest]:
128        # If tests contain failed subtests (e.g testing CTE outputs) we don't want
129        # to report it as different test failures
130        test_name_to_test = {
131            test.test_name: test
132            for test, _ in self.failures + self.errors
133            if isinstance(test, ModelTest)
134        }
135        return list(test_name_to_test.values())