misc.python.materialize.mzcompose.test_result

  1# Copyright Materialize, Inc. and contributors. All rights reserved.
  2#
  3# Use of this software is governed by the Business Source License
  4# included in the LICENSE file at the root of this repository.
  5#
  6# As of the Change Date specified in that file, in accordance with
  7# the Business Source License, use of this software will be governed
  8# by the Apache License, Version 2.0.
  9
 10from __future__ import annotations
 11
 12import re
 13from dataclasses import dataclass
 14
 15from materialize import MZ_ROOT
 16from materialize.ui import CommandFailureCausedUIError, UIError
 17from materialize.util import filter_cmd
 18
 19PEM_CONTENT_RE = r"-----BEGIN ([A-Z ]+)-----[^-]+-----END [A-Z ]+-----"
 20PEM_CONTENT_REPLACEMENT = r"<\1>"
 21
 22
 23@dataclass
 24class TestResult:
 25    __test__ = False
 26
 27    duration: float
 28    errors: list[TestFailureDetails]
 29
 30    def is_successful(self) -> bool:
 31        return len(self.errors) == 0
 32
 33
 34@dataclass
 35class TestFailureDetails:
 36    __test__ = False
 37
 38    message: str
 39    details: str | None
 40    additional_details_header: str | None = None
 41    additional_details: str | None = None
 42    test_class_name_override: str | None = None
 43    """The test class usually describes the framework."""
 44    test_case_name_override: str | None = None
 45    """The test case usually describes the workflow, unless more fine-grained information is available."""
 46    location: str | None = None
 47    """depending on the check, this may either be a file name or a path"""
 48    line_number: int | None = None
 49
 50    def location_as_file_name(self) -> str | None:
 51        if self.location is None:
 52            return None
 53
 54        if "/" in self.location:
 55            return self.location[self.location.rindex("/") + 1 :]
 56
 57        return self.location
 58
 59
 60class FailedTestExecutionError(UIError):
 61    """
 62    An UIError that is caused by a failing test.
 63    """
 64
 65    def __init__(
 66        self,
 67        errors: list[TestFailureDetails],
 68        error_summary: str = "At least one test failed",
 69    ):
 70        super().__init__(error_summary)
 71        self.errors = errors
 72
 73
 74def try_determine_errors_from_cmd_execution(
 75    e: CommandFailureCausedUIError, test_context: str | None
 76) -> list[TestFailureDetails]:
 77    # Combine both streams: testdrive prints each marked error to stdout and
 78    # the final error report to stderr, so either stream alone is incomplete.
 79    output_parts = [s for s in [e.stdout, e.stderr] if s]
 80    output = "\n".join(output_parts) if output_parts else None
 81
 82    if "running docker compose failed" in str(e):
 83        return [determine_error_from_docker_compose_failure(e, output, test_context)]
 84
 85    if output is None:
 86        return []
 87
 88    error_chunks = extract_error_chunks_from_output(output)
 89
 90    fallback_file_path = try_determine_error_location_from_cmd(e.cmd)
 91    if fallback_file_path is not None and ":" in fallback_file_path:
 92        parts = fallback_file_path.split(":")
 93        fallback_file_path, fallback_line_number = parts[0], int(parts[1])
 94    else:
 95        fallback_line_number = None
 96
 97    collected_errors = []
 98    for chunk in error_chunks:
 99        match = re.search(r"([^.]+\.td):(\d+):\d+:", chunk)
100        if match is not None:
101            # for .td files like Postgres CDC, file_path will just contain the file name
102            file_path = match.group(1)
103            line_number = int(match.group(2))
104        else:
105            # for .py files like platform checks, file_path will be a path
106            file_path = fallback_file_path
107            line_number = fallback_line_number
108
109        message = (
110            f"Executing {file_path if file_path is not None else 'command'} failed!"
111        )
112
113        failure_details = TestFailureDetails(
114            message,
115            details=chunk,
116            test_case_name_override=test_context,
117            location=file_path,
118            line_number=line_number,
119        )
120
121        if failure_details not in collected_errors:
122            collected_errors.append(failure_details)
123
124    return collected_errors
125
126
127def determine_error_from_docker_compose_failure(
128    e: CommandFailureCausedUIError, output: str | None, test_context: str | None
129) -> TestFailureDetails:
130    command = to_sanitized_command_str(e.cmd)
131    context_prefix = f"{test_context}: " if test_context is not None else ""
132    return TestFailureDetails(
133        f"{context_prefix}Docker compose failed: {command}",
134        details=output,
135        test_case_name_override=test_context,
136        location=None,
137        line_number=None,
138    )
139
140
141def try_determine_error_location_from_cmd(cmd: list[str]) -> str | None:
142    root_path_as_string = f"{MZ_ROOT}/"
143    for cmd_part in cmd:
144        if type(cmd_part) == str and cmd_part.startswith("--source="):
145            return cmd_part.removeprefix("--source=").replace(root_path_as_string, "")
146
147    return None
148
149
150def extract_error_chunks_from_output(output: str) -> list[str]:
151    pos = output.find("+++ !!! Error Report")
152    if pos == -1:
153        return []
154
155    error_output = output[:pos]
156    # Testdrive prints "^^^ +++" *before* each error, so everything up to the
157    # first marker is regular output, not an error.
158    error_chunks = error_output.split("^^^ +++")[1:]
159
160    return [chunk.strip() for chunk in error_chunks if len(chunk.strip()) > 0]
161
162
163def to_sanitized_command_str(cmd: list[str]) -> str:
164    # Ensure all elements are strings (cmd may contain Path objects)
165    str_cmd = [str(x) for x in cmd]
166    command_str = " ".join(filter_cmd(str_cmd))
167    return re.sub(PEM_CONTENT_RE, PEM_CONTENT_REPLACEMENT, command_str)