# 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.

"""Sysroot service unittest."""

import contextlib
import datetime
import operator
import os
from pathlib import Path
import shutil
from typing import Optional, Union
from unittest import mock

from chromite.api.gen.chromiumos import prebuilts_cloud_pb2
from chromite.lib import build_target_lib
from chromite.lib import chroot_lib
from chromite.lib import constants
from chromite.lib import cpupower_helper
from chromite.lib import cros_build_lib
from chromite.lib import cros_sdk_lib
from chromite.lib import cros_test_lib
from chromite.lib import goma_lib
from chromite.lib import gs
from chromite.lib import osutils
from chromite.lib import partial_mock
from chromite.lib import portage_util
from chromite.lib import sysroot_lib
from chromite.lib.parser import package_info
from chromite.service import binhost
from chromite.service import sdk
from chromite.service import sysroot
from chromite.utils import os_util


class SetupBoardRunConfigTest(cros_test_lib.TestCase):
    """Tests for the SetupBoardRunConfig class."""

    def testGetUpdateChrootArgs(self) -> None:
        """Test the update chroot args conversion method."""
        # False/0/None tests.
        instance = sysroot.SetupBoardRunConfig(
            usepkg=False, jobs=None, update_toolchain=False, backtrack=1
        )
        args = instance.GetUpdateChrootArgs("board").GetArgList()
        self.assertIn("--backtrack=1", args)
        self.assertIn("--nousepkg", args)
        self.assertNotIn("--usepkg", args)
        self.assertFalse(any(x.startswith("--jobs") for x in args))

        # True/set values tests.
        instance = sysroot.SetupBoardRunConfig(
            usepkg=True, jobs=1, update_toolchain=True
        )
        args = instance.GetUpdateChrootArgs("board").GetArgList()
        self.assertIn("--usepkg", args)
        self.assertIn("--jobs=1", args)
        self.assertNotIn("--nousepkg", args)


class SetupBoardTest(cros_test_lib.MockTestCase):
    """Tests for SetupBoard."""

    def setUp(self) -> None:
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)

    def testFullRun(self) -> None:
        """Test a regular full run.

        This method just checks that it's trying to create the sysroot and
        install the toolchain by default.
        """
        target_sysroot = sysroot_lib.Sysroot("/build/board")
        create_mock = self.PatchObject(
            sysroot, "Create", return_value=target_sysroot
        )
        install_toolchain_mock = self.PatchObject(sysroot, "InstallToolchain")

        sysroot.SetupBoard(build_target_lib.BuildTarget("board"))

        create_mock.assert_called_once()
        install_toolchain_mock.assert_called_once()

    def testRegenConfigs(self) -> None:
        """Test the regen configs install prevention."""
        target_sysroot = sysroot_lib.Sysroot("/build/board")
        create_mock = self.PatchObject(
            sysroot, "Create", return_value=target_sysroot
        )
        install_toolchain_mock = self.PatchObject(sysroot, "InstallToolchain")

        target = build_target_lib.BuildTarget("board")
        configs = sysroot.SetupBoardRunConfig(regen_configs=True)

        sysroot.SetupBoard(target, run_configs=configs)

        # Should still try to create the sysroot, but should not try to install
        # the toolchain.
        create_mock.assert_called_once()
        install_toolchain_mock.assert_not_called()


class CreateTest(cros_test_lib.RunCommandTempDirTestCase):
    """Create function tests."""

    def setUp(self) -> None:
        # Avoid sudo password prompt for config writing.
        self.PatchObject(os_util, "is_root_user", return_value=True)

        # It has to be run inside the chroot.
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)

        # Don't write to the chroot.
        self.PatchObject(cros_sdk_lib.ChrootUpdater, "ApplyUpdates")

        # Don't test sdk.Update internals -- sdk_unittest.py handles that.
        result = sdk.UpdateResult(return_code=0)
        self.update_mock = self.PatchObject(sdk, "Update", return_value=result)

        # A board we have a sysroot for already.
        self.board = "board"
        self.sysroot_path = os.path.join(self.tempdir, "build", self.board)
        self.build_target = build_target_lib.BuildTarget(
            self.board, build_root=self.sysroot_path
        )
        # A board we don't have a sysroot for yet.
        self.unbuilt_board = "board2"
        self.unbuilt_path = os.path.join(
            self.tempdir, "build", self.unbuilt_board
        )
        self.unbuilt_target = build_target_lib.BuildTarget(
            self.unbuilt_board, build_root=self.unbuilt_path
        )

        # Create the sysroot.
        osutils.SafeMakedirs(self.sysroot_path)

    def testUpdateChroot(self) -> None:
        """Test the update_chroot related handling."""
        # Prevent it from doing anything else for this test.
        self.PatchObject(sysroot, "_CreateSysrootSkeleton")
        self.PatchObject(sysroot, "_InstallConfigs")
        self.PatchObject(sysroot, "_InstallPortageConfigs")
        self.PatchObject(
            cros_sdk_lib,
            "ChrootReadWrite",
            side_effect=contextlib.contextmanager(lambda path="": (yield)),
        )

        # Make sure we have a board we haven't setup to avoid triggering the
        # existing sysroot logic. That is entirely unrelated to the chroot
        # update.
        target = self.unbuilt_target

        # Test no update case.
        config = sysroot.SetupBoardRunConfig(upgrade_chroot=False)
        get_args_patch = self.PatchObject(config, "GetUpdateChrootArgs")

        sysroot.Create(target, config, None)

        # The update chroot args not being fetched is a
        # strong enough signal that the update wasn't run.
        get_args_patch.assert_not_called()

        # Test update case.
        config = sysroot.SetupBoardRunConfig(upgrade_chroot=True)

        sysroot.Create(target, config, None)

        self.update_mock.assert_called_once()

    def test_update_chroot_failure(self) -> None:
        """Test failure handling when update chroot fails."""
        failed_pkgs = [
            package_info.parse("foo/bar-1.2-r3"),
            package_info.parse("cat/pkg-1.2-r3"),
        ]
        result = sdk.UpdateResult(return_code=1, failed_pkgs=failed_pkgs)
        self.update_mock.return_value = result

        try:
            sysroot.Create(
                self.unbuilt_target, sysroot.SetupBoardRunConfig(), None
            )
            self.fail("Should have raised an UpdateChrootError.")
        except sysroot.UpdateChrootError as e:
            self.assertSequenceEqual(
                sorted(failed_pkgs), sorted(e.failed_packages)
            )

    def testForce(self) -> None:
        """Test the force flag."""
        # Prevent it from doing anything else for this test.
        self.PatchObject(sysroot, "_CreateSysrootSkeleton")
        self.PatchObject(sysroot, "_InstallConfigs")
        self.PatchObject(sysroot, "_InstallPortageConfigs")
        self.PatchObject(
            cros_sdk_lib,
            "ChrootReadWrite",
            side_effect=contextlib.contextmanager(lambda path="": (yield)),
        )

        delete_patch = self.PatchObject(sysroot_lib.Sysroot, "Delete")

        config = sysroot.SetupBoardRunConfig(force=False)
        sysroot.Create(self.build_target, config, None)
        delete_patch.assert_not_called()
        self.update_mock.assert_called_once()
        self.update_mock.reset_mock()

        config = sysroot.SetupBoardRunConfig(force=True)
        sysroot.Create(self.build_target, config, None)
        delete_patch.assert_called_once()
        self.update_mock.assert_called_once()


