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/image.py b/api/controller/image.py
index 10a8edf..d95c574 100644
--- a/api/controller/image.py
+++ b/api/controller/image.py
@@ -32,8 +32,8 @@
if TYPE_CHECKING:
- from chromite.api import api_config
- from chromite.api.gen.chromite.api import image_pb2
+ from chromite.api import api_config
+ from chromite.api.gen.chromite.api import image_pb2
# The image.proto ImageType enum ids.
_BASE_ID = common_pb2.IMAGE_TYPE_BASE
@@ -94,412 +94,451 @@
# Core are the build/dev/test images.
# Use "latest" until we do a better job of passing through image directories,
# e.g. for artifacts.
-LOCATION_CORE = 'latest'
+LOCATION_CORE = "latest"
# The factory_install image.
-LOCATION_FACTORY = 'factory_shim'
+LOCATION_FACTORY = "factory_shim"
class ImageTypes(NamedTuple):
- """Parsed image types."""
- images: Set[str]
- vms: Set[int]
- mod_images: Set[int]
+ """Parsed image types."""
- @property
- def core_images(self) -> List[str]:
- """The core images (base/dev/test) as a list."""
- return list(self.images - {_IMAGE_MAPPING[_FACTORY_ID]}) or []
+ images: Set[str]
+ vms: Set[int]
+ mod_images: Set[int]
- @property
- def has_factory(self) -> bool:
- """Whether the factory image is present."""
- return _IMAGE_MAPPING[_FACTORY_ID] in self.images
+ @property
+ def core_images(self) -> List[str]:
+ """The core images (base/dev/test) as a list."""
+ return list(self.images - {_IMAGE_MAPPING[_FACTORY_ID]}) or []
- @property
- def factory(self) -> List[str]:
- """A list with the factory type if set."""
- return [_IMAGE_MAPPING[_FACTORY_ID]] if self.has_factory else []
+ @property
+ def has_factory(self) -> bool:
+ """Whether the factory image is present."""
+ return _IMAGE_MAPPING[_FACTORY_ID] in self.images
+
+ @property
+ def factory(self) -> List[str]:
+ """A list with the factory type if set."""
+ return [_IMAGE_MAPPING[_FACTORY_ID]] if self.has_factory else []
-def _add_image_to_proto(output_proto, path: Union['Path', str], image_type: int,
- board: str):
- """Quick helper function to add a new image to the output proto."""
- new_image = output_proto.images.add()
- new_image.path = str(path)
- new_image.type = image_type
- new_image.build_target.name = board
+def _add_image_to_proto(
+ output_proto, path: Union["Path", str], image_type: int, board: str
+):
+ """Quick helper function to add a new image to the output proto."""
+ new_image = output_proto.images.add()
+ new_image.path = str(path)
+ new_image.type = image_type
+ new_image.build_target.name = board
def ExampleGetResponse():
- """Give an example response to assemble upstream in caller artifacts."""
- uabs = common_pb2.UploadedArtifactsByService
- cabs = common_pb2.ArtifactsByService
- return uabs.Sysroot(artifacts=[
- uabs.Image.ArtifactPaths(
- artifact_type=cabs.Image.ArtifactType.DLC_IMAGE,
- paths=[
- common_pb2.Path(
- path='/tmp/dlc/dlc.img', location=common_pb2.Path.OUTSIDE)
- ])
- ])
+ """Give an example response to assemble upstream in caller artifacts."""
+ uabs = common_pb2.UploadedArtifactsByService
+ cabs = common_pb2.ArtifactsByService
+ return uabs.Sysroot(
+ artifacts=[
+ uabs.Image.ArtifactPaths(
+ artifact_type=cabs.Image.ArtifactType.DLC_IMAGE,
+ paths=[
+ common_pb2.Path(
+ path="/tmp/dlc/dlc.img",
+ location=common_pb2.Path.OUTSIDE,
+ )
+ ],
+ )
+ ]
+ )
-def GetArtifacts(in_proto: common_pb2.ArtifactsByService.Image,
- chroot: chroot_lib.Chroot, sysroot_class: sysroot_lib.Sysroot,
- build_target: build_target_lib.BuildTarget,
- output_dir) -> list:
- """Builds and copies images to specified output_dir.
+def GetArtifacts(
+ in_proto: common_pb2.ArtifactsByService.Image,
+ chroot: chroot_lib.Chroot,
+ sysroot_class: sysroot_lib.Sysroot,
+ build_target: build_target_lib.BuildTarget,
+ output_dir,
+) -> list:
+ """Builds and copies images to specified output_dir.
- Copies (after optionally bundling) all required images into the output_dir,
- returning a mapping of image type to a list of (output_dir) paths to
- the desired files. Note that currently it is only processing one image (DLC),
- but the future direction is to process all required images. Required images
- are located within output_artifact.artifact_type.
+ Copies (after optionally bundling) all required images into the output_dir,
+ returning a mapping of image type to a list of (output_dir) paths to
+ the desired files. Note that currently it is only processing one image (DLC),
+ but the future direction is to process all required images. Required images
+ are located within output_artifact.artifact_type.
- Args:
- in_proto: Proto request defining reqs.
- chroot: The chroot proto used for these artifacts.
- sysroot_class: The sysroot proto used for these artifacts.
- build_target: The build target used for these artifacts.
- output_dir: The path to write artifacts to.
+ Args:
+ in_proto: Proto request defining reqs.
+ chroot: The chroot proto used for these artifacts.
+ sysroot_class: The sysroot proto used for these artifacts.
+ build_target: The build target used for these artifacts.
+ output_dir: The path to write artifacts to.
- Returns:
- A list of dictionary mappings of ArtifactType to list of paths.
- """
- base_path = chroot.full_path(sysroot_class.path)
- board = build_target.name
- factory_shim_location = Path(
- image_lib.GetLatestImageLink(board, pointer=LOCATION_FACTORY))
+ Returns:
+ A list of dictionary mappings of ArtifactType to list of paths.
+ """
+ base_path = chroot.full_path(sysroot_class.path)
+ board = build_target.name
+ factory_shim_location = Path(
+ image_lib.GetLatestImageLink(board, pointer=LOCATION_FACTORY)
+ )
- generated = []
- dlc_func = functools.partial(image.copy_dlc_image, base_path)
- license_func = functools.partial(
- image.copy_license_credits, board, symlink=LOCATION_CORE)
- factory_image_func = functools.partial(
- image.create_factory_image_zip,
- chroot,
- sysroot_class,
- factory_shim_location,
- packages_service.determine_full_version(),
- )
- stripped_packags_func = functools.partial(
- image.create_stripped_packages_tar,
- chroot,
- build_target,
- )
- artifact_types = {
- in_proto.ArtifactType.DLC_IMAGE: dlc_func,
- in_proto.ArtifactType.LICENSE_CREDITS: license_func,
- in_proto.ArtifactType.FACTORY_IMAGE: factory_image_func,
- in_proto.ArtifactType.STRIPPED_PACKAGES: stripped_packags_func,
- }
+ generated = []
+ dlc_func = functools.partial(image.copy_dlc_image, base_path)
+ license_func = functools.partial(
+ image.copy_license_credits, board, symlink=LOCATION_CORE
+ )
+ factory_image_func = functools.partial(
+ image.create_factory_image_zip,
+ chroot,
+ sysroot_class,
+ factory_shim_location,
+ packages_service.determine_full_version(),
+ )
+ stripped_packags_func = functools.partial(
+ image.create_stripped_packages_tar,
+ chroot,
+ build_target,
+ )
+ artifact_types = {
+ in_proto.ArtifactType.DLC_IMAGE: dlc_func,
+ in_proto.ArtifactType.LICENSE_CREDITS: license_func,
+ in_proto.ArtifactType.FACTORY_IMAGE: factory_image_func,
+ in_proto.ArtifactType.STRIPPED_PACKAGES: stripped_packags_func,
+ }
- for output_artifact in in_proto.output_artifacts:
- for artifact_type, func in artifact_types.items():
- if artifact_type in output_artifact.artifact_types:
- result = func(output_dir)
- if result:
- generated.append({
- 'paths': [result] if isinstance(result, str) else result,
- 'type': artifact_type,
- })
+ for output_artifact in in_proto.output_artifacts:
+ for artifact_type, func in artifact_types.items():
+ if artifact_type in output_artifact.artifact_types:
+ result = func(output_dir)
+ if result:
+ generated.append(
+ {
+ "paths": [result]
+ if isinstance(result, str)
+ else result,
+ "type": artifact_type,
+ }
+ )
- return generated
+ return generated
def _CreateResponse(_input_proto, output_proto, _config):
- """Set output_proto success field on a successful Create response."""
- output_proto.success = True
+ """Set output_proto success field on a successful Create response."""
+ output_proto.success = True
@faux.success(_CreateResponse)
@faux.empty_completed_unsuccessfully_error
-@validate.require('build_target.name')
+@validate.require("build_target.name")
@validate.validation_complete
@metrics_lib.collect_metrics
-def Create(input_proto: 'image_pb2.CreateImageRequest',
- output_proto: 'image_pb2.CreateImageResult',
- _config: 'api_config.ApiConfig'):
- """Build images.
+def Create(
+ input_proto: "image_pb2.CreateImageRequest",
+ output_proto: "image_pb2.CreateImageResult",
+ _config: "api_config.ApiConfig",
+):
+ """Build images.
- Args:
- input_proto: The input message.
- output_proto: The output message.
- _config: The API call config.
- """
- board = input_proto.build_target.name
+ Args:
+ input_proto: The input message.
+ output_proto: The output message.
+ _config: The API call config.
+ """
+ board = input_proto.build_target.name
- # Build the base image if no images provided.
- to_build = input_proto.image_types or [_BASE_ID]
+ # Build the base image if no images provided.
+ to_build = input_proto.image_types or [_BASE_ID]
- image_types = _ParseImagesToCreate(to_build)
- build_config = _ParseCreateBuildConfig(input_proto)
- factory_build_config = build_config._replace(
- symlink=LOCATION_FACTORY, output_dir_suffix=LOCATION_FACTORY)
+ image_types = _ParseImagesToCreate(to_build)
+ build_config = _ParseCreateBuildConfig(input_proto)
+ factory_build_config = build_config._replace(
+ symlink=LOCATION_FACTORY, output_dir_suffix=LOCATION_FACTORY
+ )
- # Try building the core and factory images.
- # Sorted isn't really necessary here, but it's much easier to test.
- core_result = image.Build(
- board, sorted(image_types.core_images), config=build_config)
- logging.debug('Core Result Images: %s', core_result.images)
+ # Try building the core and factory images.
+ # Sorted isn't really necessary here, but it's much easier to test.
+ core_result = image.Build(
+ board, sorted(image_types.core_images), config=build_config
+ )
+ logging.debug("Core Result Images: %s", core_result.images)
- factory_result = image.Build(
- board, image_types.factory, config=factory_build_config)
- logging.debug('Factory Result Images: %s', factory_result.images)
+ factory_result = image.Build(
+ board, image_types.factory, config=factory_build_config
+ )
+ logging.debug("Factory Result Images: %s", factory_result.images)
- # A successful run will have no images missing, will have run at least one
- # of the two image sets, and neither attempt errored. The no error condition
- # should be redundant with no missing images, but is cheap insurance.
- all_built = core_result.all_built and factory_result.all_built
- one_ran = core_result.build_run or factory_result.build_run
- no_errors = not core_result.run_error and not factory_result.run_error
- output_proto.success = success = all_built and one_ran and no_errors
+ # A successful run will have no images missing, will have run at least one
+ # of the two image sets, and neither attempt errored. The no error condition
+ # should be redundant with no missing images, but is cheap insurance.
+ all_built = core_result.all_built and factory_result.all_built
+ one_ran = core_result.build_run or factory_result.build_run
+ no_errors = not core_result.run_error and not factory_result.run_error
+ output_proto.success = success = all_built and one_ran and no_errors
- if success:
- # Success! We need to record the images we built in the output.
- all_images = {**core_result.images, **factory_result.images}
- for img_name, img_path in all_images.items():
- _add_image_to_proto(output_proto, img_path, _IMAGE_MAPPING[img_name],
- board)
+ if success:
+ # Success! We need to record the images we built in the output.
+ all_images = {**core_result.images, **factory_result.images}
+ for img_name, img_path in all_images.items():
+ _add_image_to_proto(
+ output_proto, img_path, _IMAGE_MAPPING[img_name], board
+ )
- # Build and record VMs as necessary.
- for vm_type in image_types.vms:
- is_test = vm_type in [_TEST_VM_ID, _TEST_GUEST_VM_ID]
- img_type = _IMAGE_MAPPING[_TEST_ID if is_test else _BASE_ID]
- img_dir = core_result.images[img_type].parent.resolve()
- try:
- if vm_type in [_BASE_GUEST_VM_ID, _TEST_GUEST_VM_ID]:
- vm_path = image.CreateGuestVm(
- board, is_test=is_test, image_dir=img_dir)
- else:
- vm_path = image.CreateVm(
- board,
- disk_layout=build_config.disk_layout,
- is_test=is_test,
- image_dir=img_dir)
- except image.ImageToVmError as e:
- cros_build_lib.Die(e)
+ # Build and record VMs as necessary.
+ for vm_type in image_types.vms:
+ is_test = vm_type in [_TEST_VM_ID, _TEST_GUEST_VM_ID]
+ img_type = _IMAGE_MAPPING[_TEST_ID if is_test else _BASE_ID]
+ img_dir = core_result.images[img_type].parent.resolve()
+ try:
+ if vm_type in [_BASE_GUEST_VM_ID, _TEST_GUEST_VM_ID]:
+ vm_path = image.CreateGuestVm(
+ board, is_test=is_test, image_dir=img_dir
+ )
+ else:
+ vm_path = image.CreateVm(
+ board,
+ disk_layout=build_config.disk_layout,
+ is_test=is_test,
+ image_dir=img_dir,
+ )
+ except image.ImageToVmError as e:
+ cros_build_lib.Die(e)
- _add_image_to_proto(output_proto, vm_path, vm_type, board)
+ _add_image_to_proto(output_proto, vm_path, vm_type, board)
- # Build and record any mod images.
- for mod_type in image_types.mod_images:
- if mod_type == _RECOVERY_ID:
- base_image_path = core_result.images[constants.IMAGE_TYPE_BASE]
- # For ChromeOS Flex special case.
- if build_config.base_is_recovery:
- result = image.CopyBaseToRecovery(
- board=board, image_path=base_image_path)
- else:
- result = image.BuildRecoveryImage(
- board=board, image_path=base_image_path)
- if result.all_built:
- _add_image_to_proto(output_proto,
- result.images[_IMAGE_MAPPING[mod_type]], mod_type,
- board)
- else:
- cros_build_lib.Die('Failed to create recovery image.')
- else:
- cros_build_lib.Die('_RECOVERY_ID is the only mod_image_type.')
+ # Build and record any mod images.
+ for mod_type in image_types.mod_images:
+ if mod_type == _RECOVERY_ID:
+ base_image_path = core_result.images[constants.IMAGE_TYPE_BASE]
+ # For ChromeOS Flex special case.
+ if build_config.base_is_recovery:
+ result = image.CopyBaseToRecovery(
+ board=board, image_path=base_image_path
+ )
+ else:
+ result = image.BuildRecoveryImage(
+ board=board, image_path=base_image_path
+ )
+ if result.all_built:
+ _add_image_to_proto(
+ output_proto,
+ result.images[_IMAGE_MAPPING[mod_type]],
+ mod_type,
+ board,
+ )
+ else:
+ cros_build_lib.Die("Failed to create recovery image.")
+ else:
+ cros_build_lib.Die("_RECOVERY_ID is the only mod_image_type.")
- # Read metric events log and pipe them into output_proto.events.
- deserialize_metrics_log(output_proto.events, prefix=board)
- return controller.RETURN_CODE_SUCCESS
+ # Read metric events log and pipe them into output_proto.events.
+ deserialize_metrics_log(output_proto.events, prefix=board)
+ return controller.RETURN_CODE_SUCCESS
- else:
- # Failure, include all of the failed packages in the output when available.
- packages = core_result.failed_packages + factory_result.failed_packages
- if not packages:
- return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY
+ else:
+ # Failure, include all of the failed packages in the output when available.
+ packages = core_result.failed_packages + factory_result.failed_packages
+ if not packages:
+ return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY
- for package in packages:
- current = output_proto.failed_packages.add()
- controller_util.serialize_package_info(package, current)
+ for package in packages:
+ current = output_proto.failed_packages.add()
+ controller_util.serialize_package_info(package, current)
- return controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE
+ return controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE
def _ParseImagesToCreate(to_build: List[int]) -> ImageTypes:
- """Helper function to parse the image types to build.
+ """Helper function to parse the image types to build.
- This function expresses the dependencies of each image type and adds
- the requisite image types if they're not explicitly defined.
+ This function expresses the dependencies of each image type and adds
+ the requisite image types if they're not explicitly defined.
- Args:
- to_build: The image type list.
+ Args:
+ to_build: The image type list.
- Returns:
- ImageTypes: The parsed images to build.
- """
- image_types = set()
- vm_types = set()
- mod_image_types = set()
- for current in to_build:
- # Find out if it's a special case (vm, img mod), or just any old image.
- if current in _VM_IMAGE_MAPPING:
- vm_types.add(current)
- # Make sure we build the image required to build the VM.
- image_types.add(_VM_IMAGE_MAPPING[current])
- elif current in _MOD_IMAGE_MAPPING:
- mod_image_types.add(current)
- image_types.add(_MOD_IMAGE_MAPPING[current])
- elif current in _IMAGE_MAPPING:
- image_types.add(_IMAGE_MAPPING[current])
- else:
- # Not expected, but at least it will be obvious if this comes up.
- cros_build_lib.Die(
- "The service's known image types do not match those in image.proto. "
- 'Unknown Enum ID: %s' % current)
+ Returns:
+ ImageTypes: The parsed images to build.
+ """
+ image_types = set()
+ vm_types = set()
+ mod_image_types = set()
+ for current in to_build:
+ # Find out if it's a special case (vm, img mod), or just any old image.
+ if current in _VM_IMAGE_MAPPING:
+ vm_types.add(current)
+ # Make sure we build the image required to build the VM.
+ image_types.add(_VM_IMAGE_MAPPING[current])
+ elif current in _MOD_IMAGE_MAPPING:
+ mod_image_types.add(current)
+ image_types.add(_MOD_IMAGE_MAPPING[current])
+ elif current in _IMAGE_MAPPING:
+ image_types.add(_IMAGE_MAPPING[current])
+ else:
+ # Not expected, but at least it will be obvious if this comes up.
+ cros_build_lib.Die(
+ "The service's known image types do not match those in image.proto. "
+ "Unknown Enum ID: %s" % current
+ )
- # We can only build one type of these images at a time since image_to_vm.sh
- # uses the default path if a name is not provided.
- if vm_types.issuperset({_BASE_VM_ID, _TEST_VM_ID}):
- cros_build_lib.Die('Cannot create more than one VM.')
+ # We can only build one type of these images at a time since image_to_vm.sh
+ # uses the default path if a name is not provided.
+ if vm_types.issuperset({_BASE_VM_ID, _TEST_VM_ID}):
+ cros_build_lib.Die("Cannot create more than one VM.")
- return ImageTypes(
- images=image_types, vms=vm_types, mod_images=mod_image_types)
+ return ImageTypes(
+ images=image_types, vms=vm_types, mod_images=mod_image_types
+ )
def _ParseCreateBuildConfig(input_proto):
- """Helper to parse the image build config for Create."""
- enable_rootfs_verification = not input_proto.disable_rootfs_verification
- version = input_proto.version or None
- disk_layout = input_proto.disk_layout or None
- builder_path = input_proto.builder_path or None
- base_is_recovery = input_proto.base_is_recovery or False
- return image.BuildConfig(
- enable_rootfs_verification=enable_rootfs_verification,
- replace=True,
- version=version,
- disk_layout=disk_layout,
- builder_path=builder_path,
- symlink=LOCATION_CORE,
- base_is_recovery=base_is_recovery,
- )
+ """Helper to parse the image build config for Create."""
+ enable_rootfs_verification = not input_proto.disable_rootfs_verification
+ version = input_proto.version or None
+ disk_layout = input_proto.disk_layout or None
+ builder_path = input_proto.builder_path or None
+ base_is_recovery = input_proto.base_is_recovery or False
+ return image.BuildConfig(
+ enable_rootfs_verification=enable_rootfs_verification,
+ replace=True,
+ version=version,
+ disk_layout=disk_layout,
+ builder_path=builder_path,
+ symlink=LOCATION_CORE,
+ base_is_recovery=base_is_recovery,
+ )
def _SignerTestResponse(_input_proto, output_proto, _config):
- """Set output_proto success field on a successful SignerTest response."""
- output_proto.success = True
- return controller.RETURN_CODE_SUCCESS
+ """Set output_proto success field on a successful SignerTest response."""
+ output_proto.success = True
+ return controller.RETURN_CODE_SUCCESS
@faux.success(_SignerTestResponse)
@faux.empty_completed_unsuccessfully_error
-@validate.exists('image.path')
+@validate.exists("image.path")
@validate.validation_complete
-def SignerTest(input_proto: 'image_pb2.ImageTestRequest',
- output_proto: 'image_pb2.ImageTestRequest',
- _config: 'api_config.ApiConfig'):
- """Run image tests.
+def SignerTest(
+ input_proto: "image_pb2.ImageTestRequest",
+ output_proto: "image_pb2.ImageTestRequest",
+ _config: "api_config.ApiConfig",
+):
+ """Run image tests.
- Args:
- input_proto: The input message.
- output_proto: The output message.
- _config: The API call config.
- """
- image_path = input_proto.image.path
+ Args:
+ input_proto: The input message.
+ output_proto: The output message.
+ _config: The API call config.
+ """
+ image_path = input_proto.image.path
- result = image_lib.SecurityTest(image=image_path)
- output_proto.success = result
- if result:
- return controller.RETURN_CODE_SUCCESS
- else:
- return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY
+ result = image_lib.SecurityTest(image=image_path)
+ output_proto.success = result
+ if result:
+ return controller.RETURN_CODE_SUCCESS
+ else:
+ return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY
def _TestResponse(_input_proto, output_proto, _config):
- """Set output_proto success field on a successful Test response."""
- output_proto.success = True
- return controller.RETURN_CODE_SUCCESS
+ """Set output_proto success field on a successful Test response."""
+ output_proto.success = True
+ return controller.RETURN_CODE_SUCCESS
@faux.success(_TestResponse)
@faux.empty_completed_unsuccessfully_error
-@validate.require('build_target.name', 'result.directory')
-@validate.exists('image.path')
-def Test(input_proto: 'image_pb2.ImageTestRequest',
- output_proto: 'image_pb2.ImageTestResult',
- config: 'api_config.ApiConfig'):
- """Run image tests.
+@validate.require("build_target.name", "result.directory")
+@validate.exists("image.path")
+def Test(
+ input_proto: "image_pb2.ImageTestRequest",
+ output_proto: "image_pb2.ImageTestResult",
+ config: "api_config.ApiConfig",
+):
+ """Run image tests.
- Args:
- input_proto: The input message.
- output_proto: The output message.
- config: The API call config.
- """
- image_path = input_proto.image.path
- board = input_proto.build_target.name
- result_directory = input_proto.result.directory
+ Args:
+ input_proto: The input message.
+ output_proto: The output message.
+ config: The API call config.
+ """
+ image_path = input_proto.image.path
+ board = input_proto.build_target.name
+ result_directory = input_proto.result.directory
- if not os.path.isfile(image_path) or not image_path.endswith('.bin'):
- cros_build_lib.Die(
- 'The image.path must be an existing image file with a .bin extension.')
+ if not os.path.isfile(image_path) or not image_path.endswith(".bin"):
+ cros_build_lib.Die(
+ "The image.path must be an existing image file with a .bin extension."
+ )
- if config.validate_only:
- return controller.RETURN_CODE_VALID_INPUT
+ if config.validate_only:
+ return controller.RETURN_CODE_VALID_INPUT
- success = image.Test(board, result_directory, image_dir=image_path)
- output_proto.success = success
+ success = image.Test(board, result_directory, image_dir=image_path)
+ output_proto.success = success
- if success:
- return controller.RETURN_CODE_SUCCESS
- else:
- return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY
+ if success:
+ return controller.RETURN_CODE_SUCCESS
+ else:
+ return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY
@faux.empty_success
@faux.empty_completed_unsuccessfully_error
-@validate.require('gs_image_dir', 'sysroot.build_target.name')
-def PushImage(input_proto: 'image_pb2.PushImageRequest',
- _output_proto: 'image_pb2.PushImageResponse',
- config: 'api.config.ApiConfig'):
- """Push artifacts from the archive bucket to the release bucket.
+@validate.require("gs_image_dir", "sysroot.build_target.name")
+def PushImage(
+ input_proto: "image_pb2.PushImageRequest",
+ _output_proto: "image_pb2.PushImageResponse",
+ config: "api.config.ApiConfig",
+):
+ """Push artifacts from the archive bucket to the release bucket.
- Wraps chromite/scripts/pushimage.py.
+ Wraps chromite/scripts/pushimage.py.
- Args:
- input_proto: Input proto.
- _output_proto: Output proto.
- config: The API call config.
+ Args:
+ input_proto: Input proto.
+ _output_proto: Output proto.
+ config: The API call config.
- Returns:
- A controller return code (e.g. controller.RETURN_CODE_SUCCESS).
- """
- sign_types = []
- if input_proto.sign_types:
- for sign_type in input_proto.sign_types:
- if sign_type not in SUPPORTED_IMAGE_TYPES:
- logging.error('unsupported sign type %g', sign_type)
- return controller.RETURN_CODE_INVALID_INPUT
- sign_types.append(SUPPORTED_IMAGE_TYPES[sign_type])
+ Returns:
+ A controller return code (e.g. controller.RETURN_CODE_SUCCESS).
+ """
+ sign_types = []
+ if input_proto.sign_types:
+ for sign_type in input_proto.sign_types:
+ if sign_type not in SUPPORTED_IMAGE_TYPES:
+ logging.error("unsupported sign type %g", sign_type)
+ return controller.RETURN_CODE_INVALID_INPUT
+ sign_types.append(SUPPORTED_IMAGE_TYPES[sign_type])
- # If configured for validation only we're done here.
- if config.validate_only:
- return controller.RETURN_CODE_VALID_INPUT
+ # If configured for validation only we're done here.
+ if config.validate_only:
+ return controller.RETURN_CODE_VALID_INPUT
- kwargs = {}
- if input_proto.profile.name:
- kwargs['profile'] = input_proto.profile.name
- if input_proto.dest_bucket:
- kwargs['dest_bucket'] = input_proto.dest_bucket
- if input_proto.channels:
- kwargs['force_channels'] = [
- common_pb2.Channel.Name(channel).lower()[len('channel_'):]
- for channel in input_proto.channels
- ]
- try:
- channel_to_uris = pushimage.PushImage(
- input_proto.gs_image_dir,
- input_proto.sysroot.build_target.name,
- dry_run=input_proto.dryrun,
- sign_types=sign_types,
- **kwargs)
- except Exception:
- logging.error('PushImage failed: ', exc_info=True)
- return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY
- if channel_to_uris:
- for uris in channel_to_uris.values():
- for uri in uris:
- _output_proto.instructions.add().instructions_file_path = uri
- return controller.RETURN_CODE_SUCCESS
+ kwargs = {}
+ if input_proto.profile.name:
+ kwargs["profile"] = input_proto.profile.name
+ if input_proto.dest_bucket:
+ kwargs["dest_bucket"] = input_proto.dest_bucket
+ if input_proto.channels:
+ kwargs["force_channels"] = [
+ common_pb2.Channel.Name(channel).lower()[len("channel_") :]
+ for channel in input_proto.channels
+ ]
+ try:
+ channel_to_uris = pushimage.PushImage(
+ input_proto.gs_image_dir,
+ input_proto.sysroot.build_target.name,
+ dry_run=input_proto.dryrun,
+ sign_types=sign_types,
+ **kwargs,
+ )
+ except Exception:
+ logging.error("PushImage failed: ", exc_info=True)
+ return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY
+ if channel_to_uris:
+ for uris in channel_to_uris.values():
+ for uri in uris:
+ _output_proto.instructions.add().instructions_file_path = uri
+ return controller.RETURN_CODE_SUCCESS