Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def process_build(self, build):
"""
Code review workflow to load all necessary information from Phabricator builds
received from the webserver
"""
assert build is not None, "Invalid payload"
assert isinstance(build, PhabricatorBuild)
# Update its state
self.update_state(build)
if build.state == PhabricatorBuildState.Public:
# Check if the author is not blacklisted
if self.is_blacklisted(build.revision):
return
# When the build is public, load needed details
try:
self.load_patches_stack(build)
logger.info("Loaded stack of patches", build=str(build))
except Exception as e:
logger.warning(
"Failed to load build details", build=str(build), error=str(e)
)
return
# Then send the build toward next stage
async def start_risk_analysis(self, build: PhabricatorBuild):
"""
Run risk analysis by triggering a Taskcluster hook
"""
assert build.state == PhabricatorBuildState.Public
try:
if not self.should_run_risk_analysis(build):
return
task = self.community_tc["hooks"].triggerHook(
"project-relman",
"bugbug-classify-patch",
{
"PHABRICATOR_DEPLOYMENT": self.phabricator_deployment,
"DIFF_ID": build.diff_id,
},
)
task_id = task["status"]["taskId"]
logger.info("Triggered a new risk analysis task", id=task_id)
# Send task to monitoring