api: controller: Add controller for CopyBot
Add a controller and mock data for the CopyBot build API.
BUG=b:248325989
TEST=./api/contrib/call_scripts/copybot__run_copybot
Change-Id: Ic07b2ba8a77220e7cbdbcf777ef22c6a808cc6c2
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/chromite/+/4235244
Commit-Queue: Tristan Honscheid <honscheid@google.com>
Reviewed-by: Jack Rosenthal <jrosenth@chromium.org>
Tested-by: Tristan Honscheid <honscheid@google.com>
diff --git a/api/controller/copybot.py b/api/controller/copybot.py
new file mode 100644
index 0000000..7e11a68
--- /dev/null
+++ b/api/controller/copybot.py
@@ -0,0 +1,98 @@
+# Copyright 2023 The ChromiumOS Authors
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Copybot Controller.
+
+Handles the endpoint for running copybot and generating the protobuf.
+"""
+
+from pathlib import Path
+import tempfile
+
+from chromite.third_party.google.protobuf import json_format
+
+from chromite.api import controller
+from chromite.api import faux
+from chromite.api import validate
+from chromite.api.gen.chromite.api import copybot_pb2
+from chromite.lib import constants
+from chromite.lib import cros_build_lib
+
+
+def _MockSuccess(_input_proto, _output_proto, _config_proto):
+ """Mock success output for the RunCopybot endpoint."""
+
+ # Successful response is the default protobuf, so no need to fill it out.
+
+
+@faux.success(_MockSuccess)
+@faux.empty_error
+@validate.validation_complete
+def RunCopybot(input_proto, output_proto, _config_proto):
+ """Run copybot. Translate all fields in the input protobuf to CLI args."""
+
+ cmd = [
+ Path(constants.SOURCE_ROOT)
+ / "src/platform/dev/contrib/copybot/copybot.py"
+ ]
+
+ if input_proto.topic:
+ cmd.extend(["--topic", input_proto.topic])
+
+ for label in input_proto.labels:
+ cmd.extend(["--label", label.label])
+
+ for reviewer in input_proto.reviewers:
+ cmd.extend(["--re", reviewer.user])
+
+ for cc in input_proto.ccs:
+ cmd.extend(["--cc", cc.user])
+
+ if input_proto.prepend_subject:
+ cmd.extend(["--prepend-subject", input_proto.prepend_subject])
+
+ if (
+ input_proto.merge_conflict_behavior
+ == copybot_pb2.RunCopybotRequest.MERGE_CONFLICT_BEHAVIOR_SKIP
+ ):
+ cmd.extend(["--merge-conflict-behavior", "SKIP"])
+
+ if (
+ input_proto.merge_conflict_behavior
+ == copybot_pb2.RunCopybotRequest.MERGE_CONFLICT_BEHAVIOR_FAIL
+ ):
+ cmd.extend(["--merge-conflict-behavior", "FAIL"])
+
+ for exclude in input_proto.exclude_file_patterns:
+ cmd.extend(["--exclude-file-pattern", exclude.pattern])
+
+ for ph in input_proto.keep_pseudoheaders:
+ cmd.extend(["--keep-pseudoheader", ph.name])
+
+ if input_proto.add_signed_off_by:
+ cmd.append("--add-signed-off-by")
+
+ if input_proto.dry_run:
+ cmd.append("--dry-run")
+
+ cmd.append(f"{input_proto.upstream.url}:{input_proto.upstream.branch}")
+ cmd.append(f"{input_proto.downstream.url}:{input_proto.downstream.branch}")
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ json_output_path = Path(temp_dir) / "copybot_output.json"
+ cmd.extend(["--json-out", json_output_path])
+
+ try:
+ cros_build_lib.run(cmd)
+ except cros_build_lib.RunCommandError:
+ # In case of failure, load details about the error from CopyBot's JSON
+ # output into the output protobuf. (If CopyBot ran successfully, the
+ # default values are simply used). CopyBot's output matches the JSON
+ # representation of the RunCopybotResponse protobuf.
+
+ if not json_output_path.exists():
+ return controller.RETURN_CODE_UNRECOVERABLE
+
+ json_format.Parse(json_output_path.read_text(), output_proto)
+ return controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE