# -*- coding: utf-8 -*-

import argparse

import base64

import csv

import io

import json

import os

import pickle

import re

import subprocess

from datetime import datetime

from logging import INFO, basicConfig, getLogger

from typing import cast

import dateutil.parser

import hglib

import matplotlib

import numpy as np

import requests

import shap

import tenacity

from dateutil.relativedelta import relativedelta

from libmozdata import vcs_map

from libmozdata.phabricator import PhabricatorAPI

from scipy.stats import spearmanr

from bugbug import db, repository, test_scheduling

from bugbug.model import Model, get_transformer_pipeline

from bugbug.models.regressor import RegressorModel

from bugbug.models.testfailure import TestFailureModel

from bugbug.utils import (

download_check_etag,

download_model,

get_secret,

to_array,

zstd_decompress,

)

basicConfig(level=INFO)

logger = getLogger(__name__)

URL = "https://community-tc.services.mozilla.com/api/index/v1/task/project.bugbug.train_{model_name}.latest/artifacts/public/{file_name}"

PAST_BUGS_BY_FUNCTION_URL = "https://community-tc.services.mozilla.com/api/index/v1/task/project.bugbug.past_bugs_by_unit.latest/artifacts/public/past_fixed_bugs_by_function.json.zst"

PHAB_PROD = "prod"

PHAB_DEV = "dev"

# ------------------------------------------------------------------------------

# Copied from https://github.com/mozilla-conduit/lando-api/blob/4b583f9d773dfc8c3e8c39e3d3b7385568d744df/landoapi/commit_message.py

SPECIFIER = r"(?:r|a|sr|rs|ui-r)[=?]"

R_SPECIFIER = r"\br[=?]"

R_SPECIFIER_RE = re.compile(R_SPECIFIER)

LIST = r"[;,\/\\]\s*"

# Note that we only allows a subset of legal IRC-nick characters.

# Specifically, we do not allow [ \ ] ^ ` { | }

IRC_NICK = r"[a-zA-Z0-9\-\_]+"

# fmt: off

REVIEWERS_RE = re.compile(

r"([\s\(\.\[;,])" # before "r" delimiter

+ r"(" + SPECIFIER + r")" # flag

+ r"(" # capture all reviewers

+ r"#?" # Optional "#" group reviewer prefix # noqa: E131

+ IRC_NICK # reviewer

+ r"!?" # Optional "!" blocking indicator

+ r"(?:" # additional reviewers

+ LIST # delimiter # noqa: E131

+ r"(?![a-z0-9\.\-]+[=?])" # don"t extend match into next flag

+ r"#?" # Optional "#" group reviewer prefix

+ IRC_NICK # reviewer

+ r"!?" # Optional "!" blocking indicator

+ r")*"

+ r")?"

)

# fmt: on

def replace_reviewers(commit_description, reviewers):

if not reviewers:

reviewers_str = ""

else:

reviewers_str = "r=" + ",".join(reviewers)

if commit_description == "":

return reviewers_str

commit_description = commit_description.splitlines()

commit_summary = commit_description.pop(0)

commit_description = "\n".join(commit_description)

if not R_SPECIFIER_RE.search(commit_summary):

commit_summary += " " + reviewers_str

else:

# replace the first r? with the reviewer list, and all subsequent

# occurrences with a marker to mark the blocks we need to remove

# later

d = {"first": True}

def replace_first_reviewer(matchobj):

if R_SPECIFIER_RE.match(matchobj.group(2)):

if d["first"]:

d["first"] = False

return matchobj.group(1) + reviewers_str

else:

return "\0"

else:

return matchobj.group(0)

commit_summary = re.sub(REVIEWERS_RE, replace_first_reviewer, commit_summary)

# remove marker values as well as leading separators. this allows us

# to remove runs of multiple reviewers and retain the trailing

# separator.

commit_summary = re.sub(LIST + "\0", "", commit_summary)

commit_summary = re.sub("\0", "", commit_summary)

if commit_description == "":

return commit_summary.strip()

else:

return commit_summary.strip() + "\n" + commit_description

# ------------------------------------------------------------------------------

class CommitClassifier(object):

def __init__(

self,

model_name: str,

repo_dir: str,

git_repo_dir: str,

method_defect_predictor_dir: str,

use_single_process: bool,

skip_feature_importance: bool,

phabricator_deployment: str | None = None,

diff_id: int | None = None,

):

self.model_name = model_name

self.repo_dir = repo_dir

self.model = Model.load(download_model(model_name))

assert self.model is not None

self.git_repo_dir = git_repo_dir

if git_repo_dir:

self.clone_git_repo(

"hg::https://hg.mozilla.org/mozilla-unified", git_repo_dir

)

self.revision = None

if diff_id is not None:

assert phabricator_deployment is not None

with hglib.open(self.repo_dir) as hg:

self.apply_phab(hg, phabricator_deployment, diff_id)

self.revision = hg.log(revrange="not public()")[0].node.decode("utf-8")

self.method_defect_predictor_dir = method_defect_predictor_dir

if method_defect_predictor_dir:

self.clone_git_repo(

"https://github.com/lucapascarella/MethodDefectPredictor",

method_defect_predictor_dir,

"8cc47f47ffb686a29324435a0151b5fabd37f865",

)

self.use_single_process = use_single_process

self.skip_feature_importance = skip_feature_importance

if model_name == "regressor":

self.use_test_history = False

model_data_X_path = f"{model_name}model_data_X"

updated = download_check_etag(

URL.format(model_name=model_name, file_name=f"{model_data_X_path}.zst")

)

if updated:

zstd_decompress(model_data_X_path)

assert os.path.exists(model_data_X_path), "Decompressed X dataset exists"

model_data_y_path = f"{model_name}model_data_y"

updated = download_check_etag(

URL.format(model_name=model_name, file_name=f"{model_data_y_path}.zst")

)

if updated:

zstd_decompress(model_data_y_path)

assert os.path.exists(model_data_y_path), "Decompressed y dataset exists"

with open(model_data_X_path, "rb") as fb:

self.X = to_array(pickle.load(fb))

with open(model_data_y_path, "rb") as fb:

self.y = to_array(pickle.load(fb))

past_bugs_by_function_path = "data/past_fixed_bugs_by_function.json"

download_check_etag(

PAST_BUGS_BY_FUNCTION_URL, path=f"{past_bugs_by_function_path}.zst"

)

zstd_decompress(past_bugs_by_function_path)

assert os.path.exists(past_bugs_by_function_path)

with open(past_bugs_by_function_path, "r") as f:

self.past_bugs_by_function = json.load(f)

if model_name == "testlabelselect":

self.use_test_history = True

assert db.download_support_file(

test_scheduling.TEST_LABEL_SCHEDULING_DB,

test_scheduling.PAST_FAILURES_LABEL_DB,

)

self.past_failures_data = test_scheduling.PastFailures("label", True)

self.testfailure_model = cast(

TestFailureModel, TestFailureModel.load(download_model("testfailure"))

)

assert self.testfailure_model is not None

def clone_git_repo(self, repo_url, repo_dir, rev="origin/branches/default/tip"):

logger.info("Cloning %s...", repo_url)

if not os.path.exists(repo_dir):

for attempt in tenacity.Retrying(

wait=tenacity.wait_exponential(multiplier=2, min=2),

stop=tenacity.stop_after_attempt(7),

):

with attempt:

subprocess.run(

["git", "clone", "--quiet", repo_url, repo_dir],

check=True,

)

for attempt in tenacity.Retrying(

wait=tenacity.wait_exponential(multiplier=2, min=2),

stop=tenacity.stop_after_attempt(7),

):

with attempt:

subprocess.run(

["git", "fetch"],

cwd=repo_dir,

capture_output=True,

check=True,

)

subprocess.run(

["git", "checkout", rev], cwd=repo_dir, capture_output=True, check=True

)

def update_commit_db(self):

repository.clone(

self.repo_dir, "https://hg.mozilla.org/mozilla-unified", update=True

)

assert db.download(repository.COMMITS_DB, support_files_too=True)

for commit in repository.get_commits():

pass

repository.download_commits(

self.repo_dir,

rev_start="children({})".format(commit["node"]),

use_single_process=self.use_single_process,

)

def has_revision(self, hg, revision):

if not revision:

return False

try:

hg.identify(revision)

return True

except hglib.error.CommandError:

return False

def apply_phab(self, hg, phabricator_deployment, diff_id):