class CreateSimpleChromeSysrootTest(cros_test_lib.RunCommandTempDirTestCase):
    """Tests for CreateSimpleChromeSysroot."""

    def setUp(self) -> None:
        self.source_root = os.path.join(self.tempdir, "source_root")
        osutils.SafeMakedirs(self.source_root)
        self.PatchObject(constants, "SOURCE_ROOT", new=self.source_root)

        # Configure {chroot,out}_path.
        self.chroot_path = os.path.join(self.tempdir, "chroot_dir")
        self.out_path = self.tempdir / "out_dir"

        # Create output dir.
        self.output_dir = os.path.join(self.tempdir, "output_dir")
        osutils.SafeMakedirs(self.output_dir)

        # Create chroot and build_target objs.
        self.chroot = chroot_lib.Chroot(
            path=self.chroot_path, out_path=self.out_path
        )
        self.build_target = build_target_lib.BuildTarget("target")

        # Create the tmp dir.
        osutils.SafeMakedirs(self.chroot.tmp)

    def testCreateSimpleChromeSysroot(self) -> None:
        # Mock the artifact copy.
        tar_dest = os.path.join(self.output_dir, constants.CHROME_SYSROOT_TAR)
        self.PatchObject(shutil, "copy", return_value=tar_dest)
        # Call service, verify arguments passed to run.
        sysroot.CreateSimpleChromeSysroot(
            self.chroot, None, self.build_target, self.output_dir
        )

        self.rc.assertCommandCalled(
            [
                "cros_generate_sysroot",
                "--out-dir",
                mock.ANY,
                "--board",
                self.build_target.name,
                "--package",
                "chromeos-base/chromeos-chrome",
                "--out-file",
                "sysroot_chromeos-base_chromeos-chrome.tar.xz",
                "--deps-only",
            ],
            enter_chroot=True,
            cwd=self.source_root,
            chroot_args=mock.ANY,
            extra_env=mock.ANY,
        )


class CreateFuzzerSysrootTest(cros_test_lib.RunCommandTempDirTestCase):
    """Tests for CreateFuzzerSysroot."""

    def setUp(self) -> None:
        self.source_root = os.path.join(self.tempdir, "source_root")
        osutils.SafeMakedirs(self.source_root)
        self.PatchObject(constants, "SOURCE_ROOT", new=self.source_root)

        # Configure {chroot,out}_path.
        self.chroot_path = os.path.join(self.tempdir, "chroot_dir")
        self.out_path = self.tempdir / "out_dir"

        # Create output dir.
        self.output_dir = os.path.join(self.tempdir, "output_dir")
        osutils.SafeMakedirs(self.output_dir)

        # Create chroot and build_target objs.
        self.chroot = chroot_lib.Chroot(
            path=self.chroot_path, out_path=self.out_path
        )
        self.build_target = build_target_lib.BuildTarget("target")

        # Create the tmp dir.
        osutils.SafeMakedirs(self.chroot.tmp)

    def testCreateFuzzerSysroot(self) -> None:
        """Test the CreateFuzzerSysroot function under normal operation."""
        # Mock the artifact copy.
        tar_dest = os.path.join(self.output_dir, constants.CHROME_SYSROOT_TAR)
        self.PatchObject(shutil, "copy", return_value=tar_dest)
        # Call service, verify arguments passed to run.
        sysroot.CreateFuzzerSysroot(
            self.chroot, None, self.build_target, self.output_dir
        )

        self.rc.assertCommandCalled(
            [
                "cros_generate_sysroot",
                "--out-dir",
                mock.ANY,
                "--board",
                self.build_target.name,
                "--package",
                "virtual/target-fuzzers",
                "--out-file",
                "sysroot_virtual_target-os.tar.xz",
            ],
            enter_chroot=True,
            cwd=self.source_root,
            chroot_args=mock.ANY,
            extra_env=mock.ANY,
        )


class ArchiveChromeEbuildEnvTest(cros_test_lib.MockTempDirTestCase):
    """ArchiveChromeEbuildEnv tests."""

    def setUp(self) -> None:
        # Create the chroot and sysroot instances.
        self.chroot_path = os.path.join(self.tempdir, "chroot_dir")
        self.out_path = self.tempdir / "out_dir"
        self.chroot = chroot_lib.Chroot(
            path=self.chroot_path, out_path=self.out_path
        )
        self.sysroot_path = os.path.join(self.chroot_path, "sysroot_dir")
        self.sysroot = sysroot_lib.Sysroot(self.sysroot_path)

        # Create the output directory.
        self.output_dir = os.path.join(self.tempdir, "output_dir")
        osutils.SafeMakedirs(self.output_dir)

        # The sysroot's /var/db/pkg prefix for the chrome package directories.
        var_db_pkg = self.chroot.full_path(
            self.sysroot_path, portage_util.VDB_PATH
        )
        # Create the var/db/pkg dir so we have that much for no-chrome tests.
        osutils.SafeMakedirs(var_db_pkg)

        # Two versions of chrome to test the multiple version checks/handling.
        chrome_v1 = "%s-1.0.0-r1" % constants.CHROME_PN
        chrome_v2 = "%s-2.0.0-r1" % constants.CHROME_PN

        # Build the two chrome version paths.
        chrome_cat_dir = os.path.join(var_db_pkg, constants.CHROME_CN)
        self.chrome_v1_dir = os.path.join(chrome_cat_dir, chrome_v1)
        self.chrome_v2_dir = os.path.join(chrome_cat_dir, chrome_v2)

        # Directory tuple for verifying the result archive contents.
        self.expected_archive_contents = cros_test_lib.Directory(
            "./", "environment"
        )

        # Create a environment.bz2 file to put into folders.
        env_file = os.path.join(self.tempdir, "environment")
        osutils.Touch(env_file)
        cros_build_lib.run(["bzip2", env_file])
        self.env_bz2 = "%s.bz2" % env_file

    def _CreateChromeDir(self, path: str, populate: bool = True) -> None:
        """Setup a chrome package directory.

        Args:
            path: The full chrome package path.
            populate: Whether to include the environment bz2.
        """
        osutils.SafeMakedirs(path)
        if populate:
            shutil.copy(self.env_bz2, path)

    def testSingleChromeVersion(self) -> None:
        """Test a successful single-version run."""
        self._CreateChromeDir(self.chrome_v1_dir)

        created = sysroot.CreateChromeEbuildEnv(
            self.chroot, self.sysroot, None, self.output_dir
        )

        self.assertStartsWith(created, self.output_dir)
        cros_test_lib.VerifyTarball(created, self.expected_archive_contents)

    def testMultipleChromeVersions(self) -> None:
        """Test a successful multiple version run."""
        # Create both directories, but don't populate the v1 dir so it'll hit an
        # error if the wrong one is used.
        self._CreateChromeDir(self.chrome_v1_dir, populate=False)
        self._CreateChromeDir(self.chrome_v2_dir)

        created = sysroot.CreateChromeEbuildEnv(
            self.chroot, self.sysroot, None, self.output_dir
        )

        self.assertStartsWith(created, self.output_dir)
        cros_test_lib.VerifyTarball(created, self.expected_archive_contents)

    def testNoChrome(self) -> None:
        """Test no version of chrome present."""
        self.assertIsNone(
            sysroot.CreateChromeEbuildEnv(
                self.chroot, self.sysroot, None, self.output_dir
            )
        )


