blob: 862bb4c1756e2b2bb21e502f9f7f5864120a8a4c [file] [log] [blame] [edit]
# Copyright 2019 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Image service tests."""
import os
from pathlib import Path
from typing import List, Optional
from unittest import mock
from chromite.api import api_config
from chromite.api import controller
from chromite.api.controller import image as image_controller
from chromite.api.gen.chromite.api import image_pb2
from chromite.api.gen.chromite.api import sysroot_pb2
from chromite.api.gen.chromiumos import common_pb2
from chromite.api.gen.chromiumos import signing_pb2
from chromite.lib import build_target_lib
from chromite.lib import chroot_lib
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import image_lib
from chromite.lib import osutils
from chromite.lib import sysroot_lib
from chromite.service import image as image_service
class CreateTest(cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin):
"""Create image tests."""
def setUp(self) -> None:
self.response = image_pb2.CreateImageResult()
def _GetRequest(
self,
board=None,
types=None,
version=None,
builder_path=None,
disable_rootfs_verification=False,
base_is_recovery=False,
):
"""Helper to build a request instance."""
return image_pb2.CreateImageRequest(
build_target={"name": board},
image_types=types,
disable_rootfs_verification=disable_rootfs_verification,
version=version,
builder_path=builder_path,
base_is_recovery=base_is_recovery,
)
def testValidateOnly(self) -> None:
"""Verify a validate-only call does not execute any logic."""
patch = self.PatchObject(image_service, "Build")
request = self._GetRequest(board="board")
image_controller.Create(
request, self.response, self.validate_only_config
)
patch.assert_not_called()
def testMockCall(self) -> None:
"""Test mock call does not execute any logic, returns mocked value."""
patch = self.PatchObject(image_service, "Build")
request = self._GetRequest(board="board")
image_controller.Create(request, self.response, self.mock_call_config)
patch.assert_not_called()
self.assertEqual(self.response.success, True)
def testMockError(self) -> None:
"""Test that mock call does not execute any logic, returns error."""
patch = self.PatchObject(image_service, "Build")
request = self._GetRequest(board="board")
rc = image_controller.Create(
request, self.response, self.mock_error_config
)
patch.assert_not_called()
self.assertEqual(controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY, rc)
def testNoBoard(self) -> None:
"""Test no board given fails."""
request = self._GetRequest()
# No board should cause it to fail.
with self.assertRaises(cros_build_lib.DieSystemExit):
image_controller.Create(request, self.response, self.api_config)
def testNoTypeSpecified(self) -> None:
"""Test the image type default."""
request = self._GetRequest(board="board")
# Failed result to avoid the success handling logic.
result = image_service.BuildResult([constants.IMAGE_TYPE_BASE])
result.return_code = 1
build_patch = self.PatchObject(
image_service, "Build", return_value=result
)
image_controller.Create(request, self.response, self.api_config)
build_patch.assert_any_call(
"board", [constants.IMAGE_TYPE_BASE], config=mock.ANY
)
def testSingleTypeSpecified(self) -> None:
"""Test it's properly using a specified type."""
request = self._GetRequest(
board="board", types=[common_pb2.IMAGE_TYPE_DEV]
)
# Failed result to avoid the success handling logic.
result = image_service.BuildResult([constants.IMAGE_TYPE_DEV])
result.return_code = 1
build_patch = self.PatchObject(
image_service, "Build", return_value=result
)
image_controller.Create(request, self.response, self.api_config)
build_patch.assert_any_call(
"board", [constants.IMAGE_TYPE_DEV], config=mock.ANY
)
def testMultipleAndImpliedTypes(self) -> None:
"""Test multiple types and implied type handling."""
# The TEST_VM type should force it to build the test image.
types = [common_pb2.IMAGE_TYPE_BASE, common_pb2.IMAGE_TYPE_TEST_VM]
expected_images = [constants.IMAGE_TYPE_BASE, constants.IMAGE_TYPE_TEST]
request = self._GetRequest(board="board", types=types)
# Failed result to avoid the success handling logic.
result = image_service.BuildResult(expected_images)
result.return_code = 1
build_patch = self.PatchObject(
image_service, "Build", return_value=result
)
image_controller.Create(request, self.response, self.api_config)
build_patch.assert_any_call("board", expected_images, config=mock.ANY)
def testRecoveryImpliedTypes(self) -> None:
"""Test implied type handling of recovery images."""
# The TEST_VM type should force it to build the test image.
types = [common_pb2.IMAGE_TYPE_RECOVERY]
request = self._GetRequest(board="board", types=types)
# Failed result to avoid the success handling logic.
result = image_service.BuildResult([])
result.return_code = 1
build_patch = self.PatchObject(
image_service, "Build", return_value=result
)
image_controller.Create(request, self.response, self.api_config)
build_patch.assert_any_call(
"board", [constants.IMAGE_TYPE_BASE], config=mock.ANY
)
def testFailedPackageHandling(self) -> None:
"""Test failed packages are populated correctly."""
result = image_service.BuildResult([])
result.return_code = 1
result.failed_packages = ["foo/bar", "cat/pkg"]
expected_packages = [("foo", "bar"), ("cat", "pkg")]
self.PatchObject(image_service, "Build", return_value=result)
request = self._GetRequest(board="board")
rc = image_controller.Create(request, self.response, self.api_config)
self.assertEqual(
controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc
)
for package in self.response.failed_packages:
self.assertIn(
(package.category, package.package_name), expected_packages
)
def testNoPackagesFailureHandling(self) -> None:
"""Test failed packages are populated correctly."""
result = image_service.BuildResult([])
result.return_code = 1
self.PatchObject(image_service, "Build", return_value=result)
request = image_pb2.CreateImageRequest()
request.build_target.name = "board"
rc = image_controller.Create(request, self.response, self.api_config)
self.assertTrue(rc)
self.assertNotEqual(
controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc
)
self.assertFalse(self.response.failed_packages)
def testFactory(self) -> None:
"""Test it's properly building factory."""
request = self._GetRequest(
board="board",
types=[
common_pb2.IMAGE_TYPE_FACTORY,
common_pb2.IMAGE_TYPE_NETBOOT,
],
)
factory_path = self.tempdir / "factory-shim"
factory_path.touch()
result = image_service.BuildResult([constants.IMAGE_TYPE_FACTORY_SHIM])
result.add_image(constants.IMAGE_TYPE_FACTORY_SHIM, factory_path)
result.return_code = 0
build_patch = self.PatchObject(
image_service, "Build", return_value=result
)
netboot_patch = self.PatchObject(image_service, "create_netboot_kernel")
image_controller.Create(request, self.response, self.api_config)
build_patch.assert_any_call(
"board", [constants.IMAGE_TYPE_FACTORY_SHIM], config=mock.ANY
)
netboot_patch.assert_any_call("board", os.path.dirname(factory_path))
def testFlexor(self) -> None:
"""Test it's properly building flexor."""
request = self._GetRequest(
board="board",
types=[
common_pb2.IMAGE_TYPE_FLEXOR_KERNEL,
],
)
flexor_path = self.tempdir / "flexor_vmlinuz"
flexor_path.touch()
result = image_service.BuildResult([constants.IMAGE_TYPE_FLEXOR_KERNEL])
result.add_image(constants.IMAGE_TYPE_FLEXOR_KERNEL, flexor_path)
result.return_code = 0
build_patch = self.PatchObject(
image_service, "Build", return_value=result
)
image_controller.Create(request, self.response, self.api_config)
build_patch.assert_any_call(
"board", [constants.IMAGE_TYPE_FLEXOR_KERNEL], config=mock.ANY
)
class GetArtifactsTest(
cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
):
"""GetArtifacts function tests."""
# pylint: disable=line-too-long
_artifact_funcs = {
common_pb2.ArtifactsByService.Image.ArtifactType.DLC_IMAGE: image_service.copy_dlc_image,
common_pb2.ArtifactsByService.Image.ArtifactType.LICENSE_CREDITS: image_service.copy_license_credits,
common_pb2.ArtifactsByService.Image.ArtifactType.FACTORY_IMAGE: image_service.create_factory_image_zip,
common_pb2.ArtifactsByService.Image.ArtifactType.STRIPPED_PACKAGES: image_service.create_stripped_packages_tar,
common_pb2.ArtifactsByService.Image.ArtifactType.IMAGE_SCRIPTS: image_service.create_image_scripts_archive,
}
# pylint: enable=line-too-long
def setUp(self) -> None:
self._mocks = {}
for artifact, func in self._artifact_funcs.items():
self._mocks[artifact] = self.PatchObject(
image_service, func.__name__
)
self.chroot = chroot_lib.Chroot(
path=self.tempdir / "chroot",
chrome_root=self.tempdir,
out_path=self.tempdir / "out",
)
board = "chell"
sysroot_path = "/build/%s" % board
self.sysroot_class = sysroot_lib.Sysroot(sysroot_path)
self.build_target = build_target_lib.BuildTarget(board)
def _InputProto(
self,
artifact_types=_artifact_funcs.keys(),
):
"""Helper to build an input proto instance."""
return common_pb2.ArtifactsByService.Image(
output_artifacts=[
common_pb2.ArtifactsByService.Image.ArtifactInfo(
artifact_types=artifact_types
)
]
)
def testNoArtifacts(self) -> None:
"""Test GetArtifacts with no artifact types."""
in_proto = self._InputProto(artifact_types=[])
image_controller.GetArtifacts(
in_proto, self.chroot, self.sysroot_class, self.build_target, ""
)
for _, patch in self._mocks.items():
patch.assert_not_called()
def testArtifactsSuccess(self) -> None:
"""Test GetArtifacts with all artifact types."""
image_controller.GetArtifacts(
self._InputProto(),
self.chroot,
self.sysroot_class,
self.build_target,
"",
)
for _, patch in self._mocks.items():
patch.assert_called_once()
def testArtifactsException(self) -> None:
"""Test with all artifact types when one type throws an exception."""
self._mocks[
common_pb2.ArtifactsByService.Image.ArtifactType.STRIPPED_PACKAGES
].side_effect = Exception("foo bar")
generated = image_controller.GetArtifacts(
self._InputProto(),
self.chroot,
self.sysroot_class,
self.build_target,
"",
)
for _, patch in self._mocks.items():
patch.assert_called_once()
found_artifact = False
for data in generated:
artifact_type = (
common_pb2.ArtifactsByService.Image.ArtifactType.Name(
data["type"]
)
)
if artifact_type == "STRIPPED_PACKAGES":
found_artifact = True
self.assertTrue(data["failed"])
self.assertEqual(data["failure_reason"], "foo bar")
self.assertTrue(found_artifact)
class RecoveryImageTest(
cros_test_lib.RunCommandTempDirTestCase, api_config.ApiConfigMixin
):
"""Recovery image tests."""
def setUp(self) -> None:
self.response = image_pb2.CreateImageResult()
self.types = [
common_pb2.IMAGE_TYPE_BASE,
common_pb2.IMAGE_TYPE_RECOVERY,
]
self.build_result = self._CreateMockBuildResult(
[common_pb2.IMAGE_TYPE_BASE]
)
self.PatchObject(
image_service,
"Build",
side_effect=[
self.build_result,
self._CreateMockBuildResult([common_pb2.IMAGE_TYPE_FACTORY]),
],
)
self.copy_image_mock = self.PatchObject(
image_service,
"CopyBaseToRecovery",
side_effect=[
self._CreateMockBuildResult([common_pb2.IMAGE_TYPE_RECOVERY]),
],
)
self.recov_image_mock = self.PatchObject(
image_service,
"BuildRecoveryImage",
side_effect=[
self._CreateMockBuildResult([common_pb2.IMAGE_TYPE_RECOVERY]),
],
)
def _GetRequest(
self,
board=None,
types=None,
version=None,
builder_path=None,
disable_rootfs_verification=False,
base_is_recovery=False,
):
"""Helper to build a request instance."""
return image_pb2.CreateImageRequest(
build_target={"name": board},
image_types=types,
disable_rootfs_verification=disable_rootfs_verification,
version=version,
builder_path=builder_path,
base_is_recovery=base_is_recovery,
)
def _CreateMockBuildResult(
self, image_types: List[int]
) -> Optional[image_service.BuildResult]:
"""Helper to create Mock `cros build-image` results.
Args:
image_types: A list of image types for which the mock BuildResult
has to be created.
Returns:
A list of mock BuildResult.
"""
image_types_names = [
image_controller.SUPPORTED_IMAGE_TYPES[x]
for x in image_types
if image_controller.SUPPORTED_IMAGE_TYPES[x]
in constants.IMAGE_TYPE_TO_NAME
]
if not image_types_names:
if (
common_pb2.IMAGE_TYPE_FACTORY in image_types
and len(image_types) == 1
):
image_types_names.append(constants.IMAGE_TYPE_FACTORY_SHIM)
else:
return None
_build_result = image_service.BuildResult(image_types_names)
_build_result.return_code = 0
for image_type in image_types_names:
test_image = (
Path(self.tempdir) / constants.IMAGE_TYPE_TO_NAME[image_type]
)
test_image.touch()
_build_result.add_image(image_type, test_image)
return _build_result
def testBaseIsRecoveryTrue(self) -> None:
"""Test that cp is called."""
request = self._GetRequest(
board="board", types=self.types, base_is_recovery=True
)
image_controller.Create(request, self.response, self.api_config)
self.copy_image_mock.assert_called_with(
board="board",
image_path=self.build_result.images[constants.IMAGE_TYPE_BASE],
)
def testBaseIsRecoveryFalse(self) -> None:
"""Test that mod_image_for_recovery.sh is called."""
request = self._GetRequest(
board="board", types=self.types, base_is_recovery=False
)
image_controller.Create(request, self.response, self.api_config)
self.recov_image_mock.assert_called_with(
board="board",
image_path=self.build_result.images[constants.IMAGE_TYPE_BASE],
)
class ImageSignerTestTest(
cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
):
"""Image signer test tests."""
def setUp(self) -> None:
self.image_path = os.path.join(self.tempdir, "image.bin")
self.result_directory = os.path.join(self.tempdir, "results")
osutils.SafeMakedirs(self.result_directory)
osutils.Touch(self.image_path)
def testValidateOnly(self) -> None:
"""Sanity check that validate-only calls don't execute any logic."""
patch = self.PatchObject(image_lib, "SecurityTest", return_value=True)
request = image_pb2.TestImageRequest()
request.image.path = self.image_path
response = image_pb2.TestImageResult()
image_controller.SignerTest(
request, response, self.validate_only_config
)
patch.assert_not_called()
def testMockCall(self) -> None:
"""Test mock call does not execute any logic, returns mocked value."""
patch = self.PatchObject(image_lib, "SecurityTest", return_value=True)
request = image_pb2.TestImageRequest()
request.image.path = self.image_path
response = image_pb2.TestImageResult()
image_controller.SignerTest(request, response, self.mock_call_config)
patch.assert_not_called()
self.assertEqual(response.success, True)
def testMockError(self) -> None:
"""Test that mock call does not execute any logic, returns error."""
patch = self.PatchObject(image_lib, "SecurityTest", return_value=True)
request = image_pb2.TestImageRequest()
request.image.path = self.image_path
response = image_pb2.TestImageResult()
rc = image_controller.SignerTest(
request, response, self.mock_error_config
)
patch.assert_not_called()
self.assertEqual(controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY, rc)
def testSignerTestNoImage(self) -> None:
"""Test function argument validation."""
request = image_pb2.TestImageRequest()
response = image_pb2.TestImageResult()
# Nothing provided.
with self.assertRaises(cros_build_lib.DieSystemExit):
image_controller.SignerTest(request, response, self.api_config)
def testSignerTestSuccess(self) -> None:
"""Test successful call handling."""
self.PatchObject(image_lib, "SecurityTest", return_value=True)
request = image_pb2.TestImageRequest()
request.image.path = self.image_path
response = image_pb2.TestImageResult()
image_controller.SignerTest(request, response, self.api_config)
def testSignerTestFailure(self) -> None:
"""Test function output tests."""
request = image_pb2.TestImageRequest()
request.image.path = self.image_path
response = image_pb2.TestImageResult()
self.PatchObject(image_lib, "SecurityTest", return_value=False)
image_controller.SignerTest(request, response, self.api_config)
self.assertFalse(response.success)
class ImageTestTest(
cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
):
"""Image test tests."""
def setUp(self) -> None:
self.image_path = os.path.join(self.tempdir, "image.bin")
self.board = "board"
self.result_directory = os.path.join(self.tempdir, "results")
osutils.SafeMakedirs(self.result_directory)
osutils.Touch(self.image_path)
def testValidateOnly(self) -> None:
"""Verify a validate-only call does not execute any logic."""
patch = self.PatchObject(image_service, "Test")
request = image_pb2.TestImageRequest()
request.image.path = self.image_path
request.build_target.name = self.board
request.result.directory = self.result_directory
response = image_pb2.TestImageResult()
image_controller.Test(request, response, self.validate_only_config)
patch.assert_not_called()
def testMockCall(self) -> None:
"""Test mock call does not execute any logic, returns mocked value."""
patch = self.PatchObject(image_service, "Test")
request = image_pb2.TestImageRequest()
request.image.path = self.image_path
request.build_target.name = self.board
request.result.directory = self.result_directory
response = image_pb2.TestImageResult()
image_controller.Test(request, response, self.mock_call_config)
patch.assert_not_called()
self.assertEqual(response.success, True)
def testMockError(self) -> None:
"""Test that mock call does not execute any logic, returns error."""
patch = self.PatchObject(image_service, "Test")
request = image_pb2.TestImageRequest()
request.image.path = self.image_path
request.build_target.name = self.board
request.result.directory = self.result_directory
response = image_pb2.TestImageResult()
rc = image_controller.Test(request, response, self.mock_error_config)
patch.assert_not_called()
self.assertEqual(controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY, rc)
def testTestArgumentValidation(self) -> None:
"""Test function argument validation tests."""
self.PatchObject(image_service, "Test", return_value=True)
request = image_pb2.TestImageRequest()
response = image_pb2.TestImageResult()
# Nothing provided.
with self.assertRaises(cros_build_lib.DieSystemExit):
image_controller.Test(request, response, self.api_config)
# Just one argument.
request.build_target.name = self.board
with self.assertRaises(cros_build_lib.DieSystemExit):
image_controller.Test(request, response, self.api_config)
# Two arguments provided.
request.result.directory = self.result_directory
with self.assertRaises(cros_build_lib.DieSystemExit):
image_controller.Test(request, response, self.api_config)
# Invalid image path.
request.image.path = "/invalid/image/path"
with self.assertRaises(cros_build_lib.DieSystemExit):
image_controller.Test(request, response, self.api_config)
# All valid arguments.
request.image.path = self.image_path
image_controller.Test(request, response, self.api_config)
def testTestOutputHandling(self) -> None:
"""Test function output tests."""
request = image_pb2.TestImageRequest()
request.image.path = self.image_path
request.build_target.name = self.board
request.result.directory = self.result_directory
response = image_pb2.TestImageResult()
self.PatchObject(image_service, "Test", return_value=True)
image_controller.Test(request, response, self.api_config)
self.assertTrue(response.success)
self.PatchObject(image_service, "Test", return_value=False)
image_controller.Test(request, response, self.api_config)
self.assertFalse(response.success)
class PushImageTest(
cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
):
"""Push image test."""
def setUp(self) -> None:
"""Set up."""
self.run_push_image_mock = self.PatchObject(
image_service, "run_push_image", return_value={}
)
def _GetRequest(
self,
gs_image_dir="gs://chromeos-image-archive/atlas-release/R89-13604.0.0",
build_target_name="atlas",
profile="foo",
sign_types=None,
dryrun=True,
channels=None,
):
return image_pb2.PushImageRequest(
gs_image_dir=gs_image_dir,
sysroot=sysroot_pb2.Sysroot(
build_target=common_pb2.BuildTarget(name=build_target_name)
),
profile=common_pb2.Profile(name=profile),
sign_types=sign_types,
dryrun=dryrun,
channels=channels,
)
def _GetResponse(self):
return image_pb2.PushImageResponse()
def testValidateOnly(self) -> None:
"""Check that a validate only call does not execute any logic."""
req = self._GetRequest(
sign_types=[
common_pb2.IMAGE_TYPE_RECOVERY,
common_pb2.IMAGE_TYPE_FACTORY,
common_pb2.IMAGE_TYPE_FIRMWARE,
common_pb2.IMAGE_TYPE_ACCESSORY_USBPD,
common_pb2.IMAGE_TYPE_ACCESSORY_RWSIG,
common_pb2.IMAGE_TYPE_BASE,
common_pb2.IMAGE_TYPE_GSC_FIRMWARE,
common_pb2.IMAGE_TYPE_HPS_FIRMWARE,
]
)
rc = image_controller.PushImage(
req, self._GetResponse(), self.validate_only_config
)
self.run_push_image_mock.assert_not_called()
self.assertEqual(rc, controller.RETURN_CODE_VALID_INPUT)
def testValidateOnlyInvalid(self) -> None:
"""Check that validate call rejects invalid sign types."""
# Pass unsupported image type.
req = self._GetRequest(sign_types=[common_pb2.IMAGE_TYPE_DLC])
with self.assertRaises(cros_build_lib.DieSystemExit):
image_controller.PushImage(
req, self._GetResponse(), self.validate_only_config
)
self.run_push_image_mock.assert_not_called()
def testMockCall(self) -> None:
"""Test mock call does not execute any logic, returns mocked value."""
rc = image_controller.PushImage(
self._GetRequest(), self._GetResponse(), self.mock_call_config
)
self.run_push_image_mock.assert_not_called()
self.assertEqual(controller.RETURN_CODE_SUCCESS, rc)
def testMockError(self) -> None:
"""Test that mock call does not execute any logic, returns error."""
rc = image_controller.PushImage(
self._GetRequest(), self._GetResponse(), self.mock_error_config
)
self.run_push_image_mock.assert_not_called()
self.assertEqual(controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY, rc)
def testNoBuildTarget(self) -> None:
"""Test no build target given fails."""
request = self._GetRequest(build_target_name="")
with self.assertRaises(cros_build_lib.DieSystemExit):
image_controller.PushImage(
request, self._GetResponse(), self.api_config
)
def testNoGsImageDir(self) -> None:
"""Test no image dir given fails."""
request = self._GetRequest(gs_image_dir="")
with self.assertRaises(cros_build_lib.DieSystemExit):
image_controller.PushImage(
request, self._GetResponse(), self.api_config
)
def testCallCorrect(self) -> None:
"""Check that a call is called with the correct parameters."""
request = self._GetRequest(
dryrun=False,
profile="profile",
sign_types=[common_pb2.IMAGE_TYPE_RECOVERY],
channels=[common_pb2.CHANNEL_DEV, common_pb2.CHANNEL_CANARY],
)
request.dest_bucket = "gs://foo"
image_controller.PushImage(
request, self._GetResponse(), self.api_config
)
expected_args = image_service.PushImageArguments(
request.gs_image_dir,
build_target_lib.BuildTarget(
request.sysroot.build_target.name, profile="profile"
),
sign_types=["recovery"],
dryrun=request.dryrun,
channels=["dev", "canary"],
destination_bucket=request.dest_bucket,
yes=True,
)
self.run_push_image_mock.assert_called_with(expected_args)
def testOutput(self) -> None:
"""Check that a call populates the response object."""
request = self._GetRequest(
profile="",
sign_types=[common_pb2.IMAGE_TYPE_RECOVERY],
channels=[common_pb2.CHANNEL_DEV, common_pb2.CHANNEL_CANARY],
)
request.dest_bucket = "gs://foo"
response = self._GetResponse()
self.run_push_image_mock.return_value = {
"dev": ["gs://dev/instr1", "gs://dev/instr2"],
"canary": ["gs://canary/instr1"],
}
image_controller.PushImage(request, response, self.api_config)
self.assertEqual(
sorted([i.instructions_file_path for i in response.instructions]),
sorted(
["gs://dev/instr1", "gs://dev/instr2", "gs://canary/instr1"]
),
)
class SignImageTest(
cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
):
"""Sign image test."""
def testValidateOnly(self) -> None:
"""Check that a validate only call does not execute any logic."""
req = image_pb2.SignImageRequest(
archive_dir=str(self.tempdir),
result_path=common_pb2.ResultPath(
path=common_pb2.Path(
path="/path/to/outside",
location=common_pb2.Path.OUTSIDE,
)
),
tmp_path="/path/to/docker/tmp",
)
resp = image_pb2.SignImageResponse()
rc = image_controller.SignImage(req, resp, self.validate_only_config)
self.assertEqual(rc, controller.RETURN_CODE_VALID_INPUT)
@mock.patch.object(image_controller.image, "SignImage")
def testSuccess(self, mock_sign_image: mock.MagicMock) -> None:
"""Check that the endpoint finishes successfully."""
docker_image = "us-docker.pkg.dev/chromeos-bot/signing/signing:16963491"
req = image_pb2.SignImageRequest(
archive_dir=str(self.tempdir),
result_path=common_pb2.ResultPath(
path=common_pb2.Path(
path="/tmp/result_path",
location=common_pb2.Path.Location.OUTSIDE,
)
),
tmp_path="/path/to/docker/tmp",
docker_image=docker_image,
)
resp = image_pb2.SignImageResponse()
build_target_signed_artifacts = signing_pb2.BuildTargetSignedArtifacts(
archive_artifacts=[
signing_pb2.ArchiveArtifacts(
input_archive_name="foo",
)
]
)
mock_sign_image.return_value = build_target_signed_artifacts
rc = image_controller.SignImage(req, resp, self.api_config)
self.assertEqual(rc, controller.RETURN_CODE_SUCCESS)
self.assertEqual(
resp,
image_pb2.SignImageResponse(
output_archive_dir=str(self.tempdir),
signed_artifacts=build_target_signed_artifacts,
),
)
mock_sign_image.assert_called_with(
mock.ANY,
str(self.tempdir),
Path("/tmp/result_path"),
"/path/to/docker/tmp",
docker_image,
)