if phabricator_deployment == PHAB_PROD:

api_key = get_secret("PHABRICATOR_TOKEN")

url = get_secret("PHABRICATOR_URL")

else:

api_key = get_secret("PHABRICATOR_DEV_TOKEN")

url = get_secret("PHABRICATOR_DEV_URL")

phabricator_api = PhabricatorAPI(api_key=api_key, url=url)

# Get the stack of patches

stack = phabricator_api.load_patches_stack(diff_id)

assert len(stack) > 0, "No patches to apply"

# Find the first unknown base revision

needed_stack = []

revisions = {}

for patch in reversed(stack):

needed_stack.insert(0, patch)

# Stop as soon as a base revision is available

if self.has_revision(hg, patch.base_revision):

logger.info(

"Stopping at diff %s and revision %s", patch.id, patch.base_revision

)

break

if not needed_stack:

logger.info("All the patches are already applied")

return

# Load all the diff revisions

diffs = phabricator_api.search_diffs(diff_phid=[p.phid for p in stack])

revisions = {

diff["phid"]: phabricator_api.load_revision(

rev_phid=diff["revisionPHID"], attachments={"reviewers": True}

)

for diff in diffs

}

# Update repo to base revision

hg_base = needed_stack[0].base_revision

if not self.has_revision(hg, hg_base):

logger.warning("Missing base revision {} from Phabricator".format(hg_base))

hg_base = "default"

if hg_base:

hg.update(rev=hg_base, clean=True)

logger.info("Updated repo to %s", hg_base)

if self.git_repo_dir and hg_base != "default":

try:

self.git_base = tuple(

vcs_map.mercurial_to_git(self.git_repo_dir, [hg_base])

)[0]

subprocess.run(

["git", "checkout", "-b", "analysis_branch", self.git_base],

check=True,

cwd=self.git_repo_dir,

)

logger.info("Updated git repo to %s", self.git_base)

except Exception as e:

logger.info(

"Updating git repo to Mercurial %s failed: %s", hg_base, e

)

def load_user(phid):

if phid.startswith("PHID-USER"):

return phabricator_api.load_user(user_phid=phid)

elif phid.startswith("PHID-PROJ"):

# TODO: Support group reviewers somehow.

logger.info("Skipping group reviewer %s", phid)

else:

raise ValueError(f"Unsupported reviewer {phid}")

for patch in needed_stack:

revision = revisions[patch.phid]

message = "{}\n\n{}".format(

revision["fields"]["title"], revision["fields"]["summary"]

)

author_name = None

author_email = None

if patch.commits:

author_name = patch.commits[0]["author"]["name"]

author_email = patch.commits[0]["author"]["email"]

if author_name is None:

author = load_user(revision["fields"]["authorPHID"])

author_name = author["fields"]["realName"]

# XXX: Figure out a way to know the email address of the author.

author_email = author["fields"]["username"]

reviewers = list(

filter(

None,

(

load_user(reviewer["reviewerPHID"])

for reviewer in revision["attachments"]["reviewers"][

"reviewers"

]

),

)

)

reviewers = set(reviewer["fields"]["username"] for reviewer in reviewers)

if len(reviewers):

message = replace_reviewers(message, reviewers)

logger.info(

"Applying %s from revision %s: %s", patch.phid, revision["id"], message

)

hg.import_(

patches=io.BytesIO(patch.patch.encode("utf-8")),

message=message.encode("utf-8"),

user=f"{author_name} <{author_email}>".encode("utf-8"),

)

latest_rev = repository.get_revs(hg, -1)[0]

if self.git_repo_dir:

subprocess.run(

["git", "cinnabar", "fetch", f"hg::{self.repo_dir}", latest_rev],

check=True,

cwd=self.git_repo_dir,

)

def generate_feature_importance_data(self, probs, importance):

_X = get_transformer_pipeline(self.clf).transform(self.X)

X_shap_values = shap.TreeExplainer(

self.clf.named_steps["estimator"]

).shap_values(_X)

pred_class = self.model.le.inverse_transform([probs[0].argmax()])[0]

features = []

for i, (val, feature_index, is_positive) in enumerate(

importance["importances"]["classes"][pred_class][0]

):

name = importance["feature_legend"][str(i + 1)]