class GenerateArchiveTest(cros_test_lib.RunCommandTempDirTestCase):
    """Tests for GenerateArchive."""

    def setUp(self) -> None:
        self.chroot_path = os.path.join(self.tempdir, "chroot_dir")

    def testCreateSimpleChromeSysroot(self) -> None:
        # A board for which we will create a simple chrome sysroot.
        target = "board"
        pkg_list = ["virtual/target-fuzzers"]

        # Call service, verify arguments passed to run.
        sysroot.GenerateArchive(self.chroot_path, target, pkg_list)
        self.rc.assertCommandCalled(
            [
                "cros_generate_sysroot",
                "--out-file",
                constants.TARGET_SYSROOT_TAR,
                "--out-dir",
                mock.ANY,
                "--board",
                target,
                "--package",
                "virtual/target-fuzzers",
            ],
            cwd=constants.SOURCE_ROOT,
        )


class InstallToolchainTest(cros_test_lib.MockTempDirTestCase):
    """Tests for InstallToolchain."""

    def setUp(self) -> None:
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)
        # A board we have a sysroot for already.
        self.board = "board"
        self.sysroot_path = os.path.join(self.tempdir, "build", self.board)
        self.build_target = build_target_lib.BuildTarget(
            self.board, build_root=self.sysroot_path
        )
        self.sysroot = sysroot_lib.Sysroot(self.sysroot_path)

        # A board we don't have a sysroot for yet.
        self.unbuilt_board = "board2"
        self.unbuilt_path = os.path.join(
            self.tempdir, "build", self.unbuilt_board
        )
        self.unbuilt_target = build_target_lib.BuildTarget(
            self.unbuilt_board, build_root=self.unbuilt_path
        )
        self.unbuilt_sysroot = sysroot_lib.Sysroot(self.unbuilt_path)

        osutils.SafeMakedirs(self.sysroot_path)

    def testNoSysroot(self) -> None:
        """Test handling of no sysroot."""
        with self.assertRaises(ValueError):
            sysroot.InstallToolchain(
                self.unbuilt_target,
                self.unbuilt_sysroot,
                sysroot.SetupBoardRunConfig(),
            )

    def testLocalBuild(self) -> None:
        """Test the local build logic."""
        update_patch = self.PatchObject(self.sysroot, "UpdateToolchain")

        # Test False.
        config = sysroot.SetupBoardRunConfig(usepkg=False, local_build=False)
        sysroot.InstallToolchain(self.build_target, self.sysroot, config)
        update_patch.assert_called_with(self.board, local_init=False)

        # Test usepkg True.
        update_patch.reset_mock()
        config = sysroot.SetupBoardRunConfig(usepkg=True, local_build=False)
        sysroot.InstallToolchain(self.build_target, self.sysroot, config)
        update_patch.assert_called_with(self.board, local_init=True)

        # Test local_build True.
        update_patch.reset_mock()
        config = sysroot.SetupBoardRunConfig(usepkg=False, local_build=True)
        sysroot.InstallToolchain(self.build_target, self.sysroot, config)
        update_patch.assert_called_with(self.board, local_init=True)


class BuildPackagesRunConfigTest(
    cros_test_lib.RunCommandTestCase, cros_test_lib.LoggingTestCase
):
    """Tests for the BuildPackagesRunConfig."""

    def testGetBuildPackagesExtraEnv(self) -> None:
        """Test the `cros build-packages` extra env."""
        # Test the default config.
        instance = sysroot.BuildPackagesRunConfig()

        extra_env = instance.GetExtraEnv()

        self.assertNotIn("USE_GOMA", extra_env)
        self.assertNotIn("USE_REMOTEEXEC", extra_env)

        # Test when use_flags are specified.
        use_flags = ["flag1", "flag2"]
        instance = sysroot.BuildPackagesRunConfig(use_flags=use_flags)

        extra_env = instance.GetExtraEnv()

        self.assertEqual(extra_env.get("USE"), instance.GetUseFlags())

    def testGetPackages(self) -> None:
        """Test getting packages for the config."""
        # Test the default config.
        instance = sysroot.BuildPackagesRunConfig()

        packages = instance.GetPackages()

        self.assertIn("virtual/target-os", packages)
        self.assertIn("virtual/target-os-dev", packages)
        self.assertIn("virtual/target-os-factory", packages)
        self.assertIn("virtual/target-os-test", packages)
        self.assertIn("chromeos-base/autotest-all", packages)

        # Test when packages are specified.
        test_packages = ["test/package"]
        instance = sysroot.BuildPackagesRunConfig(packages=test_packages)

        packages = instance.GetPackages()

        self.assertEqual(packages, test_packages)

    def testGetForceLocalBuildPackages(self) -> None:
        """Test getting force local build packages for the config."""
        test_sysroot_path = "/sysroot/path"
        test_sysroot = sysroot_lib.Sysroot(test_sysroot_path)
        get_reverse_dependencies_mock = self.PatchObject(
            portage_util, "GetReverseDependencies"
        )

        # Test the default config.
        instance = sysroot.BuildPackagesRunConfig()
        self.rc.AddCmdResult(
            ["cros_list_modified_packages", "--sysroot", test_sysroot.path],
            stdout="",
        )

        packages = instance.GetForceLocalBuildPackages(test_sysroot)

        # Test when there are cros_workon packages and reverse dependencies
        # but skipping base install packages and their reverse dependencies.
        instance = sysroot.BuildPackagesRunConfig(incremental_build=False)
        test_cros_list_modified_packages_output = (
            "test/package1 test/package2\n"
        )
        self.rc.AddCmdResult(
            ["cros_list_modified_packages", "--sysroot", test_sysroot.path],
            stdout=test_cros_list_modified_packages_output,
        )
        test_reverse_dependencies = [
            package_info.parse("test/package3-1.0"),
            package_info.parse("test/package4-2.0-r1"),
        ]
        get_reverse_dependencies_mock.return_value = test_reverse_dependencies

        packages = instance.GetForceLocalBuildPackages(test_sysroot)

        self.assertIn("test/package1", packages)
        self.assertIn("test/package2", packages)
        self.assertIn("test/package3", packages)
        self.assertIn("test/package4", packages)

        # Test base install packages and their reverse dependency packages
        # but skipping cros workon packages and their reverse dependencies.
        instance = sysroot.BuildPackagesRunConfig(workon=False)
        test_emerge_output = """\
[binary N] test/package1 ... to /build/sysroot/
[ebuild r U] chromeos-base/tast-build-deps ... to /build/sysroot/
[binary U] chromeos-base/chromeos-chrome ... /build/sysroot/"""
        self.rc.AddCmdResult(
            partial_mock.ListRegex(
                f"parallel_emerge --sysroot={test_sysroot.path}"
            ),
            stdout=test_emerge_output,
        )
        test_reverse_dependencies = [
            package_info.parse("virtual/package3-1.0"),
            package_info.parse("test/package4-2.0-r1"),
        ]
        get_reverse_dependencies_mock.return_value = test_reverse_dependencies

        packages = instance.GetForceLocalBuildPackages(test_sysroot)

        self.assertIn("chromeos-base/tast-build-deps", packages)
        self.assertIn("test/package4", packages)

        # Test base install packages and their reverse dependencies are skipped
        # when --no-withrevdeps is specified.
        instance = sysroot.BuildPackagesRunConfig(incremental_build=False)

        with cros_test_lib.LoggingCapturer() as logs:
            instance.GetForceLocalBuildPackages(test_sysroot)

            self.AssertLogsContain(
                logs,
                "Starting reverse dependency calculations...",
                inverted=True,
            )

    def testGetEmergeFlags(self) -> None:
        """Test building the emerge flags."""
        # Test the default config.
        instance = sysroot.BuildPackagesRunConfig()

        flags = instance.GetEmergeFlags()

        self.assertIn("--with-test-deps", flags)
        self.assertIn("--getbinpkg", flags)
        self.assertIn("--with-bdeps", flags)
        self.assertIn("--usepkg", flags)

        # Test when use_any_chrome is specified.
        instance = sysroot.BuildPackagesRunConfig(use_any_chrome=True)

        flags = instance.GetEmergeFlags()

        self.assertIn(
            "--force-remote-binary=chromeos-base/chromeos-chrome", flags
        )
        self.assertIn("--force-remote-binary=chromeos-base/chrome-icu", flags)

        # Test when usepkgonly is specified.
        instance = sysroot.BuildPackagesRunConfig(usepkgonly=True)

        flags = instance.GetEmergeFlags()

        self.assertIn("--usepkgonly", flags)

        # Test when jobs is specified.
        instance = sysroot.BuildPackagesRunConfig(jobs=10)

        flags = instance.GetEmergeFlags()

        self.assertIn("--jobs=10", flags)

    def testGetBuildPackagesRemoteExec(self) -> None:
        """Test the `cros build-packages` with remote execution."""
        # Test the default config.
        instance = sysroot.BuildPackagesRunConfig()

        instance.use_remoteexec = True
        reproxy_cfg_file = "reproxy_release.cfg"
        instance.reproxy_cfg_file = reproxy_cfg_file

        extra_env = instance.GetExtraEnv()

        self.assertNotIn("USE_GOMA", extra_env)
        self.assertEqual(extra_env.get("USE_REMOTEEXEC"), "true")
        self.assertEqual(extra_env.get("REPROXY_CFG_FILE"), reproxy_cfg_file)


