Formatting: Format all python code with black.
This CL is probably not what you're looking for, it's only
automated formatting. Ignore it with
`git blame --ignore-rev <revision>` for this commit.
BUG=b:233893248
TEST=CQ
Change-Id: I66591d7a738d241aed3290138c0f68065ab10a6d
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/chromite/+/3879174
Reviewed-by: Mike Frysinger <vapier@chromium.org>
Tested-by: Alex Klein <saklein@chromium.org>
diff --git a/api/controller/binhost.py b/api/controller/binhost.py
index e79917a..8702c47 100644
--- a/api/controller/binhost.py
+++ b/api/controller/binhost.py
@@ -22,252 +22,269 @@
if TYPE_CHECKING:
- from chromite.api import api_config
+ from chromite.api import api_config
_OVERLAY_TYPE_TO_NAME = {
binhost_pb2.OVERLAYTYPE_PUBLIC: constants.PUBLIC_OVERLAYS,
binhost_pb2.OVERLAYTYPE_PRIVATE: constants.PRIVATE_OVERLAYS,
binhost_pb2.OVERLAYTYPE_BOTH: constants.BOTH_OVERLAYS,
- binhost_pb2.OVERLAYTYPE_NONE: None
+ binhost_pb2.OVERLAYTYPE_NONE: None,
}
def _GetBinhostsResponse(_input_proto, output_proto, _config):
- """Add fake binhosts to a successful response."""
- new_binhost = output_proto.binhosts.add()
- new_binhost.uri = ('gs://cr-prebuilt/board/amd64-generic/'
- 'paladin-R66-17.0.0-rc2/packages/')
- new_binhost.package_index = 'Packages'
+ """Add fake binhosts to a successful response."""
+ new_binhost = output_proto.binhosts.add()
+ new_binhost.uri = (
+ "gs://cr-prebuilt/board/amd64-generic/"
+ "paladin-R66-17.0.0-rc2/packages/"
+ )
+ new_binhost.package_index = "Packages"
@faux.success(_GetBinhostsResponse)
@faux.empty_error
-@validate.require('build_target.name')
+@validate.require("build_target.name")
@validate.validation_complete
def GetBinhosts(input_proto, output_proto, _config):
- """Get a list of binhosts."""
- build_target = controller_util.ParseBuildTarget(input_proto.build_target)
+ """Get a list of binhosts."""
+ build_target = controller_util.ParseBuildTarget(input_proto.build_target)
- binhosts = binhost.GetBinhosts(build_target)
+ binhosts = binhost.GetBinhosts(build_target)
- for current in binhosts:
- new_binhost = output_proto.binhosts.add()
- new_binhost.uri = current
- new_binhost.package_index = 'Packages'
+ for current in binhosts:
+ new_binhost = output_proto.binhosts.add()
+ new_binhost.uri = current
+ new_binhost.package_index = "Packages"
def _GetPrivatePrebuiltAclArgsResponse(_input_proto, output_proto, _config):
- """Add fake acls to a successful response."""
- new_arg = output_proto.args.add()
- new_arg.arg = '-g'
- new_arg.value = 'group1:READ'
+ """Add fake acls to a successful response."""
+ new_arg = output_proto.args.add()
+ new_arg.arg = "-g"
+ new_arg.value = "group1:READ"
@faux.success(_GetPrivatePrebuiltAclArgsResponse)
@faux.empty_error
-@validate.require('build_target.name')
+@validate.require("build_target.name")
@validate.validation_complete
def GetPrivatePrebuiltAclArgs(input_proto, output_proto, _config):
- """Get the ACL args from the files in the private overlays."""
- build_target = controller_util.ParseBuildTarget(input_proto.build_target)
+ """Get the ACL args from the files in the private overlays."""
+ build_target = controller_util.ParseBuildTarget(input_proto.build_target)
- try:
- args = binhost.GetPrebuiltAclArgs(build_target)
- except binhost.Error as e:
- cros_build_lib.Die(e)
+ try:
+ args = binhost.GetPrebuiltAclArgs(build_target)
+ except binhost.Error as e:
+ cros_build_lib.Die(e)
- for arg, value in args:
- new_arg = output_proto.args.add()
- new_arg.arg = arg
- new_arg.value = value
+ for arg, value in args:
+ new_arg = output_proto.args.add()
+ new_arg.arg = arg
+ new_arg.value = value
def _PrepareBinhostUploadsResponse(_input_proto, output_proto, _config):
- """Add fake binhost upload targets to a successful response."""
- output_proto.uploads_dir = '/upload/directory'
- output_proto.upload_targets.add().path = 'upload_target'
+ """Add fake binhost upload targets to a successful response."""
+ output_proto.uploads_dir = "/upload/directory"
+ output_proto.upload_targets.add().path = "upload_target"
@faux.success(_PrepareBinhostUploadsResponse)
@faux.empty_error
-@validate.require('uri')
+@validate.require("uri")
def PrepareBinhostUploads(
input_proto: binhost_pb2.PrepareBinhostUploadsRequest,
output_proto: binhost_pb2.PrepareBinhostUploadsResponse,
- config: 'api_config.ApiConfig'):
- """Return a list of files to upload to the binhost.
+ config: "api_config.ApiConfig",
+):
+ """Return a list of files to upload to the binhost.
- See BinhostService documentation in api/proto/binhost.proto.
+ See BinhostService documentation in api/proto/binhost.proto.
- Args:
- input_proto: The input proto.
- output_proto: The output proto.
- config: The API call config.
- """
- if input_proto.sysroot.build_target.name:
- build_target_msg = input_proto.sysroot.build_target
- else:
- build_target_msg = input_proto.build_target
- sysroot_path = input_proto.sysroot.path
+ Args:
+ input_proto: The input proto.
+ output_proto: The output proto.
+ config: The API call config.
+ """
+ if input_proto.sysroot.build_target.name:
+ build_target_msg = input_proto.sysroot.build_target
+ else:
+ build_target_msg = input_proto.build_target
+ sysroot_path = input_proto.sysroot.path
- if not sysroot_path and not build_target_msg.name:
- cros_build_lib.Die('Sysroot.path is required.')
+ if not sysroot_path and not build_target_msg.name:
+ cros_build_lib.Die("Sysroot.path is required.")
- build_target = controller_util.ParseBuildTarget(build_target_msg)
- chroot = controller_util.ParseChroot(input_proto.chroot)
+ build_target = controller_util.ParseBuildTarget(build_target_msg)
+ chroot = controller_util.ParseChroot(input_proto.chroot)
- if not sysroot_path:
- sysroot_path = build_target.root
- sysroot = sysroot_lib.Sysroot(sysroot_path)
+ if not sysroot_path:
+ sysroot_path = build_target.root
+ sysroot = sysroot_lib.Sysroot(sysroot_path)
- uri = input_proto.uri
- # For now, we enforce that all input URIs are Google Storage buckets.
- if not gs.PathIsGs(uri):
- raise ValueError('Upload URI %s must be Google Storage.' % uri)
+ uri = input_proto.uri
+ # For now, we enforce that all input URIs are Google Storage buckets.
+ if not gs.PathIsGs(uri):
+ raise ValueError("Upload URI %s must be Google Storage." % uri)
- if config.validate_only:
- return controller.RETURN_CODE_VALID_INPUT
+ if config.validate_only:
+ return controller.RETURN_CODE_VALID_INPUT
- parsed_uri = urllib.parse.urlparse(uri)
- upload_uri = gs.GetGsURL(parsed_uri.netloc, for_gsutil=True).rstrip('/')
- upload_path = parsed_uri.path.lstrip('/')
+ parsed_uri = urllib.parse.urlparse(uri)
+ upload_uri = gs.GetGsURL(parsed_uri.netloc, for_gsutil=True).rstrip("/")
+ upload_path = parsed_uri.path.lstrip("/")
- # Read all packages and update the index. The index must be uploaded to the
- # binhost for Portage to use it, so include it in upload_targets.
- uploads_dir = binhost.GetPrebuiltsRoot(chroot, sysroot, build_target)
- index_path = binhost.UpdatePackageIndex(
- uploads_dir, upload_uri, upload_path, sudo=True)
- upload_targets = binhost.GetPrebuiltsFiles(uploads_dir)
- assert index_path.startswith(uploads_dir), (
- 'expected index_path to start with uploads_dir')
- upload_targets.append(index_path[len(uploads_dir):])
+ # Read all packages and update the index. The index must be uploaded to the
+ # binhost for Portage to use it, so include it in upload_targets.
+ uploads_dir = binhost.GetPrebuiltsRoot(chroot, sysroot, build_target)
+ index_path = binhost.UpdatePackageIndex(
+ uploads_dir, upload_uri, upload_path, sudo=True
+ )
+ upload_targets = binhost.GetPrebuiltsFiles(uploads_dir)
+ assert index_path.startswith(
+ uploads_dir
+ ), "expected index_path to start with uploads_dir"
+ upload_targets.append(index_path[len(uploads_dir) :])
- output_proto.uploads_dir = uploads_dir
- for upload_target in upload_targets:
- output_proto.upload_targets.add().path = upload_target.strip('/')
+ output_proto.uploads_dir = uploads_dir
+ for upload_target in upload_targets:
+ output_proto.upload_targets.add().path = upload_target.strip("/")
-def _PrepareDevInstallBinhostUploadsResponse(_input_proto, output_proto,
- _config):
- """Add fake binhost files to a successful response."""
- output_proto.upload_targets.add().path = 'app-arch/zip-3.0-r3.tbz2'
- output_proto.upload_targets.add().path = 'virtual/python-enum34-1.tbz2'
- output_proto.upload_targets.add().path = 'Packages'
+def _PrepareDevInstallBinhostUploadsResponse(
+ _input_proto, output_proto, _config
+):
+ """Add fake binhost files to a successful response."""
+ output_proto.upload_targets.add().path = "app-arch/zip-3.0-r3.tbz2"
+ output_proto.upload_targets.add().path = "virtual/python-enum34-1.tbz2"
+ output_proto.upload_targets.add().path = "Packages"
@faux.success(_PrepareDevInstallBinhostUploadsResponse)
@faux.empty_error
-@validate.require('uri', 'sysroot.path')
-@validate.exists('uploads_dir')
+@validate.require("uri", "sysroot.path")
+@validate.exists("uploads_dir")
def PrepareDevInstallBinhostUploads(
input_proto: binhost_pb2.PrepareDevInstallBinhostUploadsRequest,
output_proto: binhost_pb2.PrepareDevInstallBinhostUploadsResponse,
- config: 'api_config.ApiConfig'):
- """Return a list of files to upload to the binhost"
+ config: "api_config.ApiConfig",
+):
+ """Return a list of files to upload to the binhost"
- The files will also be copied to the uploads_dir.
- See BinhostService documentation in api/proto/binhost.proto.
+ The files will also be copied to the uploads_dir.
+ See BinhostService documentation in api/proto/binhost.proto.
- Args:
- input_proto: The input proto.
- output_proto: The output proto.
- config: The API call config.
- """
- sysroot_path = input_proto.sysroot.path
+ Args:
+ input_proto: The input proto.
+ output_proto: The output proto.
+ config: The API call config.
+ """
+ sysroot_path = input_proto.sysroot.path
- chroot = controller_util.ParseChroot(input_proto.chroot)
- sysroot = sysroot_lib.Sysroot(sysroot_path)
+ chroot = controller_util.ParseChroot(input_proto.chroot)
+ sysroot = sysroot_lib.Sysroot(sysroot_path)
- uri = input_proto.uri
- # For now, we enforce that all input URIs are Google Storage buckets.
- if not gs.PathIsGs(uri):
- raise ValueError('Upload URI %s must be Google Storage.' % uri)
+ uri = input_proto.uri
+ # For now, we enforce that all input URIs are Google Storage buckets.
+ if not gs.PathIsGs(uri):
+ raise ValueError("Upload URI %s must be Google Storage." % uri)
- if config.validate_only:
- return controller.RETURN_CODE_VALID_INPUT
+ if config.validate_only:
+ return controller.RETURN_CODE_VALID_INPUT
- parsed_uri = urllib.parse.urlparse(uri)
- upload_uri = gs.GetGsURL(parsed_uri.netloc, for_gsutil=True).rstrip('/')
- upload_path = parsed_uri.path.lstrip('/')
+ parsed_uri = urllib.parse.urlparse(uri)
+ upload_uri = gs.GetGsURL(parsed_uri.netloc, for_gsutil=True).rstrip("/")
+ upload_path = parsed_uri.path.lstrip("/")
- # Calculate the filename for the to-be-created Packages file, which will
- # contain only devinstall packages.
- devinstall_package_index_path = os.path.join(input_proto.uploads_dir,
- 'Packages')
- upload_targets_list = binhost.ReadDevInstallFilesToCreatePackageIndex(
- chroot, sysroot, devinstall_package_index_path, upload_uri, upload_path)
+ # Calculate the filename for the to-be-created Packages file, which will
+ # contain only devinstall packages.
+ devinstall_package_index_path = os.path.join(
+ input_proto.uploads_dir, "Packages"
+ )
+ upload_targets_list = binhost.ReadDevInstallFilesToCreatePackageIndex(
+ chroot, sysroot, devinstall_package_index_path, upload_uri, upload_path
+ )
- package_dir = chroot.full_path(sysroot.path, 'packages')
- for upload_target in upload_targets_list:
- # Copy each package to target/category/package
- upload_target = upload_target.strip('/')
- category = upload_target.split(os.sep)[0]
- target_dir = os.path.join(input_proto.uploads_dir, category)
- if not os.path.exists(target_dir):
- os.makedirs(target_dir)
- full_src_pkg_path = os.path.join(package_dir, upload_target)
- full_target_src_path = os.path.join(input_proto.uploads_dir, upload_target)
- shutil.copyfile(full_src_pkg_path, full_target_src_path)
- output_proto.upload_targets.add().path = upload_target
- output_proto.upload_targets.add().path = 'Packages'
+ package_dir = chroot.full_path(sysroot.path, "packages")
+ for upload_target in upload_targets_list:
+ # Copy each package to target/category/package
+ upload_target = upload_target.strip("/")
+ category = upload_target.split(os.sep)[0]
+ target_dir = os.path.join(input_proto.uploads_dir, category)
+ if not os.path.exists(target_dir):
+ os.makedirs(target_dir)
+ full_src_pkg_path = os.path.join(package_dir, upload_target)
+ full_target_src_path = os.path.join(
+ input_proto.uploads_dir, upload_target
+ )
+ shutil.copyfile(full_src_pkg_path, full_target_src_path)
+ output_proto.upload_targets.add().path = upload_target
+ output_proto.upload_targets.add().path = "Packages"
def _SetBinhostResponse(_input_proto, output_proto, _config):
- """Add fake binhost file to a successful response."""
- output_proto.output_file = '/path/to/BINHOST.conf'
+ """Add fake binhost file to a successful response."""
+ output_proto.output_file = "/path/to/BINHOST.conf"
@faux.success(_SetBinhostResponse)
@faux.empty_error
-@validate.require('build_target.name', 'key', 'uri')
+@validate.require("build_target.name", "key", "uri")
@validate.validation_complete
-def SetBinhost(input_proto: binhost_pb2.SetBinhostRequest,
- output_proto: binhost_pb2.SetBinhostResponse,
- _config: 'api_config.ApiConfig'):
- """Set the URI for a given binhost key and build target.
+def SetBinhost(
+ input_proto: binhost_pb2.SetBinhostRequest,
+ output_proto: binhost_pb2.SetBinhostResponse,
+ _config: "api_config.ApiConfig",
+):
+ """Set the URI for a given binhost key and build target.
- See BinhostService documentation in api/proto/binhost.proto.
+ See BinhostService documentation in api/proto/binhost.proto.
- Args:
- input_proto: The input proto.
- output_proto: The output proto.
- _config: The API call config.
- """
- target = input_proto.build_target.name
- key = binhost_pb2.BinhostKey.Name(input_proto.key)
- uri = input_proto.uri
- private = input_proto.private
+ Args:
+ input_proto: The input proto.
+ output_proto: The output proto.
+ _config: The API call config.
+ """
+ target = input_proto.build_target.name
+ key = binhost_pb2.BinhostKey.Name(input_proto.key)
+ uri = input_proto.uri
+ private = input_proto.private
- output_proto.output_file = binhost.SetBinhost(
- target, key, uri, private=private)
+ output_proto.output_file = binhost.SetBinhost(
+ target, key, uri, private=private
+ )
def _RegenBuildCacheResponse(_input_proto, output_proto, _config):
- """Add fake binhosts cache path to a successful response."""
- output_proto.modified_overlays.add().path = '/path/to/BuildCache'
+ """Add fake binhosts cache path to a successful response."""
+ output_proto.modified_overlays.add().path = "/path/to/BuildCache"
@faux.success(_RegenBuildCacheResponse)
@faux.empty_error
-@validate.require('overlay_type')
-@validate.is_in('overlay_type', _OVERLAY_TYPE_TO_NAME)
+@validate.require("overlay_type")
+@validate.is_in("overlay_type", _OVERLAY_TYPE_TO_NAME)
@validate.validation_complete
-def RegenBuildCache(input_proto: binhost_pb2.RegenBuildCacheRequest,
- output_proto: binhost_pb2.RegenBuildCacheResponse,
- _config: 'api_config.ApiConfig'):
- """Regenerate the Build Cache for a build target.
+def RegenBuildCache(
+ input_proto: binhost_pb2.RegenBuildCacheRequest,
+ output_proto: binhost_pb2.RegenBuildCacheResponse,
+ _config: "api_config.ApiConfig",
+):
+ """Regenerate the Build Cache for a build target.
- See BinhostService documentation in api/proto/binhost.proto.
+ See BinhostService documentation in api/proto/binhost.proto.
- Args:
- input_proto: The input proto.
- output_proto: The output proto.
- _config: The API call config.
- """
- chroot = controller_util.ParseChroot(input_proto.chroot)
- overlay_type = input_proto.overlay_type
- overlays = binhost.RegenBuildCache(chroot,
- _OVERLAY_TYPE_TO_NAME[overlay_type])
+ Args:
+ input_proto: The input proto.
+ output_proto: The output proto.
+ _config: The API call config.
+ """
+ chroot = controller_util.ParseChroot(input_proto.chroot)
+ overlay_type = input_proto.overlay_type
+ overlays = binhost.RegenBuildCache(
+ chroot, _OVERLAY_TYPE_TO_NAME[overlay_type]
+ )
- for overlay in overlays:
- output_proto.modified_overlays.add().path = overlay
+ for overlay in overlays:
+ output_proto.modified_overlays.add().path = overlay