value = importance["importances"]["values"][0, int(feature_index)]

shap.summary_plot(

X_shap_values[:, int(feature_index)].reshape(_X.shape[0], 1),

_X[:, int(feature_index)].reshape(_X.shape[0], 1),

feature_names=[""],

plot_type="layered_violin",

show=False,

)

matplotlib.pyplot.xlabel("Impact on model output")

img = io.BytesIO()

matplotlib.pyplot.savefig(img, bbox_inches="tight")

matplotlib.pyplot.clf()

img.seek(0)

base64_img = base64.b64encode(img.read()).decode("ascii")

X = _X[:, int(feature_index)]

y = self.y[X != 0]

X = X[X != 0]

spearman = spearmanr(X, y)

buggy_X = X[y == 1]

clean_X = X[y == 0]

median = np.median(X)

median_clean = np.median(clean_X)

median_buggy = np.median(buggy_X)

perc_buggy_values_higher_than_median = (

buggy_X >= median

).sum() / buggy_X.shape[0]

perc_buggy_values_lower_than_median = (

buggy_X < median

).sum() / buggy_X.shape[0]

perc_clean_values_higher_than_median = (

clean_X > median

).sum() / clean_X.shape[0]

perc_clean_values_lower_than_median = (

clean_X <= median

).sum() / clean_X.shape[0]

logger.info("Feature: {}".format(name))

logger.info("Shap value: {}{}".format("+" if (is_positive) else "-", val))

logger.info("spearman: %f", spearman)

logger.info("value: %f", value)

logger.info("overall mean: %f", np.mean(X))

logger.info("overall median: %f", np.median(X))

logger.info("mean for y == 0: %f", np.mean(clean_X))

logger.info("mean for y == 1: %f", np.mean(buggy_X))

logger.info("median for y == 0: %f", np.median(clean_X))

logger.info("median for y == 1: %f", np.median(buggy_X))

logger.info(

"perc_buggy_values_higher_than_median: %f",

perc_buggy_values_higher_than_median,

)

logger.info(

"perc_buggy_values_lower_than_median: %f",

perc_buggy_values_lower_than_median,

)

logger.info(

"perc_clean_values_higher_than_median: %f",

perc_clean_values_higher_than_median,

)

logger.info(

"perc_clean_values_lower_than_median: %f",

perc_clean_values_lower_than_median,

)

features.append(

{

"index": i + 1,

"name": name,

"shap": float(f"{'+' if (is_positive) else '-'}{val}"),

"value": importance["importances"]["values"][0, int(feature_index)],

"spearman": spearman,

"median": median,

"median_bug_introducing": median_buggy,

"median_clean": median_clean,

"perc_buggy_values_higher_than_median": perc_buggy_values_higher_than_median,

"perc_buggy_values_lower_than_median": perc_buggy_values_lower_than_median,

"perc_clean_values_higher_than_median": perc_clean_values_higher_than_median,

"perc_clean_values_lower_than_median": perc_clean_values_lower_than_median,

"plot": base64_img,

}

)

# Group together features that are very similar to each other, so we can simplify the explanation

# to users.

attributes = ["Total", "Maximum", "Minimum", "Average"]

already_added = set()

feature_groups = []

for i1, f1 in enumerate(features):

if i1 in already_added:

continue

feature_groups.append([f1])

for j, f2 in enumerate(features[i1 + 1 :]):

i2 = j + i1 + 1

f1_name = f1["name"]

for attribute in attributes:

if f1_name.startswith(attribute):

f1_name = f1_name[len(attribute) + 1 :]

break

f2_name = f2["name"]

for attribute in attributes:

if f2_name.startswith(attribute):

f2_name = f2_name[len(attribute) + 1 :]

break

if f1_name != f2_name:

continue

already_added.add(i2)

feature_groups[-1].append(f2)

# Pick a representative example from each group.

features = []

for feature_group in feature_groups:

shap_sum = sum(f["shap"] for f in feature_group)

# Only select easily explainable features from the group.

selected = [

f

for f in feature_group

if (

f["shap"] > 0

and abs(f["value"] - f["median_bug_introducing"])

< abs(f["value"] - f["median_clean"])

)

or (

f["shap"] < 0

and abs(f["value"] - f["median_clean"])

< abs(f["value"] - f["median_bug_introducing"])

)

]