class BuildPackagesTest(
    cros_test_lib.RunCommandTestCase, cros_test_lib.LoggingTestCase
):
    """Test BuildPackages function."""

    def setUp(self) -> None:
        # Currently just used to keep the parallel emerge status file from being
        # created in the chroot. This probably isn't strictly necessary, but
        # since we can otherwise run this test without a chroot existing at all
        # and without touching the chroot folder, it's better to keep it out of
        # there all together.
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=True)

        self.board = "board"
        self.target = build_target_lib.BuildTarget(self.board)
        self.sysroot = sysroot_lib.Sysroot(self.target.root)
        self.build_target_name_mock = self.PatchObject(
            sysroot_lib.Sysroot, "build_target_name", return_value=self.board
        )
        self.binhost_lookup_service_data = (
            prebuilts_cloud_pb2.BinhostLookupServiceData(
                snapshot_shas=["test_snapshot_sha"], private=True
            )
        )

        self.base_command = [
            constants.CHROMITE_BIN_DIR / "parallel_emerge",
            f"--sysroot={self.sysroot.path}",
            f"--root={self.sysroot.path}",
        ]

        # Prevent the test from switching the cpu governor.
        self.PatchObject(cpupower_helper, "ModifyCpuGovernor")
        # Prevent the test from remove files in the system.
        self.PatchObject(cros_build_lib, "ClearShadowLocks")
        self.portageq_envvar_mock = self.PatchObject(
            portage_util, "PortageqEnvvar", return_value="gs://fake/binhost"
        )
        self.lookup_binhosts_mock = self.PatchObject(binhost, "lookup_binhosts")
        self.PatchObject(portage_util, "RegenDependencyCache")
        self.installed_packages_mock = self.PatchObject(
            portage_util.PortageDB, "InstalledPackages"
        )
        self.clean_outdated_binpkgs_mock = self.PatchObject(
            portage_util, "CleanOutdatedBinaryPackages"
        )
        # Prevent the test from performing gsutil actions.
        self.get_creation_time_since_mock = self.PatchObject(
            gs.GSContext,
            "GetCreationTimeSince",
            return_value=datetime.timedelta(days=10),
        )
        self.PatchObject(gs.GSContext, "InitializeCache")

    def testSuccess(self) -> None:
        """Test successful run."""
        config = sysroot.BuildPackagesRunConfig()
        self.PatchObject(
            config, "GetForceLocalBuildPackages", return_value=["test/package1"]
        )
        sdk_pkgs = " ".join(
            sysroot._CRITICAL_SDK_PACKAGES  # pylint: disable=protected-access
        )

        with cros_test_lib.LoggingCapturer() as logs:
            sysroot.BuildPackages(self.target, self.sysroot, config)

            # The rest of the command's args we test in
            # BuildPackagesRunConfigTest, so just make sure we're calling the
            # right command and pass the args not handled by the run config.
            self.assertCommandContains(self.base_command)
            self.assertCommandContains(
                [
                    "--reinstall-atoms=test/package1",
                    "--usepkg-exclude=test/package1",
                ]
            )
            self.assertCommandContains(
                [
                    f"--useoldpkg-atoms={sdk_pkgs}",
                    f"--rebuild-exclude={sdk_pkgs}",
                ]
            )

            # Verify the extra environment variables are passed correctly.
            self.assertCommandContains([f"PKGDIR={self.sysroot.path}/packages"])

            # Check and the logs contain the expected output.
            self.AssertLogsContain(logs, "PORTAGE_BINHOST")
            self.AssertLogsContain(logs, "Rebuilding Portage cache.")
            self.AssertLogsContain(logs, "Cleaning stale binpkgs.")
            self.AssertLogsContain(logs, "Merging board packages now.")

    def testLogBinhostAgeThresholds(self) -> None:
        """Test the log output from _LogBinhostAge with different thresholds."""
        config = sysroot.BuildPackagesRunConfig()

        with cros_test_lib.LoggingCapturer() as logs:
            self.lookup_binhosts_mock.return_value = []
            # check for when the binhost was created within the threshold
            sysroot.BuildPackages(self.target, self.sysroot, config)
            self.get_creation_time_since_mock.assert_called_with(
                path="gs://fake/binhost/Packages", since_date=mock.ANY
            )
            self.AssertLogsContain(
                logs,
                "PORTAGE_BINHOST gs://fake/binhost was created 10 days ago.",
            )

            # check for when the binhost was created outside of the threshold
            self.get_creation_time_since_mock.return_value = datetime.timedelta(
                days=31
            )
            sysroot.BuildPackages(self.target, self.sysroot, config)
            self.AssertLogsContain(
                logs,
                "PORTAGE_BINHOST gs://fake/binhost was created more"
                " than 30 days ago. Please repo sync for the latest build"
                " artifacts.",
            )
            self.AssertLogsContain(
                logs,
                "PORTAGE_BINHOST gs://fake/binhost was created 31 days ago.",
            )

    def testLogBinHostAgeUrls(self) -> None:
        """Test the log output from _LogBinhostAge with different binhosts."""
        config = sysroot.BuildPackagesRunConfig()

        with cros_test_lib.LoggingCapturer() as logs:
            self.lookup_binhosts_mock.return_value = []
            # check that the gs url is formatted correctly when the binhost
            # has a trailing slash
            self.portageq_envvar_mock.return_value = "gs://fake/binhost/"
            sysroot.BuildPackages(self.target, self.sysroot, config)
            self.get_creation_time_since_mock.assert_called_with(
                path="gs://fake/binhost/Packages", since_date=mock.ANY
            )
            self.AssertLogsContain(
                logs,
                "PORTAGE_BINHOST gs://fake/binhost/ was created 10 days ago.",
            )

            # check for when there is a GSNoSuchKey error for a malformed
            # binhost url
            self.get_creation_time_since_mock.side_effect = gs.GSNoSuchKey(
                "Err"
            )
            sysroot.BuildPackages(self.target, self.sysroot, config)
            self.AssertLogsContain(
                logs,
                "PORTAGE_BINHOST gs://fake/binhost/",
            )
            self.AssertLogsContain(
                logs,
                "Error getting the binhost age",
            )

    def testLookupService(self) -> None:
        """Test that binhosts from the lookup service are passed to portage."""
        config = sysroot.BuildPackagesRunConfig()
        self.lookup_binhosts_mock.return_value = ["gs://AAAA", "gs://BBBB"]

        sysroot.BuildPackages(self.target, self.sysroot, config)

        self.assertCommandContains(["PORTAGE_BINHOST=gs://AAAA gs://BBBB"])

    def testEcleanBinpkgs(self) -> None:
        """Test that eclean is called with the expected packages."""

        def assert_file_contents(
            sysroot_path: Union[str, os.PathLike],
            deep: bool,
            exclusion_file: Optional[Union[str, os.PathLike]] = None,
        ) -> None:
            if exclusion_file:
                contents = osutils.ReadFile(exclusion_file)
                self.assertEqual("cross-dev/package", contents)

            self.assertEqual(self.sysroot.path, sysroot_path)
            self.assertTrue(deep)

        self.PatchObject(os.path, "exists", return_value=True)
        self.installed_packages_mock.return_value = [
            portage_util.InstalledPackage(None, "", "test", "package-1.0"),
            portage_util.InstalledPackage(None, "", "cross-dev", "package-1.0"),
        ]
        self.clean_outdated_binpkgs_mock.side_effect = assert_file_contents
        config = sysroot.BuildPackagesRunConfig()

        sysroot.BuildPackages(self.target, self.sysroot, config)

    def testInstallDebugSymbols(self) -> None:
        """Test cros_install_debug_syms is called with the expected args."""
        config = sysroot.BuildPackagesRunConfig(install_debug_symbols=True)

        with cros_test_lib.LoggingCapturer() as logs:
            sysroot.BuildPackages(self.target, self.sysroot, config)

            self.assertCommandContains(
                [
                    constants.CHROMITE_BIN_DIR / "cros_install_debug_syms",
                    f"--board={self.build_target_name_mock}",
                    "--all",
                ]
            )
            self.AssertLogsContain(logs, "Fetching the debug symbols.")

    def testPackageFailure(self) -> None:
        """Test package failure handling."""
        failed = ["cat/pkg", "foo/bar"]
        cpvs = [package_info.SplitCPV(p, strict=False) for p in failed]
        self.PatchObject(
            portage_util, "ParseDieHookStatusFile", return_value=cpvs
        )
        config = sysroot.BuildPackagesRunConfig()

        self.rc.AddCmdResult(
            partial_mock.ListRegex("emerge.*--useoldpkg-atoms"), returncode=1
        )
        with self.assertRaises(sysroot_lib.PackageInstallError) as e:
            sysroot.BuildPackages(self.target, self.sysroot, config)
        self.assertEqual(cpvs, e.exception.failed_packages)
        self.assertCommandContains(self.base_command)

    def testEnsureBootstrapSuccess(self) -> None:
        """Ensure ensure_bootstrap gets called."""
        config = sysroot.BuildPackagesRunConfig()
        sysroot.BuildPackages(self.target, self.sysroot, config)
        self.assertCommandCalled(
            [constants.DEPOT_TOOLS_DIR / "ensure_bootstrap"]
        )

    def testEnsureBootstrapFailure(self) -> None:
        """Ensure BuildPackages fails when ensure_bootstrap fails."""
        self.rc.AddCmdResult(
            partial_mock.ListRegex("ensure_bootstrap"), returncode=1
        )
        config = sysroot.BuildPackagesRunConfig()
        with self.assertRaises(cros_build_lib.RunCommandError):
            sysroot.BuildPackages(self.target, self.sysroot, config)
        self.assertCommandCalled(
            [constants.DEPOT_TOOLS_DIR / "ensure_bootstrap"]
        )


