1from __future__ import annotations
2
3import json
4import os
5from pathlib import Path
6from typing import Any
7
8from tools.stats.import_test_stats import (
9    ADDITIONAL_CI_FILES_FOLDER,
10    TD_HEURISTIC_PREVIOUSLY_FAILED,
11    TD_HEURISTIC_PREVIOUSLY_FAILED_ADDITIONAL,
12)
13from tools.testing.target_determination.heuristics.interface import (
14    HeuristicInterface,
15    TestPrioritizations,
16)
17from tools.testing.target_determination.heuristics.utils import (
18    python_test_file_to_test_name,
19)
20from tools.testing.test_run import TestRun
21
22
23REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent.parent
24
25
26class PreviouslyFailedInPR(HeuristicInterface):
27    def __init__(self, **kwargs: dict[str, Any]) -> None:
28        super().__init__(**kwargs)
29
30    def get_prediction_confidence(self, tests: list[str]) -> TestPrioritizations:
31        critical_tests = get_previous_failures() | read_additional_test_failures_file()
32        return TestPrioritizations(
33            tests, {TestRun(test): 1 for test in critical_tests if test in tests}
34        )
35
36
37def get_previous_failures() -> set[str]:
38    path = REPO_ROOT / ADDITIONAL_CI_FILES_FOLDER / TD_HEURISTIC_PREVIOUSLY_FAILED
39    if not os.path.exists(path):
40        print(f"could not find path {path}")
41        return set()
42    with open(path) as f:
43        return python_test_file_to_test_name(
44            _parse_prev_failing_test_files(json.load(f))
45        )
46
47
48def _parse_prev_failing_test_files(last_failed_tests: dict[str, bool]) -> set[str]:
49    prioritized_tests = set()
50
51    # The keys are formatted as "test_file.py::test_class::test_method[params]"
52    # We just need the test_file part
53    for test in last_failed_tests:
54        parts = test.split("::")
55        if len(parts) > 1:
56            test_file = parts[0]
57            prioritized_tests.add(test_file)
58
59    return prioritized_tests
60
61
62def gen_additional_test_failures_file(tests: list[str]) -> None:
63    # Segfaults usually result in no xml and some tests don't run through pytest
64    # (ex doctests).  In these cases, there will be no entry in the pytest
65    # cache, so we should generate a separate file for them and upload it to s3
66    # along with the pytest cache
67    pytest_cache_dir = REPO_ROOT / ".pytest_cache"
68    if not os.path.exists(pytest_cache_dir):
69        os.makedirs(pytest_cache_dir)
70    with open(pytest_cache_dir / TD_HEURISTIC_PREVIOUSLY_FAILED_ADDITIONAL, "w") as f:
71        json.dump(tests, f, indent=2)
72
73
74def read_additional_test_failures_file() -> set[str]:
75    path = (
76        REPO_ROOT
77        / ADDITIONAL_CI_FILES_FOLDER
78        / TD_HEURISTIC_PREVIOUSLY_FAILED_ADDITIONAL
79    )
80    if not os.path.exists(path):
81        print(f"could not find path {path}")
82        return set()
83    with open(path) as f:
84        s = set(json.load(f))
85        print(f"additional failures: {s}")
86        return s
87