# If there are no easily explainable features in the group, select all features of the group.

if len(selected) == 0:

selected = feature_group

def feature_sort_key(f):

if f["shap"] > 0 and f["spearman"][0] > 0:

return f["perc_buggy_values_higher_than_median"]

elif f["shap"] > 0 and f["spearman"][0] < 0:

return f["perc_buggy_values_lower_than_median"]

elif f["shap"] < 0 and f["spearman"][0] > 0:

return f["perc_clean_values_lower_than_median"]

elif f["shap"] < 0 and f["spearman"][0] < 0:

return f["perc_clean_values_higher_than_median"]

feature = max(selected, key=feature_sort_key)

feature["shap"] = shap_sum

for attribute in attributes:

if feature["name"].startswith(attribute):

feature["name"] = feature["name"][len(attribute) + 1 :].capitalize()

break

features.append(feature)

with open("importances.json", "w") as f:

json.dump(features, f)

def classify(

self,

revision: str | None = None,

runnable_jobs_path: str | None = None,

) -> None:

self.update_commit_db()

if self.revision is not None:

assert revision is None

revision = self.revision

commits = repository.download_commits(

self.repo_dir,

rev_start=revision,

save=False,

use_single_process=self.use_single_process,

)

else:

assert revision is not None

commits = tuple(

commit

for commit in repository.get_commits()

if commit["node"] == revision

)

# The commit to analyze was not in our DB, let's mine it.

if len(commits) == 0:

commits = repository.download_commits(

self.repo_dir,

revs=[revision.encode("ascii")],

save=False,

use_single_process=self.use_single_process,

)

assert len(commits) > 0, "There are no commits to analyze"

if not self.use_test_history:

self.classify_regressor(commits)

else:

self.classify_test_select(commits, runnable_jobs_path)

def classify_regressor(self, commits: tuple[repository.CommitDict, ...]) -> None:

# We use "clean" (or "dirty") commits as the background dataset for feature importance.

# This way, we can see the features which are most important in differentiating

# the current commit from the "clean" (or "dirty") commits.

probs, importance = self.model.classify(

commits[-1],

probabilities=True,

importances=True,

background_dataset=lambda v: self.X[self.y != v],

importance_cutoff=0.05,

)

if not self.skip_feature_importance:

self.generate_feature_importance_data(probs, importance)

results = {

"probs": probs[0].tolist(),

}

if self.model_name == "regressor":

results["risk_band"] = RegressorModel.find_risk_band(probs[0][1])

with open("results.json", "w") as f:

json.dump(results, f)

if self.model_name == "regressor" and self.method_defect_predictor_dir:

self.classify_methods(commits[-1])

def classify_test_select(self, commits, runnable_jobs_path):

testfailure_probs = self.testfailure_model.classify(

commits[-1], probabilities=True

)

logger.info("Test failure risk: %f", testfailure_probs[0][1])

if not runnable_jobs_path:

runnable_jobs = {}

elif runnable_jobs_path.startswith("http"):

r = requests.get(runnable_jobs_path)

r.raise_for_status()

runnable_jobs = r.json()

else:

with open(runnable_jobs_path, "r") as f:

runnable_jobs = json.load(f)

# XXX: Remove tasks which are not in runnable jobs right away, so we avoid classifying them.

# XXX: Consider using mozilla-central built-in rules to filter some of the tasks out, e.g. SCHEDULES.

selected_tasks = list(

self.model.select_tests(

commits, float(get_secret("TEST_SELECTION_CONFIDENCE_THRESHOLD"))

).keys()

)

# XXX: For now, only restrict to linux64 test tasks (as for runnable jobs above, we could remove these right away).

selected_tasks = [

t for t in selected_tasks if t.startswith("test-linux1804-64/")

]

with open("failure_risk", "w") as f:

f.write(

"1"

if testfailure_probs[0][1]

> float(get_secret("TEST_FAILURE_CONFIDENCE_THRESHOLD"))

else "0"

)

# This should be kept in sync with the test scheduling history retriever script.

cleaned_selected_tasks = []

if len(runnable_jobs) > 0:

for selected_task in selected_tasks:

if (

selected_task.startswith("test-linux64")

and selected_task not in runnable_jobs

):