class GatherSymbolFilesTest(cros_test_lib.MockTempDirTestCase):
    """Base class for testing GatherSymbolFiles."""

    SLIM_CONTENT = """
some junk
"""

    FAT_CONTENT = """
STACK CFI 1234
some junk
STACK CFI 1234
"""

    def createSymbolFile(self, filename, content=FAT_CONTENT, size=0) -> None:
        """Create a symbol file using content with minimum size."""
        osutils.SafeMakedirs(os.path.dirname(filename))

        # If a file size is given, force that to be the minimum file size.
        # Create a sparse file so large files are practical.
        with open(filename, "w+b") as f:
            f.truncate(size)
            f.seek(0)
            f.write(content.encode("utf-8"))

    def test_ListOutputOfGatherSymbolFiles(self) -> None:
        """Mimic how the controller materializes output of GatherSymbolFiles."""
        # Create directory with some symbol files.
        tar_tmp_dir = os.path.join(self.tempdir, "tar_tmp")
        output_dir = os.path.join(self.tempdir, "output")
        input_dir = os.path.join(self.tempdir, "input")
        osutils.SafeMakedirs(output_dir)
        self.createSymbolFile(os.path.join(input_dir, "a/b/c/file1.sym"))
        self.createSymbolFile(os.path.join(input_dir, "a/b/c/d/file2.sym"))
        self.createSymbolFile(os.path.join(input_dir, "a/file3.sym"))
        self.createSymbolFile(os.path.join(input_dir, "a/b/c/d/e/file1.sym"))

        # Call sysroot.GatherSymbolFiles to find symbol files under self.tempdir
        # and copy them to output_dir.
        symbol_files = list(
            sysroot.GatherSymbolFiles(tar_tmp_dir, output_dir, [input_dir])
        )
        self.assertEqual(len(symbol_files), 4)

    def test_GatherSymbolFiles(self) -> None:
        """Test that files are found and copied."""
        # Create directory with some symbol files.
        tar_tmp_dir = os.path.join(self.tempdir, "tar_tmp")
        output_dir = os.path.join(self.tempdir, "output")
        input_dir = os.path.join(self.tempdir, "input")
        osutils.SafeMakedirs(output_dir)
        self.createSymbolFile(os.path.join(input_dir, "a/b/c/file1.sym"))
        self.createSymbolFile(os.path.join(input_dir, "a/b/c/d/file2.sym"))
        self.createSymbolFile(os.path.join(input_dir, "a/file3.sym"))
        self.createSymbolFile(os.path.join(input_dir, "a/b/c/d/e/file1.sym"))

        # Call sysroot.GatherSymbolFiles to find symbol files under self.tempdir
        # and copy them to output_dir.
        symbol_files = list(
            sysroot.GatherSymbolFiles(tar_tmp_dir, output_dir, [input_dir])
        )

        # Construct the expected symbol files. Note that the SymbolFileTuple
        # field source_file_name is the full path to where a symbol file was
        # found, while relative_path is the relative path (from the search)
        # where it is created in the output directory.
        expected_symbol_files = [
            sysroot.SymbolFileTuple(
                source_file_name=os.path.join(input_dir, "a/b/c/file1.sym"),
                relative_path="a/b/c/file1.sym",
            ),
            sysroot.SymbolFileTuple(
                source_file_name=os.path.join(input_dir, "a/b/c/d/file2.sym"),
                relative_path="a/b/c/d/file2.sym",
            ),
            sysroot.SymbolFileTuple(
                source_file_name=os.path.join(input_dir, "a/file3.sym"),
                relative_path="a/file3.sym",
            ),
            sysroot.SymbolFileTuple(
                source_file_name=os.path.join(input_dir, "a/b/c/d/e/file1.sym"),
                relative_path="a/b/c/d/e/file1.sym",
            ),
        ]

        # Sort symbol_files and expected_output_files by the relative_path
        # attribute.
        symbol_files = sorted(
            symbol_files, key=operator.attrgetter("relative_path")
        )
        expected_symbol_files = sorted(
            expected_symbol_files, key=operator.attrgetter("relative_path")
        )
        # Compare the files to the expected files. This verifies the size and
        # contents, and on failure shows the full contents.
        self.assertEqual(symbol_files, expected_symbol_files)

        # Verify that the files in output_dir match the SymbolFile
        # relative_paths.
        files_in_output_dir = self.getFilesWithRelativeDir(output_dir)
        files_in_output_dir.sort()
        symbol_file_relative_paths = [obj.relative_path for obj in symbol_files]
        symbol_file_relative_paths.sort()
        self.assertEqual(files_in_output_dir, symbol_file_relative_paths)

        # Verify that the display_name of each symbol does not contain pathsep.
        symbol_file_relative_paths = [
            os.path.basename(obj.relative_path) for obj in symbol_files
        ]
        for display_name in symbol_file_relative_paths:
            self.assertEqual(-1, display_name.find(os.path.sep))

    def test_GatherSymbolTarFiles(self) -> None:
        """Test that symbol files in tar files are extracted."""
        output_dir = os.path.join(self.tempdir, "output")
        osutils.SafeMakedirs(output_dir)

        # Set up test input directory.
        tarball_dir = os.path.join(self.tempdir, "some/path")
        files_in_tarball = [
            "dir1/fileZ.sym",
            "dir2/fileY.sym",
            "dir2/fileX.sym",
            "fileA.sym",
            "fileB.sym",
            "fileC.sym",
        ]
        for filename in files_in_tarball:
            self.createSymbolFile(os.path.join(tarball_dir, filename))
        temp_tarball_file_path = os.path.join(self.tempdir, "symfiles.tar")
        cros_build_lib.CreateTarball(temp_tarball_file_path, tarball_dir)
        # Now that we've created the tarball, remove the .sym files in
        # the tarball dir and move the tarball to that dir.
        for filename in files_in_tarball:
            os.remove(os.path.join(tarball_dir, filename))
        tarball_path = os.path.join(tarball_dir, "symfiles.tar")
        shutil.move(temp_tarball_file_path, tarball_path)

        # Execute sysroot.GatherSymbolFiles where the path contains the tarball.
        symbol_files = list(
            sysroot.GatherSymbolFiles(tarball_dir, output_dir, [tarball_path])
        )

        self.assertEqual(len(symbol_files), 6)
        # Verify the symbol_file relative_paths.
        symbol_file_relative_paths = [obj.relative_path for obj in symbol_files]
        symbol_file_relative_paths.sort()
        self.assertEqual(
            symbol_file_relative_paths,
            [
                "dir1/fileZ.sym",
                "dir2/fileX.sym",
                "dir2/fileY.sym",
                "fileA.sym",
                "fileB.sym",
                "fileC.sym",
            ],
        )
        # Verify the symbol_file source_file_names.
        symbol_file_source_file_names = [
            obj.source_file_name for obj in symbol_files
        ]
        symbol_file_source_file_names.sort()
        # Note that the expected symbol_file_source_names are the full path to
        # the tarfile followed by the relative path, such as
        # /tmp/chromite.test2ng92vzo/some/path/symfiles.tar/dir1/fileZ.sym
        expected_symbol_file_source_names = [
            os.path.join(tarball_path, "dir1/fileZ.sym"),
            os.path.join(tarball_path, "dir2/fileX.sym"),
            os.path.join(tarball_path, "dir2/fileY.sym"),
            os.path.join(tarball_path, "fileA.sym"),
            os.path.join(tarball_path, "fileB.sym"),
            os.path.join(tarball_path, "fileC.sym"),
        ]
        self.assertEqual(
            symbol_file_source_file_names, expected_symbol_file_source_names
        )

        # Verify that the files in output_dir match the SymbolFile
        # relative_paths.
        files_in_output_dir = self.getFilesWithRelativeDir(output_dir)
        files_in_output_dir.sort()
        symbol_file_relative_paths = [obj.relative_path for obj in symbol_files]
        symbol_file_relative_paths.sort()
        self.assertEqual(files_in_output_dir, symbol_file_relative_paths)
        # Verify that the display_name of each symbol does not contain pathsep.
        symbol_file_relative_paths = [
            os.path.basename(obj.relative_path) for obj in symbol_files
        ]
        for display_name in symbol_file_relative_paths:
            self.assertEqual(-1, display_name.find(os.path.sep))

    def test_GatherSymbolTarFilesWithNonSymFiles(self) -> None:
        """Test that non-symbol files in tar files are not extracted."""
        output_dir = os.path.join(self.tempdir, "output")
        osutils.SafeMakedirs(output_dir)

        # Set up test input directory.
        tarball_dir = os.path.join(self.tempdir, "some/path")
        files_in_tarball = [
            "dir1/fileU.sym",
            "dir1/fileU.txt",
            "fileD.sym",
            "fileD.txt",
        ]
        for filename in files_in_tarball:
            # we don't care about file contents, so we are using
            # createSymbolFile for files whether they end with .sym or not.
            self.createSymbolFile(os.path.join(tarball_dir, filename))
        temp_tarball_file_path = os.path.join(self.tempdir, "symfiles.tar")
        cros_build_lib.CreateTarball(temp_tarball_file_path, tarball_dir)
        # Now that we've created the tarball, remove the .sym files in
        # the tarball dir and move the tarball to that dir.
        for filename in files_in_tarball:
            os.remove(os.path.join(tarball_dir, filename))
        tarball_path = os.path.join(tarball_dir, "symfiles.tar")
        shutil.move(temp_tarball_file_path, tarball_path)

        # Execute sysroot.GatherSymbolFiles where the path contains the tarball.
        symbol_files = list(
            sysroot.GatherSymbolFiles(tarball_dir, output_dir, [tarball_path])
        )

        # Verify the symbol_file relative_paths only has .sym files.
        symbol_file_relative_paths = [obj.relative_path for obj in symbol_files]
        symbol_file_relative_paths.sort()
        self.assertEqual(
            symbol_file_relative_paths, ["dir1/fileU.sym", "fileD.sym"]
        )
        for symfile in symbol_file_relative_paths:
            extension = symfile.split(".")[1]
            self.assertEqual(extension, "sym")

    def test_GatherSymbolFileFullFilePaths(self) -> None:
        """Test full filepaths (.sym and .txt) only gather .sym files."""
        tar_tmp_dir = os.path.join(self.tempdir, "tar_tmp")
        output_dir = os.path.join(self.tempdir, "output")
        input_dir = os.path.join(self.tempdir, "input")
        osutils.SafeMakedirs(output_dir)
        # We don't care about contents, so use createSymbolFiles for all files.
        self.createSymbolFile(os.path.join(input_dir, "a_file.sym"))
        self.createSymbolFile(os.path.join(input_dir, "a_file.txt"))

        # Call sysroot.GatherSymbolFiles with full paths to files, some of which
        # don't end in .sym, verify that only .sym files get copied to
        # output_dir.
        symbol_files = list(
            sysroot.GatherSymbolFiles(
                tar_tmp_dir,
                output_dir,
                [
                    os.path.join(input_dir, "a_file.sym"),
                    os.path.join(input_dir, "a_file.txt"),
                ],
            )
        )

        # Construct the expected symbol files. Note that the SymbolFileTuple
        # field source_file_name is the full path to where a symbol file was
        # found, while relative_path is the relative path (from the search)
        # where it is created in the output directory.
        expected_symbol_files = [
            sysroot.SymbolFileTuple(
                source_file_name=os.path.join(input_dir, "a_file.sym"),
                relative_path="a_file.sym",
            )
        ]

        # Compare the files to the expected files. This verifies the size and
        # contents, and on failure shows the full contents.
        self.assertEqual(symbol_files, expected_symbol_files)
        # Verify that only a_file.sym is in the output_dir.
        files_in_output_dir = self.getFilesWithRelativeDir(output_dir)
        self.assertEqual(files_in_output_dir, ["a_file.sym"])

    def getFilesWithRelativeDir(self, dest_dir):
        """Find all files below dest_dir using dir relative to dest_dir."""
        relative_files = []
        for path, __, files in os.walk(dest_dir):
            for filename in files:
                fullpath = os.path.join(path, filename)
                relpath = os.path.relpath(fullpath, dest_dir)
                relative_files.append(relpath)
        return relative_files


