Quelle  high_freq_skipfails.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import datetime
import logging
import os
import sys
from pathlib import Path
from typing import Optional, TypedDict

import requests
from intermittent_failures import IntermittentFailuresFetcher
from mozci.util.taskcluster import get_task
from mozinfo.platforminfo import PlatformInfo
from skipfails import Skipfails

ERROR = "error"
USER_AGENT = "mach-manifest-high-freq-skipfails/1.0"


class FailureByBug(TypedDict):
    task_id: str
    bug_id: int
    job_id: int
    tree: str


class BugSuggestion(TypedDict):
    path_end: Optional[str]


class TestInfoAllTestsItem(TypedDict):
    manifest: list[str]
    test: str


class TestInfoAllTests(TypedDict):
    tests: dict[str, list[TestInfoAllTestsItem]]


class HighFreqSkipfails:
    "mach manifest high-freq-skip-fails implementation: Update manifests to skip failing tests by looking at recent failures"

    def __init__(self, command_context=None, failures: int = 30, days: int = 7) -> None:
        self.command_context = command_context
        if self.command_context is not None:
            self.topsrcdir = self.command_context.topsrcdir
        else:
            self.topsrcdir = Path(__file__).parent.parent
        self.topsrcdir = os.path.normpath(self.topsrcdir)
        self.component = "high-freq-skip-fails"

        self.failures = failures
        self.days = days

        self.fetcher = IntermittentFailuresFetcher(
            days=days, threshold=failures, verbose=False
        )

        self.start_date = datetime.datetime.now()
        self.start_date = self.start_date - datetime.timedelta(days=self.days)
        self.end_date = datetime.datetime.now()
        self.test_info_all_tests: Optional[TestInfoAllTests] = None

    def error(self, e):
        if self.command_context is not None:
            self.command_context.log(
                logging.ERROR, self.component, {ERROR: str(e)}, "ERROR: {error}"
            )
        else:
            print(f"ERROR: {e}", file=sys.stderr, flush=True)

    def info(self, e):
        if self.command_context is not None:
            self.command_context.log(
                logging.INFO, self.component, {ERROR: str(e)}, "INFO: {error}"
            )
        else:
            print(f"INFO: {e}", file=sys.stderr, flush=True)

    def run(self):
        self.info(
            f"Fetching bugs with failure count above {self.failures} in the last {self.days} days..."
        )
        bug_list = self.fetcher.get_single_tracking_bugs_with_paths()
        if len(bug_list) == 0:
            self.info(
                f"Could not find bugs wih at least {self.failures} failures in the last {self.days}"
            )
            return
        self.info(f"Found {len(bug_list)} bugs to inspect")

        self.info("Fetching test_info_all_tests and caching it...")
        self.test_info_all_tests = self.get_test_info_all_tests()

        manifest_errors: set[tuple[int, str]] = set()

        task_data: dict[str, tuple[int, str, str]] = {}
        for bug_id, test_path in bug_list:
            self.info(f"Getting failures for bug '{bug_id}'...")
            failures_by_bug = self.get_failures_by_bug(bug_id)
            self.info(f"Found {len(failures_by_bug)} failures")
            manifest = self.get_manifest_from_path(test_path)
            if manifest:
                self.info(f"Found manifest '{manifest}' for path '{test_path}'")
                for failure in failures_by_bug:
                    task_data[failure["task_id"]] = (bug_id, test_path, manifest)
            else:
                manifest_errors.add((bug_id, test_path))
                self.error(f"Could not find manifest for path '{test_path}'")

        skipfails = Skipfails(self.command_context, "", True, "disable", True)

        task_list = self.get_task_list([task_id for task_id in task_data])
        for task_id, task in task_list:
            test_setting = task.get("extra", {}).get("test-setting", {})
            if not test_setting:
                continue
            platform_info = PlatformInfo(test_setting)
            (bug_id, test_path, raw_manifest) = task_data[task_id]

            kind, manifest = skipfails.get_kind_manifest(raw_manifest)
            if kind is None or manifest is None:
                self.error(f"Could not resolve kind for manifest {raw_manifest}")
                continue
            skipfails.skip_failure(
                manifest,
                kind,
                test_path,
                task_id,
                platform_info,
                str(bug_id),
                high_freq=True,
            )

        if len(manifest_errors) > 0:
            self.info("\nExecution complete\n")
            self.info("Script encountered errors while fetching manifests:")
            for bug_id, test_path in manifest_errors:
                self.info(
                    f"Bug {bug_id}: Could not find manifest for path '{test_path}'"
                )

    def get_manifest_from_path(self, path: Optional[str]) -> Optional[str]:
        manifest: Optional[str] = None
        if path is not None and self.test_info_all_tests is not None:
            for test_list in self.test_info_all_tests["tests"].values():
                for test in test_list:
                    # FIXME
                    # in case of wpt, we have an incoming path that is a subset of the full test["test"], for example, path could be:
                    # /navigation-api/ordering-and-transition/location-href-canceled.html
                    # but full path as found in test_info_all_tests is:
                    # testing/web-platform/tests/navigation-api/ordering-and-transition/location-href-canceled.html
                    # unfortunately in this case manifest ends up being: /navigation-api/ordering-and-transition
                    if test["test"] == path:
                        manifest = test["manifest"][0]
                        break
                if manifest is not None:
                    break
        return manifest

    #################
    #   API Calls   #
    #################

    def get_failures_by_bug(self, bug: int, branch="trunk") -> list[FailureByBug]:
        url = f"https://treeherder.mozilla.org/api/failuresbybug/?startday={self.start_date.date()}&endday={self.end_date.date()}&tree={branch}&bug={bug}"
        response = requests.get(url, headers={"User-agent": USER_AGENT})
        json_data = response.json()
        return json_data

    def get_task_list(
        self, task_id_list: list[str], branch="trunk"
    ) -> list[tuple[str, object]]:
        retVal = []
        for tid in task_id_list:
            task = get_task(tid)
            retVal.append((tid, task))
        return retVal

    def get_test_info_all_tests(self) -> TestInfoAllTests:
        url = "https://firefox-ci-tc.services.mozilla.com/api/index/v1/task/gecko.v2.mozilla-central.latest.source.test-info-all/artifacts/public/test-info-all-tests.json"
        response = requests.get(url, headers={"User-agent": USER_AGENT})
        json_data = response.json()
        return json_data