# Copyright 2015 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Routines and a delegate for dealing with locally worked on packages."""

import collections
import glob
import logging
import os
from pathlib import Path
import re
from typing import Iterable, List, Set

from chromite.lib import build_target_lib
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import dependency_graph
from chromite.lib import git
from chromite.lib import osutils
from chromite.lib import path_util
from chromite.lib import portage_util
from chromite.lib import sysroot_lib
from chromite.lib.parser import package_info


if cros_build_lib.IsInsideChroot():
    from chromite.lib import depgraph

# A package is a canonical CP atom.
# A package may have 0 or more repositories, given as strings.
# Each repository may be mapped into our workspace at some path.
PackageInfo = collections.namedtuple(
    "PackageInfo", ("package", "repos", "src_paths")
)


# Vars that we want to pass through for the user.
EBUILD_PASS_THROUGH_VARS = frozenset(
    [
        # Common test vars.
        "GTEST_ARGS",
        # Platform eclass vars.
        "P2_TEST_FILTER",
        "P2_VMODULE",
    ]
)


def _IsWorkonEbuild(include_chrome, ebuild_path, ebuild_contents=None):
    """Returns True iff the ebuild at |ebuild_path| is a workon ebuild.

    This means roughly that the ebuild is compatible with our cros_workon based
    system.  For most packages, this means that it inherits the cros-workon
    overlay.

    Args:
        include_chrome: True iff we should include Chrome and chromium-source
            packages.
        ebuild_path: path an ebuild in question.
        ebuild_contents: None, or the contents of the ebuild at |ebuild_path|.
            If None, _IsWorkonEbuild will read the contents of the ebuild when
            necessary.

    Returns:
        True iff the ebuild can be used with cros_workon.
    """
    # TODO(rcui): remove special casing of chromeos-chrome here when we make it
    # inherit from cros-workon / chromium-source class (chromium-os:19259).
    if (
        include_chrome
        and portage_util.EbuildToCP(ebuild_path) == constants.CHROME_CP
    ):
        return True

    workon_eclasses = "cros-workon"
    if include_chrome:
        workon_eclasses += "|chromium-source"

    ebuild_contents = ebuild_contents or osutils.ReadFile(ebuild_path)
    if re.search(
        "^inherit (.|\\\n)*(%s)" % workon_eclasses, ebuild_contents, re.M
    ):
        return True

    return False


def _GetLinesFromFile(path: Path, line_prefix, line_suffix) -> Set[str]:
    """Get a unique set of lines from a file, stripping off a prefix and suffix.

    Rejects lines that do not start with |line_prefix| or end with
    |line_suffix|.
    Returns an empty set if the file at |path| does not exist.
    Discards duplicate lines.

    Args:
        path: path to file.
        line_prefix: prefix of line to look for and strip if found.
        line_suffix: suffix of line to look for and strip if found.

    Returns:
        A set of filtered lines from the file at |path|.
    """
    if not path.exists():
        return set()

    # Note that there is an opportunity to race with the file system here.
    lines = set()
    for line in path.read_text(encoding="utf-8").splitlines():
        if not line.startswith(line_prefix) or not line.endswith(line_suffix):
            logging.warning("Filtering out malformed line: %s", line)
            continue
        lines.add(line[len(line_prefix) : -len(line_suffix)])

    return lines


def _WriteLinesToFile(path, lines, line_prefix, line_suffix) -> None:
    """Write a set of lines to a file, adding prefixes, suffixes and newlines.

    Args:
        path: path to file.
        lines: iterable of lines to write.
        line_prefix: string to prefix each line with.
        line_suffix: string to append to each line before a newline.
    """
    contents = "".join(
        ["%s%s%s\n" % (line_prefix, line, line_suffix) for line in lines]
    )
    if not contents:
        osutils.SafeUnlink(path)
    else:
        osutils.WriteFile(path, contents, makedirs=True)


def GetWorkonPath(source_root=constants.SOURCE_ROOT, sub_path=None):
    """Get the path to files related to packages we're working locally on.

    Args:
        source_root: path to source root inside chroot.
        sub_path: optional path to file relative to the workon root directory.

    Returns:
        path to the workon root directory or file within the root directory.
    """
    ret = os.path.join(source_root, ".config/cros_workon")
    if sub_path:
        ret = os.path.join(ret, sub_path)

    return ret


class WorkonError(Exception):
    """Raised when invariants of the WorkonHelper are violated."""


def _FilterWorkonOnlyEbuilds(ebuilds):
    """Filter a list of ebuild paths to only with those no stable version.

    Args:
        ebuilds: list of string paths to ebuild files (e.g.
            ['/prefix/sys-app/app/app-9999.ebuild'])

    Returns:
        list of ebuild paths meeting this criterion.
    """
    result = []
    for ebuild_path in ebuilds:
        ebuild_pattern = os.path.join(os.path.dirname(ebuild_path), "*.ebuild")
        stable_ebuilds = [
            path
            for path in glob.glob(ebuild_pattern)
            if not path.endswith("-9999.ebuild")
        ]
        if not stable_ebuilds:
            result.append(ebuild_path)

    return result


