| # 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. | |
| """Unittests for Binhost operations.""" | |
| import os | |
| from pathlib import Path | |
| from unittest import mock | |
| from chromite.api import api_config | |
| from chromite.api.controller import binhost | |
| from chromite.api.gen.chromite.api import binhost_pb2 | |
| from chromite.api.gen.chromiumos import common_pb2 | |
| from chromite.lib import binpkg | |
| 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 osutils | |
| from chromite.service import binhost as binhost_service | |
| class GetBinhostsTest(cros_test_lib.MockTestCase, api_config.ApiConfigMixin): | |
| """Unittests for GetBinhosts.""" | |
| def setUp(self): | |
| self.response = binhost_pb2.BinhostGetResponse() | |
| def testValidateOnly(self): | |
| """Check that a validate only call does not execute any logic.""" | |
| patch = self.PatchObject(binhost_service, "GetBinhosts") | |
| request = binhost_pb2.BinhostGetRequest() | |
| request.build_target.name = "target" | |
| binhost.GetBinhosts(request, self.response, self.validate_only_config) | |
| patch.assert_not_called() | |
| def testMockCall(self): | |
| """Test a mock call does not execute logic, returns mocked value.""" | |
| patch = self.PatchObject(binhost_service, "GetBinhosts") | |
| input_proto = binhost_pb2.BinhostGetRequest() | |
| input_proto.build_target.name = "target" | |
| binhost.GetBinhosts(input_proto, self.response, self.mock_call_config) | |
| self.assertEqual(len(self.response.binhosts), 1) | |
| self.assertEqual(self.response.binhosts[0].package_index, "Packages") | |
| patch.assert_not_called() | |
| def testGetBinhosts(self): | |
| """GetBinhosts calls service with correct args.""" | |
| # pylint: disable=line-too-long | |
| binhost_list = [ | |
| f"{constants.TRASH_BUCKET}/board/amd64-generic/paladin-R66-17.0.0-rc2/packages/", | |
| f"{constants.TRASH_BUCKET}/board/eve/paladin-R66-17.0.0-rc2/packages/", | |
| ] | |
| # pylint: enable=line-too-long | |
| get_binhost = self.PatchObject( | |
| binhost_service, "GetBinhosts", return_value=binhost_list | |
| ) | |
| input_proto = binhost_pb2.BinhostGetRequest() | |
| input_proto.build_target.name = "target" | |
| binhost.GetBinhosts(input_proto, self.response, self.api_config) | |
| self.assertEqual(len(self.response.binhosts), 2) | |
| self.assertEqual(self.response.binhosts[0].package_index, "Packages") | |
| get_binhost.assert_called_once_with(mock.ANY) | |
| class GetPrivatePrebuiltAclArgsTest( | |
| cros_test_lib.MockTestCase, api_config.ApiConfigMixin | |
| ): | |
| """Unittests for GetPrivatePrebuiltAclArgs.""" | |
| def setUp(self): | |
| self.response = binhost_pb2.AclArgsResponse() | |
| def testValidateOnly(self): | |
| """Check that a validate only call does not execute any logic.""" | |
| patch = self.PatchObject(binhost_service, "GetPrebuiltAclArgs") | |
| request = binhost_pb2.AclArgsRequest() | |
| request.build_target.name = "target" | |
| binhost.GetPrivatePrebuiltAclArgs( | |
| request, self.response, self.validate_only_config | |
| ) | |
| patch.assert_not_called() | |
| def testMockCall(self): | |
| """Test a mock call does not execute logic, returns mocked value.""" | |
| patch = self.PatchObject(binhost_service, "GetPrebuiltAclArgs") | |
| input_proto = binhost_pb2.AclArgsRequest() | |
| input_proto.build_target.name = "target" | |
| binhost.GetPrivatePrebuiltAclArgs( | |
| input_proto, self.response, self.mock_call_config | |
| ) | |
| self.assertEqual(len(self.response.args), 1) | |
| self.assertEqual(self.response.args[0].arg, "-g") | |
| self.assertEqual(self.response.args[0].value, "group1:READ") | |
| patch.assert_not_called() | |
| def testGetPrivatePrebuiltAclArgs(self): | |
| """GetPrivatePrebuildAclsArgs calls service with correct args.""" | |
| argvalue_list = [["-g", "group1:READ"]] | |
| get_binhost = self.PatchObject( | |
| binhost_service, "GetPrebuiltAclArgs", return_value=argvalue_list | |
| ) | |
| input_proto = binhost_pb2.AclArgsRequest() | |
| input_proto.build_target.name = "target" | |
| binhost.GetPrivatePrebuiltAclArgs( | |
| input_proto, self.response, self.api_config | |
| ) | |
| self.assertEqual(len(self.response.args), 1) | |
| self.assertEqual(self.response.args[0].arg, "-g") | |
| self.assertEqual(self.response.args[0].value, "group1:READ") | |
| get_binhost.assert_called_once_with(mock.ANY) | |
| class PrepareBinhostUploadsTest( | |
| cros_test_lib.MockTestCase, api_config.ApiConfigMixin | |
| ): | |
| """Unittests for PrepareBinhostUploads.""" | |
| def setUp(self): | |
| self.PatchObject( | |
| binhost_service, | |
| "GetPrebuiltsRoot", | |
| return_value="/build/target/packages", | |
| ) | |
| self.PatchObject( | |
| binhost_service, | |
| "GetPrebuiltsFiles", | |
| return_value=["foo.tbz2", "bar.tbz2"], | |
| ) | |
| self.PatchObject( | |
| binhost_service, | |
| "UpdatePackageIndex", | |
| return_value="/build/target/packages/Packages", | |
| ) | |
| self.response = binhost_pb2.PrepareBinhostUploadsResponse() | |
| def testValidateOnly(self): | |
| """Check that a validate only call does not execute any logic.""" | |
| patch = self.PatchObject(binhost_service, "GetPrebuiltsRoot") | |
| request = binhost_pb2.PrepareBinhostUploadsRequest() | |
| request.build_target.name = "target" | |
| request.uri = "gs://chromeos-prebuilt/target" | |
| rc = binhost.PrepareBinhostUploads( | |
| request, self.response, self.validate_only_config | |
| ) | |
| patch.assert_not_called() | |
| self.assertEqual(rc, 0) | |
| def testMockCall(self): | |
| """Test a mock call does not execute logic, returns mocked value.""" | |
| patch = self.PatchObject(binhost_service, "GetPrebuiltsRoot") | |
| request = binhost_pb2.PrepareBinhostUploadsRequest() | |
| request.build_target.name = "target" | |
| request.uri = "gs://chromeos-prebuilt/target" | |
| rc = binhost.PrepareBinhostUploads( | |
| request, self.response, self.mock_call_config | |
| ) | |
| self.assertEqual(self.response.uploads_dir, "/upload/directory") | |
| self.assertEqual(self.response.upload_targets[0].path, "upload_target") | |
| patch.assert_not_called() | |
| self.assertEqual(rc, 0) | |
| def testPrepareBinhostUploads(self): | |
| """PrepareBinhostUploads returns Packages and tar files.""" | |
| input_proto = binhost_pb2.PrepareBinhostUploadsRequest() | |
| input_proto.build_target.name = "target" | |
| input_proto.uri = "gs://chromeos-prebuilt/target" | |
| binhost.PrepareBinhostUploads( | |
| input_proto, self.response, self.api_config | |
| ) | |
| self.assertEqual(self.response.uploads_dir, "/build/target/packages") | |
| self.assertCountEqual( | |
| [ut.path for ut in self.response.upload_targets], | |
| ["Packages", "foo.tbz2", "bar.tbz2"], | |
| ) | |
| def testPrepareBinhostUploadsNonGsUri(self): | |
| """PrepareBinhostUploads dies when URI does not point to GS.""" | |
| input_proto = binhost_pb2.PrepareBinhostUploadsRequest() | |
| input_proto.build_target.name = "target" | |
| input_proto.uri = "https://foo.bar" | |
| with self.assertRaises(ValueError): | |
| binhost.PrepareBinhostUploads( | |
| input_proto, self.response, self.api_config | |
| ) | |
| class UpdatePackageIndexTest( | |
| cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin | |
| ): | |
| """Unit tests for BinhostService/UpdatePackageIndex.""" | |
| def setUp(self): | |
| self._original_pkg_index = binpkg.PackageIndex() | |
| self._original_pkg_index.header["A"] = "B" | |
| self._original_pkg_index.packages = [ | |
| { | |
| "CPV": "foo/bar", | |
| "KEY": "value", | |
| }, | |
| { | |
| "CPV": "cat/pkg", | |
| "KEY": "also_value", | |
| }, | |
| ] | |
| self._pkg_index_fp = os.path.join( | |
| self.tempdir, | |
| "path/to/packages/Packages", | |
| ) | |
| def _write_original_package_index(self): | |
| """Write the package index to the tempdir. | |
| Note that if an input_proto specifies location=INSIDE, then they will | |
| not be able to find the written file, since the tempdir isn't actually | |
| inside a chroot. | |
| """ | |
| osutils.Touch(self._pkg_index_fp, makedirs=True) | |
| self._original_pkg_index.WriteFile(self._pkg_index_fp) | |
| def testValidateOnly(self): | |
| """Check that a validate only call does not execute any logic.""" | |
| self._write_original_package_index() | |
| patch = self.PatchObject(binpkg.PackageIndex, "ReadFilePath") | |
| request = binhost_pb2.UpdatePackageIndexRequest( | |
| package_index_file=common_pb2.Path( | |
| path=self._pkg_index_fp, | |
| location=common_pb2.Path.Location.OUTSIDE, | |
| ), | |
| set_upload_location=True, | |
| ) | |
| response = binhost_pb2.UpdatePackageIndexResponse() | |
| binhost.UpdatePackageIndex(request, response, self.validate_only_config) | |
| patch.assert_not_called() | |
| def testMustProvideSomeCommand(self): | |
| """Test that an error is raised if no update types are specified.""" | |
| self._write_original_package_index() | |
| request = binhost_pb2.UpdatePackageIndexRequest( | |
| package_index_file=common_pb2.Path( | |
| path=self._pkg_index_fp, | |
| location=common_pb2.Path.OUTSIDE, | |
| ), | |
| uri="gs://chromeos-prebuilt/board/amd64-host/packages", | |
| ) | |
| response = binhost_pb2.UpdatePackageIndexResponse() | |
| with self.assertRaises(cros_build_lib.DieSystemExit): | |
| binhost.UpdatePackageIndex(request, response, self.api_config) | |
| def testSetUploadLocation(self): | |
| """Test setting the package upload location in the index file. | |
| This test includes correctly parsing the input uri. | |
| """ | |
| # Arrange | |
| self._write_original_package_index() | |
| # Act | |
| request = binhost_pb2.UpdatePackageIndexRequest( | |
| package_index_file=common_pb2.Path( | |
| path=self._pkg_index_fp, | |
| location=common_pb2.Path.Location.OUTSIDE, | |
| ), | |
| set_upload_location=True, | |
| uri="gs://chromeos-prebuilt/board/amd64-host/packages/", | |
| ) | |
| response = binhost_pb2.UpdatePackageIndexResponse() | |
| binhost.UpdatePackageIndex(request, response, self.api_config) | |
| # Assert | |
| new_pkg_index = binpkg.PackageIndex() | |
| new_pkg_index.ReadFilePath(self._pkg_index_fp) | |
| self.assertEqual(new_pkg_index.header["URI"], "gs://chromeos-prebuilt") | |
| self.assertDictEqual( | |
| new_pkg_index.packages[0], | |
| { | |
| "CPV": "cat/pkg", | |
| "KEY": "also_value", | |
| "PATH": "board/amd64-host/packages/cat/pkg.tbz2", | |
| }, | |
| ) | |
| self.assertDictEqual( | |
| new_pkg_index.packages[1], | |
| { | |
| "CPV": "foo/bar", | |
| "KEY": "value", | |
| "PATH": "board/amd64-host/packages/foo/bar.tbz2", | |
| }, | |
| ) | |
| class SetBinhostTest(cros_test_lib.MockTestCase, api_config.ApiConfigMixin): | |
| """Unittests for SetBinhost.""" | |
| def setUp(self): | |
| self.response = binhost_pb2.SetBinhostResponse() | |
| def testValidateOnly(self): | |
| """Check that a validate only call does not execute any logic.""" | |
| patch = self.PatchObject(binhost_service, "SetBinhost") | |
| request = binhost_pb2.SetBinhostRequest() | |
| request.build_target.name = "target" | |
| request.key = binhost_pb2.POSTSUBMIT_BINHOST | |
| request.uri = "gs://chromeos-prebuilt/target" | |
| binhost.SetBinhost(request, self.response, self.validate_only_config) | |
| patch.assert_not_called() | |
| def testMockCall(self): | |
| """Test a mock call does not execute logic, returns mocked value.""" | |
| patch = self.PatchObject(binhost_service, "SetBinhost") | |
| request = binhost_pb2.SetBinhostRequest() | |
| request.build_target.name = "target" | |
| request.key = binhost_pb2.POSTSUBMIT_BINHOST | |
| request.uri = "gs://chromeos-prebuilt/target" | |
| request.max_uris = 4 | |
| binhost.SetBinhost(request, self.response, self.mock_call_config) | |
| patch.assert_not_called() | |
| self.assertEqual(self.response.output_file, "/path/to/BINHOST.conf") | |
| def testSetBinhost(self): | |
| """SetBinhost calls service with correct args.""" | |
| set_binhost = self.PatchObject( | |
| binhost_service, "SetBinhost", return_value="/path/to/BINHOST.conf" | |
| ) | |
| input_proto = binhost_pb2.SetBinhostRequest() | |
| input_proto.build_target.name = "target" | |
| input_proto.private = True | |
| input_proto.key = binhost_pb2.POSTSUBMIT_BINHOST | |
| input_proto.uri = "gs://chromeos-prebuilt/target" | |
| input_proto.max_uris = 4 | |
| binhost.SetBinhost(input_proto, self.response, self.api_config) | |
| self.assertEqual(self.response.output_file, "/path/to/BINHOST.conf") | |
| set_binhost.assert_called_once_with( | |
| "target", | |
| "POSTSUBMIT_BINHOST", | |
| "gs://chromeos-prebuilt/target", | |
| private=True, | |
| max_uris=4, | |
| ) | |
| class GetBinhostConfPathTest( | |
| cros_test_lib.MockTestCase, api_config.ApiConfigMixin | |
| ): | |
| """Unittests for GetBinhostConfPath.""" | |
| def setUp(self): | |
| self.response = binhost_pb2.GetBinhostConfPathResponse() | |
| def testValidateOnly(self): | |
| """Check that a validate only call does not execute any logic.""" | |
| patch = self.PatchObject(binhost_service, "GetBinhostConfPath") | |
| request = binhost_pb2.GetBinhostConfPathRequest() | |
| request.build_target.name = "target" | |
| request.key = binhost_pb2.POSTSUBMIT_BINHOST | |
| binhost.GetBinhostConfPath( | |
| request, self.response, self.validate_only_config | |
| ) | |
| patch.assert_not_called() | |
| def testMockCall(self): | |
| """Test a mock call does not execute logic, returns mocked value.""" | |
| patch = self.PatchObject(binhost_service, "GetBinhostConfPath") | |
| request = binhost_pb2.GetBinhostConfPathRequest() | |
| request.build_target.name = "target" | |
| request.key = binhost_pb2.POSTSUBMIT_BINHOST | |
| binhost.GetBinhostConfPath( | |
| request, self.response, self.mock_call_config | |
| ) | |
| patch.assert_not_called() | |
| self.assertEqual(self.response.conf_path, "/path/to/BINHOST.conf") | |
| def testGetBinhostConfPath(self): | |
| """GetBinhostConfPath calls service with correct args.""" | |
| get_binhost_conf_path = self.PatchObject( | |
| binhost_service, | |
| "GetBinhostConfPath", | |
| return_value="/path/to/BINHOST.conf", | |
| ) | |
| input_proto = binhost_pb2.GetBinhostConfPathRequest() | |
| input_proto.build_target.name = "target" | |
| input_proto.private = True | |
| input_proto.key = binhost_pb2.POSTSUBMIT_BINHOST | |
| binhost.GetBinhostConfPath(input_proto, self.response, self.api_config) | |
| self.assertEqual(self.response.conf_path, "/path/to/BINHOST.conf") | |
| get_binhost_conf_path.assert_called_once_with( | |
| "target", | |
| "POSTSUBMIT_BINHOST", | |
| True, | |
| ) | |
| class RegenBuildCacheTest( | |
| cros_test_lib.MockTestCase, api_config.ApiConfigMixin | |
| ): | |
| """Unittests for RegenBuildCache.""" | |
| def setUp(self): | |
| self.response = binhost_pb2.RegenBuildCacheResponse() | |
| def testValidateOnly(self): | |
| """Check that a validate only call does not execute any logic.""" | |
| patch = self.PatchObject(binhost_service, "RegenBuildCache") | |
| request = binhost_pb2.RegenBuildCacheRequest() | |
| request.overlay_type = binhost_pb2.OVERLAYTYPE_BOTH | |
| binhost.RegenBuildCache( | |
| request, self.response, self.validate_only_config | |
| ) | |
| patch.assert_not_called() | |
| def testMockCall(self): | |
| """Test a mock call does not execute logic, returns mocked value.""" | |
| patch = self.PatchObject(binhost_service, "RegenBuildCache") | |
| request = binhost_pb2.RegenBuildCacheRequest() | |
| request.overlay_type = binhost_pb2.OVERLAYTYPE_BOTH | |
| binhost.RegenBuildCache(request, self.response, self.mock_call_config) | |
| patch.assert_not_called() | |
| self.assertEqual(len(self.response.modified_overlays), 1) | |
| self.assertEqual( | |
| self.response.modified_overlays[0].path, "/path/to/BuildCache" | |
| ) | |
| def testRegenBuildCache(self): | |
| """RegenBuildCache calls service with the correct args.""" | |
| regen_cache = self.PatchObject(binhost_service, "RegenBuildCache") | |
| input_proto = binhost_pb2.RegenBuildCacheRequest() | |
| input_proto.overlay_type = binhost_pb2.OVERLAYTYPE_BOTH | |
| binhost.RegenBuildCache(input_proto, self.response, self.api_config) | |
| regen_cache.assert_called_once_with(mock.ANY, "both") | |
| def testRequiresOverlayType(self): | |
| """RegenBuildCache dies if overlay_type not specified.""" | |
| regen_cache = self.PatchObject(binhost_service, "RegenBuildCache") | |
| input_proto = binhost_pb2.RegenBuildCacheRequest() | |
| input_proto.overlay_type = binhost_pb2.OVERLAYTYPE_UNSPECIFIED | |
| with self.assertRaises(cros_build_lib.DieSystemExit): | |
| binhost.RegenBuildCache(input_proto, self.response, self.api_config) | |
| regen_cache.assert_not_called() | |
| class PrepareChromeBinhostUploadsTest( | |
| cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin | |
| ): | |
| """Tests for BinhostService/PrepareChromeBinhostUploads.""" | |
| def setUp(self): | |
| self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False) | |
| self.create_chrome_package_index_mock = self.PatchObject( | |
| binhost_service, "CreateChromePackageIndex" | |
| ) | |
| self.chroot = chroot_lib.Chroot( | |
| path=self.tempdir / "chroot", | |
| out_path=self.tempdir / "out", | |
| ) | |
| self.sysroot_path = "build/target" | |
| self.uploads_dir = self.tempdir / "uploads_dir" | |
| self.input_proto = binhost_pb2.PrepareChromeBinhostUploadsRequest() | |
| self.input_proto.uri = "gs://chromeos-prebuilt/target" | |
| self.input_proto.chroot.path = str(self.chroot.path) | |
| self.input_proto.chroot.out_path = str(self.chroot.out_path) | |
| self.input_proto.sysroot.path = self.sysroot_path | |
| self.input_proto.uploads_dir = str(self.uploads_dir) | |
| self.response = binhost_pb2.PrepareChromeBinhostUploadsResponse() | |
| self.packages_path = Path( | |
| self.chroot.full_path(self.sysroot_path, "packages") | |
| ) | |
| self.chrome_packages_path = self.packages_path / constants.CHROME_CN | |
| osutils.Touch( | |
| self.chrome_packages_path / "chromeos-chrome-100-r1.tbz2", | |
| makedirs=True, | |
| ) | |
| osutils.Touch( | |
| self.chrome_packages_path / "chrome-icu-100-r1.tbz2", | |
| makedirs=True, | |
| ) | |
| osutils.Touch( | |
| self.chrome_packages_path / "chromeos-lacros-100-r1.tbz2", | |
| makedirs=True, | |
| ) | |
| def testValidateOnly(self): | |
| """Check that a validate only call does not execute any logic.""" | |
| binhost.PrepareChromeBinhostUploads( | |
| self.input_proto, self.response, self.validate_only_config | |
| ) | |
| self.create_chrome_package_index_mock.assert_not_called() | |
| def testMockCall(self): | |
| """Test a mock call does not execute logic, returns mocked value.""" | |
| binhost.PrepareChromeBinhostUploads( | |
| self.input_proto, self.response, self.mock_call_config | |
| ) | |
| self.assertEqual(len(self.response.upload_targets), 4) | |
| self.assertEqual(self.response.upload_targets[3].path, "Packages") | |
| self.create_chrome_package_index_mock.assert_not_called() | |
| def testChromeUpload(self): | |
| """Test uploads of Chrome prebuilts.""" | |
| expected_upload_targets = [ | |
| "chromeos-base/chromeos-chrome-100-r1.tbz2", | |
| "chromeos-base/chrome-icu-100-r1.tbz2", | |
| "chromeos-base/chromeos-lacros-100-r1.tbz2", | |
| ] | |
| self.create_chrome_package_index_mock.return_value = ( | |
| expected_upload_targets | |
| ) | |
| binhost.PrepareChromeBinhostUploads( | |
| self.input_proto, self.response, self.api_config | |
| ) | |
| self.assertCountEqual( | |
| [target.path for target in self.response.upload_targets], | |
| expected_upload_targets + ["Packages"], | |
| ) | |
| def testPrepareBinhostUploadsNonGsUri(self): | |
| """PrepareBinhostUploads dies when URI does not point to GS.""" | |
| self.input_proto.uri = "https://foo.bar" | |
| with self.assertRaises(ValueError): | |
| binhost.PrepareChromeBinhostUploads( | |
| self.input_proto, self.response, self.api_config | |
| ) |