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    output = e.stderr or e.stdout
 78
 79    if "running docker compose failed" in str(e):
 80        return [determine_error_from_docker_compose_failure(e, output, test_context)]
 81
 82    if output is None:
 83        return []
 84
 85    error_chunks = extract_error_chunks_from_output(output)
 86
 87    fallback_file_path = try_determine_error_location_from_cmd(e.cmd)
 88    if fallback_file_path is not None and ":" in fallback_file_path:
 89        parts = fallback_file_path.split(":")
 90        fallback_file_path, fallback_line_number = parts[0], int(parts[1])
 91    else:
 92        fallback_line_number = None
 93
 94    collected_errors = []
 95    for chunk in error_chunks:
 96        match = re.search(r"([^.]+\.td):(\d+):\d+:", chunk)
 97        if match is not None:
 98            # for .td files like Postgres CDC, file_path will just contain the file name
 99            file_path = match.group(1)
100            line_number = int(match.group(2))
101        else:
102            # for .py files like platform checks, file_path will be a path
103            file_path = fallback_file_path
104            line_number = fallback_line_number
105
106        message = (
107            f"Executing {file_path if file_path is not None else 'command'} failed!"
108        )
109
110        failure_details = TestFailureDetails(
111            message,
112            details=chunk,
113            test_case_name_override=test_context,
114            location=file_path,
115            line_number=line_number,
116        )
117
118        if failure_details not in collected_errors:
119            collected_errors.append(failure_details)
120
121    return collected_errors
122
123
124def determine_error_from_docker_compose_failure(
125    e: CommandFailureCausedUIError, output: str | None, test_context: str | None
126) -> TestFailureDetails:
127    command = to_sanitized_command_str(e.cmd)
128    context_prefix = f"{test_context}: " if test_context is not None else ""
129    return TestFailureDetails(
130        f"{context_prefix}Docker compose failed: {command}",
131        details=output,
132        test_case_name_override=test_context,
133        location=None,
134        line_number=None,
135    )
136
137
138def try_determine_error_location_from_cmd(cmd: list[str]) -> str | None:
139    root_path_as_string = f"{MZ_ROOT}/"
140    for cmd_part in cmd:
141        if type(cmd_part) == str and cmd_part.startswith("--source="):
142            return cmd_part.removeprefix("--source=").replace(root_path_as_string, "")
143
144    return None
145
146
147def extract_error_chunks_from_output(output: str) -> list[str]:
148    pos = output.find("+++ !!! Error Report")
149    if pos == -1:
150        return []
151
152    error_output = output[: pos - 1]
153    error_chunks = error_output.split("^^^ +++")
154
155    return [chunk.strip() for chunk in error_chunks if len(chunk.strip()) > 0]
156
157
158def to_sanitized_command_str(cmd: list[str]) -> str:
159    # Ensure all elements are strings (cmd may contain Path objects)
160    str_cmd = [str(x) for x in cmd]
161    command_str = " ".join(filter_cmd(str_cmd))
162    return re.sub(PEM_CONTENT_RE, PEM_CONTENT_REPLACEMENT, command_str)