def ListAllWorkedOnAtoms(src_root=constants.SOURCE_ROOT):
    """Get a list of all atoms we're currently working on.

    Args:
        src_root: path to source root inside chroot.

    Returns:
        Dictionary of atoms marked as worked on (e.g. ['chromeos-base/shill'])
        for each system.
    """
    workon_dir = GetWorkonPath(source_root=src_root)
    if not os.path.isdir(workon_dir):
        return {}

    system_to_atoms = {}
    for file_name in os.listdir(workon_dir):
        if file_name.endswith(".mask"):
            continue
        file_contents = osutils.ReadFile(os.path.join(workon_dir, file_name))

        atoms = []
        for line in file_contents.splitlines():
            match = re.match("=(.*)-9999", line)
            if match:
                atoms.append(match.group(1))
        if atoms:
            system_to_atoms[os.path.basename(file_name)] = atoms

    return system_to_atoms


class WorkonHelper:
    """Delegate that knows how to mark packages as being worked on locally.

    This class assumes that we're executing in the build root.
    """

    def __init__(
        self,
        sysroot,
        friendly_name=None,
        verbose=False,
        src_root=constants.SOURCE_ROOT,
    ) -> None:
        """Construct an instance.

        Args:
            sysroot: path to sysroot to work on packages within.
            friendly_name: friendly name of the system (e.g. 'host', <board
                name>, or a brick friendly name). Defaults to 'host' if sysroot
                is '/' or the last component of the sysroot path.
            verbose: boolean True iff we should print a lot more command output.
                This is intended for debugging, and you should never cause a
                script to depend on behavior enabled by this flag.
            src_root: path to source root inside chroot.
        """
        self._sysroot = sysroot
        if friendly_name:
            self._system = friendly_name
        else:
            self._system = (
                "host"
                if sysroot == "/"
                else os.path.basename(sysroot.rstrip("/"))
            )
        self._verbose = verbose
        self._src_root = src_root
        self._cached_overlays = None
        self._cached_arch = None
        self._depgraph = None

        profile = os.path.join(self._sysroot, "etc", "portage")
        self._unmasked_symlink = os.path.join(
            profile, "package.unmask", "cros-workon"
        )
        self._keywords_symlink = os.path.join(
            profile, "package.accept_keywords", "cros-workon"
        )
        self._masked_symlink = os.path.join(
            profile, "package.mask", "cros-workon"
        )

        # Clobber and re-create the WORKON_FILE symlinks every time. This is a
        # trivial operation and eliminates all kinds of corner cases as well as
        # any possible future renames of WORKON_FILE.
        # In particular, we build the chroot as a board (amd64-host), bundle it
        # and unpack it on /. After unpacking, the symlinks will point to
        # .config/cros_workon/amd64-host instead of .config/cros_workon/host.
        # Regenerating the symlinks here corrects it. crbug.com/23096.
        # Note: This is currently also relied upon as an indirect fix for
        # crbug.com/679831. Search the bug number for instance(s).
        self._RefreshSymlinks()

    @property
    def workon_file_path(self):
        """Returns path to the file holding our currently worked on atoms."""
        return GetWorkonPath(source_root=self._src_root, sub_path=self._system)

    @property
    def masked_file_path(self):
        """Returns path to file masking non-9999 ebuilds for worked on atoms."""
        return self.workon_file_path + ".mask"

    @property
    def _arch(self):
        if self._cached_arch is None:
            self._cached_arch = sysroot_lib.Sysroot(
                self._sysroot
            ).GetStandardField(sysroot_lib.STANDARD_FIELD_ARCH)

        return self._cached_arch

    @property
    def _overlays(self):
        """Returns overlays installed for the selected system."""
        if self._cached_overlays is None:
            sysroot = sysroot_lib.Sysroot(self._sysroot)
            portdir_overlay = sysroot.GetStandardField("PORTDIR_OVERLAY")
            self._cached_overlays = [
                x.strip() for x in portdir_overlay.splitlines()
            ]

        return self._cached_overlays

    def _SetWorkedOnAtoms(self, atoms) -> None:
        """Sets the unmasked atoms.

        This will generate both the unmasked atom list and the masked atoms list
        as the two files mention the same atom list.

        Args:
            atoms: Atoms to unmask.
        """
        _WriteLinesToFile(self.workon_file_path, atoms, "=", "-9999")
        _WriteLinesToFile(self.masked_file_path, atoms, "<", "-9999")
        self._RefreshSymlinks()

    def _RefreshSymlinks(self) -> None:
        """Recreates the symlinks.

        This will create the symlinks needed:
        * package.mask/cros-workon: list of packages to mask.
        * package.keywords/cros-workon: list of hidden packages to accept.
        """
        if not os.path.exists(self._sysroot):
            return

        # We used to generate this, but stopped as it's not actually needed.
        # Clean up any old links that might still be around.
        osutils.SafeUnlink(self._unmasked_symlink, sudo=True)

        for target, symlink in (
            (self.masked_file_path, self._masked_symlink),
            (self.workon_file_path, self._keywords_symlink),
        ):
            if not os.path.exists(target):
                logging.debug("Config %s doesn't exist; ignoring.", target)
            elif not os.path.exists(symlink):
                osutils.SafeMakedirs(os.path.dirname(symlink), sudo=True)
                osutils.SafeSymlink(target, symlink, sudo=True)
            else:
                logging.debug(
                    "Symlink %s already exists. Don't recreate it.", symlink
                )

    def _AtomsToEbuilds(self, atoms):
        """Maps a list of CP atoms to a list of corresponding -9999 ebuilds.

        Args:
            atoms: iterable of portage atoms (e.g. ['sys-apps/dbus']).

        Returns:
            list of ebuilds corresponding to those atoms.
        """
        atoms_to_ebuilds = {atom: None for atom in atoms}

        for overlay in self._overlays:
            ebuild_paths = glob.glob(
                os.path.join(overlay, "*-*", "*", "*-9999.ebuild")
            )
            for ebuild_path in ebuild_paths:
                atom = portage_util.EbuildToCP(ebuild_path)
                if atom in atoms_to_ebuilds:
                    atoms_to_ebuilds[atom] = ebuild_path

        ebuilds = []
        for atom, ebuild in atoms_to_ebuilds.items():
            if ebuild is None:
                raise WorkonError("Could not find ebuild for atom %s" % atom)
            ebuilds.append(ebuild)

        return ebuilds

    def _GetCanonicalAtom(self, package_fragment: str, find_stale=False):
        """Transform a pkg source path or name fragment to the canonical atom.

        If there are multiple atoms that a package fragment could map to,
        picks an arbitrary one and prints a warning.

        Args:
            package_fragment: Package source path or name fragment.
            find_stale: if True, allow stale (missing) worked on package.

        Returns:
            string canonical atom name (e.g. 'sys-apps/dbus')
        """
        # Attempt to not hit portage if at all possible for speed.
        if package_fragment in self._GetWorkedOnAtoms():
            return package_fragment

        # Ask portage directly what it thinks about that package.
        ebuild_path = self._FindEbuildForPackage(package_fragment)

        # If portage didn't know about that package, try and autocomplete it.
        if ebuild_path is None:
            possible_atoms = set()
            for ebuild in self._GetWorkonEbuilds(filter_on_arch=False):
                pkg_atom = portage_util.EbuildToCP(ebuild)
                if package_fragment in pkg_atom:
                    possible_atoms.add(pkg_atom)

            # Also autocomplete from the worked-on list, in case the ebuild was
            # deleted.
            if find_stale:
                for pkg_atom in self._GetWorkedOnAtoms():
                    if package_fragment in pkg_atom:
                        possible_atoms.add(pkg_atom)

            if not possible_atoms:
                # Try finding the packages affected by a given path.
                path_atoms = sorted(self._GetPathAtoms(package_fragment))
                if not path_atoms:
                    logging.warning(
                        'Could not find canonical package for "%s"',
                        package_fragment,
                    )
                    return None

                if len(path_atoms) > 1:
                    logging.warning(
                        "Multiple affected packages found for %s:",
                        package_fragment,
                    )
                    for p in path_atoms:
                        logging.warning("  %s", p)
                    logging.warning("Using %s", path_atoms[0])
                    logging.notice(
                        "cros workon start command for the rest of the "
                        "packages:"
                    )
                    logging.notice(
                        f"cros workon -b {self._system} start "
                        f'{" ".join(path_atoms[1:])}'
                    )

                logging.notice(
                    "Package %s found for path %s",
                    path_atoms[0],
                    package_fragment,
                )
                return path_atoms[0]

            # We want some consistent order for making our selection below.
            possible_atoms = sorted(possible_atoms)
            if len(possible_atoms) > 1:
                logging.warning("Multiple autocompletes found:")
                for possible_atom in possible_atoms:
                    logging.warning("  %s", possible_atom)
            # If only provided a $PN, prefer that first.
            pn_possible_atoms = [
                x
                for x in possible_atoms
                if package_info.parse(x).package == package_fragment
            ]
            if pn_possible_atoms:
                autocompleted_package = portage_util.EbuildToCP(
                    pn_possible_atoms[0]
                )
            else:
                autocompleted_package = portage_util.EbuildToCP(
                    possible_atoms[0]
                )
            # Sanity check to avoid infinite loop.
            if package_fragment == autocompleted_package:
                logging.error("Resolved %s to itself", package_fragment)
                return None
            logging.info(
                'Autocompleted "%s" to: "%s"',
                package_fragment,
                autocompleted_package,
            )

            return self._GetCanonicalAtom(autocompleted_package)

        if not _IsWorkonEbuild(True, ebuild_path):
            msg = (
                "In order to cros_workon a package, it must have a -9999 "
                "ebuild that inherits from cros-workon.\n"
            )
            if "-9999" in ebuild_path:
                msg += (
                    '"%s" is a -9999 ebuild, make sure it inherits from '
                    "cros-workon.\n" % ebuild_path
                )
            else:
                msg += '"%s" is not a -9999 ebuild.\n' % ebuild_path

            logging.warning(msg)
            return None

        return portage_util.EbuildToCP(ebuild_path)

    def _GetCanonicalAtoms(
        self, package_fragments: Iterable[str], find_stale=False
    ):
        """Transforms a list of package name fragments into a list of CP atoms.

        Args:
            package_fragments: list of package source paths and/or name
                fragments.
            find_stale: if True, allow stale (missing) worked on package.

        Returns:
            list of canonical portage atoms corresponding to the given
            fragments.
        """
        if not package_fragments:
            raise WorkonError("No packages specified")

        atoms = []
        for package_fragment in package_fragments:
            atom = self._GetCanonicalAtom(
                package_fragment, find_stale=find_stale
            )
            if atom is None:
                raise WorkonError("Error parsing package list")
            atoms.append(atom)

        return atoms

    def _GetDepGraph(self):
        """Get the dependency graph."""
        if self._depgraph:
            return self._depgraph

        try:
            # Get the graph for our target.
            if self._sysroot == build_target_lib.get_default_sysroot_path():
                self._depgraph = depgraph.get_sdk_dependency_graph(
                    with_src_paths=True
                )
            else:
                self._depgraph = depgraph.get_build_target_dependency_graph(
                    self._sysroot, with_src_paths=True
                )
        except dependency_graph.Error as e:
            logging.error(e)
            raise WorkonError("Error generating dependency graph.")

        return self._depgraph

    def _GetPathAtoms(self, raw_path) -> Iterable:
        """Get workon atoms affected by the current path."""
        # Make sure we're in a source path.
        path = Path(raw_path).resolve()
        if not path.exists():
            # The path doesn't exist. To avoid the long lookup when the dev
            # misspells a package name, lets just assume it's that and return no
            # packages.
            logging.warning(
                "%s (%s) does not exist. Is it a misspelled package?",
                path,
                raw_path,
            )
            return []

        try:
            path = path.relative_to(constants.SOURCE_ROOT)
        except ValueError as e:
            logging.error(e)
            raise WorkonError(
                "Current path not in the source root: "
                f"{path} not in {constants.SOURCE_ROOT}"
            )

        graph = self._GetDepGraph()
        # Get the relevant packages from the dep graph.
        logging.debug("Getting packages relevant to %s", path)
        relevant_atoms = set(x.atom for x in graph.get_relevant_nodes([path]))

        if not relevant_atoms:
            return []

        logging.debug("Found relevant packages: %s", relevant_atoms)

        # Filter out any non-cros-workon packages.
        canonical = [self._GetCanonicalAtom(x) for x in relevant_atoms]
        workon_atoms = [x for x in canonical if x]

        logging.debug("Found relevant workon packages: %s", workon_atoms)

        return workon_atoms

    def _GetWorkedOnAtoms(self) -> Set[str]:
        """Returns the set of package atoms that we're currently working on."""
        return _GetLinesFromFile(Path(self.workon_file_path), "=", "-9999")

    def _FindEbuildForPackage(self, package):
        """Find an ebuild for a given atom (accepting even masked ebuilds).

        Args:
            package: package string.

        Returns:
            path to ebuild for given package.
        """
        return portage_util.FindEbuildForPackage(
            package,
            self._sysroot,
            include_masked=True,
            extra_env={"ACCEPT_KEYWORDS": "~%s" % self._arch},
        )

    def _GetWorkonEbuilds(
        self, filter_workon=False, filter_on_arch=True, include_chrome=True
    ):
        """Get a list of all cros-workon ebuilds in the current system.

        Args:
            filter_workon: True iff we should filter the list of ebuilds to
                those packages which define only a workon ebuild (i.e. no stable
                version).
            filter_on_arch: True iff we should only return ebuilds which are
                marked as unstable for the architecture of the system we're
                interested in.
            include_chrome: True iff we should also include chromeos-chrome and
                related ebuilds.  These ebuilds can be worked on, but don't work
                like normal cros-workon ebuilds.

        Returns:
            list of paths to ebuilds meeting the above criteria.
        """
        result = []
        if filter_on_arch:
            keyword_pat = re.compile(
                r'^KEYWORDS=".*~(\*|%s).*"$' % self._arch, re.M
            )

        for overlay in self._overlays:
            ebuild_paths = glob.glob(
                os.path.join(overlay, "*-*", "*", "*-9999.ebuild")
            )
            for ebuild_path in ebuild_paths:
                ebuild_contents = osutils.ReadFile(ebuild_path)
                if filter_on_arch and not keyword_pat.search(ebuild_contents):
                    continue
                if not _IsWorkonEbuild(
                    include_chrome, ebuild_path, ebuild_contents=ebuild_contents
                ):
                    continue
                result.append(ebuild_path)

        if filter_workon:
            result = _FilterWorkonOnlyEbuilds(result)

        return result

    def _GetLiveAtoms(self, filter_workon=False):
        """Get a list of atoms currently marked as being locally compiled.

        Args:
            filter_workon: True iff the list should be filtered to only those
                atoms without a stable version (i.e. the -9999 ebuild is the
                only ebuild).

        Returns:
            list of canonical portage atoms.
        """
        atoms = self._GetWorkedOnAtoms()

        if filter_workon:
            ebuilds = _FilterWorkonOnlyEbuilds(self._AtomsToEbuilds(atoms))
            return [portage_util.EbuildToCP(ebuild) for ebuild in ebuilds]

        return atoms

    def _AddProjectsToPartialManifests(self, atoms) -> None:
        """Add projects corresponding to a list of atoms to the local manifest.

        If we mark projects as workon that we don't have in our local checkout,
        it is convenient to have them added to the manifest.  Note that users
        will need to `repo sync` to pull down repositories added in this way.

        Args:
            atoms: iterable of atoms to ensure are in the manifest.
        """
        # If this is a Cog checkout, we do not care about local manifests at
        # this time, as external access to Cider is a long ways down the road.
        # A strategy to equate a Cog superproject checkout to a local manifest
        # is also a requirement to re-enable this.
        # TODO: b/334950349 - Support workon modifications of local manifests
        if path_util.DetermineCheckout().type == path_util.CheckoutType.CITC:
            return

        manifest = git.ManifestCheckout.Cached(self._src_root)

        should_repo_sync = False
        ebuilds = portage_util.FindEbuildsForPackages(atoms, self._sysroot)
        for ebuild_path in ebuilds.values():
            infos = portage_util.GetRepositoryForEbuild(
                ebuild_path, self._sysroot
            )
            if not infos:
                logging.warning(
                    "Unable to determine project information for %s. "
                    "If you already have the project(s) for the package "
                    "synced, no action is needed.",
                    ebuild_path,
                )
                continue

            for info in infos:
                if not info.project or manifest.FindCheckouts(info.project):
                    continue

                cmd = ["loman", "add", "--workon", info.project]
                try:
                    cros_build_lib.dbg_run(cmd)
                    should_repo_sync = True
                except cros_build_lib.RunCommandError as e:
                    logging.warning(
                        "Error adding project %s for %s to the manifest. "
                        "If you already have the project synced, no "
                        "action is needed.",
                        info.project,
                        ebuild_path,
                    )
                    logging.debug(e)

        if should_repo_sync:
            print('Please run "repo sync" now.')

    def ListAtoms(self, use_all=False, use_workon_only=False):
        """Returns a list of interesting atoms.

        By default, return a list of the atoms marked as being locally worked on
        for the system in question.

        Args:
            use_all: If true, return a list of all atoms we could possibly work
                on for the system in question.
            use_workon_only: If true, return a list of all atoms we could
                possibly work on that have no stable ebuild.

        Returns:
            a list of atoms (e.g. ['chromeos-base/shill', 'sys-apps/dbus']).
        """
        if use_workon_only or use_all:
            ebuilds = self._GetWorkonEbuilds(filter_workon=use_workon_only)
            packages = [portage_util.EbuildToCP(ebuild) for ebuild in ebuilds]
        else:
            packages = self._GetLiveAtoms()

        return sorted(packages)

    def StartWorkingOnPackages(
        self,
        packages,
        use_all: bool = False,
        use_workon_only: bool = False,
        quiet: bool = False,
    ) -> None:
        """Mark a list of packages as being worked on locally.

        Args:
            packages: list of package name fragments. While each fragment could
                be a complete portage atom, this helper will attempt to infer
                intent by looking for fragments in a list of all possible atoms
                for the system in question.
            use_all: True iff we should ignore the package list, and instead
                consider all possible atoms that we could mark as worked on
                locally.
            use_workon_only: True iff we should ignore the package list, and
                instead consider all possible atoms for the system in question
                that define only the -9999 ebuild.
            quiet: Does not log the started atoms when True. Used to avoid
                confusion in cases where the list must be toggled to compute the
                new packages.
        """
        if not os.path.exists(self._sysroot):
            raise WorkonError("Sysroot %s is not setup." % self._sysroot)

        if use_all or use_workon_only:
            ebuilds = self._GetWorkonEbuilds(filter_workon=use_workon_only)
            atoms = [portage_util.EbuildToCP(ebuild) for ebuild in ebuilds]
        else:
            atoms = self._GetCanonicalAtoms(packages)
        atoms = set(atoms)
        logging.debug("Atoms: %s", sorted(atoms))

        # Read out what atoms we're already working on.
        existing_atoms = self._GetWorkedOnAtoms()

        # Warn the user if they're requested to work on an atom that's already
        # marked as being worked on.
        for atom in atoms & existing_atoms:
            logging.warning("Already working on %s", atom)

        # If we have no new atoms to work on, we can quit now.
        new_atoms = atoms - existing_atoms
        if not new_atoms:
            return

        # Write out all these atoms to the appropriate files.
        current_atoms = new_atoms | existing_atoms
        self._SetWorkedOnAtoms(current_atoms)

        self._AddProjectsToPartialManifests(new_atoms)

        if not quiet:
            # Legacy scripts used single quotes in their output, and we carry on
            # this honorable tradition.
            logging.notice(
                "Started working on '%s' for '%s'",
                " ".join(new_atoms),
                self._system,
            )

    def StopWorkingOnPackages(
        self,
        packages,
        use_all: bool = False,
        use_workon_only: bool = False,
        quiet: bool = False,
    ) -> None:
        """Stop working on a list of pkgs currently marked as locally worked on.

        Args:
            packages: list of package name fragments.  These will be mapped to
                canonical portage atoms via the same process as
                StartWorkingOnPackages().
            use_all: True iff instead of the provided package list, we should
                just stop working on all currently worked on atoms for the
                system in question.
            use_workon_only: True iff instead of the provided package list, we
                should stop working on all currently worked on atoms that define
                only a -9999 ebuild.
            quiet: Does not log the started atoms when True. Used to avoid
                confusion in cases where the list must be toggled to compute the
                new packages.
        """
        if use_all or use_workon_only:
            atoms = self._GetLiveAtoms(filter_workon=use_workon_only)
        else:
            atoms = self._GetCanonicalAtoms(packages, find_stale=True)

        current_atoms = self._GetWorkedOnAtoms()
        stopped_atoms = []
        for atom in atoms:
            if not atom in current_atoms:
                logging.warning("Not working on %s", atom)
                continue

            current_atoms.discard(atom)
            stopped_atoms.append(atom)

        self._SetWorkedOnAtoms(current_atoms)

        if stopped_atoms and not quiet:
            # Legacy scripts used single quotes in their output, and we carry on
            # this honorable tradition.
            logging.notice(
                "Stopped working on '%s' for '%s'",
                " ".join(stopped_atoms),
                self._system,
            )

    def GetWorkonAtom(self, package: str) -> str:
        """Find the canonical atom name for the workon package.

        The function finds the canonical atom name (e.g. 'sys-apps/dbus')
        for the given workon package. The function will throw WorkonError
        if the package does not exist or is not on the current workon list
        for the board.

        Args:
            package: The package to check in workon list for board.

        Returns:
            The canonical portage atom corresponding to the given fragment
            (e.g. 'sys-apps/dbus')
        """
        atom = self._GetCanonicalAtom(package, find_stale=False)

        if not atom:
            raise WorkonError(f"error looking up package {package}")

        if not atom in self._GetWorkedOnAtoms():
            raise WorkonError(
                f"run 'cros workon --board {self._system} "
                f"start {package}' first!"
            )

        return atom

    def _AssertNotHost(self) -> None:
        """Fails if the sysroot is for host."""
        if self._sysroot == build_target_lib.get_default_sysroot_path():
            raise WorkonError("cannot run for host. expecting board target.")

    def BuildPackage(
        self,
        package: str,
        clean: bool = False,
        test: bool = False,
    ) -> None:
        """Build a workon package.

        Args:
            package: The name of the package to build. This must be a workon
            else an error is raised.
            clean: True if we want to re run configure and prepare steps.
            test: True if we want to run tests.
        """
        cros_build_lib.AssertInsideChroot()
        self._AssertNotHost()

        atom = self.GetWorkonAtom(package)
        pkgfile = self._FindEbuildForPackage(package)

        if not pkgfile:
            raise WorkonError(f"cannot find ebuild for {package}")

        workpath = Path(
            sysroot_lib.Sysroot(self._sysroot).JoinPath(
                "tmp", "portage", f"{atom}-9999"
            )
        )
        features = os.environ.get("FEATURES", "").split()
        features.append("-noauto")

        if test:
            features.append("test")
            (workpath / ".tested").unlink(missing_ok=True)

        workdir = workpath / "work" / f"{atom}-9999"

        ebuild_vars = osutils.SourceEnvironment(
            pkgfile, ["CROS_WORKON_OUTOFTREE_BUILD"]
        )

        if not (
            workdir.is_symlink()
            or ebuild_vars.get("CROS_WORKON_OUTOFTREE_BUILD", "") == "1"
        ):
            logging.warning("Cleaning up stale workdir: %s", workdir)
            clean = True

        if not clean:
            (workpath / ".compiled").unlink(missing_ok=True)
            envf = workpath / "temp" / "environment"
            if envf.exists():
                lines = [
                    (
                        f"declare -x {v}="
                        f"{cros_build_lib.ShellQuote(os.environ.get(v, ''))}"
                        f"{os.linesep}"
                    )
                    for v in EBUILD_PASS_THROUGH_VARS
                ]

                osutils.WriteFile(
                    envf,
                    lines,
                    encoding="utf-8",
                    mode="a",
                )

        cmd = [f"ebuild-{self._system}", pkgfile]
        if clean:
            cmd.append("clean")
        cmd.append("test" if test else "compile")

        cros_build_lib.run(
            cmd,
            print_cmd=True,
            extra_env={
                "FEATURES": " ".join(features).strip(),
                "SANDBOX_WRITE": self._src_root,
                "CROS_WORKON_INPLACE": "1",
            },
        )

    def InstallPackage(self, package: str) -> None:
        """Install a workon package.

        Args:
            package: The name of the package to be installed. This must be a
            workon else an error is raised.
        """
        cros_build_lib.AssertInsideChroot()
        self._AssertNotHost()

        atom = self.GetWorkonAtom(package)
        cros_build_lib.run(
            [f"emerge-{self._system}", "--nodeps", atom], print_cmd=True
        )

    def ScrubPackage(self, package: str) -> None:
        """Scrub a workon package.

        Args:
            package: The name of the package to be scrubbed.
        """
        cros_build_lib.AssertInsideChroot()
        self._AssertNotHost()

        atom = self.GetWorkonAtom(package)
        self.RunCommandInAtomSourceDirectory(atom, ["git", "clean", "-dxf"])

    def GetPackageInfo(self, packages, use_all=False, use_workon_only=False):
        """Get information about packages.

        Args:
            packages: list of package name fragments.  These will be mapped to
                canonical portage atoms via the same process as
                StartWorkingOnPackages().
            use_all: True iff we should ignore the package list, and instead
                consider all possible workon-able atoms.
            use_workon_only: True iff we should ignore the package list, and
                instead consider all possible atoms for the system in question
                that define only the -9999 ebuild.

        Returns:
            Returns a list of PackageInfo tuples.
        """
        if use_all or use_workon_only:
            # You can't use info to find the source code from Chrome, since that
            # workflow is different.
            ebuilds = self._GetWorkonEbuilds(
                filter_workon=use_workon_only, include_chrome=False
            )
        else:
            atoms = self._GetCanonicalAtoms(packages)
            ebuilds = [self._FindEbuildForPackage(atom) for atom in atoms]

        build_root = self._src_root
        src_root = os.path.join(build_root, "src")
        manifest = git.ManifestCheckout.Cached(build_root)

        ebuild_to_repos = {}
        ebuild_to_src_paths = collections.defaultdict(list)

        for ebuild in ebuilds:
            ebuild_obj = portage_util.EBuild(ebuild)
            workon_vars = ebuild_obj.cros_workon_vars
            projects = workon_vars.project if workon_vars else []
            ebuild_to_repos[ebuild] = projects
            if ebuild_obj.is_manually_uprevved:
                # Manually uprevved ebuild is pinned to a specific git sha1, so
                # change in that repo matter to the ebuild.
                continue
            src_paths = ebuild_obj.GetSourceInfo(src_root, manifest).srcdirs
            src_paths = [
                os.path.relpath(path, build_root) for path in src_paths
            ]
            ebuild_to_src_paths[ebuild] = src_paths

        result = []
        for ebuild in ebuilds:
            package = portage_util.EbuildToCP(ebuild)
            repos = ebuild_to_repos.get(ebuild, [])
            src_paths = ebuild_to_src_paths.get(ebuild, [])
            result.append(PackageInfo(package, repos, src_paths))

        result.sort()
        return result

    def RunCommandInAtomSourceDirectory(self, atom, command: List[str]) -> None:
        """Run a command in the source directory of an atom.

        Args:
            atom: string atom to run the command in (e.g.
                'chromeos-base/shill').
            command: Command to run in the source directory of |atom|.
        """
        logging.info('Running "%s" on %s', command, atom)
        ebuild_path = self._FindEbuildForPackage(atom)
        if ebuild_path is None:
            raise WorkonError("Error looking for atom %s" % atom)

        for info in portage_util.GetRepositoryForEbuild(
            ebuild_path, self._sysroot
        ):
            cros_build_lib.run(command, cwd=info.srcdir, print_cmd=False)

    def RunCommandInPackages(
        self, packages, command: List[str], use_all=False, use_workon_only=False
    ) -> None:
        """Run a command in the source directory of a list of packages.

        Args:
            packages: list of package name fragments.
            command: Command to run in the source directory of |atom|.
            use_all: True iff we should ignore the package list, and instead
                consider all possible workon-able atoms.
            use_workon_only: True iff we should ignore the package list, and
                instead consider all possible atoms for the system in question
                that define only the -9999 ebuild.
        """
        if use_all or use_workon_only:
            atoms = self._GetLiveAtoms(filter_workon=use_workon_only)
        else:
            atoms = self._GetCanonicalAtoms(packages)
        for atom in atoms:
            self.RunCommandInAtomSourceDirectory(atom, command)

    def InstalledWorkonAtoms(self):
        """Returns the set of installed cros_workon packages."""
        installed_cp = set()
        for pkg in portage_util.PortageDB(self._sysroot).InstalledPackages():
            installed_cp.add("%s/%s" % (pkg.category, pkg.package))

        return set(a for a in self.ListAtoms(use_all=True) if a in installed_cp)