class GenerateBreakpadSymbolsTest(cros_test_lib.RunCommandTempDirTestCase):
    """Base class for testing GenerateBreakpadSymbols."""

    def setUp(self) -> None:
        self.chroot_dir = os.path.join(self.tempdir, "chroot_dir")
        osutils.SafeMakedirs(self.chroot_dir)

    def test_generateBreakpadSymbols(self) -> None:
        """Verify calling the service layer invokes the script as expected."""
        chroot = chroot_lib.Chroot(self.chroot_dir)
        build_target = build_target_lib.BuildTarget("board")

        # Call the method being tested.
        sysroot.GenerateBreakpadSymbols(chroot, build_target, False, False, [])

        self.rc.assertCommandContains(
            [
                "cros_generate_breakpad_symbols",
                "--board=board",
                "--jobs",
                mock.ANY,
                "--exclude-dir=firmware",
            ],
            enter_chroot=True,
            chroot_args=["--chroot", mock.ANY],
        )

    def test_generateBreakpadSymbolsWithDebug(self) -> None:
        """Verify that calling with debug invokes the script as expected."""
        chroot = chroot_lib.Chroot(self.chroot_dir)
        build_target = build_target_lib.BuildTarget("board")

        # Call the method being tested.
        sysroot.GenerateBreakpadSymbols(chroot, build_target, True, False, [])

        self.rc.assertCommandContains(
            [
                "cros_generate_breakpad_symbols",
                "--debug",
                "--board=board",
                "--jobs",
                mock.ANY,
                "--exclude-dir=firmware",
            ],
            enter_chroot=True,
            chroot_args=["--chroot", mock.ANY],
        )

    def test_generateBreakpadSymbolsWithIgnoreErrors(self) -> None:
        """Verify that calling with debug invokes the script as expected."""
        chroot = chroot_lib.Chroot(self.chroot_dir)
        build_target = build_target_lib.BuildTarget("board")

        # Call the method being tested.
        sysroot.GenerateBreakpadSymbols(chroot, build_target, False, True, [])

        self.rc.assertCommandContains(
            [
                "cros_generate_breakpad_symbols",
                "--ignore_errors",
                "--board=board",
                "--jobs",
                mock.ANY,
                "--exclude-dir=firmware",
            ],
            enter_chroot=True,
            chroot_args=["--chroot", mock.ANY],
        )

    def test_generateBreakpadSymbolsWithIgnoreIgnoreExpectedFiles(self) -> None:
        """Verify that calling with debug invokes the script as expected."""
        chroot = chroot_lib.Chroot(self.chroot_dir)
        build_target = build_target_lib.BuildTarget("board")

        # Call the method being tested.
        sysroot.GenerateBreakpadSymbols(
            chroot, build_target, False, False, ["ASH_CHROME", "LIBC"]
        )

        self.rc.assertCommandContains(
            [
                "cros_generate_breakpad_symbols",
                "--ignore_expected_file=ASH_CHROME",
                "--ignore_expected_file=LIBC",
                "--board=board",
                "--jobs",
                mock.ANY,
                "--exclude-dir=firmware",
            ],
            enter_chroot=True,
            chroot_args=["--chroot", mock.ANY],
        )