selected_task = selected_task.replace(

"test-linux64-", "test-linux1804-64-"

)

if (

selected_task.startswith("test-linux1804-64-")

and selected_task not in runnable_jobs

):

selected_task = selected_task.replace(

"test-linux1804-64-", "test-linux64-"

)

if selected_task in runnable_jobs:

cleaned_selected_tasks.append(selected_task)

# It isn't worth running the build associated to the tests, if we only run three test tasks.

if len(cleaned_selected_tasks) < 3:

cleaned_selected_tasks = []

with open("selected_tasks", "w") as f:

f.writelines(

f"{selected_task}\n" for selected_task in cleaned_selected_tasks

)

def classify_methods(self, commit):

# Get commit hash from 4 months before the analysis time.

# The method-level analyzer needs 4 months of history.

stop_hash = None

four_months_ago = datetime.utcnow() - relativedelta(months=4)

for commit in repository.get_commits():

if dateutil.parser.parse(commit["pushdate"]) >= four_months_ago:

stop_hash = tuple(

vcs_map.mercurial_to_git(self.git_repo_dir, [commit["node"]])

)[0]

break

assert stop_hash is not None

p = subprocess.run(

[

"git",

"rev-list",

"-n",

"1",

"HEAD",

],

check=True,

capture_output=True,

cwd=self.git_repo_dir,

)

start_hash = p.stdout.decode().strip()

# Run the method-level analyzer.

subprocess.run(

[

"python3",

"tester.py",

"--repo",

self.git_repo_dir,

"--start",

start_hash,

"--stop",

stop_hash,

"--output",

os.path.abspath("method_level.csv"),

],

cwd=self.method_defect_predictor_dir,

)

method_level_results = []

try:

with open("method_level.csv", "r") as f:

reader = csv.DictReader(f)

for item in reader:

item["past_bugs"] = []

method_level_results.append(item)

except FileNotFoundError:

# No methods were classified.

pass

for method_level_result in method_level_results:

method_level_result_path = method_level_result["file_name"]

if method_level_result_path not in self.past_bugs_by_function:

continue

for path, functions in commit["functions"].items():

if method_level_result_path != path:

continue

for function in functions:

function_name = function["name"]

if function_name not in self.past_bugs_by_function[path]:

continue

if method_level_result["method_name"].endswith(function_name):

method_level_result["past_bugs"] = [

"Bug {} - {}".format(bug["id"], bug["summary"])

for bug in self.past_bugs_by_function[path][function_name][

-3:

]

]

with open("method_level.json", "w") as f:

json.dump(method_level_results, f)

def main() -> None:

description = "Classify a commit"

parser = argparse.ArgumentParser(description=description)

parser.add_argument("model", help="Which model to use for evaluation")

parser.add_argument(

"repo_dir",

help="Path to a Gecko repository. If no repository exists, it will be cloned to this location.",

)

parser.add_argument(

"--phabricator-deployment",

help="Which Phabricator deployment to hit.",

type=str,

choices=[PHAB_PROD, PHAB_DEV],

)

parser.add_argument("--diff-id", help="diff ID to analyze.", type=int)

parser.add_argument("--revision", help="revision to analyze.", type=str)

parser.add_argument(

"--runnable-jobs",

help="Path or URL to a file containing runnable jobs.",

type=str,

)

parser.add_argument(

"--git_repo_dir", help="Path where the git repository will be cloned."

)

parser.add_argument(

"--method_defect_predictor_dir",

help="Path where the git repository will be cloned.",

)

parser.add_argument(

"--use-single-process",

action="store_true",

help="Whether to use a single process.",

)

parser.add_argument(

"--skip-feature-importance",

action="store_true",

help="Whether to skip feature importance calculation.",

)

args = parser.parse_args()

if args.revision is not None:

assert args.phabricator_deployment is None

assert args.diff_id is None

if args.diff_id is not None:

assert args.phabricator_deployment is not None

assert args.revision is None

classifier = CommitClassifier(

args.model,

args.repo_dir,

args.git_repo_dir,

args.method_defect_predictor_dir,

args.use_single_process,

args.skip_feature_importance,

args.phabricator_deployment,

args.diff_id,

)

classifier.classify(args.revision, args.runnable_jobs)

if __name__ == "__main__":

main()