in bot/code_review_bot/revisions.py [0:0]
def from_try_task(try_task: dict, decision_task: dict, phabricator: PhabricatorAPI):
"""
Load identifiers from Phabricator, using the remote task description
"""
# Load build target phid from the task env
code_review = try_task["extra"]["code-review"]
build_target_phid = code_review.get("phabricator-diff") or code_review.get(
"phabricator-build-target"
)
assert (
build_target_phid is not None
), "Missing phabricator-build-target or phabricator-diff declaration"
assert build_target_phid.startswith("PHID-HMBT-")
# And get the diff from the phabricator api
buildable = phabricator.find_target_buildable(build_target_phid)
diff_phid = buildable["fields"]["objectPHID"]
assert diff_phid.startswith("PHID-DIFF-")
# Load diff details to get the diff revision
# We also load the commits list in order to get the email of the author of the
# patch for sending email if builds are failing.
diffs = phabricator.search_diffs(
diff_phid=diff_phid, attachments={"commits": True}
)
assert len(diffs) == 1, f"No diff available for {diff_phid}"
diff = diffs[0]
diff_id = diff["id"]
phid = diff["revisionPHID"]
revision = phabricator.load_revision(phid)
# Load repository detailed information
repos = phabricator.request(
"diffusion.repository.search",
constraints={"phids": [revision["fields"]["repositoryPHID"]]},
)
assert len(repos["data"]) == 1, "Repository not found on Phabricator"
# Load target patch from Phabricator for Try mode
patch = phabricator.load_raw_diff(diff_id)
# The parent decision task should exist
assert decision_task is not None, "Missing parent decision task"
logger.info("Found decision task", name=decision_task["metadata"]["name"])
# Match the decision task environment to get the mercurial information
decision_env = decision_task["payload"]["env"]
head_repository = base_repository = head_changeset = base_changeset = (
repository_try_name
) = None
for prefix in settings.decision_env_prefixes:
head_repository_key = f"{prefix}_HEAD_REPOSITORY"
base_repository_key = f"{prefix}_BASE_REPOSITORY"
head_changeset_key = f"{prefix}_HEAD_REV"
base_changeset_key = f"{prefix}_BASE_REV"
if (
head_repository_key not in decision_env
or base_repository_key not in decision_env
or head_changeset_key not in decision_env
or base_changeset_key not in decision_env
):
continue
head_repository = decision_env[head_repository_key]
base_repository = decision_env[base_repository_key]
head_changeset = decision_env[head_changeset_key]
base_changeset = decision_env[base_changeset_key]
repository_try_name = (
urllib.parse.urlparse(head_repository)
.path.rstrip("/")
.rsplit("/", 1)[-1]
)
break
# Check mercurial information were properly retrieved
assert all(
attr is not None
for attr in [
head_repository,
base_repository,
head_changeset,
base_changeset,
]
), "Unsupported parent decision task, missing mercurial information in its environment"
logger.info(
"Using mercurial changeset",
head_changeset=head_changeset,
head_repository=head_repository,
base_repository=base_repository,
)
# Build a revision without repositories as they are retrieved later
# when analyzing the full task group
return Revision(
phabricator_id=revision["id"],
phabricator_phid=phid,
diff_id=diff_id,
diff_phid=diff_phid,
build_target_phid=build_target_phid,
revision=revision,
phabricator_repository=repos["data"][0],
diff=diff,
url="https://{}/D{}".format(phabricator.hostname, revision["id"]),
patch=patch,
head_changeset=head_changeset,
base_changeset=base_changeset,
head_repository=head_repository,
repository_try_name=repository_try_name,
base_repository=base_repository,
)