class BundleDebugSymbolsTest(cros_test_lib.MockTempDirTestCase):
    """Unittests for BundleDebugSymbols."""

    def setUp(self) -> None:
        # Configure {chroot,out,syroot}_path.
        self.chroot_path = os.path.join(self.tempdir, "chroot_dir")
        self.out_path = self.tempdir / "out_dir"
        self.sysroot_path = os.path.join(self.chroot_path, "build/target")

        # Has to be run outside the chroot.
        self.PatchObject(cros_build_lib, "IsInsideChroot", return_value=False)

        osutils.SafeMakedirs(self.sysroot_path)

        # Create output dir.
        self.output_dir = os.path.join(self.tempdir, "output_dir")
        osutils.SafeMakedirs(self.output_dir)

        # Create chroot, sysroot, and build_target objs.
        self.chroot = chroot_lib.Chroot(
            path=self.chroot_path, out_path=self.out_path
        )
        self.sysroot = sysroot_lib.Sysroot(path=self.sysroot_path)
        self.build_target = build_target_lib.BuildTarget("target")

        # Create the tmp dir.
        osutils.SafeMakedirs(self.chroot.tmp)

    def testBundleBreakpadDebugSymbols(self) -> None:
        """BundleBreakpadSymbols calls cbuildbot/commands with correct args."""
        # Patch service layer functions.
        generate_breakpad_symbols_patch = self.PatchObject(
            sysroot,
            "GenerateBreakpadSymbols",
            return_value=cros_build_lib.CompletedProcess(
                returncode=0, stdout=""
            ),
        )
        gather_symbol_files_patch = self.PatchObject(
            sysroot,
            "GatherSymbolFiles",
            return_value=[
                sysroot.SymbolFileTuple(
                    source_file_name="path/to/source/file1.sym",
                    relative_path="file1.sym",
                )
            ],
        )

        tar_file = sysroot.BundleBreakpadSymbols(
            self.chroot,
            self.sysroot,
            self.build_target,
            self.output_dir,
            False,
            ["ASH_CHROME"],
        )

        # Verify mock objects were called.
        generate_breakpad_symbols_patch.assert_called_with(
            self.chroot,
            self.build_target,
            debug=True,
            ignore_errors=False,
            ignore_expected_files=["ASH_CHROME"],
        )
        gather_symbol_files_patch.assert_called()

        # Verify response proto contents and output directory contents.
        self.assertIsNotNone(tar_file)
        self.assertTrue(tar_file.endswith("/output_dir/debug_breakpad.tar.xz"))

    def testBundleDebugSymbols(self) -> None:
        """BundleDebugSymbols calls cbuildbot/commands with correct args."""
        # Patch service layer functions.
        self.PatchObject(os.path, "exists", return_value=True)
        self.PatchObject(os.path, "isdir", return_value=True)

        create_tarball_patch = self.PatchObject(
            cros_build_lib,
            "CreateTarball",
            return_value=cros_build_lib.CompletedProcess(
                returncode=0, stdout=""
            ),
        )

        tar_file = sysroot.BundleDebugSymbols(
            self.chroot,
            self.sysroot,
            build_target_lib.BuildTarget(None),
            self.output_dir,
        )
        create_tarball_patch.assert_called()

        # Verify response contents.
        self.assertIsNotNone(tar_file)
        self.assertTrue(tar_file.endswith("/output_dir/debug.tgz"))


