datastore/ci/jobs/scripts/functional_tests_results.py

import dataclasses
import re
import runpy
import signal
import traceback
from pathlib import Path
from typing import List, Optional

from praktika.result import Result

OK_SIGN = "[ OK "
FAIL_SIGN = "[ FAIL "
UNKNOWN_SIGN = "[ UNKNOWN "
SKIPPED_SIGN = "[ SKIPPED "
NOT_FAILED_SIGN = "[ NOT_FAILED "
HUNG_SIGN = "Found hung queries in processlist"
DATABASE_SIGN = "Database: "

# Pick up the runner exit codes straight from `tests/datastore-test` so
# the contract has a single source of truth.
_clickhouse_test = Path(__file__).resolve().parents[3] / "tests" / "datastore-test"
_clickhouse_test_globals = runpy.run_path(str(_clickhouse_test))
STOP_TESTING_EXIT_CODE = _clickhouse_test_globals["STOP_TESTING_EXIT_CODE"]
GLOBAL_TIME_LIMIT_EXIT_CODE = _clickhouse_test_globals["GLOBAL_TIME_LIMIT_EXIT_CODE"]
MAX_FAILURES_EXIT_CODE = _clickhouse_test_globals["MAX_FAILURES_EXIT_CODE"]

# Exit codes that mean the run was aborted mid-flight, so per-test results
# (if any) are incomplete and we cannot trust which test "caused" the
# failure. `STOP_TESTING_EXIT_CODE` is the in-band signal — the parent
# raised `StopTesting` and reached the outer handler. The kill-by-signal
# variants cover the out-of-band cases where the parent was killed before
# it could exit through that handler (currently reachable via the
# worker -> parent SIGTERM feedback loop in `stop_tests`: each worker the
# parent terminates re-broadcasts SIGTERM to the whole process group via
# `killpg`, hitting the parent before it can `sys.exit(STOP_TESTING_EXIT_CODE)`;
# also covers external kills like job-level timeouts and runner shutdown).
#
# Both `128 + N` (bash's convention when its child died from signal N) and
# the negative form `-N` are included: `Shell.run` wraps the command in
# `bash -c`, so most kills surface as `128 + N` via bash's exit status, but
# `Shell._check_timeout` calls `os.killpg` on the whole group, so the
# wrapper bash can itself die from the signal — and Python's
# `subprocess.Popen.returncode` reports that as `-N`, not `128 + N`.
#
# Exit code 1 is deliberately NOT in this set: it is set by end-of-run
# checks (final hung-check, `runner_process_killed`, `total_tests_run == 0`)
# that run AFTER all tests have finished. Per-test results in that case are
# complete and authoritative and must not be demoted.
# `MAX_FAILURES_EXIT_CODE` is likewise excluded: the run stopped early because
# too many tests failed, but the server is alive and those failures are real
# and already attributed - so they must be reported as FAIL, not "Server died".
ABORTED_RUN_EXIT_CODES = frozenset(
    {
        STOP_TESTING_EXIT_CODE,
        128 + signal.SIGTERM,  # 143
        128 + signal.SIGKILL,  # 137
        -signal.SIGTERM,  # -15
        -signal.SIGKILL,  # -9
    }
)

SUCCESS_FINISH_SIGNS = ["All tests have finished", "No tests were run"]

RETRIES_SIGN = "Some tests were restarted"

# Regex pattern to match test result lines.
# The shape `name: [ STATUS ] N.NN sec.` is specific enough that we don't pin
# the leading timestamp or the counter - the bounded `^.{0,36}?` lets through
# any expected framing (raw=0, `ts`=20, `[YYYY-MM-DD HH:MM:SS] `=22) but rules
# out matches embedded deeper in an error/exception message (see PR #88825).
# Test names can contain letters, digits, underscores, hyphens, and dots.
TEST_RESULT_PATTERN = re.compile(
    r"^.{0,36}?"
    r"([\w\-\.]+):\s+(\[ (?:OK|FAIL|SKIPPED|UNKNOWN|NOT_FAILED) \])\s+([\d.]+) sec\."
)


