api/controller/binhost_unittest.py - third_party/chromite - Git at Google

 # 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) -> None:
         self.response = binhost_pb2.BinhostGetResponse()

     def testValidateOnly(self) -> None:
         """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) -> None:
         """Test a mock call does not execute logic, returns mocked value."""
         patch = self.PatchObject(binhost_service, "GetBinhosts")

         request = binhost_pb2.BinhostGetRequest()
         request.build_target.name = "target"

         binhost.GetBinhosts(request, 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) -> None:
         """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
         )

         request = binhost_pb2.BinhostGetRequest()
         request.build_target.name = "target"

         binhost.GetBinhosts(request, 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) -> None:
         self.response = binhost_pb2.AclArgsResponse()

     def testValidateOnly(self) -> None:
         """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) -> None:
         """Test a mock call does not execute logic, returns mocked value."""
         patch = self.PatchObject(binhost_service, "GetPrebuiltAclArgs")

         request = binhost_pb2.AclArgsRequest()
         request.build_target.name = "target"

         binhost.GetPrivatePrebuiltAclArgs(
             request, 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) -> None:
         """GetPrivatePrebuildAclsArgs calls service with correct args."""
         argvalue_list = [["-g", "group1:READ"]]
         get_binhost = self.PatchObject(
             binhost_service, "GetPrebuiltAclArgs", return_value=argvalue_list
         )

         request = binhost_pb2.AclArgsRequest()
         request.build_target.name = "target"

         binhost.GetPrivatePrebuiltAclArgs(
             request, 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) -> None:
         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) -> None:
         """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) -> None:
         """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) -> None:
         """PrepareBinhostUploads returns Packages and tar files."""
         request = binhost_pb2.PrepareBinhostUploadsRequest()
         request.build_target.name = "target"
         request.uri = "gs://chromeos-prebuilt/target"
         binhost.PrepareBinhostUploads(request, 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) -> None:
         """PrepareBinhostUploads dies when URI does not point to GS."""
         request = binhost_pb2.PrepareBinhostUploadsRequest()
         request.build_target.name = "target"
         request.uri = "https://foo.bar"
         with self.assertRaises(ValueError):
             binhost.PrepareBinhostUploads(
                 request, self.response, self.api_config
             )


 class UpdatePackageIndexTest(
     cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
 ):
     """Unit tests for BinhostService/UpdatePackageIndex."""

     def setUp(self) -> None:
         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) -> None:
         """Write the package index to the tempdir.

         Note that if an request 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) -> None:
         """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) -> None:
         """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) -> None:
         """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) -> None:
         self.response = binhost_pb2.SetBinhostResponse()

     def testValidateOnly(self) -> None:
         """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) -> None:
         """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) -> None:
         """SetBinhost calls service with correct args."""
         set_binhost = self.PatchObject(
             binhost_service, "SetBinhost", return_value="/path/to/BINHOST.conf"
         )

         request = binhost_pb2.SetBinhostRequest()
         request.build_target.name = "target"
         request.private = True
         request.key = binhost_pb2.POSTSUBMIT_BINHOST
         request.uri = "gs://chromeos-prebuilt/target"
         request.max_uris = 4
         binhost.SetBinhost(request, 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) -> None:
         self.response = binhost_pb2.GetBinhostConfPathResponse()

     def testValidateOnly(self) -> None:
         """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) -> None:
         """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) -> None:
         """GetBinhostConfPath calls service with correct args."""
         get_binhost_conf_path = self.PatchObject(
             binhost_service,
             "GetBinhostConfPath",
             return_value="/path/to/BINHOST.conf",
         )
         request = binhost_pb2.GetBinhostConfPathRequest()
         request.build_target.name = "target"
         request.private = True
         request.key = binhost_pb2.POSTSUBMIT_BINHOST
         binhost.GetBinhostConfPath(request, 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) -> None:
         self.response = binhost_pb2.RegenBuildCacheResponse()

     def testValidateOnly(self) -> None:
         """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) -> None:
         """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) -> None:
         """RegenBuildCache calls service with the correct args."""
         regen_cache = self.PatchObject(binhost_service, "RegenBuildCache")

         request = binhost_pb2.RegenBuildCacheRequest()
         request.overlay_type = binhost_pb2.OVERLAYTYPE_BOTH

         binhost.RegenBuildCache(request, self.response, self.api_config)
         regen_cache.assert_called_once_with(mock.ANY, "both")

     def testRequiresOverlayType(self) -> None:
         """RegenBuildCache dies if overlay_type not specified."""
         regen_cache = self.PatchObject(binhost_service, "RegenBuildCache")

         request = binhost_pb2.RegenBuildCacheRequest()
         request.overlay_type = binhost_pb2.OVERLAYTYPE_UNSPECIFIED

         with self.assertRaises(cros_build_lib.DieSystemExit):
             binhost.RegenBuildCache(request, 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) -> None:
         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.request = binhost_pb2.PrepareChromeBinhostUploadsRequest()
         self.request.uri = "gs://chromeos-prebuilt/target"
         self.request.chroot.path = str(self.chroot.path)
         self.request.chroot.out_path = str(self.chroot.out_path)
         self.request.sysroot.path = self.sysroot_path
         self.request.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) -> None:
         """Check that a validate only call does not execute any logic."""
         binhost.PrepareChromeBinhostUploads(
             self.request, self.response, self.validate_only_config
         )

         self.create_chrome_package_index_mock.assert_not_called()

     def testMockCall(self) -> None:
         """Test a mock call does not execute logic, returns mocked value."""
         binhost.PrepareChromeBinhostUploads(
             self.request, 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) -> None:
         """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.request, self.response, self.api_config
         )

         self.assertCountEqual(
             [target.path for target in self.response.upload_targets],
             expected_upload_targets + ["Packages"],
         )

     def testPrepareBinhostUploadsNonGsUri(self) -> None:
         """PrepareBinhostUploads dies when URI does not point to GS."""
         self.request.uri = "https://foo.bar"

         with self.assertRaises(ValueError):
             binhost.PrepareChromeBinhostUploads(
                 self.request, self.response, self.api_config
             )
	# 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) -> None:
	self.response = binhost_pb2.BinhostGetResponse()

	def testValidateOnly(self) -> None:
	"""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) -> None:
	"""Test a mock call does not execute logic, returns mocked value."""
	patch = self.PatchObject(binhost_service, "GetBinhosts")

	request = binhost_pb2.BinhostGetRequest()
	request.build_target.name = "target"

	binhost.GetBinhosts(request, 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) -> None:
	"""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
	)

	request = binhost_pb2.BinhostGetRequest()
	request.build_target.name = "target"

	binhost.GetBinhosts(request, 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) -> None:
	self.response = binhost_pb2.AclArgsResponse()

	def testValidateOnly(self) -> None:
	"""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) -> None:
	"""Test a mock call does not execute logic, returns mocked value."""
	patch = self.PatchObject(binhost_service, "GetPrebuiltAclArgs")

	request = binhost_pb2.AclArgsRequest()
	request.build_target.name = "target"

	binhost.GetPrivatePrebuiltAclArgs(
	request, 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) -> None:
	"""GetPrivatePrebuildAclsArgs calls service with correct args."""
	argvalue_list = [["-g", "group1:READ"]]
	get_binhost = self.PatchObject(
	binhost_service, "GetPrebuiltAclArgs", return_value=argvalue_list
	)

	request = binhost_pb2.AclArgsRequest()
	request.build_target.name = "target"

	binhost.GetPrivatePrebuiltAclArgs(
	request, 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) -> None:
	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) -> None:
	"""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) -> None:
	"""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) -> None:
	"""PrepareBinhostUploads returns Packages and tar files."""
	request = binhost_pb2.PrepareBinhostUploadsRequest()
	request.build_target.name = "target"
	request.uri = "gs://chromeos-prebuilt/target"
	binhost.PrepareBinhostUploads(request, 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) -> None:
	"""PrepareBinhostUploads dies when URI does not point to GS."""
	request = binhost_pb2.PrepareBinhostUploadsRequest()
	request.build_target.name = "target"
	request.uri = "https://foo.bar"
	with self.assertRaises(ValueError):
	binhost.PrepareBinhostUploads(
	request, self.response, self.api_config
	)


	class UpdatePackageIndexTest(
	cros_test_lib.MockTempDirTestCase, api_config.ApiConfigMixin
	):
	"""Unit tests for BinhostService/UpdatePackageIndex."""

	def setUp(self) -> None:
	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) -> None:
	"""Write the package index to the tempdir.

	Note that if an request 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) -> None:
	"""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) -> None:
	"""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) -> None:
	"""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) -> None:
	self.response = binhost_pb2.SetBinhostResponse()

	def testValidateOnly(self) -> None:
	"""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) -> None:
	"""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) -> None:
	"""SetBinhost calls service with correct args."""
	set_binhost = self.PatchObject(
	binhost_service, "SetBinhost", return_value="/path/to/BINHOST.conf"
	)

	request = binhost_pb2.SetBinhostRequest()
	request.build_target.name = "target"
	request.private = True
	request.key = binhost_pb2.POSTSUBMIT_BINHOST
	request.uri = "gs://chromeos-prebuilt/target"
	request.max_uris = 4
	binhost.SetBinhost(request, 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) -> None:
	self.response = binhost_pb2.GetBinhostConfPathResponse()

	def testValidateOnly(self) -> None:
	"""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) -> None:
	"""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) -> None:
	"""GetBinhostConfPath calls service with correct args."""
	get_binhost_conf_path = self.PatchObject(
	binhost_service,
	"GetBinhostConfPath",
	return_value="/path/to/BINHOST.conf",
	)
	request = binhost_pb2.GetBinhostConfPathRequest()
	request.build_target.name = "target"
	request.private = True
	request.key = binhost_pb2.POSTSUBMIT_BINHOST
	binhost.GetBinhostConfPath(request, 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) -> None:
	self.response = binhost_pb2.RegenBuildCacheResponse()

	def testValidateOnly(self) -> None:
	"""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) -> None:
	"""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) -> None:
	"""RegenBuildCache calls service with the correct args."""
	regen_cache = self.PatchObject(binhost_service, "RegenBuildCache")

	request = binhost_pb2.RegenBuildCacheRequest()
	request.overlay_type = binhost_pb2.OVERLAYTYPE_BOTH

	binhost.RegenBuildCache(request, self.response, self.api_config)
	regen_cache.assert_called_once_with(mock.ANY, "both")

	def testRequiresOverlayType(self) -> None:
	"""RegenBuildCache dies if overlay_type not specified."""
	regen_cache = self.PatchObject(binhost_service, "RegenBuildCache")

	request = binhost_pb2.RegenBuildCacheRequest()
	request.overlay_type = binhost_pb2.OVERLAYTYPE_UNSPECIFIED

	with self.assertRaises(cros_build_lib.DieSystemExit):
	binhost.RegenBuildCache(request, 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) -> None:
	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.request = binhost_pb2.PrepareChromeBinhostUploadsRequest()
	self.request.uri = "gs://chromeos-prebuilt/target"
	self.request.chroot.path = str(self.chroot.path)
	self.request.chroot.out_path = str(self.chroot.out_path)
	self.request.sysroot.path = self.sysroot_path
	self.request.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) -> None:
	"""Check that a validate only call does not execute any logic."""
	binhost.PrepareChromeBinhostUploads(
	self.request, self.response, self.validate_only_config
	)

	self.create_chrome_package_index_mock.assert_not_called()

	def testMockCall(self) -> None:
	"""Test a mock call does not execute logic, returns mocked value."""
	binhost.PrepareChromeBinhostUploads(
	self.request, 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) -> None:
	"""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.request, self.response, self.api_config
	)

	self.assertCountEqual(
	[target.path for target in self.response.upload_targets],
	expected_upload_targets + ["Packages"],
	)

	def testPrepareBinhostUploadsNonGsUri(self) -> None:
	"""PrepareBinhostUploads dies when URI does not point to GS."""
	self.request.uri = "https://foo.bar"

	with self.assertRaises(ValueError):
	binhost.PrepareChromeBinhostUploads(
	self.request, self.response, self.api_config
	)