class WorkonScope:
    """Context manager to assist managing workon status for packages."""

    def __init__(
        self,
        build_target: build_target_lib.BuildTarget,
        pkgs: Iterable[str] = tuple(),
    ) -> None:
        """Construct an instance.

        Args:
            build_target: The build target (board) being built.
            pkgs: The workon packages to be used in the context manager dunder
                methods.
        """
        self.helper = WorkonHelper(build_target.root, build_target.name)
        self.pkgs = pkgs
        self.target = build_target
        self.stop_packages = []
        self.start_packages = []
        self.before_workon = self.helper.ListAtoms()

    def __enter__(self: "WorkonScope") -> "WorkonScope":
        """Commence context manager tasks for starting and stopping packages.

        Returns:
            The initialized WorkonScope context manager.
        """
        self.start(self.pkgs)
        after_workon = self.helper.ListAtoms()

        # Stop = the set we actually started. Preserves workon started status
        # for any in the packages that were already worked on.
        self.stop_packages = sorted(set(after_workon) - set(self.before_workon))
        return self

    def __exit__(self, exc_type, exc_val, tb) -> None:
        """Clean up context manager tasks for starting and stopping packages.

        Args:
            exc_type: The exception type passed when the runtime context raises
                an exception.
            exc_val: The exception value raised by the runtime context.
            tb: The exception traceback raised by the runtime context.

        Raises:
            Any exception raised in the runtime context will be raised here
            after cleanup. Beyond that, all WorkonHelper methods are expected to
            be safe operations.
        """
        # Reset the environment.
        logging.notice("Restoring cros_workon status.")
        if self.stop_packages:
            # Stop the packages we started.
            logging.info("Stopping workon packages previously started.")
            try:
                self.stop(self.stop_packages)
            except WorkonError:
                to_stop = sorted(
                    set(self.stop_packages) - set(self.helper.ListAtoms())
                )
                logging.critical(
                    "Unable to stop started packages. Please stop the "
                    "following packages: %s",
                    " ".join(to_stop),
                )
        else:
            logging.info("No packages needed to be stopped.")
        if self.start_packages:
            # Stop the packages we started.
            logging.info("Restarting workon packages previously stopped.")
            try:
                self._start_packages(self.start_packages)
            except WorkonError:
                to_start = sorted(
                    set(self.start_packages) - set(self.helper.ListAtoms())
                )
                logging.critical(
                    "Unable to start stopped packages. Please start the "
                    "following packages: %s",
                    " ".join(to_start),
                )
        else:
            logging.info("No packages needed to be restarted.")

    def _start_packages(self, pkgs: Iterable[str]) -> None:
        """Wrapper for self.WorkonHelper.StartWorkingOnPackages."""
        self.helper.StartWorkingOnPackages(pkgs)

    def _stop_packages(self, pkgs: Iterable[str]) -> None:
        self.helper.StopWorkingOnPackages(pkgs)

    def start(self, pkgs: Iterable[str]) -> None:
        """Helper to allow the context manager to explicitly start packages.

        Invocations of this method will track started packages and stop them
        when __exit__ is invoked, even when explicitly called by a client in a
        runtime context.

        Args:
            pkgs: A list of package name fragments.
        """
        if pkgs:
            logging.debug(
                "cros-workon-%s start %s", self.target.name, " ".join(pkgs)
            )
            self._start_packages(pkgs)
            after_workon = self.helper.ListAtoms()
            self.stop_packages = sorted(
                set(after_workon) - set(self.before_workon)
            )

    def stop(self, pkgs: Iterable[str]) -> None:
        """Helper to allow the context manager to explicitly stop packages.

        If a package is stopped that was marked as workon before entering the
        runtime context, that package will be restarted on exit.

        Args:
            pkgs: A list of package name fragments.
        """
        if pkgs:
            logging.debug(
                "cros-workon-%s stop %s", self.target.name, " ".join(pkgs)
            )
            self._stop_packages(pkgs)
            to_restart = set(pkgs) & set(self.before_workon)
            self.start_packages = sorted(to_restart | set(self.start_packages))