class FTResultsProcessor:
    @dataclasses.dataclass
    class Summary:
        total: int
        skipped: int
        unknown: int
        failed: int
        success: int
        test_results: List[Result]
        hung: bool = False
        retries: bool = False
        success_finish: bool = False
        test_end: bool = True

    def __init__(self, wd):
        self.tests_output_file = f"{wd}/test_result.txt"
        self.debug_files = []

    def _process_test_output(self):
        total = 0
        skipped = 0
        unknown = 0
        failed = 0
        success = 0
        hung = False
        retries = False
        success_finish = False
        test_results = []
        test_end = True

        with open(self.tests_output_file, "r", encoding="utf-8") as test_file:
            for line in test_file:
                original_line = line
                line = line.strip()

                if any(s in line for s in SUCCESS_FINISH_SIGNS):
                    success_finish = True
                # Ignore hung check report, since it may be quite large.
                # (and may break python parser which has limit of 128KiB for each row).
                if HUNG_SIGN in line:
                    hung = True
                    break
                if RETRIES_SIGN in line:
                    retries = True

                # Use regex to match test result lines precisely
                # This prevents false matches from error messages containing test patterns
                match = TEST_RESULT_PATTERN.match(line)
                if match:
                    test_name = match.group(1)
                    status_marker = match.group(2)
                    test_time = match.group(3)

                    if test_name == "+":
                        # TODO: investigate and remove
                        # https://github.com/ClickHouse/ClickHouse/issues/81888
                        print(
                            f"ERROR: incorrect test name: {test_name} in line:\n{line}"
                        )
                        continue

                    total += 1
                    if FAIL_SIGN in status_marker:
                        failed += 1
                        test_results.append((test_name, "FAIL", test_time, []))
                    elif UNKNOWN_SIGN in status_marker:
                        unknown += 1
                        test_results.append((test_name, "FAIL", test_time, []))
                    elif NOT_FAILED_SIGN in status_marker:
                        # Test was on a blacklist (expected to fail) but passed -
                        # the blacklist needs updating. Surface as a failure.
                        failed += 1
                        test_results.append((test_name, "NOT_FAILED", test_time, []))
                    elif SKIPPED_SIGN in status_marker:
                        skipped += 1
                        test_results.append((test_name, "SKIPPED", test_time, []))
                    else:
                        success += int(OK_SIGN in status_marker)
                        test_results.append((test_name, "OK", test_time, []))
                    test_end = False
                elif (
                    len(test_results) > 0
                    and test_results[-1][1] in ("FAIL", "SKIPPED", "NOT_FAILED")
                    and not test_end
                ):
                    test_results[-1][3].append(original_line)
                # Database printed after everything else in case of failures,
                # so this is a stop marker for capturing test output.
                #
                # And it is handled after everything else to include line with database into the report.
                if DATABASE_SIGN in line:
                    test_end = True

        test_results_ = []
        for test in test_results:
            try:
                test_results_.append(
                    Result(
                        name=test[0],
                        status=test[1],
                        start_time=None,
                        duration=float(test[2]),
                        info="".join(test[3])[:16384],
                    )
                )
            except Exception:
                print(f"ERROR: Failed to parse test results: {test}")
                traceback.print_exc()
                self.debug_files.append(self.tests_output_file)
                if test[0] == "+":
                    # TODO: investigate and remove
                    # https://github.com/ClickHouse/ClickHouse/issues/81888
                    continue
                test_results_.append(
                    Result(
                        name=test[0],
                        status=Result.Status.ERROR,
                        start_time=None,
                        duration=None,
                        info=f"test results parse failure:\n{traceback.print_exc()}",
                    )
                )
        test_results = test_results_

        s = self.Summary(
            total=total,
            skipped=skipped,
            unknown=unknown,
            failed=failed,
            success=success,
            test_results=test_results,
            hung=hung,
            success_finish=success_finish,
            retries=retries,
        )

        return s

    def run(self, task_name="Tests", runner_exit_code: Optional[int] = None):
        state = Result.Status.OK
        s = self._process_test_output()
        test_results = s.test_results

        if s.failed != 0 or s.unknown != 0:
            state = Result.Status.FAIL

        info = ""
        if s.hung:
            state = Result.Status.FAIL
            test_results.append(
                Result("Some queries hung", Result.Status.FAIL, info="Some queries hung")
            )
        elif runner_exit_code in ABORTED_RUN_EXIT_CODES:
            state = Result.Status.FAIL
            failed_results = [r for r in test_results if r.is_failure()]
            if len(failed_results) > 1:
                # Multiple tests failed when the server died - this is a parallel
                # run where we can't tell which test (if any) caused the crash.
                # Mark them all as UNKNOWN so they don't pollute failure reports.
                # The actual failure is captured by the "Server died" / LOGICAL_ERROR
                # entry added from the server log.
                for result in failed_results:
                    result.status = Result.Status.UNKNOWN
            elif len(failed_results) == 1:
                # Single test failed - sequential run, this test is the culprit.
                failed_results[0].status = Result.Status.ERROR
            test_results.append(Result("Server died", Result.Status.FAIL, info="Server died"))
        elif runner_exit_code == MAX_FAILURES_EXIT_CODE:
            # The run stopped early because too many tests failed
            # (`--max-failures` / `--max-failures-chain`). Unlike the aborted-run
            # branch above, the server is alive and the parsed failures are real
            # and correctly attributed - so leave them as FAIL (do not demote to
            # UNKNOWN, do not synthesize a "Server died" leaf) and just add an
            # informational summary. `state` stays FAIL from the parsed failures.
            state = Result.Status.FAIL
            test_results.append(
                Result(
                    "Too many test failures",
                    Result.Status.FAIL,
                    info="Stopped early after reaching the --max-failures limit. The failing tests above are the real failures.",
                )
            )
        elif runner_exit_code == GLOBAL_TIME_LIMIT_EXIT_CODE:
            # The run stopped gracefully because the global time limit was
            # reached. This is the *expected* stop condition for the flaky and
            # targeted checks, which rerun the selected tests until the time
            # budget is exhausted - not a failure. Real test failures and hung
            # queries are still reported through the branches above (they set
            # `state` to FAIL regardless), so here we only add an informational
            # leaf and leave `state` as computed from the parsed results.
            test_results.append(
                Result(
                    "Global time limit reached",
                    Result.Status.OK,
                    info="Stopped after reaching the time budget - the expected stop condition for this check.",
                )
            )
        elif not s.success_finish:
            state = Result.Status.ERROR
            info = "The test runner was terminated unexpectedly"
        elif s.retries:
            test_results.append(
                Result("Some tests restarted", Result.Status.SKIPPED, info="Some tests restarted")
            )
        else:
            pass

        # The runner's exit code is the authoritative signal: if `datastore-test`
        # exited non-zero, the job must not report OK even when log parsing finds
        # nothing to blame. The synthetic leaf is added only when the parser
        # found nothing - otherwise the real failure already explains the result
        # and a duplicate entry is just noise.
        # `GLOBAL_TIME_LIMIT_EXIT_CODE` is excluded: it is a graceful, expected
        # stop (handled above), not a failure, even though it is non-zero.
        if runner_exit_code not in (None, 0, GLOBAL_TIME_LIMIT_EXIT_CODE):
            if state == Result.Status.OK:
                state = Result.Status.FAIL
                test_results.append(
                    Result(
                        name="datastore-test",
                        status=Result.Status.FAIL,
                        info=f"datastore-test exited with code {runner_exit_code}",
                    )
                )

        if not info:
            info = f"Failed: {s.failed}, Passed: {s.success}, Skipped: {s.skipped}"

        result = Result.create_from(
            name=task_name,
            results=test_results,
            status=state,
            files=[],
            info=info,
            with_info_from_results=False,
        )

        if not result.is_ok():
            order = {
                "FAIL": 0,
                "SERVER_DIED": 1,
                "Timeout": 2,
                "NOT_FAILED": 3,
                "BROKEN": 4,
                "UNKNOWN": 5,
                "OK": 6,
                "SKIPPED": 7,
            }
            result.results.sort(key=lambda x: order.get(x.status, -1))

        return result