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/sysroot_unittest.py b/api/controller/sysroot_unittest.py
index d41372b..62a1f49 100644
--- a/api/controller/sysroot_unittest.py
+++ b/api/controller/sysroot_unittest.py
@@ -24,824 +24,999 @@
class CreateTest(cros_test_lib.MockTestCase, api_config.ApiConfigMixin):
- """Create function tests."""
+ """Create function tests."""
- def _InputProto(self,
- build_target=None,
- profile=None,
- replace=False,
- current=False,
- package_indexes=None):
- """Helper to build and input proto instance."""
- proto = sysroot_pb2.SysrootCreateRequest()
- if build_target:
- proto.build_target.name = build_target
- if profile:
- proto.profile.name = profile
- if replace:
- proto.flags.replace = replace
- if current:
- proto.flags.chroot_current = current
- if package_indexes:
- proto.package_indexes.extend(package_indexes)
+ def _InputProto(
+ self,
+ build_target=None,
+ profile=None,
+ replace=False,
+ current=False,
+ package_indexes=None,
+ ):
+ """Helper to build and input proto instance."""
+ proto = sysroot_pb2.SysrootCreateRequest()
+ if build_target:
+ proto.build_target.name = build_target
+ if profile:
+ proto.profile.name = profile
+ if replace:
+ proto.flags.replace = replace
+ if current:
+ proto.flags.chroot_current = current
+ if package_indexes:
+ proto.package_indexes.extend(package_indexes)
- return proto
+ return proto
- def _OutputProto(self):
- """Helper to build output proto instance."""
- return sysroot_pb2.SysrootCreateResponse()
+ def _OutputProto(self):
+ """Helper to build output proto instance."""
+ return sysroot_pb2.SysrootCreateResponse()
- def testValidateOnly(self):
- """Sanity check that a validate only call does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'Create')
+ def testValidateOnly(self):
+ """Sanity check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "Create")
- board = 'board'
- profile = None
- force = False
- upgrade_chroot = True
- in_proto = self._InputProto(
- build_target=board,
- profile=profile,
- replace=force,
- current=not upgrade_chroot)
- sysroot_controller.Create(in_proto, self._OutputProto(),
- self.validate_only_config)
- patch.assert_not_called()
+ board = "board"
+ profile = None
+ force = False
+ upgrade_chroot = True
+ in_proto = self._InputProto(
+ build_target=board,
+ profile=profile,
+ replace=force,
+ current=not upgrade_chroot,
+ )
+ sysroot_controller.Create(
+ in_proto, self._OutputProto(), self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Sanity check that a mock call does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'Create')
- request = self._InputProto()
- response = self._OutputProto()
+ def testMockCall(self):
+ """Sanity check that a mock call does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "Create")
+ request = self._InputProto()
+ response = self._OutputProto()
- rc = sysroot_controller.Create(request, response, self.mock_call_config)
+ rc = sysroot_controller.Create(request, response, self.mock_call_config)
- patch.assert_not_called()
- self.assertEqual(controller.RETURN_CODE_SUCCESS, rc)
+ patch.assert_not_called()
+ self.assertEqual(controller.RETURN_CODE_SUCCESS, rc)
- def testMockError(self):
- """Sanity check that a mock error does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'Create')
- request = self._InputProto()
- response = self._OutputProto()
+ def testMockError(self):
+ """Sanity check that a mock error does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "Create")
+ request = self._InputProto()
+ response = self._OutputProto()
- rc = sysroot_controller.Create(request, response, self.mock_error_config)
+ rc = sysroot_controller.Create(
+ request, response, self.mock_error_config
+ )
- patch.assert_not_called()
- self.assertEqual(controller.RETURN_CODE_UNRECOVERABLE, rc)
+ patch.assert_not_called()
+ self.assertEqual(controller.RETURN_CODE_UNRECOVERABLE, rc)
- def testArgumentValidation(self):
- """Test the input argument validation."""
- # Error when no name provided.
- in_proto = self._InputProto()
- out_proto = self._OutputProto()
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.Create(in_proto, out_proto, self.api_config)
+ def testArgumentValidation(self):
+ """Test the input argument validation."""
+ # Error when no name provided.
+ in_proto = self._InputProto()
+ out_proto = self._OutputProto()
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.Create(in_proto, out_proto, self.api_config)
- # Valid when board passed.
- result = sysroot_lib.Sysroot('/sysroot/path')
- patch = self.PatchObject(sysroot_service, 'Create', return_value=result)
- in_proto = self._InputProto('board')
- out_proto = self._OutputProto()
- sysroot_controller.Create(in_proto, out_proto, self.api_config)
- patch.assert_called_once()
+ # Valid when board passed.
+ result = sysroot_lib.Sysroot("/sysroot/path")
+ patch = self.PatchObject(sysroot_service, "Create", return_value=result)
+ in_proto = self._InputProto("board")
+ out_proto = self._OutputProto()
+ sysroot_controller.Create(in_proto, out_proto, self.api_config)
+ patch.assert_called_once()
- def testArgumentHandling(self):
- """Test the arguments get processed and passed correctly."""
- sysroot_path = '/sysroot/path'
+ def testArgumentHandling(self):
+ """Test the arguments get processed and passed correctly."""
+ sysroot_path = "/sysroot/path"
- sysroot = sysroot_lib.Sysroot(sysroot_path)
- create_patch = self.PatchObject(
- sysroot_service, 'Create', return_value=sysroot)
- rc_patch = self.PatchObject(sysroot_service, 'SetupBoardRunConfig')
+ sysroot = sysroot_lib.Sysroot(sysroot_path)
+ create_patch = self.PatchObject(
+ sysroot_service, "Create", return_value=sysroot
+ )
+ rc_patch = self.PatchObject(sysroot_service, "SetupBoardRunConfig")
- # Default values.
- board = 'board'
- profile = None
- force = False
- upgrade_chroot = True
- in_proto = self._InputProto(
- build_target=board,
- profile=profile,
- replace=force,
- current=not upgrade_chroot)
- out_proto = self._OutputProto()
- sysroot_controller.Create(in_proto, out_proto, self.api_config)
+ # Default values.
+ board = "board"
+ profile = None
+ force = False
+ upgrade_chroot = True
+ in_proto = self._InputProto(
+ build_target=board,
+ profile=profile,
+ replace=force,
+ current=not upgrade_chroot,
+ )
+ out_proto = self._OutputProto()
+ sysroot_controller.Create(in_proto, out_proto, self.api_config)
- # Default value checks.
- rc_patch.assert_called_with(
- force=force,
- upgrade_chroot=upgrade_chroot,
- package_indexes=[],
- backtrack=sysroot_controller.DEFAULT_BACKTRACK,
- )
- self.assertEqual(board, out_proto.sysroot.build_target.name)
- self.assertEqual(sysroot_path, out_proto.sysroot.path)
+ # Default value checks.
+ rc_patch.assert_called_with(
+ force=force,
+ upgrade_chroot=upgrade_chroot,
+ package_indexes=[],
+ backtrack=sysroot_controller.DEFAULT_BACKTRACK,
+ )
+ self.assertEqual(board, out_proto.sysroot.build_target.name)
+ self.assertEqual(sysroot_path, out_proto.sysroot.path)
- # Not default values.
- create_patch.reset_mock()
- board = 'board'
- profile = 'profile'
- force = True
- upgrade_chroot = False
- package_indexes = [
- common_pb2.PackageIndexInfo(
- snapshot_sha='SHA',
- snapshot_number=5,
- build_target=common_pb2.BuildTarget(name=board),
- location='LOCATION',
- profile=common_pb2.Profile(name=profile)),
- common_pb2.PackageIndexInfo(
- snapshot_sha='SHA2',
- snapshot_number=4,
- build_target=common_pb2.BuildTarget(name=board),
- location='LOCATION2',
- profile=common_pb2.Profile(name=profile))
- ]
+ # Not default values.
+ create_patch.reset_mock()
+ board = "board"
+ profile = "profile"
+ force = True
+ upgrade_chroot = False
+ package_indexes = [
+ common_pb2.PackageIndexInfo(
+ snapshot_sha="SHA",
+ snapshot_number=5,
+ build_target=common_pb2.BuildTarget(name=board),
+ location="LOCATION",
+ profile=common_pb2.Profile(name=profile),
+ ),
+ common_pb2.PackageIndexInfo(
+ snapshot_sha="SHA2",
+ snapshot_number=4,
+ build_target=common_pb2.BuildTarget(name=board),
+ location="LOCATION2",
+ profile=common_pb2.Profile(name=profile),
+ ),
+ ]
- in_proto = self._InputProto(
- build_target=board,
- profile=profile,
- replace=force,
- current=not upgrade_chroot,
- package_indexes=package_indexes)
- out_proto = self._OutputProto()
- sysroot_controller.Create(in_proto, out_proto, self.api_config)
+ in_proto = self._InputProto(
+ build_target=board,
+ profile=profile,
+ replace=force,
+ current=not upgrade_chroot,
+ package_indexes=package_indexes,
+ )
+ out_proto = self._OutputProto()
+ sysroot_controller.Create(in_proto, out_proto, self.api_config)
- # Not default value checks.
- rc_patch.assert_called_with(
- force=force,
- package_indexes=[
- binpkg.PackageIndexInfo.from_protobuf(x) for x in package_indexes
- ],
- upgrade_chroot=upgrade_chroot,
- backtrack=sysroot_controller.DEFAULT_BACKTRACK,
- )
- self.assertEqual(board, out_proto.sysroot.build_target.name)
- self.assertEqual(sysroot_path, out_proto.sysroot.path)
+ # Not default value checks.
+ rc_patch.assert_called_with(
+ force=force,
+ package_indexes=[
+ binpkg.PackageIndexInfo.from_protobuf(x)
+ for x in package_indexes
+ ],
+ upgrade_chroot=upgrade_chroot,
+ backtrack=sysroot_controller.DEFAULT_BACKTRACK,
+ )
+ self.assertEqual(board, out_proto.sysroot.build_target.name)
+ self.assertEqual(sysroot_path, out_proto.sysroot.path)
-class GenerateArchiveTest(cros_test_lib.MockTempDirTestCase,
- api_config.ApiConfigMixin):
- """GenerateArchive function tests."""
+class GenerateArchiveTest(
+ cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
+):
+ """GenerateArchive function tests."""
- def setUp(self):
- self.chroot_path = '/path/to/chroot'
- self.board = 'board'
+ def setUp(self):
+ self.chroot_path = "/path/to/chroot"
+ self.board = "board"
- def _InputProto(self, build_target=None, chroot_path=None, pkg_list=None):
- """Helper to build and input proto instance."""
- # pkg_list will be a list of category/package strings such as
- # ['virtual/target-fuzzers'].
- if pkg_list:
- package_list = []
- for pkg in pkg_list:
- pkg_string_parts = pkg.split('/')
- package_info_msg = common_pb2.PackageInfo(
- category=pkg_string_parts[0], package_name=pkg_string_parts[1])
- package_list.append(package_info_msg)
- else:
- package_list = []
+ def _InputProto(self, build_target=None, chroot_path=None, pkg_list=None):
+ """Helper to build and input proto instance."""
+ # pkg_list will be a list of category/package strings such as
+ # ['virtual/target-fuzzers'].
+ if pkg_list:
+ package_list = []
+ for pkg in pkg_list:
+ pkg_string_parts = pkg.split("/")
+ package_info_msg = common_pb2.PackageInfo(
+ category=pkg_string_parts[0],
+ package_name=pkg_string_parts[1],
+ )
+ package_list.append(package_info_msg)
+ else:
+ package_list = []
- return sysroot_pb2.SysrootGenerateArchiveRequest(
- build_target={'name': build_target},
- chroot={'path': chroot_path},
- packages=package_list)
+ return sysroot_pb2.SysrootGenerateArchiveRequest(
+ build_target={"name": build_target},
+ chroot={"path": chroot_path},
+ packages=package_list,
+ )
- def _OutputProto(self):
- """Helper to build output proto instance."""
- return sysroot_pb2.SysrootGenerateArchiveResponse()
+ def _OutputProto(self):
+ """Helper to build output proto instance."""
+ return sysroot_pb2.SysrootGenerateArchiveResponse()
- def testValidateOnly(self):
- """Sanity check that a validate only call does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'GenerateArchive')
+ def testValidateOnly(self):
+ """Sanity check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "GenerateArchive")
- in_proto = self._InputProto(
- build_target=self.board,
- chroot_path=self.chroot_path,
- pkg_list=['virtual/target-fuzzers'])
- sysroot_controller.GenerateArchive(in_proto, self._OutputProto(),
- self.validate_only_config)
- patch.assert_not_called()
+ in_proto = self._InputProto(
+ build_target=self.board,
+ chroot_path=self.chroot_path,
+ pkg_list=["virtual/target-fuzzers"],
+ )
+ sysroot_controller.GenerateArchive(
+ in_proto, self._OutputProto(), self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Sanity check that a mock call does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'GenerateArchive')
+ def testMockCall(self):
+ """Sanity check that a mock call does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "GenerateArchive")
- in_proto = self._InputProto(
- build_target=self.board,
- chroot_path=self.chroot_path,
- pkg_list=['virtual/target-fuzzers'])
- sysroot_controller.GenerateArchive(in_proto, self._OutputProto(),
- self.mock_call_config)
- patch.assert_not_called()
+ in_proto = self._InputProto(
+ build_target=self.board,
+ chroot_path=self.chroot_path,
+ pkg_list=["virtual/target-fuzzers"],
+ )
+ sysroot_controller.GenerateArchive(
+ in_proto, self._OutputProto(), self.mock_call_config
+ )
+ patch.assert_not_called()
- def testArgumentValidation(self):
- """Test the input argument validation."""
- # Error when no build target provided.
- in_proto = self._InputProto()
- out_proto = self._OutputProto()
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.GenerateArchive(in_proto, out_proto, self.api_config)
+ def testArgumentValidation(self):
+ """Test the input argument validation."""
+ # Error when no build target provided.
+ in_proto = self._InputProto()
+ out_proto = self._OutputProto()
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.GenerateArchive(
+ in_proto, out_proto, self.api_config
+ )
- # Error when packages is not specified.
- in_proto = self._InputProto(
- build_target='board', chroot_path=self.chroot_path)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.GenerateArchive(in_proto, out_proto, self.api_config)
+ # Error when packages is not specified.
+ in_proto = self._InputProto(
+ build_target="board", chroot_path=self.chroot_path
+ )
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.GenerateArchive(
+ in_proto, out_proto, self.api_config
+ )
- # Valid when board, chroot path, and package are specified.
- patch = self.PatchObject(
- sysroot_service,
- 'GenerateArchive',
- return_value='/path/to/sysroot/tar.bz')
- in_proto = self._InputProto(
- build_target='board',
- chroot_path=self.chroot_path,
- pkg_list=['virtual/target-fuzzers'])
- out_proto = self._OutputProto()
- sysroot_controller.GenerateArchive(in_proto, out_proto, self.api_config)
- patch.assert_called_once()
+ # Valid when board, chroot path, and package are specified.
+ patch = self.PatchObject(
+ sysroot_service,
+ "GenerateArchive",
+ return_value="/path/to/sysroot/tar.bz",
+ )
+ in_proto = self._InputProto(
+ build_target="board",
+ chroot_path=self.chroot_path,
+ pkg_list=["virtual/target-fuzzers"],
+ )
+ out_proto = self._OutputProto()
+ sysroot_controller.GenerateArchive(in_proto, out_proto, self.api_config)
+ patch.assert_called_once()
-class InstallToolchainTest(cros_test_lib.MockTempDirTestCase,
- api_config.ApiConfigMixin):
- """Install toolchain function tests."""
+class InstallToolchainTest(
+ cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
+):
+ """Install toolchain function tests."""
- def setUp(self):
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=True)
- # Avoid running the portageq command.
- self.PatchObject(sysroot_controller, '_LogBinhost')
- self.board = 'board'
- self.sysroot = os.path.join(self.tempdir, 'board')
- self.invalid_sysroot = os.path.join(self.tempdir, 'invalid', 'sysroot')
- osutils.SafeMakedirs(self.sysroot)
- # Set up portage log directory.
- self.target_sysroot = sysroot_lib.Sysroot(self.sysroot)
- self.portage_dir = os.path.join(self.tempdir, 'portage_logdir')
- self.PatchObject(
- sysroot_lib.Sysroot, 'portage_logdir', new=self.portage_dir)
- osutils.SafeMakedirs(self.portage_dir)
+ def setUp(self):
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)
+ # Avoid running the portageq command.
+ self.PatchObject(sysroot_controller, "_LogBinhost")
+ self.board = "board"
+ self.sysroot = os.path.join(self.tempdir, "board")
+ self.invalid_sysroot = os.path.join(self.tempdir, "invalid", "sysroot")
+ osutils.SafeMakedirs(self.sysroot)
+ # Set up portage log directory.
+ self.target_sysroot = sysroot_lib.Sysroot(self.sysroot)
+ self.portage_dir = os.path.join(self.tempdir, "portage_logdir")
+ self.PatchObject(
+ sysroot_lib.Sysroot, "portage_logdir", new=self.portage_dir
+ )
+ osutils.SafeMakedirs(self.portage_dir)
- def _InputProto(self,
- build_target=None,
- sysroot_path=None,
- compile_source=False):
- """Helper to build an input proto instance."""
- proto = sysroot_pb2.InstallToolchainRequest()
- if build_target:
- proto.sysroot.build_target.name = build_target
- if sysroot_path:
- proto.sysroot.path = sysroot_path
- if compile_source:
- proto.flags.compile_source = compile_source
+ def _InputProto(
+ self, build_target=None, sysroot_path=None, compile_source=False
+ ):
+ """Helper to build an input proto instance."""
+ proto = sysroot_pb2.InstallToolchainRequest()
+ if build_target:
+ proto.sysroot.build_target.name = build_target
+ if sysroot_path:
+ proto.sysroot.path = sysroot_path
+ if compile_source:
+ proto.flags.compile_source = compile_source
- return proto
+ return proto
- def _OutputProto(self):
- """Helper to build output proto instance."""
- return sysroot_pb2.InstallToolchainResponse()
+ def _OutputProto(self):
+ """Helper to build output proto instance."""
+ return sysroot_pb2.InstallToolchainResponse()
- def _CreatePortageLogFile(self, log_path: Union[str, os.PathLike],
- pkg_info: package_info.PackageInfo,
- timestamp: datetime.datetime):
- """Creates a log file for testing for individual packages built by Portage.
+ def _CreatePortageLogFile(
+ self,
+ log_path: Union[str, os.PathLike],
+ pkg_info: package_info.PackageInfo,
+ timestamp: datetime.datetime,
+ ):
+ """Creates a log file for testing for individual packages built by Portage.
- Args:
- log_path: The PORTAGE_LOGDIR path.
- pkg_info: Package name used to name the log file.
- timestamp: Timestamp used to name the file.
- """
- path = os.path.join(log_path,
- f'{pkg_info.category}:{pkg_info.pvr}:' \
- f'{timestamp.strftime("%Y%m%d-%H%M%S")}.log')
- osutils.WriteFile(
- path, f'Test log file for package {pkg_info.category}/'
- f'{pkg_info.package} written to {path}')
- return path
+ Args:
+ log_path: The PORTAGE_LOGDIR path.
+ pkg_info: Package name used to name the log file.
+ timestamp: Timestamp used to name the file.
+ """
+ path = os.path.join(
+ log_path,
+ f"{pkg_info.category}:{pkg_info.pvr}:"
+ f'{timestamp.strftime("%Y%m%d-%H%M%S")}.log',
+ )
+ osutils.WriteFile(
+ path,
+ f"Test log file for package {pkg_info.category}/"
+ f"{pkg_info.package} written to {path}",
+ )
+ return path
- def testValidateOnly(self):
- """Sanity check that a validate only call does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'InstallToolchain')
+ def testValidateOnly(self):
+ """Sanity check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "InstallToolchain")
- in_proto = self._InputProto(
- build_target=self.board, sysroot_path=self.sysroot)
- sysroot_controller.InstallToolchain(in_proto, self._OutputProto(),
- self.validate_only_config)
- patch.assert_not_called()
+ in_proto = self._InputProto(
+ build_target=self.board, sysroot_path=self.sysroot
+ )
+ sysroot_controller.InstallToolchain(
+ in_proto, self._OutputProto(), self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Sanity check that a mock call does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'InstallToolchain')
- request = self._InputProto()
- response = self._OutputProto()
+ def testMockCall(self):
+ """Sanity check that a mock call does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "InstallToolchain")
+ request = self._InputProto()
+ response = self._OutputProto()
- rc = sysroot_controller.InstallToolchain(request, response,
- self.mock_call_config)
+ rc = sysroot_controller.InstallToolchain(
+ request, response, self.mock_call_config
+ )
- patch.assert_not_called()
- self.assertEqual(controller.RETURN_CODE_SUCCESS, rc)
+ patch.assert_not_called()
+ self.assertEqual(controller.RETURN_CODE_SUCCESS, rc)
- def testMockError(self):
- """Sanity check that a mock error does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'InstallToolchain')
- request = self._InputProto()
- response = self._OutputProto()
+ def testMockError(self):
+ """Sanity check that a mock error does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "InstallToolchain")
+ request = self._InputProto()
+ response = self._OutputProto()
- rc = sysroot_controller.InstallToolchain(request, response,
- self.mock_error_config)
+ rc = sysroot_controller.InstallToolchain(
+ request, response, self.mock_error_config
+ )
- patch.assert_not_called()
- self.assertEqual(controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc)
- self.assertTrue(response.failed_package_data)
+ patch.assert_not_called()
+ self.assertEqual(
+ controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc
+ )
+ self.assertTrue(response.failed_package_data)
- def testArgumentValidation(self):
- """Test the argument validation."""
- # Test errors on missing inputs.
- out_proto = self._OutputProto()
- # Both missing.
- in_proto = self._InputProto()
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.InstallToolchain(in_proto, out_proto, self.api_config)
+ def testArgumentValidation(self):
+ """Test the argument validation."""
+ # Test errors on missing inputs.
+ out_proto = self._OutputProto()
+ # Both missing.
+ in_proto = self._InputProto()
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.InstallToolchain(
+ in_proto, out_proto, self.api_config
+ )
- # Sysroot path missing.
- in_proto = self._InputProto(build_target=self.board)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.InstallToolchain(in_proto, out_proto, self.api_config)
+ # Sysroot path missing.
+ in_proto = self._InputProto(build_target=self.board)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.InstallToolchain(
+ in_proto, out_proto, self.api_config
+ )
- # Build target name missing.
- in_proto = self._InputProto(sysroot_path=self.sysroot)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.InstallToolchain(in_proto, out_proto, self.api_config)
+ # Build target name missing.
+ in_proto = self._InputProto(sysroot_path=self.sysroot)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.InstallToolchain(
+ in_proto, out_proto, self.api_config
+ )
- # Both provided, but invalid sysroot path.
- in_proto = self._InputProto(
- build_target=self.board, sysroot_path=self.invalid_sysroot)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.InstallToolchain(in_proto, out_proto, self.api_config)
+ # Both provided, but invalid sysroot path.
+ in_proto = self._InputProto(
+ build_target=self.board, sysroot_path=self.invalid_sysroot
+ )
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.InstallToolchain(
+ in_proto, out_proto, self.api_config
+ )
- def testSuccessOutputHandling(self):
- """Test the output is processed and recorded correctly."""
- self.PatchObject(sysroot_service, 'InstallToolchain')
- out_proto = self._OutputProto()
- in_proto = self._InputProto(
- build_target=self.board, sysroot_path=self.sysroot)
+ def testSuccessOutputHandling(self):
+ """Test the output is processed and recorded correctly."""
+ self.PatchObject(sysroot_service, "InstallToolchain")
+ out_proto = self._OutputProto()
+ in_proto = self._InputProto(
+ build_target=self.board, sysroot_path=self.sysroot
+ )
- rc = sysroot_controller.InstallToolchain(in_proto, out_proto,
- self.api_config)
- self.assertFalse(rc)
- self.assertFalse(out_proto.failed_package_data)
+ rc = sysroot_controller.InstallToolchain(
+ in_proto, out_proto, self.api_config
+ )
+ self.assertFalse(rc)
+ self.assertFalse(out_proto.failed_package_data)
- def testErrorOutputHandling(self):
- """Test the error output is processed and recorded correctly."""
- out_proto = self._OutputProto()
- in_proto = self._InputProto(
- build_target=self.board, sysroot_path=self.sysroot)
+ def testErrorOutputHandling(self):
+ """Test the error output is processed and recorded correctly."""
+ out_proto = self._OutputProto()
+ in_proto = self._InputProto(
+ build_target=self.board, sysroot_path=self.sysroot
+ )
- err_pkgs = ['cat/pkg-1.0-r1', 'cat2/pkg2-1.0-r1']
- err_cpvs = [package_info.parse(pkg) for pkg in err_pkgs]
- expected = [('cat', 'pkg'), ('cat2', 'pkg2')]
+ err_pkgs = ["cat/pkg-1.0-r1", "cat2/pkg2-1.0-r1"]
+ err_cpvs = [package_info.parse(pkg) for pkg in err_pkgs]
+ expected = [("cat", "pkg"), ("cat2", "pkg2")]
- new_logs = {}
- for i, pkg in enumerate(err_pkgs):
- self._CreatePortageLogFile(self.portage_dir, err_cpvs[i],
- datetime.datetime(2021, 6, 9, 13, 37, 0))
- new_logs[pkg] = self._CreatePortageLogFile(
- self.portage_dir, err_cpvs[i],
- datetime.datetime(2021, 6, 9, 16, 20, 0))
+ new_logs = {}
+ for i, pkg in enumerate(err_pkgs):
+ self._CreatePortageLogFile(
+ self.portage_dir,
+ err_cpvs[i],
+ datetime.datetime(2021, 6, 9, 13, 37, 0),
+ )
+ new_logs[pkg] = self._CreatePortageLogFile(
+ self.portage_dir,
+ err_cpvs[i],
+ datetime.datetime(2021, 6, 9, 16, 20, 0),
+ )
- err = sysroot_lib.ToolchainInstallError(
- 'Error', cros_build_lib.CompletedProcess(), tc_info=err_cpvs)
- self.PatchObject(sysroot_service, 'InstallToolchain', side_effect=err)
+ err = sysroot_lib.ToolchainInstallError(
+ "Error", cros_build_lib.CompletedProcess(), tc_info=err_cpvs
+ )
+ self.PatchObject(sysroot_service, "InstallToolchain", side_effect=err)
- rc = sysroot_controller.InstallToolchain(in_proto, out_proto,
- self.api_config)
- self.assertEqual(controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc)
- self.assertTrue(out_proto.failed_package_data)
- # This needs to return 2 to indicate the available error response.
- self.assertEqual(controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc)
- for data in out_proto.failed_package_data:
- package = controller_util.deserialize_package_info(data.name)
- cat_pkg = (data.name.category, data.name.package_name)
- self.assertIn(cat_pkg, expected)
- self.assertEqual(data.log_path.path, new_logs[package.cpvr])
+ rc = sysroot_controller.InstallToolchain(
+ in_proto, out_proto, self.api_config
+ )
+ self.assertEqual(
+ controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc
+ )
+ self.assertTrue(out_proto.failed_package_data)
+ # This needs to return 2 to indicate the available error response.
+ self.assertEqual(
+ controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc
+ )
+ for data in out_proto.failed_package_data:
+ package = controller_util.deserialize_package_info(data.name)
+ cat_pkg = (data.name.category, data.name.package_name)
+ self.assertIn(cat_pkg, expected)
+ self.assertEqual(data.log_path.path, new_logs[package.cpvr])
-class InstallPackagesTest(cros_test_lib.MockTempDirTestCase,
- api_config.ApiConfigMixin):
- """InstallPackages tests."""
+class InstallPackagesTest(
+ cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
+):
+ """InstallPackages tests."""
- def setUp(self):
- self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=True)
- # Avoid running the portageq command.
- self.PatchObject(sysroot_controller, '_LogBinhost')
- self.build_target = 'board'
- self.sysroot = os.path.join(self.tempdir, 'build', 'board')
- osutils.SafeMakedirs(self.sysroot)
- # Set up portage log directory.
- self.target_sysroot = sysroot_lib.Sysroot(self.sysroot)
- self.portage_dir = os.path.join(self.tempdir, 'portage_logdir')
- self.PatchObject(
- sysroot_lib.Sysroot, 'portage_logdir', new=self.portage_dir)
- osutils.SafeMakedirs(self.portage_dir)
- # Set up goma directories.
- self.goma_dir = os.path.join(self.tempdir, 'goma_dir')
- osutils.SafeMakedirs(self.goma_dir)
- self.goma_out_dir = os.path.join(self.tempdir, 'goma_out_dir')
- osutils.SafeMakedirs(self.goma_out_dir)
- os.environ['GLOG_log_dir'] = self.goma_dir
+ def setUp(self):
+ self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)
+ # Avoid running the portageq command.
+ self.PatchObject(sysroot_controller, "_LogBinhost")
+ self.build_target = "board"
+ self.sysroot = os.path.join(self.tempdir, "build", "board")
+ osutils.SafeMakedirs(self.sysroot)
+ # Set up portage log directory.
+ self.target_sysroot = sysroot_lib.Sysroot(self.sysroot)
+ self.portage_dir = os.path.join(self.tempdir, "portage_logdir")
+ self.PatchObject(
+ sysroot_lib.Sysroot, "portage_logdir", new=self.portage_dir
+ )
+ osutils.SafeMakedirs(self.portage_dir)
+ # Set up goma directories.
+ self.goma_dir = os.path.join(self.tempdir, "goma_dir")
+ osutils.SafeMakedirs(self.goma_dir)
+ self.goma_out_dir = os.path.join(self.tempdir, "goma_out_dir")
+ osutils.SafeMakedirs(self.goma_out_dir)
+ os.environ["GLOG_log_dir"] = self.goma_dir
- def _InputProto(self,
- build_target=None,
- sysroot_path=None,
- build_source=False,
- goma_dir=None,
- goma_log_dir=None,
- goma_stats_file=None,
- goma_counterz_file=None,
- package_indexes=None,
- packages=None):
- """Helper to build an input proto instance."""
- instance = sysroot_pb2.InstallPackagesRequest()
+ def _InputProto(
+ self,
+ build_target=None,
+ sysroot_path=None,
+ build_source=False,
+ goma_dir=None,
+ goma_log_dir=None,
+ goma_stats_file=None,
+ goma_counterz_file=None,
+ package_indexes=None,
+ packages=None,
+ ):
+ """Helper to build an input proto instance."""
+ instance = sysroot_pb2.InstallPackagesRequest()
- if build_target:
- instance.sysroot.build_target.name = build_target
- if sysroot_path:
- instance.sysroot.path = sysroot_path
- if build_source:
- instance.flags.build_source = build_source
- if goma_dir:
- instance.goma_config.goma_dir = goma_dir
- if goma_log_dir:
- instance.goma_config.log_dir.dir = goma_log_dir
- if goma_stats_file:
- instance.goma_config.stats_file = goma_stats_file
- if goma_counterz_file:
- instance.goma_config.counterz_file = goma_counterz_file
- if package_indexes:
- instance.package_indexes.extend(package_indexes)
- if packages:
- for pkg in packages:
- pkg_info = package_info.parse(pkg)
- pkg_info_msg = instance.packages.add()
- controller_util.serialize_package_info(pkg_info, pkg_info_msg)
- return instance
+ if build_target:
+ instance.sysroot.build_target.name = build_target
+ if sysroot_path:
+ instance.sysroot.path = sysroot_path
+ if build_source:
+ instance.flags.build_source = build_source
+ if goma_dir:
+ instance.goma_config.goma_dir = goma_dir
+ if goma_log_dir:
+ instance.goma_config.log_dir.dir = goma_log_dir
+ if goma_stats_file:
+ instance.goma_config.stats_file = goma_stats_file
+ if goma_counterz_file:
+ instance.goma_config.counterz_file = goma_counterz_file
+ if package_indexes:
+ instance.package_indexes.extend(package_indexes)
+ if packages:
+ for pkg in packages:
+ pkg_info = package_info.parse(pkg)
+ pkg_info_msg = instance.packages.add()
+ controller_util.serialize_package_info(pkg_info, pkg_info_msg)
+ return instance
- def _OutputProto(self):
- """Helper to build an empty output proto instance."""
- return sysroot_pb2.InstallPackagesResponse()
+ def _OutputProto(self):
+ """Helper to build an empty output proto instance."""
+ return sysroot_pb2.InstallPackagesResponse()
- def _CreateGomaLogFile(self, goma_log_dir: Union[str, os.PathLike], name: str,
- timestamp: datetime.datetime):
- """Creates a log file for testing.
+ def _CreateGomaLogFile(
+ self,
+ goma_log_dir: Union[str, os.PathLike],
+ name: str,
+ timestamp: datetime.datetime,
+ ):
+ """Creates a log file for testing.
- Args:
- goma_log_dir: Directory where the file will be created.
- name: Log file 'base' name that is combined with the timestamp.
- timestamp: Timestamp that is written to the file.
- """
- path = os.path.join(
- goma_log_dir,
- '%s.host.log.INFO.%s' % (name, timestamp.strftime('%Y%m%d-%H%M%S.%f')))
- osutils.WriteFile(
- path, timestamp.strftime('Goma log file created at: %Y/%m/%d %H:%M:%S'))
+ Args:
+ goma_log_dir: Directory where the file will be created.
+ name: Log file 'base' name that is combined with the timestamp.
+ timestamp: Timestamp that is written to the file.
+ """
+ path = os.path.join(
+ goma_log_dir,
+ "%s.host.log.INFO.%s"
+ % (name, timestamp.strftime("%Y%m%d-%H%M%S.%f")),
+ )
+ osutils.WriteFile(
+ path,
+ timestamp.strftime("Goma log file created at: %Y/%m/%d %H:%M:%S"),
+ )
- def _CreatePortageLogFile(self, log_path: Union[str, os.PathLike],
- pkg_info: package_info.PackageInfo,
- timestamp: datetime.datetime):
- """Creates a log file for testing for individual packages built by Portage.
+ def _CreatePortageLogFile(
+ self,
+ log_path: Union[str, os.PathLike],
+ pkg_info: package_info.PackageInfo,
+ timestamp: datetime.datetime,
+ ):
+ """Creates a log file for testing for individual packages built by Portage.
- Args:
- log_path: The PORTAGE_LOGDIR path.
- pkg_info: Package name used to name the log file.
- timestamp: Timestamp used to name the file.
- """
- path = os.path.join(log_path,
- f'{pkg_info.category}:{pkg_info.pvr}:' \
- f'{timestamp.strftime("%Y%m%d-%H%M%S")}.log')
- osutils.WriteFile(
- path, f'Test log file for package {pkg_info.category}/'
- f'{pkg_info.package} written to {path}')
- return path
+ Args:
+ log_path: The PORTAGE_LOGDIR path.
+ pkg_info: Package name used to name the log file.
+ timestamp: Timestamp used to name the file.
+ """
+ path = os.path.join(
+ log_path,
+ f"{pkg_info.category}:{pkg_info.pvr}:"
+ f'{timestamp.strftime("%Y%m%d-%H%M%S")}.log',
+ )
+ osutils.WriteFile(
+ path,
+ f"Test log file for package {pkg_info.category}/"
+ f"{pkg_info.package} written to {path}",
+ )
+ return path
- def testValidateOnly(self):
- """Sanity check that a validate only call does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'BuildPackages')
+ def testValidateOnly(self):
+ """Sanity check that a validate only call does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "BuildPackages")
- in_proto = self._InputProto(
- build_target=self.build_target, sysroot_path=self.sysroot)
- sysroot_controller.InstallPackages(in_proto, self._OutputProto(),
- self.validate_only_config)
- patch.assert_not_called()
+ in_proto = self._InputProto(
+ build_target=self.build_target, sysroot_path=self.sysroot
+ )
+ sysroot_controller.InstallPackages(
+ in_proto, self._OutputProto(), self.validate_only_config
+ )
+ patch.assert_not_called()
- def testMockCall(self):
- """Sanity check that a mock call does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'BuildPackages')
- request = self._InputProto()
- response = self._OutputProto()
+ def testMockCall(self):
+ """Sanity check that a mock call does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "BuildPackages")
+ request = self._InputProto()
+ response = self._OutputProto()
- rc = sysroot_controller.InstallPackages(request, response,
- self.mock_call_config)
+ rc = sysroot_controller.InstallPackages(
+ request, response, self.mock_call_config
+ )
- patch.assert_not_called()
- self.assertEqual(controller.RETURN_CODE_SUCCESS, rc)
+ patch.assert_not_called()
+ self.assertEqual(controller.RETURN_CODE_SUCCESS, rc)
- def testMockError(self):
- """Sanity check that a mock error does not execute any logic."""
- patch = self.PatchObject(sysroot_service, 'BuildPackages')
- request = self._InputProto()
- response = self._OutputProto()
+ def testMockError(self):
+ """Sanity check that a mock error does not execute any logic."""
+ patch = self.PatchObject(sysroot_service, "BuildPackages")
+ request = self._InputProto()
+ response = self._OutputProto()
- rc = sysroot_controller.InstallPackages(request, response,
- self.mock_error_config)
+ rc = sysroot_controller.InstallPackages(
+ request, response, self.mock_error_config
+ )
- patch.assert_not_called()
- self.assertEqual(controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc)
- self.assertTrue(response.failed_package_data)
+ patch.assert_not_called()
+ self.assertEqual(
+ controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc
+ )
+ self.assertTrue(response.failed_package_data)
- def testArgumentValidationAllMissing(self):
- """Test missing all arguments."""
- out_proto = self._OutputProto()
- in_proto = self._InputProto()
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.InstallPackages(in_proto, out_proto, self.api_config)
+ def testArgumentValidationAllMissing(self):
+ """Test missing all arguments."""
+ out_proto = self._OutputProto()
+ in_proto = self._InputProto()
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
- def testArgumentValidationNoSysroot(self):
- """Test missing sysroot path."""
- out_proto = self._OutputProto()
- in_proto = self._InputProto(build_target=self.build_target)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.InstallPackages(in_proto, out_proto, self.api_config)
+ def testArgumentValidationNoSysroot(self):
+ """Test missing sysroot path."""
+ out_proto = self._OutputProto()
+ in_proto = self._InputProto(build_target=self.build_target)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
- def testArgumentValidationNoBuildTarget(self):
- """Test missing build target name."""
- out_proto = self._OutputProto()
- in_proto = self._InputProto(sysroot_path=self.sysroot)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.InstallPackages(in_proto, out_proto, self.api_config)
+ def testArgumentValidationNoBuildTarget(self):
+ """Test missing build target name."""
+ out_proto = self._OutputProto()
+ in_proto = self._InputProto(sysroot_path=self.sysroot)
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
- def testArgumentValidationInvalidSysroot(self):
- """Test sysroot that hasn't had the toolchain installed."""
- out_proto = self._OutputProto()
- in_proto = self._InputProto(
- build_target=self.build_target, sysroot_path=self.sysroot)
- self.PatchObject(
- sysroot_lib.Sysroot, 'IsToolchainInstalled', return_value=False)
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.InstallPackages(in_proto, out_proto, self.api_config)
+ def testArgumentValidationInvalidSysroot(self):
+ """Test sysroot that hasn't had the toolchain installed."""
+ out_proto = self._OutputProto()
+ in_proto = self._InputProto(
+ build_target=self.build_target, sysroot_path=self.sysroot
+ )
+ self.PatchObject(
+ sysroot_lib.Sysroot, "IsToolchainInstalled", return_value=False
+ )
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
- def testArgumentValidationInvalidPackage(self):
- out_proto = self._OutputProto()
- in_proto = self._InputProto(
- build_target=self.build_target,
- sysroot_path=self.sysroot,
- packages=['package-1.0.0-r2'])
- with self.assertRaises(cros_build_lib.DieSystemExit):
- sysroot_controller.InstallPackages(in_proto, out_proto, self.api_config)
+ def testArgumentValidationInvalidPackage(self):
+ out_proto = self._OutputProto()
+ in_proto = self._InputProto(
+ build_target=self.build_target,
+ sysroot_path=self.sysroot,
+ packages=["package-1.0.0-r2"],
+ )
+ with self.assertRaises(cros_build_lib.DieSystemExit):
+ sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
- def testSuccessOutputHandling(self):
- """Test successful call output handling."""
- # Prevent argument validation error.
- self.PatchObject(
- sysroot_lib.Sysroot, 'IsToolchainInstalled', return_value=True)
+ def testSuccessOutputHandling(self):
+ """Test successful call output handling."""
+ # Prevent argument validation error.
+ self.PatchObject(
+ sysroot_lib.Sysroot, "IsToolchainInstalled", return_value=True
+ )
- in_proto = self._InputProto(
- build_target=self.build_target, sysroot_path=self.sysroot)
- out_proto = self._OutputProto()
- self.PatchObject(sysroot_service, 'BuildPackages')
+ in_proto = self._InputProto(
+ build_target=self.build_target, sysroot_path=self.sysroot
+ )
+ out_proto = self._OutputProto()
+ self.PatchObject(sysroot_service, "BuildPackages")
- rc = sysroot_controller.InstallPackages(in_proto, out_proto,
- self.api_config)
- self.assertFalse(rc)
- self.assertFalse(out_proto.failed_package_data)
+ rc = sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
+ self.assertFalse(rc)
+ self.assertFalse(out_proto.failed_package_data)
- def testSuccessPackageIndexes(self):
- """Test successful call with package_indexes."""
- # Prevent argument validation error.
- self.PatchObject(
- sysroot_lib.Sysroot, 'IsToolchainInstalled', return_value=True)
- package_indexes = [
- common_pb2.PackageIndexInfo(
- snapshot_sha='SHA',
- snapshot_number=5,
- build_target=common_pb2.BuildTarget(name='board'),
- location='LOCATION',
- profile=common_pb2.Profile(name='profile')),
- common_pb2.PackageIndexInfo(
- snapshot_sha='SHA2',
- snapshot_number=4,
- build_target=common_pb2.BuildTarget(name='board'),
- location='LOCATION2',
- profile=common_pb2.Profile(name='profile'))
- ]
+ def testSuccessPackageIndexes(self):
+ """Test successful call with package_indexes."""
+ # Prevent argument validation error.
+ self.PatchObject(
+ sysroot_lib.Sysroot, "IsToolchainInstalled", return_value=True
+ )
+ package_indexes = [
+ common_pb2.PackageIndexInfo(
+ snapshot_sha="SHA",
+ snapshot_number=5,
+ build_target=common_pb2.BuildTarget(name="board"),
+ location="LOCATION",
+ profile=common_pb2.Profile(name="profile"),
+ ),
+ common_pb2.PackageIndexInfo(
+ snapshot_sha="SHA2",
+ snapshot_number=4,
+ build_target=common_pb2.BuildTarget(name="board"),
+ location="LOCATION2",
+ profile=common_pb2.Profile(name="profile"),
+ ),
+ ]
- in_proto = self._InputProto(
- build_target=self.build_target,
- sysroot_path=self.sysroot,
- package_indexes=package_indexes)
+ in_proto = self._InputProto(
+ build_target=self.build_target,
+ sysroot_path=self.sysroot,
+ package_indexes=package_indexes,
+ )
- out_proto = self._OutputProto()
- rc_patch = self.PatchObject(sysroot_service, 'BuildPackagesRunConfig')
- self.PatchObject(sysroot_service, 'BuildPackages')
+ out_proto = self._OutputProto()
+ rc_patch = self.PatchObject(sysroot_service, "BuildPackagesRunConfig")
+ self.PatchObject(sysroot_service, "BuildPackages")
- rc = sysroot_controller.InstallPackages(in_proto, out_proto,
- self.api_config)
- self.assertFalse(rc)
- rc_patch.assert_called_with(
- use_any_chrome=False,
- usepkg=True,
- install_debug_symbols=True,
- packages=[],
- package_indexes=[
- binpkg.PackageIndexInfo.from_protobuf(x) for x in package_indexes
- ],
- use_flags=[],
- use_goma=False,
- use_remoteexec=False,
- incremental_build=False,
- dryrun=False,
- backtrack=sysroot_controller.DEFAULT_BACKTRACK,
- )
+ rc = sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
+ self.assertFalse(rc)
+ rc_patch.assert_called_with(
+ use_any_chrome=False,
+ usepkg=True,
+ install_debug_symbols=True,
+ packages=[],
+ package_indexes=[
+ binpkg.PackageIndexInfo.from_protobuf(x)
+ for x in package_indexes
+ ],
+ use_flags=[],
+ use_goma=False,
+ use_remoteexec=False,
+ incremental_build=False,
+ dryrun=False,
+ backtrack=sysroot_controller.DEFAULT_BACKTRACK,
+ )
- def testSuccessWithGomaLogs(self):
- """Test successful call with goma."""
- self._CreateGomaLogFile(self.goma_dir, 'compiler_proxy',
- datetime.datetime(2018, 9, 21, 12, 0, 0))
- self._CreateGomaLogFile(self.goma_dir, 'compiler_proxy-subproc',
- datetime.datetime(2018, 9, 21, 12, 1, 0))
- self._CreateGomaLogFile(self.goma_dir, 'gomacc',
- datetime.datetime(2018, 9, 21, 12, 2, 0))
+ def testSuccessWithGomaLogs(self):
+ """Test successful call with goma."""
+ self._CreateGomaLogFile(
+ self.goma_dir,
+ "compiler_proxy",
+ datetime.datetime(2018, 9, 21, 12, 0, 0),
+ )
+ self._CreateGomaLogFile(
+ self.goma_dir,
+ "compiler_proxy-subproc",
+ datetime.datetime(2018, 9, 21, 12, 1, 0),
+ )
+ self._CreateGomaLogFile(
+ self.goma_dir, "gomacc", datetime.datetime(2018, 9, 21, 12, 2, 0)
+ )
- # Prevent argument validation error.
- self.PatchObject(
- sysroot_lib.Sysroot, 'IsToolchainInstalled', return_value=True)
+ # Prevent argument validation error.
+ self.PatchObject(
+ sysroot_lib.Sysroot, "IsToolchainInstalled", return_value=True
+ )
- in_proto = self._InputProto(
- build_target=self.build_target,
- sysroot_path=self.sysroot,
- goma_dir=self.goma_dir,
- goma_log_dir=self.goma_out_dir)
+ in_proto = self._InputProto(
+ build_target=self.build_target,
+ sysroot_path=self.sysroot,
+ goma_dir=self.goma_dir,
+ goma_log_dir=self.goma_out_dir,
+ )
- out_proto = self._OutputProto()
- self.PatchObject(sysroot_service, 'BuildPackages')
+ out_proto = self._OutputProto()
+ self.PatchObject(sysroot_service, "BuildPackages")
- rc = sysroot_controller.InstallPackages(in_proto, out_proto,
- self.api_config)
- self.assertFalse(rc)
- self.assertFalse(out_proto.failed_package_data)
- self.assertCountEqual(out_proto.goma_artifacts.log_files, [
- 'compiler_proxy-subproc.host.log.INFO.20180921-120100.000000.gz',
- 'compiler_proxy.host.log.INFO.20180921-120000.000000.gz',
- 'gomacc.host.log.INFO.20180921-120200.000000.tar.gz'
- ])
+ rc = sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
+ self.assertFalse(rc)
+ self.assertFalse(out_proto.failed_package_data)
+ self.assertCountEqual(
+ out_proto.goma_artifacts.log_files,
+ [
+ "compiler_proxy-subproc.host.log.INFO.20180921-120100.000000.gz",
+ "compiler_proxy.host.log.INFO.20180921-120000.000000.gz",
+ "gomacc.host.log.INFO.20180921-120200.000000.tar.gz",
+ ],
+ )
- def testSuccessWithGomaLogsAndStatsCounterzFiles(self):
- """Test successful call with goma including stats and counterz files."""
- self._CreateGomaLogFile(self.goma_dir, 'compiler_proxy',
- datetime.datetime(2018, 9, 21, 12, 0, 0))
- self._CreateGomaLogFile(self.goma_dir, 'compiler_proxy-subproc',
- datetime.datetime(2018, 9, 21, 12, 1, 0))
- self._CreateGomaLogFile(self.goma_dir, 'gomacc',
- datetime.datetime(2018, 9, 21, 12, 2, 0))
- # Create stats and counterz files.
- osutils.WriteFile(
- os.path.join(self.goma_dir, 'stats.binaryproto'),
- 'File: stats.binaryproto')
- osutils.WriteFile(
- os.path.join(self.goma_dir, 'counterz.binaryproto'),
- 'File: counterz.binaryproto')
+ def testSuccessWithGomaLogsAndStatsCounterzFiles(self):
+ """Test successful call with goma including stats and counterz files."""
+ self._CreateGomaLogFile(
+ self.goma_dir,
+ "compiler_proxy",
+ datetime.datetime(2018, 9, 21, 12, 0, 0),
+ )
+ self._CreateGomaLogFile(
+ self.goma_dir,
+ "compiler_proxy-subproc",
+ datetime.datetime(2018, 9, 21, 12, 1, 0),
+ )
+ self._CreateGomaLogFile(
+ self.goma_dir, "gomacc", datetime.datetime(2018, 9, 21, 12, 2, 0)
+ )
+ # Create stats and counterz files.
+ osutils.WriteFile(
+ os.path.join(self.goma_dir, "stats.binaryproto"),
+ "File: stats.binaryproto",
+ )
+ osutils.WriteFile(
+ os.path.join(self.goma_dir, "counterz.binaryproto"),
+ "File: counterz.binaryproto",
+ )
- # Prevent argument validation error.
- self.PatchObject(
- sysroot_lib.Sysroot, 'IsToolchainInstalled', return_value=True)
+ # Prevent argument validation error.
+ self.PatchObject(
+ sysroot_lib.Sysroot, "IsToolchainInstalled", return_value=True
+ )
- in_proto = self._InputProto(
- build_target=self.build_target,
- sysroot_path=self.sysroot,
- goma_dir=self.goma_dir,
- goma_log_dir=self.goma_out_dir,
- goma_stats_file='stats.binaryproto',
- goma_counterz_file='counterz.binaryproto')
+ in_proto = self._InputProto(
+ build_target=self.build_target,
+ sysroot_path=self.sysroot,
+ goma_dir=self.goma_dir,
+ goma_log_dir=self.goma_out_dir,
+ goma_stats_file="stats.binaryproto",
+ goma_counterz_file="counterz.binaryproto",
+ )
- out_proto = self._OutputProto()
- self.PatchObject(sysroot_service, 'BuildPackages')
+ out_proto = self._OutputProto()
+ self.PatchObject(sysroot_service, "BuildPackages")
- rc = sysroot_controller.InstallPackages(in_proto, out_proto,
- self.api_config)
- self.assertFalse(rc)
- self.assertFalse(out_proto.failed_package_data)
- self.assertCountEqual(out_proto.goma_artifacts.log_files, [
- 'compiler_proxy-subproc.host.log.INFO.20180921-120100.000000.gz',
- 'compiler_proxy.host.log.INFO.20180921-120000.000000.gz',
- 'gomacc.host.log.INFO.20180921-120200.000000.tar.gz'
- ])
- # Verify that the output dir has 5 files -- since there should be 3 log
- # files, the stats file, and the counterz file.
- output_files = os.listdir(self.goma_out_dir)
- self.assertCountEqual(output_files, [
- 'stats.binaryproto', 'counterz.binaryproto',
- 'compiler_proxy-subproc.host.log.INFO.20180921-120100.000000.gz',
- 'compiler_proxy.host.log.INFO.20180921-120000.000000.gz',
- 'gomacc.host.log.INFO.20180921-120200.000000.tar.gz'
- ])
- self.assertEqual(out_proto.goma_artifacts.counterz_file,
- 'counterz.binaryproto')
- self.assertEqual(out_proto.goma_artifacts.stats_file, 'stats.binaryproto')
+ rc = sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
+ self.assertFalse(rc)
+ self.assertFalse(out_proto.failed_package_data)
+ self.assertCountEqual(
+ out_proto.goma_artifacts.log_files,
+ [
+ "compiler_proxy-subproc.host.log.INFO.20180921-120100.000000.gz",
+ "compiler_proxy.host.log.INFO.20180921-120000.000000.gz",
+ "gomacc.host.log.INFO.20180921-120200.000000.tar.gz",
+ ],
+ )
+ # Verify that the output dir has 5 files -- since there should be 3 log
+ # files, the stats file, and the counterz file.
+ output_files = os.listdir(self.goma_out_dir)
+ self.assertCountEqual(
+ output_files,
+ [
+ "stats.binaryproto",
+ "counterz.binaryproto",
+ "compiler_proxy-subproc.host.log.INFO.20180921-120100.000000.gz",
+ "compiler_proxy.host.log.INFO.20180921-120000.000000.gz",
+ "gomacc.host.log.INFO.20180921-120200.000000.tar.gz",
+ ],
+ )
+ self.assertEqual(
+ out_proto.goma_artifacts.counterz_file, "counterz.binaryproto"
+ )
+ self.assertEqual(
+ out_proto.goma_artifacts.stats_file, "stats.binaryproto"
+ )
- def testFailureMissingGomaStatsCounterzFiles(self):
- """Test successful call with goma including stats and counterz files."""
- self._CreateGomaLogFile(self.goma_dir, 'compiler_proxy',
- datetime.datetime(2018, 9, 21, 12, 0, 0))
- self._CreateGomaLogFile(self.goma_dir, 'compiler_proxy-subproc',
- datetime.datetime(2018, 9, 21, 12, 1, 0))
- self._CreateGomaLogFile(self.goma_dir, 'gomacc',
- datetime.datetime(2018, 9, 21, 12, 2, 0))
- # Note that stats and counterz files are not created, but are specified in
- # the proto below.
+ def testFailureMissingGomaStatsCounterzFiles(self):
+ """Test successful call with goma including stats and counterz files."""
+ self._CreateGomaLogFile(
+ self.goma_dir,
+ "compiler_proxy",
+ datetime.datetime(2018, 9, 21, 12, 0, 0),
+ )
+ self._CreateGomaLogFile(
+ self.goma_dir,
+ "compiler_proxy-subproc",
+ datetime.datetime(2018, 9, 21, 12, 1, 0),
+ )
+ self._CreateGomaLogFile(
+ self.goma_dir, "gomacc", datetime.datetime(2018, 9, 21, 12, 2, 0)
+ )
+ # Note that stats and counterz files are not created, but are specified in
+ # the proto below.
- # Prevent argument validation error.
- self.PatchObject(
- sysroot_lib.Sysroot, 'IsToolchainInstalled', return_value=True)
+ # Prevent argument validation error.
+ self.PatchObject(
+ sysroot_lib.Sysroot, "IsToolchainInstalled", return_value=True
+ )
- in_proto = self._InputProto(
- build_target=self.build_target,
- sysroot_path=self.sysroot,
- goma_dir=self.goma_dir,
- goma_log_dir=self.goma_out_dir,
- goma_stats_file='stats.binaryproto',
- goma_counterz_file='counterz.binaryproto')
+ in_proto = self._InputProto(
+ build_target=self.build_target,
+ sysroot_path=self.sysroot,
+ goma_dir=self.goma_dir,
+ goma_log_dir=self.goma_out_dir,
+ goma_stats_file="stats.binaryproto",
+ goma_counterz_file="counterz.binaryproto",
+ )
- out_proto = self._OutputProto()
- self.PatchObject(sysroot_service, 'BuildPackages')
+ out_proto = self._OutputProto()
+ self.PatchObject(sysroot_service, "BuildPackages")
- rc = sysroot_controller.InstallPackages(in_proto, out_proto,
- self.api_config)
- self.assertFalse(rc)
- self.assertFalse(out_proto.failed_package_data)
- self.assertCountEqual(out_proto.goma_artifacts.log_files, [
- 'compiler_proxy-subproc.host.log.INFO.20180921-120100.000000.gz',
- 'compiler_proxy.host.log.INFO.20180921-120000.000000.gz',
- 'gomacc.host.log.INFO.20180921-120200.000000.tar.gz'
- ])
- self.assertFalse(out_proto.goma_artifacts.counterz_file)
- self.assertFalse(out_proto.goma_artifacts.stats_file)
+ rc = sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
+ self.assertFalse(rc)
+ self.assertFalse(out_proto.failed_package_data)
+ self.assertCountEqual(
+ out_proto.goma_artifacts.log_files,
+ [
+ "compiler_proxy-subproc.host.log.INFO.20180921-120100.000000.gz",
+ "compiler_proxy.host.log.INFO.20180921-120000.000000.gz",
+ "gomacc.host.log.INFO.20180921-120200.000000.tar.gz",
+ ],
+ )
+ self.assertFalse(out_proto.goma_artifacts.counterz_file)
+ self.assertFalse(out_proto.goma_artifacts.stats_file)
- def testFailureOutputHandling(self):
- """Test failed package handling."""
- # Prevent argument validation error.
- self.PatchObject(
- sysroot_lib.Sysroot, 'IsToolchainInstalled', return_value=True)
+ def testFailureOutputHandling(self):
+ """Test failed package handling."""
+ # Prevent argument validation error.
+ self.PatchObject(
+ sysroot_lib.Sysroot, "IsToolchainInstalled", return_value=True
+ )
- in_proto = self._InputProto(
- build_target=self.build_target, sysroot_path=self.sysroot)
- out_proto = self._OutputProto()
+ in_proto = self._InputProto(
+ build_target=self.build_target, sysroot_path=self.sysroot
+ )
+ out_proto = self._OutputProto()
- # Failed package info and expected list for verification.
- err_pkgs = ['cat/pkg-1.0-r3', 'cat2/pkg2-1.0-r1']
- err_cpvs = [package_info.parse(cpv) for cpv in err_pkgs]
- expected = [('cat', 'pkg'), ('cat2', 'pkg2')]
+ # Failed package info and expected list for verification.
+ err_pkgs = ["cat/pkg-1.0-r3", "cat2/pkg2-1.0-r1"]
+ err_cpvs = [package_info.parse(cpv) for cpv in err_pkgs]
+ expected = [("cat", "pkg"), ("cat2", "pkg2")]
- new_logs = {}
- for i, pkg in enumerate(err_pkgs):
- self._CreatePortageLogFile(self.portage_dir, err_cpvs[i],
- datetime.datetime(2021, 6, 9, 13, 37, 0))
- new_logs[pkg] = self._CreatePortageLogFile(
- self.portage_dir, err_cpvs[i],
- datetime.datetime(2021, 6, 9, 16, 20, 0))
- # Force error to be raised with the packages.
- error = sysroot_lib.PackageInstallError(
- 'Error', cros_build_lib.CompletedProcess(), packages=err_cpvs)
- self.PatchObject(sysroot_service, 'BuildPackages', side_effect=error)
+ new_logs = {}
+ for i, pkg in enumerate(err_pkgs):
+ self._CreatePortageLogFile(
+ self.portage_dir,
+ err_cpvs[i],
+ datetime.datetime(2021, 6, 9, 13, 37, 0),
+ )
+ new_logs[pkg] = self._CreatePortageLogFile(
+ self.portage_dir,
+ err_cpvs[i],
+ datetime.datetime(2021, 6, 9, 16, 20, 0),
+ )
+ # Force error to be raised with the packages.
+ error = sysroot_lib.PackageInstallError(
+ "Error", cros_build_lib.CompletedProcess(), packages=err_cpvs
+ )
+ self.PatchObject(sysroot_service, "BuildPackages", side_effect=error)
- rc = sysroot_controller.InstallPackages(in_proto, out_proto,
- self.api_config)
- # This needs to return 2 to indicate the available error response.
- self.assertEqual(controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc)
- for data in out_proto.failed_package_data:
- package = controller_util.deserialize_package_info(data.name)
- cat_pkg = (data.name.category, data.name.package_name)
- self.assertIn(cat_pkg, expected)
- self.assertEqual(data.log_path.path, new_logs[package.cpvr])
+ rc = sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
+ # This needs to return 2 to indicate the available error response.
+ self.assertEqual(
+ controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc
+ )
+ for data in out_proto.failed_package_data:
+ package = controller_util.deserialize_package_info(data.name)
+ cat_pkg = (data.name.category, data.name.package_name)
+ self.assertIn(cat_pkg, expected)
+ self.assertEqual(data.log_path.path, new_logs[package.cpvr])
- def testNoPackageFailureOutputHandling(self):
- """Test failure handling without packages to report."""
- # Prevent argument validation error.
- self.PatchObject(
- sysroot_lib.Sysroot, 'IsToolchainInstalled', return_value=True)
+ def testNoPackageFailureOutputHandling(self):
+ """Test failure handling without packages to report."""
+ # Prevent argument validation error.
+ self.PatchObject(
+ sysroot_lib.Sysroot, "IsToolchainInstalled", return_value=True
+ )
- in_proto = self._InputProto(
- build_target=self.build_target, sysroot_path=self.sysroot)
- out_proto = self._OutputProto()
+ in_proto = self._InputProto(
+ build_target=self.build_target, sysroot_path=self.sysroot
+ )
+ out_proto = self._OutputProto()
- # Force error to be raised with no packages.
- error = sysroot_lib.PackageInstallError(
- 'Error', cros_build_lib.CompletedProcess(), packages=[])
- self.PatchObject(sysroot_service, 'BuildPackages', side_effect=error)
+ # Force error to be raised with no packages.
+ error = sysroot_lib.PackageInstallError(
+ "Error", cros_build_lib.CompletedProcess(), packages=[]
+ )
+ self.PatchObject(sysroot_service, "BuildPackages", side_effect=error)
- rc = sysroot_controller.InstallPackages(in_proto, out_proto,
- self.api_config)
- # All we really care about is it's not 0 or 2 (response available), so
- # test for that rather than a specific return code.
- self.assertTrue(rc)
- self.assertNotEqual(controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE,
- rc)
+ rc = sysroot_controller.InstallPackages(
+ in_proto, out_proto, self.api_config
+ )
+ # All we really care about is it's not 0 or 2 (response available), so
+ # test for that rather than a specific return code.
+ self.assertTrue(rc)
+ self.assertNotEqual(
+ controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE, rc
+ )