def test_CollectBazelPerformanceArtifacts(monkeypatch, tmp_path) -> None:
    """CollectBazelPerformanceArtifacts copies the known set of files."""
    # Configure {chroot,out,syroot}_path.
    tempdir = tmp_path
    chroot_path = tempdir / "chroot_dir"
    out_path = tempdir / "out_dir"
    sysroot_path = chroot_path / "build" / "target"

    # Simulate being outside the chroot.
    monkeypatch.setattr(cros_build_lib, "IsInsideChroot", lambda: False)

    osutils.SafeMakedirs(sysroot_path)

    # Create output dir.
    output_dir = os.path.join(tempdir, "output_dir")
    osutils.SafeMakedirs(output_dir)

    # Create chroot, sysroot, and build_target objs.
    chroot = chroot_lib.Chroot(path=chroot_path, out_path=out_path)
    target_sysroot = sysroot_lib.Sysroot(path=sysroot_path)
    build_target = build_target_lib.BuildTarget("target")

    # Create Bazel performance files for testing
    sysroot_bazel_files = (
        sysroot.BAZEL_ALLPACKAGES_COMMAND_PROFILE_FILE,
        sysroot.BAZEL_ALLPACKAGES_EXEC_LOG_FILE,
    )
    created_files = [
        chroot.full_path(artifact) for artifact in sysroot_bazel_files
    ]
    for created_file in created_files:
        osutils.Touch(created_file, makedirs=True)

    expected_files = [
        tempdir / os.path.basename(file) for file in sysroot_bazel_files
    ]

    # Collect the files and confirm we got what we expected
    archived_files = sysroot.CollectBazelPerformanceArtifacts(
        chroot,
        target_sysroot,
        build_target,
        tempdir,
    )
    for expected_file in expected_files:
        assert str(expected_file) in archived_files
        assert expected_file.exists()


class RemoteExecutionTest(cros_test_lib.MockLoggingTestCase):
    """Unittests for remote execution context manager."""

    def setUp(self) -> None:
        self.goma_mock = self.PatchObject(goma_lib, "Goma", autospec=True)
        self.goma_instance = self.goma_mock.return_value

    def testGomaDir(self) -> None:
        """Test the case where GOMA env variable is defined."""
        os.environ.update(
            {
                "GOMA_DIR": "goma/path",
                "GOMA_TMP_DIR": "goma/tmp/dir",
                "GLOG_log_dir": "glog/log/dir",
            }
        )

        with sysroot.RemoteExecution(use_goma=True):
            self.goma_mock.assert_called_once_with(
                Path("goma/path"),
                "goma/tmp/dir",
                stage_name="BuildPackages",
                log_dir="glog/log/dir",
            )
        self.goma_instance.Restart.assert_called_once()
        self.goma_instance.Stop.assert_called_once()

    def testGomaHomeDir(self) -> None:
        """Test the case where Home Path is used."""
        self.PatchObject(Path, "home", return_value=Path("home"))

        with sysroot.RemoteExecution(use_goma=True):
            self.goma_mock.assert_called_once_with(
                Path("home/goma"),
                None,
                stage_name="BuildPackages",
                log_dir=None,
            )
        self.goma_instance.Restart.assert_called_once()
        self.goma_instance.Stop.assert_called_once()

    def testGomaException(self) -> None:
        """Test the case where GOMA interface raises exception."""
        self.goma_mock.side_effect = ValueError()

        with cros_test_lib.LoggingCapturer() as log:
            with sysroot.RemoteExecution(use_goma=True):
                self.AssertLogsMatch(log, ".*initialization error.*")
        self.goma_instance.Restart.assert_not_called()
        self.goma_instance.Stop.assert_not_called()

    def testNoRemoteExec(self) -> None:
        """Test the case where no remoteexec is requested with env variable."""
        os.environ.update(
            {
                "GOMA_DIR": "goma/path",
            }
        )

        with sysroot.RemoteExecution(use_goma=False):
            pass
        self.goma_mock.assert_not_called()

    def testNoRemoteExecNoEnv(self) -> None:
        """Test case where no remoteexec is requested without env variable."""
        with sysroot.RemoteExecution(use_goma=False):
            pass
        self.goma_mock.assert_not_called()


class ArchiveSysrootTest(cros_test_lib.TempDirTestCase):
    """ArchiveSysroot tests."""

    def setUp(self) -> None:
        chroot_path = self.tempdir / "chroot"
        self.chroot = chroot_lib.Chroot(path=chroot_path)
        sysroot_path = chroot_path / "build" / "testBoard"
        self.sysroot = sysroot_lib.Sysroot(sysroot_path)
        self.dir_structure = [
            cros_test_lib.Directory(".", []),
            cros_test_lib.Directory("test", ["foo.bar"]),
        ]

    def testArchiveSysroot(self) -> None:
        """Archive a sample folder and verify its contents."""
        cros_test_lib.CreateOnDiskHierarchy(
            self.chroot.full_path(self.sysroot.path), self.dir_structure
        )

        archive_file = sysroot.ArchiveSysroot(
            self.chroot,
            self.sysroot,
            build_target_lib.BuildTarget("testBoard"),
            self.tempdir,
        )
        self.assertTrue(Path(archive_file).exists())
        self.assertEqual(
            archive_file, self.tempdir / sysroot.SYSROOT_ARCHIVE_FILE
        )
        cros_test_lib.VerifyTarball(archive_file, self.dir_structure)

    def testArchiveSysrootFailure(self) -> None:
        """Archive a sample folder that doesnt exists."""
        archive_file = sysroot.ArchiveSysroot(
            self.chroot,
            self.sysroot,
            build_target_lib.BuildTarget("testBoard"),
            self.tempdir,
        )
        self.assertIsNone(archive_file)


class ExtractSysrootTest(cros_test_lib.TempDirTestCase):
    """ExtractSysroot tests."""

    def setUp(self) -> None:
        chroot_path = self.tempdir / "chroot"
        self.chroot = chroot_lib.Chroot(path=chroot_path)
        self.sysroot_path = chroot_path / "build" / "testBoard"
        self.sysroot = sysroot_lib.Sysroot(self.sysroot_path)
        self.sysroot_archive_path = chroot_path / "build" / "tmp"
        self.sysroot_archive = sysroot_lib.Sysroot(self.sysroot_archive_path)
        self.build_target = build_target_lib.BuildTarget("testBoard")
        self.archive_dir_structure = [
            cros_test_lib.Directory(".", []),
            cros_test_lib.Directory("test", ["foo.bar"]),
        ]
        self.sysroot_dir_structure = [
            cros_test_lib.Directory(".", ["test", "test2"]),
        ]
        self.expected_dir_structure = [
            cros_test_lib.Directory("test", ["foo.bar"]),
        ]

    def testExtractSysroot(self) -> None:
        """Extract a simple tar."""
        sysroot_archive = None
        cros_test_lib.CreateOnDiskHierarchy(
            self.sysroot_archive_path, self.archive_dir_structure
        )
        sysroot_archive = sysroot.ArchiveSysroot(
            self.chroot,
            self.sysroot_archive,
            build_target_lib.BuildTarget("tmp"),
            self.tempdir,
        )

        cros_test_lib.CreateOnDiskHierarchy(
            self.sysroot_path, self.sysroot_dir_structure
        )
        result = sysroot.ExtractSysroot(
            self.chroot, self.sysroot, sysroot_archive
        )
        self.assertTrue(Path(result).exists())
        self.assertEqual(str(result), self.sysroot.path)
        cros_test_lib.VerifyOnDiskHierarchy(
            self.sysroot_path, self.expected_dir_structure
        )