# Copyright 2020 The Pigweed Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""Calls to checkout code.

Usage:
api.checkout(remote='https://pigweed.googlesource.com/pigweed/pigweed')

The checkout recipe module is designed to handle the process of checking out
source code from version control systems, primarily Git. It's a core component
in CI/CQ, ensuring that the correct version of the code is available for
subsequent build, test, or deployment steps.

Its key functionalities include:

1. Source Code Retrieval:
   * It can perform direct Git checkouts from a specified remote URL and branch.
   * It supports checkouts using the Android Repo Tool when use_repo is true,
     allowing it to work with manifest-based projects. This includes
     initializing the repo, syncing projects, and handling manifest branches and
     groups.
2. Gerrit Integration:
   * It processes Gerrit changes provided via build inputs. When applicable, it
     applies stacks of Gerrit changes.
   * It uses the fuchsia/gerrit module to fetch change details (like branch,
     project, and revision).
   * It integrates with the pigweed/cq_deps module to resolve and apply
     dependent CLs specified in patches.json files.
3. Change Application:
   * It applies fetched Gerrit changes to the checkout. For try builds, it
     rebases changes by default (this can be overridden).
   * It applies changes to submodules recursively.
   * It determines the base revision for applied changes (e.g., remote branch
     tip, submitted commit hash, or merge-base).
 4 Submodule Management:
   * It can initialize and update Git submodules. This can be configured to
     include all submodules, a specific list, or exclude certain submodules.
   * It uses a helper script to get detailed status information about submodules
     (hash, path, remote, branch, etc.).
   * It handles matching CLs to their respective submodules based on remote URLs
     and branches.
5. Caching:
   * It can use caching to speed up the checkout process by storing and
     retrieving repository data.
6. Configuration and Options:
   * The behavior of the checkout process is controlled by a variety of options.
     These options include the remote URL, branch, whether to use repo,
     submodule handling strategies (initialize, include/exclude lists),
     timeouts for various operations, and definitions for equivalent remotes.
   * It supports rewriting Git remote URLs (e.g., for mirrors).
7. Workspace and Build System Interaction:
   * It can interact with Bazel workspaces by attempting to patch Bazel
     WORKSPACE files (eligible_workspace_paths) based on the checked-out
     changes, using functionality from the pigweed/bazel_roll module.
8. Status Reporting and Output:
   * It provides context about the checkout (CheckoutContext), including the
     root path, applied changes, and the final revision.
   * It can create a "snapshot" of the checkout, which includes the manifest
     (for repo checkouts) or submodule status, and a git log.
   * It reports the status of applied and unapplied changes, with links to
     Gerrit for easy inspection.
"""

from __future__ import annotations

import collections
import contextlib
import dataclasses
import re
import urllib
import xml.etree.ElementTree
from collections.abc import Generator, Iterator, Sequence
from typing import Any, TypedDict

from google.protobuf import json_format

from recipe_engine import config_types, recipe_api

from PB.go.chromium.org.luci.buildbucket.proto import build as build_pb
from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb
from PB.go.chromium.org.luci.scheduler.api.scheduler.v1 import (
    triggers as triggers_pb,
)
from PB.recipe_modules.pigweed.checkout.options import Options

PIGWEED_REMOTE = 'https://pigweed.googlesource.com/pigweed/pigweed'


def to_dict(obj) -> dict[str, Any]:
    try:
        # Modifications to the dict returned by the built-in vars() function
        # modify the original data structure. Always create a copy for this
        # function to return.
        return __builtins__['vars'](obj).copy()
    except TypeError:  # pragma: no cover
        keys = [x for x in obj.__slots__ if not x.startswith('__')]
        return {k: getattr(obj, k) for k in keys}


@dataclasses.dataclass
class Manifest:
    remotes: dict[str, Remote] = dataclasses.field(default_factory=dict)
    projects: list[Project] = dataclasses.field(default_factory=list)

    def dict(self) -> dict[str, Any]:
        return {
            'remotes': {k: v.dict() for k, v in self.remotes.items()},
            'projects': [x.dict() for x in self.projects],
        }


class Url:
    def __init__(self, url: str, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.url: str = url
        self.https: str | None = None

    def dict(self) -> dict[str, Any]:
        return to_dict(self)


@dataclasses.dataclass
class Remote:
    """Remote config from manifest."""

    name: str
    fetch: Url
    review: str | None = None
    revision: str | None = None
    alias: str | None = None

    def dict(self) -> dict[str, Any]:
        res = to_dict(self)
        res['fetch'] = res['fetch'].dict()
        return res


@dataclasses.dataclass
class Project:
    """Key variables describing a repository/project."""

    name: str
    path: str
    remote: str
    revision: str
    upstream: str
    url: str | None = None

    def path_object(self, root: config_types.Path) -> config_types.Path:
        return root / self.path

    def dict(self) -> dict[str, Any]:
        return to_dict(self)


def _str_or_none(x: Any | None) -> str | None:
    if x is None:
        return x
    return str(x)


def _int_or_none(x: Any | None) -> int | None:
    if x is None:
        return x
    return int(x)


@dataclasses.dataclass
class Change:
    """Data from buildbucket."""

    number: int
    remote: str | None
    ref: str | None
    rebase: bool | None = None
    project: str | None = None
    branch: str | None = None
    gerrit_name: str | None = None
    submitted: bool = False
    patchset: int | None = None
    applied: bool = dataclasses.field(default=False, repr=False)
    path: str | None = None
    base: str | None = None
    base_type: str | None = None
    is_merge: bool = False
    commit_message: str = ''
    topic: str | None = None
    current_revision: str | None = None
    primary: bool = False
    submit_requirements: dict[str, str] = dataclasses.field(
        default_factory=dict,
    )
    uploader: str | None = None

    def __post_init__(self):
        self.number = _int_or_none(self.number)
        self.remote = _str_or_none(self.remote)
        self.ref = _str_or_none(self.ref)
        self.branch = _str_or_none(self.branch)
        self.gerrit_name = _str_or_none(self.gerrit_name)
        self.patchset = _int_or_none(self.patchset)
        self.base = _str_or_none(self.base)
        self.base_type = _str_or_none(self.base_type)

    @property
    def gerrit_host(self) -> str:
        return f'https://{self.gerrit_name}-review.googlesource.com'

    @property
    def gerrit_url(self) -> str:
        if not self.number:
            return self.gitiles_url
        return f'{self.gerrit_host}/c/{self.number}'

    @property
    def gitiles_url(self) -> str:
        return f'{self.remote}/+/{self.ref}'

    @property
    def name(self) -> str:
        return f'{self.gerrit_name}:{self.number}'

    @property
    def name_with_path(self) -> str:
        return f'{self.name} ({self.path})'


@dataclasses.dataclass
class Submodule:
    """Submodule properties."""

    api: recipe_api.RecipeApi = dataclasses.field(repr=False)
    hash: str
    relative_path: str
    path: config_types.Path
    name: str
    describe: str
    remote: str
    initialized: bool
    modified: bool
    conflict: bool
    branch: str
    url: str
    update: str
    ignore: str
    shallow: bool
    fetchRecurseSubmodules: bool
    describe: str

    def __lt__(self, other: Submodule) -> bool:
        return (self.relative_path, self.url) < (
            other.relative_path,
            other.url,
        )


@dataclasses.dataclass
class StatusOfChanges:
    """Changes that were applied or not applied."""

    applied: tuple[Change, ...]
    not_applied: tuple[Change, ...]


@dataclasses.dataclass(slots=False)
class CheckoutContext:
    _api: recipe_api.RecipeApi = dataclasses.field(repr=False)
    options: Options | None = None
    changes: list[Change] | None = None  # List of triggering changes.
    top: config_types.Path = None  # Actual checkout root.
    # Logical checkout root. Usually identical to 'top', but occasionally a
    # subdirectory instead.
    root: config_types.Path = None
    # Which triggering changes were applied or not applied.
    status: StatusOfChanges | None = None
    # Remotes that should be treated identically.
    equivalent_remotes: dict[str, list[str]] | None = dataclasses.field(
        default_factory=dict
    )
    manifest: Manifest | None = None  # Parsed repo manifest.
    # Path to a JSON file containing metadata about the triggering changes.
    changes_json: config_types.Path | None = None
    bazel_overrides: dict[str, config_types.Path] = dataclasses.field(
        default_factory=dict
    )

    # Current revision number.
    def revision(self) -> str:
        if hasattr(self, '_revision'):
            return self._revision

        self._revision = self._api.checkout.get_revision(self.top)
        return self._revision

    @property
    def manifest_path(self) -> config_types.Path:
        return self.top / self.options.manifest_file

    def applied_changes(self) -> list[Change]:
        return [x for x in self.changes if x.applied]

    # Repo manifest with all projects pinned.
    def manifest_snapshot(self):
        if not self.options.use_repo:
            return None

        if hasattr(self, '_manifest_snapshot'):
            return self._manifest_snapshot

        with self._api.context(cwd=self.top):
            self._manifest_snapshot = self._api.repo.manifest_snapshot()
            return self._manifest_snapshot

    # Equivalent of manifest_snapshot() but not as strictly formatted.
    def submodule_snapshot(self):
        if self.options.use_repo:
            return None

        if hasattr(self, '_submodule_snapshot'):
            return self._submodule_snapshot

        with self._api.context(cwd=self.top):
            # To get step_test_data line to pass pylint.
            raw_io_stream_output = self._api.raw_io.test_api.stream_output_text

            self._submodule_snapshot = (
                self._api.git(
                    'submodule-status',
                    'submodule',
                    'status',
                    '--recursive',
                    stdout=self._api.raw_io.output_text(),
                    step_test_data=lambda: raw_io_stream_output(
                        'submodule status filler text',
                    ),
                    ok_ret='any',
                ).stdout.strip()
                or ''
            )
            return self._submodule_snapshot

    def snapshot_to_dir(self, directory: config_types.Path) -> None:
        self._api.file.ensure_directory('mkdir', directory)
        if self.manifest_snapshot():
            self._api.file.write_text(
                'write manifest.xml',
                directory / 'manifest.xml',
                self.manifest_snapshot(),
            )

        if self.submodule_snapshot():
            self._api.file.write_text(
                'write submodule snapshot',
                directory / 'submodules.log',
                self.submodule_snapshot(),
            )

        with self._api.context(cwd=self.top):
            log = self._api.git(
                'log',
                'log',
                '--oneline',
                '-n',
                '10',
                stdout=self._api.raw_io.output_text(),
                ok_ret='any',
            ).stdout
        self._api.file.write_text(
            'write git log',
            directory / 'git.log',
            log,
        )

    def submodules(self, recursive: bool = False) -> list[Submodule]:
        """Return data about all submodules."""

        cmd = [
            'python3',
            self._api.checkout.resource('submodule_status.py'),
            self.top,
            self._api.json.output(),
        ]

        if recursive:
            cmd.append('--recursive')

        submodules = []
        submodule_status = self._api.step(
            'submodule status',
            cmd,
            step_test_data=lambda: self._api.json.test_api.output({}),
        ).json.output
        for sub in submodule_status.values():
            sub['remote'] = self._api.sso.sso_to_https(sub['remote'])
            if sub['remote'].endswith('.git'):
                sub['remote'] = sub['remote'][:-4]
            sub['relative_path'] = sub['path']
            sub['path'] = self.top / sub['path']
            try:
                submodules.append(Submodule(self._api, **sub))
            except Exception as exc:  # pragma: no cover
                self._api.step.empty(
                    'exception',
                    status='FAILURE',
                    raise_on_failure=False,
                ).presentation.step_summary_text = repr(exc)
                raise ValueError(f'submodule {sub["path"]}: {sub}')

        return submodules

    _REMOTE_REGEX = re.compile(r'^https://(?P<host>[^/]+)/(?P<project>.+)$')

    def gerrit_host(self) -> str | None:
        match = self._REMOTE_REGEX.match(self.options.remote)
        if not match:
            return  # pragma: no cover

        gerrit_review_host = f'{match.group("host")}'
        if '-review' not in gerrit_review_host:
            gerrit_review_host = gerrit_review_host.replace('.', '-review.', 1)
        return gerrit_review_host

    def gerrit_project(self) -> str | None:
        match = self._REMOTE_REGEX.match(self.options.remote)
        if not match:
            return  # pragma: no cover

        return match.group('project')

    def remotes_equivalent(
        self,
        remote1: str | None,
        remote2: str | None,
    ) -> bool:
        return self._api.checkout.remotes_equivalent(
            self.equivalent_remotes,
            remote1,
            remote2,
        )


class CheckoutApi(recipe_api.RecipeApi):
    """Calls to checkout code."""

    Change = Change
    CheckoutContext = CheckoutContext

    def fake_context(self):  # pragma: no cover
        ctx = CheckoutContext(self.m)
        ctx.top = ctx.root = self.m.path.start_dir / 'checkout'
        ctx.options = Options(remote=self.test_api.pigweed_repo)
        return ctx

    def _read_manifest(
        self, manifest_remote: str, manifest_file: str
    ) -> Manifest:
        """Reads manifest file to get git repo locations."""

        with self.m.step.nest('read manifest') as read_step:
            manifest_text: str = self.m.file.read_text(
                'read file', manifest_file
            )
            read_step.logs['raw'] = manifest_text

            xml_tree = xml.etree.ElementTree.fromstring(manifest_text)

            manifest = Manifest()

            for remote in xml_tree.iter('remote'):
                with self.m.step.nest('log') as pres:
                    pres.step_summary_text = repr(remote.attrib)
                remote = Remote(**remote.attrib)
                if remote.fetch.startswith('..'):
                    rest = re.sub(r'^(..)(/..)*', '', remote.fetch)
                    parsed = urllib.parse.urlparse(manifest_remote)
                    remote.fetch = f'{parsed.scheme}://{parsed.netloc}' + rest
                remote.fetch = Url(remote.fetch)
                remote.fetch.https = self.m.sso.sso_to_https(remote.fetch.url)
                manifest.remotes[remote.name] = remote

            defaults = {}
            for default in xml_tree.iter('default'):
                defaults.update(default.attrib)

            for project in xml_tree.iter('project'):
                name = project.attrib['name']
                path = project.attrib.get('path', name)

                if 'remote' in project.attrib:
                    remote = project.attrib['remote']
                elif 'remote' in defaults:
                    remote = defaults['remote']
                else:  # pragma: no cover
                    assert False, f'remote not specified for {name}'

                assert remote in manifest.remotes, (
                    f'Remote {remote} does not exist'
                )

                if 'revision' in project.attrib:
                    revision = project.attrib['revision']
                elif manifest.remotes[remote].revision:
                    revision = manifest.remotes[remote].revision
                elif 'revision' in defaults:
                    revision = defaults['revision']
                else:  # pragma: no cover
                    assert False, f'revision not specified for {name}'

                if 'upstream' in project.attrib:
                    upstream = project.attrib['upstream']
                elif 'upstream' in defaults:  # pragma: no cover
                    # This is unlikely to be used and hard to test--it requires
                    # a completely separate manifest definition, otherwise the
                    # 'else' condition won't be covered. It's also simple.
                    upstream = defaults['upstream']
                else:
                    upstream = revision

                # urllib.urljoin does something different than what's desired
                # here.
                url = '/'.join(
                    (
                        manifest.remotes[remote].fetch.https.rstrip('/'),
                        name.lstrip('/'),
                    )
                )
                manifest.projects.append(
                    Project(
                        name=name,
                        path=path,
                        remote=remote,
                        revision=revision,
                        upstream=upstream,
                        url=url,
                    )
                )

            self.m.file.write_json(
                'manifest json',
                self.m.path.start_dir / 'manifest.json',
                manifest.dict(),
            )

        return manifest

    def _process_gerrit_change(
        self,
        ctx: CheckoutContext,
        host: str,
        change_id: str | int,
        project: str = 'pigweed/pigweed',
        patchset: int | None = None,
        primary: bool = False,
    ) -> Change:
        """Process a LUCI GerritChange and return a Change object."""

        host = self.m.gerrit.normalize_host(host)
        gitiles_host = host.replace(
            '-review.googlesource.com', '.googlesource.com'
        )
        gerrit_name = gitiles_host.split('.')[0]
        details = self.m.gerrit.change_details(
            'details',
            change_id=str(change_id),
            host=host,
            max_attempts=5,
            query_params=[
                'CURRENT_COMMIT',
                'CURRENT_REVISION',
                'SUBMIT_REQUIREMENTS',
            ],
            timeout=30,
            test_data=self.m.json.test_api.output(
                {
                    'branch': 'main',
                    'current_revision': 'f' * 40,
                    'revisions': {
                        'f' * 40: {
                            '_number': 3,
                            'commit': {
                                'parents': [{}],
                                'message': '',
                            },
                            'uploader': {
                                'email': 'noreply@google.com',
                            },
                        }
                    },
                    'project': project,
                    'submit_requirements': [
                        {
                            'name': 'Code-Review',
                            'status': 'OVERRIDDEN',
                        },
                    ],
                }
            ),
        ).json.output
        branch = details['branch']

        remote = f'https://{gitiles_host}/{details["project"]}'.strip('/')
        rebase = not ctx.options.force_no_rebase

        current_revision = details['revisions'][details['current_revision']]
        is_merge = len(current_revision['commit']['parents']) > 1
        if is_merge:
            rebase = False

        if not patchset:
            patchset = current_revision['_number']

        ref = f'refs/changes/{change_id % 100:02}/{change_id}/{patchset}'

        submit_requirements: dict[str, str] = {}
        for req in details.get('submit_requirements', ()):
            submit_requirements[req['name']] = req['status']

        uploader_info: dict[str, Any] = current_revision.get('real_uploader')
        if not uploader_info:
            uploader_info = current_revision.get('uploader')
            if not uploader_info and self.m.recipe_testing.enabled:
                uploader_info = {  # pragma: no cover
                    'email': 'noreply@google.com',
                }

        uploader: str | None = None
        if uploader_info:
            uploader = uploader_info['email']

        return Change(
            number=int(change_id),
            patchset=patchset,
            remote=remote,
            ref=ref,
            rebase=rebase,
            is_merge=is_merge,
            branch=branch,
            gerrit_name=gerrit_name,
            submitted=False,
            commit_message=current_revision['commit']['message'],
            project=details['project'],
            topic=details.get('topic', None) or None,
            current_revision=details['current_revision'],
            primary=primary,
            submit_requirements=submit_requirements,
            uploader=uploader,
        )

    def _process_gerrit_changes(
        self,
        ctx: CheckoutContext,
        bb_input: build_pb.Build.Input,
    ) -> Generator[Change, None, None]:
        seen = set()
        for i, change in enumerate(bb_input.gerrit_changes):
            with self.m.step.nest(str(i)):
                result = self._process_gerrit_change(
                    ctx=ctx,
                    host=change.host,
                    project=change.project,
                    change_id=change.change,
                    patchset=change.patchset,
                    primary=True,
                )
                yield result
                seen.add(result.name)

        cq_deps_result = self.m.cq_deps.resolve(
            result.gerrit_name,
            result.number,
            result.topic,
        )
        for dep in cq_deps_result.resolved:
            # dep.name should only appear in seen if there are multiple
            # gerrit_changes from buildbucket and a later one depends on an
            # earlier one. If buildbucket has multiple gerrit_changes the
            # cq_deps module is not needed here, so this is just double-checking
            # something that shouldn't happen.
            if dep.name in seen:  # pragma: no cover
                continue
            seen.add(dep.name)
            yield self._process_gerrit_change(
                ctx=ctx,
                host=dep.host,
                project=dep.project,
                change_id=dep.change,
                primary=False,
            )

        for cl in cq_deps_result.unresolved:
            yield Change(
                number=cl.change,
                remote=None,
                ref=None,
                rebase=None,
                project=None,
                branch=None,
                gerrit_name=cl.gerrit_name,
                submitted=False,
            )

    def number_for_hash(
        self,
        host: str,
        commit_hash: str,
        branch: str = 'main',
    ) -> dict[str, Any]:
        if 'github.com' in host or 'github-review' in host:
            return None  # pragma: no cover

        try:
            results = self.m.gerrit.change_query(
                'number',
                f'commit:{commit_hash}',
                host=host,
                max_attempts=5,
                timeout=30,
                test_data=self.m.json.test_api.output(
                    [
                        {
                            '_number': '1234',
                            'branch': branch,
                            'project': 'pigweed',
                        }
                    ]
                ),
            ).json.output
            # Skip this change if it didn't go through Gerrit.
            if results and len(results) == 1:
                return results[0]
        except self.m.step.StepFailure:  # pragma: no cover
            pass

        return None

    def _change_data(
        self,
        ctx: CheckoutContext,
        remote: str = None,
        branch: str = None,
    ) -> tuple[Change, ...]:
        bb_input: build_pb.Build.Input = self.m.buildbucket.build.input
        results: list[Change] = []

        triggers: dict[str, dict[str, triggers_pb.Trigger]] = (
            collections.defaultdict(dict)
        )
        for trigger in self.m.scheduler.triggers:
            gitiles: triggers_pb.GitilesTrigger = trigger.gitiles
            if gitiles:
                triggers[gitiles.repo][gitiles.revision] = trigger

        with self.m.step.nest('change data'):
            if bb_input.gerrit_changes:
                with self.m.step.nest('process gerrit changes'):
                    results.extend(self._process_gerrit_changes(ctx, bb_input))

            elif bb_input.gitiles_commit.id:
                with self.m.step.nest('process gitiles commit'):
                    commit: common_pb.GitilesCommit = bb_input.gitiles_commit
                    assert commit.host
                    if commit.project:
                        remote: str = f'https://{commit.host}/{commit.project}'

                    host: str = commit.host.replace(
                        '.googlesource.com', '-review.googlesource.com'
                    )
                    gerrit_name: str = commit.host.split('.')[0]

                    result: dict[str, Any] = self.number_for_hash(
                        host, commit.id
                    )

                    if result:
                        branch: str = result['branch']
                        if commit.id in triggers[remote]:
                            branch = triggers[remote][commit.id].gitiles.ref
                            branch = branch.removeprefix('refs/heads/')

                        results.append(
                            Change(
                                number=result['_number'],
                                remote=remote,
                                ref=commit.id,
                                rebase=False,
                                branch=branch,
                                gerrit_name=gerrit_name,
                                submitted=True,
                                project=result['project'],
                                current_revision=commit.id,
                                primary=True,
                            )
                        )

            if not results:
                # If not triggered by a gitiles_poller gitiles_commit may be
                # empty. In that case treat the most recent commit on the
                # remote as the triggering commit. This is a good assumption
                # except for Android Repo Tool projects, unless all projects
                # are pinned to commits instead of tracking branches. However,
                # even if this is wrong it's close enough to have utility.
                head: str = self.m.git.get_remote_branch_head(remote, branch)
                gerrit_name: str = urllib.parse.urlparse(remote).netloc.split(
                    '.'
                )[0]
                host: str = f'{gerrit_name}-review.googlesource.com'
                result: dict[str, Any] = self.number_for_hash(host, head)

                results.append(
                    Change(
                        number=result['_number'] if result else 0,
                        remote=remote,
                        ref=head,
                        rebase=False,
                        branch=result['branch'] if result else branch,
                        gerrit_name=gerrit_name,
                        project=None,
                        submitted=True,
                    )
                )

            with self.m.step.nest('changes'):
                for result in results:
                    with self.m.step.nest(result.name) as change_data_pres:
                        change_data_pres.step_summary_text = repr(result)

            return tuple(results)

    def _matching_branches(
        self,
        repo: str,
        branches: Sequence[str],
        name: str = 'has branch',
        **kwargs,
    ):
        """Returns the subset of the given branches that exist on gitiles."""
        matches: set[str] = set()
        with self.m.step.nest(name), self.m.context(infra_steps=True):
            for branch in branches:
                head: str = self.m.git.get_remote_branch_head(
                    repo,
                    branch,
                    step_name=f'git ls-remote {branch}',
                    step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(
                        ''
                    ),
                    **kwargs,
                )
                with self.m.step.nest('head') as pres:
                    pres.step_summary_text = repr(head)

                if head:
                    matches.add(branch)
        return sorted(matches)

    def _apply_change(
        self,
        ctx: CheckoutContext,
        change: Change,
        cwd: config_types.Path = None,
    ):
        """Applies the given change to the given directory.

        Args:
            ctx: Checkout context object.
            change: Change to apply.
            cwd: Working directory, defaults to current directory.
        """
        with self._apply_change_context(ctx=ctx, change=change, cwd=cwd):
            pass

    @contextlib.contextmanager
    def _apply_change_context(
        self,
        ctx: CheckoutContext,
        change: Change,
        cwd: config_types.Path = None,
    ) -> Iterator[None]:
        """Applies the given change to the given directory.

        Args:
            ctx: Checkout context object.
            change: Change to apply.
            cwd: Working directory, defaults to current directory.
        """
        kwargs: dict[str, Any] = {'cwd': cwd} if cwd else {}
        change.applied = True
        change.path = self.m.path.relpath(cwd or ctx.top, ctx.top)

        try:
            apply_step: str = f'apply {change.name}'
            with self.m.context(**kwargs), self.m.step.nest(apply_step) as pres:
                pres.links['gerrit'] = change.gerrit_url
                pres.links['gitiles'] = change.gitiles_url
                if cwd:
                    pres.step_summary_text = str(
                        self.m.path.relpath(cwd, ctx.top)
                    )

                with self.m.context(infra_steps=True):
                    # 'git fetch' fails if a submodule pin in the patch isn't
                    # present in the remote (for example, if the pin is only
                    # present in the uploader's workspace). Use
                    # '--no-recurse-submodules' here so 'git fetch' doesn't fail
                    # but instead 'git rebase' or 'git submodule update' fails
                    # later (important because those are not infra steps). Also
                    # don't use '--recurse-submodules' in 'git checkout' for
                    # similar reasons.
                    with self.m.default_timeout():
                        self.m.git.fetch(
                            change.remote,
                            change.ref,
                            recurse_submodules=False,
                            step_name='git fetch patch',
                        )
                    self.m.git(
                        'git checkout patch',
                        'checkout',
                        '--force',
                        '-b',
                        'working',
                        'FETCH_HEAD',
                    )

                # These remain unused if change.submitted is False.
                remote: str | None = None
                remote_branch: str | None = None

                with self.m.context(infra_steps=True):
                    # Change "https://foo.googlesource.com/bar"
                    #     to "https___foo_googlesource_com_bar".
                    # In Android Repo Tool projects, the remote for the manifest
                    # is often configured in a way that seems incorrect. Instead
                    # of relying on it, create a whole new remote every time
                    # that is always correct.
                    remote = re.sub(r'[^\w]', '_', change.remote)
                    remote_branch = '/'.join((remote, change.branch))
                    self.m.git(
                        'git remote add',
                        'remote',
                        'add',
                        remote,
                        change.remote,
                    )

                    with self.m.default_timeout():
                        self.m.git.fetch(
                            remote,
                            f'refs/heads/{change.branch}',
                            prune=False,
                            step_name='git fetch branch',
                        )

                    self.m.git(
                        'git set upstream',
                        'branch',
                        f'--set-upstream-to={remote_branch}',
                    )

                if not change.submitted:
                    with self.m.context(infra_steps=True):
                        self.m.git(
                            'pre-rebase log', 'log', '--oneline', '-n', '10'
                        )

                if change.submitted:
                    change.base = self.m.git.rev_parse(
                        'HEAD',
                        step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(
                            'HEAD_' * 8,
                        ),
                    )
                    change.base_type = 'submitted_commit_hash'

                elif change.rebase:
                    self.m.git('git rebase', 'rebase', remote_branch)

                    change.base = self.m.git.rev_parse(
                        remote_branch,
                        step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(
                            'REMOTE_BRANCH_' * 3,
                        ),
                    )
                    change.base_type = 'remote_branch_tip'

                else:
                    change.base = self.m.git(
                        'merge-base',
                        'merge-base',
                        'HEAD',
                        remote_branch,
                        stdout=self.m.raw_io.output_text(),
                        step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(
                            'MERGEBASE_' * 4,
                        ),
                    ).stdout
                    change.base_type = 'merge-base'

                with self.m.context(infra_steps=True):
                    self.m.git(
                        'post-rebase log', 'log', '--oneline', '-n', '10'
                    )

                # In most cases this is redundant or unnecessary, but it
                # shouldn't cause problems. It's necessary when a superproject
                # CL is updating a submodule pin and we need to sync the
                # submodule to the new revision.
                with self.m.default_timeout():
                    # See b/243673776 for why we detach before updating
                    # submodules.
                    self.m.git('detach', 'checkout', '--detach')
                    self.m.git.submodule_sync(recursive=True)
                    self.m.git.submodule_update(
                        recursive=True,
                        timeout=ctx.options.submodule_timeout_sec,
                    )
                    self.m.git('reattach', 'checkout', '-')

                yield

        finally:
            pass

    def _check_unapplied_changes(self, changes: Sequence[Change]):
        applied: list[Change] = []
        failed_to_apply: list[Change] = []
        if not changes:  # pragma: no cover
            return None

        def handle_unapplied_change(change):
            with self.m.step.nest(f'failed to apply {change.name}') as pres:
                pres.status = 'WARNING'
                pres.links['gerrit'] = change.gerrit_url
                pres.links['gitiles'] = change.gitiles_url
            failed_to_apply.append(change)

        with self.m.context(infra_steps=True):
            if all(not x.applied for x in changes):
                with self.m.step.nest('no changes were applied') as pres:
                    pres.status = 'FAILURE'
                    for change in changes:
                        handle_unapplied_change(change)
                    pres.properties['changes'] = [x.name for x in changes]

                raise self.m.step.InfraFailure(
                    'could not find triggering changes in checkout'
                )

            elif any(not x.applied for x in changes):
                with self.m.step.nest('some changes were not applied') as pres:
                    pres.status = 'WARNING'
                    for change in changes:
                        if change.applied:
                            applied.append(change)
                        else:
                            handle_unapplied_change(change)

            else:
                applied.extend(changes)

        with self.m.step.nest('status') as pres:
            pres.step_summary_text = (
                f'applied {applied}\nnot applied {failed_to_apply}'
            )

        return StatusOfChanges(
            applied=tuple(applied),
            not_applied=tuple(failed_to_apply),
        )

    def _cached_checkout(
        self,
        remote: str,
        path: config_types.Path,
        ref: str,
        submodules: bool,
        included_submodules: Sequence[str] | None = None,
        excluded_submodules: Sequence[str] | None = None,
        submodule_timeout_sec: int = 10 * 60,
        cache: bool = True,
        use_packfiles: bool = False,
        **kwargs,
    ):
        submodule_paths = included_submodules = included_submodules or []

        if cache:
            with self.m.step.nest('cache') as pres, self.m.cache.guard('git'):
                parsed_remote = urllib.parse.urlparse(remote)
                cache_name = (
                    parsed_remote.hostname
                    + parsed_remote.path.replace('-', '--').replace('/', '-')
                )
                cache_path = self.m.path.cache_dir / 'git' / cache_name
                self.m.file.ensure_directory('makedirs', cache_path)

                with self.m.context(cwd=cache_path):
                    dotgit = cache_path / '.git'
                    if self.m.path.exists(dotgit):  # pragma: no cover
                        hit_pres = self.m.step.empty('hit').presentation
                        hit_pres.properties[f'checkout {remote}'] = 'cache hit'
                        self.m.git.config_remove_section(
                            'remote.origin', **kwargs
                        )
                        pres.step_summary_text = 'hit'
                    else:
                        miss_pres = self.m.step.empty('miss').presentation
                        miss_pres.properties[f'checkout {remote}'] = (
                            'cache miss'
                        )
                        self.m.git.init(bare=False, **kwargs)
                        pres.step_summary_text = 'miss'

                    self.m.git.config(
                        'remote.origin.url',
                        remote,
                        step_name='remote set-url',
                        **kwargs,
                    )

                    if use_packfiles:
                        self.m.git.config(
                            'fetch.uriprotocols',
                            'https',
                            step_name='enable packfiles',
                            **kwargs,
                        )
                    else:
                        self.m.git.config(
                            'fetch.uriprotocols',
                            '',
                            step_name='disable packfiles',
                            **kwargs,
                        )

                    with self.m.default_timeout():
                        try:
                            self.m.git.fetch(
                                repository='origin',
                                prune=True,
                                tags=True,
                                recurse_submodules=submodules,
                                **kwargs,
                            )

                        # If the checkout failed save the git config. It might
                        # not be helpful, but it shouldn't hurt.
                        except self.m.step.StepFailure:
                            with self.m.step.nest('git config'):
                                self.m.git.config(
                                    '--list', '--local', step_name='local'
                                )
                                self.m.git.config(
                                    '--list', '--global', step_name='global'
                                )
                            raise

                    self.m.git.merge(ref='FETCH_HEAD', **kwargs)

                    if included_submodules and excluded_submodules:
                        raise self.m.step.InfraFailure(
                            'cannot specify both included_submodules and '
                            'excluded_submodules'
                        )

                    submodule_paths = included_submodules
                    if excluded_submodules:
                        submodule_status = self.m.git(
                            'submodule status',
                            'submodule',
                            'status',
                            stdout=self.m.raw_io.output_text(),
                            step_test_data=lambda: self.test_api.m.raw_io.stream_output_text(
                                '-0000000000000000000000000000000000000000 pigweed (abc123)\n'
                                '-1111111111111111111111111111111111111111 nanopb (heads/branch)\n'
                            ),
                        ).stdout.splitlines()

                        submodule_paths = [
                            x.split()[1] for x in submodule_status
                        ]
                        for sub in excluded_submodules:
                            if sub not in submodule_paths:
                                raise self.m.step.InfraFailure(
                                    f'excluded submodule {sub} is not a submodule'
                                )
                            with self.m.step.nest(f'excluding submodule {sub}'):
                                pass
                            submodule_paths.remove(sub)

                    for sub in submodule_paths:
                        with self.m.step.nest(f'including submodule {sub}'):
                            pass

                    self.m.git.submodule_sync(recursive=True, **kwargs)
                    if submodules or submodule_paths:
                        with self.m.default_timeout():
                            self.m.git.submodule_update(
                                recursive=True,
                                force=True,
                                paths=submodule_paths,
                                timeout=submodule_timeout_sec,
                                **kwargs,
                            )

                    if not submodules:
                        # Even though submodules weren't requested, if the cache
                        # had any active submodules we need to update them.
                        # Otherwise we'll get weird situations in rolls where an
                        # uninvolved submodule will be rolled back.
                        with self.m.default_timeout():
                            self.m.git.submodule_update(
                                recursive=True,
                                force=True,
                                init=False,
                                timeout=submodule_timeout_sec,
                                **kwargs,
                            )

            self.m.file.copytree(
                'copy from cache', cache_path, path, symlinks=True
            )

        # Deliberately not combining contexts into one line so it's obvious to
        # both devs and Python which one is "outer" and which is "inner".
        with self.m.step.nest('git checkout'):
            with self.m.default_timeout():
                self.m.git_checkout(
                    repo=remote,
                    path=path,
                    cache=False,
                    revision=ref,
                    recursive=submodules,
                    submodules=submodules,
                    submodule_force=submodules,
                    submodule_paths=submodule_paths,
                    submodule_timeout=submodule_timeout_sec,
                    step_name='',
                    use_packfiles=use_packfiles,
                )

    def _git(self, ctx: CheckoutContext):
        """Checkout code from git."""

        super_branch = self._matching_branch(ctx) or ctx.options.branch

        with self.m.context(infra_steps=True):
            cache = not ctx.options.do_not_cache
            build = self.m.buildbucket.build
            if 'pigweed.disable_git_cache' in build.input.experiments:
                cache = False

            self._cached_checkout(
                ctx.options.remote,
                path=ctx.top,
                ref=super_branch,
                cache=cache,
                submodules=ctx.options.initialize_submodules,
                submodule_timeout_sec=ctx.options.submodule_timeout_sec,
                included_submodules=ctx.options.included_submodules,
                excluded_submodules=ctx.options.excluded_submodules,
                use_packfiles=not ctx.options.do_not_use_packfiles,
            )

        with self.m.context(cwd=ctx.top):
            got_revision = None
            got_revision_type = 'no_trigger'

            submodules = []

            if ctx.options.use_trigger:
                got_revision = self.m.git.rev_parse(
                    'HEAD',
                    step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(
                        'HEAD' * 10,
                    ),
                )

                # Check for CLs for the top-level repository.
                for change in ctx.changes:
                    if ctx.remotes_equivalent(
                        ctx.options.remote, change.remote
                    ):
                        self._apply_change(ctx, change)
                        got_revision = change.base
                        got_revision_type = change.base_type

                submodules = ctx.submodules(recursive=True)

                # Check for CLs for submodules.
                # There are three rough cases:
                # 1. Zero submodules have matching remotes. In this case we do
                #    nothing.
                # 2. Exactly one submodule has a matching remote. In this case,
                #    the change is applied to that submodule, even if the branch
                #    does not match.
                # 3. Multiple submodules have matching remotes.
                #    1. Exactly one of them matches the triggering change's
                #       branch. In this case, the change is applied to this
                #       submodule.
                #    2. Zero or multiple submodules match both the remote and
                #       the branch of the triggering change. In these cases we
                #       error out.
                for change in ctx.changes:
                    with self.m.step.nest(f'matching {change.name}') as pres:
                        pres.links['gerrit'] = change.gerrit_url
                        pres.links['gitiles'] = change.gitiles_url

                        matching_submodules = []
                        for submodule in submodules:
                            if submodule.initialized:
                                if ctx.remotes_equivalent(
                                    submodule.remote, change.remote
                                ):
                                    self.m.step.empty(
                                        f'match: {submodule.path} '
                                        f'({submodule.remote})'
                                    )
                                    matching_submodules.append(submodule)
                                else:
                                    self.m.step.empty(
                                        f'no match: {submodule.path} '
                                        f'({submodule.remote})'
                                    )

                        if not matching_submodules:
                            pres.step_summary_text = 'no matching submodules'
                            continue

                        if len(matching_submodules) > 1:
                            submodule_info = ', '.join(
                                f'{self.m.path.relpath(sub.path, ctx.top)} '
                                f'(branch {sub.branch})'
                                for sub in matching_submodules
                            )

                            matching_with_branch = []
                            for submodule in matching_submodules:
                                if submodule.branch == change.branch:
                                    matching_with_branch.append(submodule)

                            if len(matching_with_branch) == 1:
                                pres.step_summary_text = (
                                    'one matching submodule'
                                )
                                matching_submodules = matching_with_branch

                            elif len(matching_with_branch) > 1:
                                pres.step_summary_text = (
                                    'too many submodules match the branch'
                                )
                                raise self.m.step.StepFailure(
                                    f'change {change.name} (branch '
                                    f'{change.branch}) matches multiple '
                                    f'submodules ({submodule_info}), but too '
                                    'many branches match'
                                )

                            else:
                                pres.step_summary_text = (
                                    'zero submodules match the branch'
                                )
                                raise self.m.step.StepFailure(
                                    f'change {change.name} '
                                    f'(branch {change.branch}) matches '
                                    f'multiple submodules ({submodule_info}) '
                                    'but no branches match'
                                )

                    if len(matching_submodules) == 1:
                        submodule = matching_submodules[0]
                        if not ctx.options.initialize_submodules:
                            with self.m.default_timeout():
                                self.m.git.submodule_sync(recursive=True)
                                self.m.git.submodule_update(
                                    paths=(submodule.path,)
                                )
                        self._apply_change(ctx, change, cwd=submodule.path)

                def _vars_primitive_only(x):
                    return {
                        k: v
                        for k, v in to_dict(x).items()
                        if isinstance(v, (int, str, bool, type(None)))
                    }

                applied_changes = [
                    _vars_primitive_only(x) for x in ctx.changes if x.applied
                ]
                ctx.changes_json = self.m.path.mkstemp()
                self.m.file.write_json(
                    'write changes.json',
                    ctx.changes_json,
                    applied_changes,
                )

            # Run git log for both the top-level checkout and every submodule.
            with self.m.step.nest('git log'):
                self.m.git(str(ctx.top), 'log', '--oneline', '-n', '10')
                for submodule in sorted(submodules):
                    with self.m.context(cwd=submodule.path):
                        self.m.git(
                            str(submodule.path),
                            'log',
                            '--oneline',
                            '-n',
                            '10',
                        )

            if got_revision:
                with self.m.step.nest('base') as pres:
                    pres.properties['got_revision'] = got_revision
                    # got_revision_type isn't needed by anything but helps
                    # explain why got_revision is the value it is.
                    pres.properties['got_revision_type'] = got_revision_type

    def _matching_branch(self, ctx: CheckoutContext):
        """Return if there are manifest branches that match the triggering CLs.

        If the triggering change is on a branch name that is also present in the
        manifest or superproject remote, use that branch when checking out the
        project.

        Args:
            ctx (CheckoutContext): Context object.

        Raises:
            StepFailure if there are multiple matching branches.

        Returns:
            One matching branch name, or None.
        """
        if not ctx.options.match_branch or not ctx.options.use_trigger:
            with self.m.step.nest('not matching branch names'):
                return

        kind = 'manifest' if ctx.options.use_repo else 'superproject'

        manifest_branch = None
        branch_names = sorted(
            set(
                x.branch
                for x in ctx.changes
                if x.branch not in ('master', 'main', None)
            )
        )

        if not branch_names:
            with self.m.step.nest('no non-standard branch names'):
                return

        with self.m.step.nest('branch names') as pres:
            pres.step_summary_text = str(branch_names)

        matching_branches = self._matching_branches(
            ctx.options.remote, branch_names, name=f'{kind} has branch'
        )
        if not matching_branches:
            with self.m.step.nest('no branch names match'):
                return

        if len(matching_branches) > 1:
            with self.m.step.nest(
                f'too many matching branches ({", ".join(matching_branches)})'
            ) as pres:
                pres.step_summary_text = (
                    "Can't figure out which {} branch to use. Remove some "
                    '"Requires:" lines to simplify the checkout.'.format(kind)
                )
                raise self.m.step.StepFailure('multiple matching branches')

        manifest_branch = matching_branches.pop()
        self.m.step(
            f'changing {kind} branch to {manifest_branch}',
            None,
        )
        return manifest_branch

    def _repo(self, ctx: CheckoutContext):
        """Checkout code from an Android Repo Tool manifest."""

        # Git makes the top-level folder, Repo requires caller to make it.
        self.m.file.ensure_directory('mkdir checkout', ctx.top)

        with self.m.context(cwd=ctx.top):
            manifest_branch = self._matching_branch(ctx) or ctx.options.branch

            with self.m.context(infra_steps=True):
                kwargs = {}
                if ctx.options.repo_init_timeout_sec:
                    kwargs['timeout'] = ctx.options.repo_init_timeout_sec
                    kwargs['attempts'] = ctx.options.number_of_attempts
                if ctx.options.manifest_groups:
                    kwargs['groups'] = ctx.options.manifest_groups

                self.m.repo.init(
                    manifest_url=ctx.options.remote,
                    manifest_branch=manifest_branch,
                    manifest_name=ctx.options.manifest_file,
                    clone_bundle=not ctx.options.repo_no_clone_bundle,
                    **kwargs,
                )

            manifests_dir = ctx.top / '.repo' / 'manifests'
            # If the triggering CL is a manifest change, apply it before running
            # sync.
            if ctx.options.use_trigger:
                for change in ctx.changes:
                    if change.remote and ctx.remotes_equivalent(
                        ctx.options.remote, change.remote
                    ):
                        with self._apply_change_context(
                            ctx, change, cwd=manifests_dir
                        ):
                            # Right now the upstream of 'working' is the local
                            # 'default' branch. 'repo sync' complains if the
                            # upstream isn't remote, so it's changed to the
                            # remote branch that's identical to 'default'.
                            self.m.git(
                                'git branch',
                                'branch',
                                f'--set-upstream-to=origin/{manifest_branch}',
                            )

            ctx.manifest = self._read_manifest(
                ctx.options.remote,
                manifests_dir / ctx.options.manifest_file,
            )

            for _, remote_host in sorted(ctx.manifest.remotes.items()):
                if remote_host.fetch.url.startswith('sso://'):
                    self.m.sso.configure_insteadof(remote_host.fetch.url)

            with self.m.context(infra_steps=True):
                kwargs = {}
                if ctx.options.repo_sync_timeout_sec:
                    kwargs['timeout'] = ctx.options.repo_sync_timeout_sec
                    kwargs['attempts'] = ctx.options.number_of_attempts
                self.m.repo.sync(
                    force_sync=True, current_branch=True, jobs=2, **kwargs
                )
                self.m.repo.start('base')

            if ctx.options.use_trigger:
                for change in ctx.changes:
                    for entry in ctx.manifest.projects:
                        if ctx.remotes_equivalent(entry.url, change.remote):
                            with self._apply_change_context(
                                ctx,
                                change,
                                cwd=entry.path_object(ctx.top),
                            ):
                                with self.m.step.nest(
                                    'compare branch name'
                                ) as pres:
                                    pres.step_summary_text = (
                                        'CL branch: {}\nupstream branch: {}'
                                    ).format(change.branch, entry.upstream)

    def _workspace(self, ctx: CheckoutContext):
        if not ctx.options.eligible_workspace_paths or all(
            x.applied for x in ctx.changes
        ):
            return

        repos_dir = self.m.path.start_dir / 'bazel_repos'
        self.m.file.ensure_directory(f'mkdir {repos_dir}', repos_dir)

        with self.m.step.nest('workspace'):
            for change in ctx.changes:
                if change.applied:
                    continue  # pragma: no cover

                with self.m.step.nest(change.name) as pres:
                    pres.links['gerrit'] = change.gerrit_url
                    pres.links['gitiles'] = change.gitiles_url

                    for workspace in ctx.options.eligible_workspace_paths:
                        workspace_path = ctx.root / workspace
                        self.m.path.mock_add_file(workspace_path)
                        if not self.m.path.isfile(workspace_path):
                            continue  # pragma: no cover

                        with self.m.step.nest(workspace) as ws_pres:
                            ws_pres.step_summary_text = str(workspace_path)

                            repos = self.m.bazel_roll.retrieve_attributes(
                                project_value=change.remote,
                                path=workspace_path,
                                matches=ctx.remotes_equivalent,
                                search_key='remote',
                            )

                            if not repos:
                                continue

                            # Things will be much simpler if we assume all
                            # entries with a url matching the change have
                            # identical remote URLs.
                            remotes = list(set(x['remote'] for x in repos))
                            assert len(remotes) == 1
                            remote = remotes[0]

                            name = remote
                            name = name.replace('http://', '')
                            name = name.replace('https://', '')
                            name = name.replace('sso://', '')
                            name = name.replace('.git.corp.google.com', '')
                            name = name.replace('.googlesource.com', '')
                            name = name.replace('/', '_')
                            path = repos_dir / name

                            self.m.git_checkout(
                                remote,
                                path=path,
                                step_name=f'checkout {name}',
                                ignore_build_input=True,
                                cache=False,
                            )

                            self._apply_change(ctx, change, path)

                            for repo in repos:
                                if 'strip_prefix' in repo:
                                    path = path / repo['strip_prefix']
                                ctx.bazel_overrides[repo['name']] = path

    def _check_gerrit_changes(self, ctx: CheckoutContext):
        """Check that we don't have multiple changes for the same repo."""
        seen = {}

        for change in ctx.changes:
            if change.submitted:
                continue

            for remote, prev_change in seen.items():
                if ctx.remotes_equivalent(change.remote, remote):
                    failure = (
                        f'Change {prev_change.name} and change {change.name} '
                        f'are from equivalent remotes ({prev_change.remote!r} '
                        f'and {change.remote!r}). This is not supported. '
                        'Instead, stack these changes, or only refer to the '
                        "top from a 'patches.json' file in another repository."
                    )
                    pres = self.m.step.empty(
                        'multiple changes for same repo',
                    ).presentation
                    pres.step_summary_text = failure
                    pres.status = 'FAILURE'
                    pres.links[prev_change.name] = prev_change.gerrit_url
                    pres.links[change.name] = change.gerrit_url
                    raise self.m.step.StepFailure(failure)

            seen[change.remote] = change

    def _configure_insteadof(self, ctx: CheckoutContext):
        """Configure git to use some urls in place of others."""
        if not ctx.options.rewrites:
            return

        with self.m.step.nest('insteadof'):
            for rewrite in ctx.options.rewrites:
                self.m.git(
                    f'{rewrite.original} to {rewrite.final}',
                    'config',
                    '--global',
                    '--add',
                    f'url.{rewrite.final}.insteadof',
                    rewrite.original,
                )

            self.m.git('rewrites', 'config', '--get-regexp', '^url.*')

    def _name(self, options: Options):
        """Turn "https://foo/bar/baz.git" into "baz"."""
        name = options.remote.rstrip('/').removesuffix('.git')
        parts = name.split('/')
        if options.use_repo and parts[-1] == 'manifest':
            parts.pop(-1)
        return f'checkout {parts[-1]}'

    class EquivalentRemotes(TypedDict):
        remotes: list[str]

    def load_equivalent_remotes(
        self,
        equivalent_remotes_input: list[EquivalentRemotes],
    ):
        equivalent_remotes = {}
        for remotes in equivalent_remotes_input:
            new_remotes = [
                self.m.sso.sso_to_https(x) for x in remotes['remotes']
            ]
            for remote in new_remotes:
                assert remote not in equivalent_remotes
                equivalent_remotes[remote] = new_remotes
        return equivalent_remotes

    def remotes_equivalent(
        self,
        equivalent_remotes: dict[str, list[str]],
        remote1: str | None,
        remote2: str | None,
    ) -> bool:
        # Sometimes remote1 or remote2 is None. In that case we shouldn't
        # convert sso to https.
        if remote1:
            remote1 = self.m.sso.sso_to_https(remote1).removesuffix('.git')
        if remote2:
            remote2 = self.m.sso.sso_to_https(remote2).removesuffix('.git')
        if remote1 == remote2:
            return True

        # TODO: b/400532999 - Remove hard-coded equivalency support.
        hardcoded_equivalent_remotes = (
            (
                'https://fuchsia.googlesource.com/third_party/pigweed.googlesource.com/pigweed/pigweed',
                'https://pigweed.googlesource.com/pigweed/pigweed',
            ),
        )

        def extend_unique(dest, to_add):
            for item in to_add:
                if item not in dest:
                    dest.append(item)

        # TODO: b/400532999 - Remove hard-coded equivalency support.
        for hardcoded_equivalent_group in hardcoded_equivalent_remotes:
            final_group = list(hardcoded_equivalent_group)
            for remote in hardcoded_equivalent_group:
                if remote in equivalent_remotes:
                    extend_unique(final_group, equivalent_remotes[remote])
            for remote in final_group:
                equivalent_remotes.setdefault(remote, [])
                extend_unique(equivalent_remotes[remote], final_group)

        return remote1 in equivalent_remotes.get(remote2, ())

    def __call__(
        self,
        options: Options,
        root: config_types.Path | None = None,
        name: str = None,
    ) -> CheckoutContext:
        """Checkout code."""

        checkout_name = name or self._name(options)

        assert options.remote

        initial_options = repr(options)
        options.manifest_file = options.manifest_file or 'default.xml'
        options.repo_init_timeout_sec = options.repo_init_timeout_sec or 20
        options.repo_sync_timeout_sec = options.repo_sync_timeout_sec or 2 * 60
        options.number_of_attempts = options.number_of_attempts or 3
        options.submodule_timeout_sec = options.submodule_timeout_sec or 10 * 60
        final_options = repr(options)

        ctx = CheckoutContext(self.m)
        ctx.options = options
        ctx.changes = []

        root = root or self.m.path.start_dir / 'co'
        ctx.top = root
        if ctx.options.root_subdirectory:
            ctx.root = ctx.top / ctx.options.root_subdirectory
        else:
            ctx.root = ctx.top

        ctx.equivalent_remotes = self.load_equivalent_remotes(
            json_format.MessageToDict(options).get('equivalentRemotes', []),
        )

        with self.m.step.nest(checkout_name) as pres:
            with self.m.step.nest('options') as options_pres:
                options_pres.step_summary_text = initial_options

            with self.m.step.nest('options with defaults') as options_pres:
                options_pres.step_summary_text = final_options

            if options.remote.endswith('.git'):
                options.remote = options.remote[:-4]

            if options.use_trigger:
                ctx.changes = self._change_data(
                    ctx, options.remote, options.branch
                )

            self._check_gerrit_changes(ctx)

            self._configure_insteadof(ctx)

            if options.use_repo:
                self._repo(ctx)

            else:
                self._git(ctx)

            self._workspace(ctx)

            if not options.use_repo:
                with self.m.context(cwd=ctx.top):
                    self.m.git.clean(force=2, recursive=True)
                    try:
                        self.m.git.status()
                    except self.m.step.StepFailure:  # pragma: no cover
                        pass

            ctx.status = self._check_unapplied_changes(ctx.changes)

            if ctx.status:
                for change in ctx.status.applied:
                    pres.links[f'applied {change.name_with_path}'] = (
                        change.gerrit_url
                    )

                for change in ctx.status.not_applied:
                    pres.links[f'failed to apply {change.name}'] = (
                        change.gerrit_url
                    )

            snapshot_dir = self.m.path.start_dir / 'snapshot'
            ctx.snapshot_to_dir(snapshot_dir)

        return ctx

    def get_revision(
        self,
        root: config_types.Path,
        name: str = 'git log',
        test_data: str = 'HASH',
    ):
        """Like self.revision, but works for secondary checkouts."""
        with self.m.context(cwd=root):
            step = self.m.git(
                name,
                'log',
                '--max-count=1',
                '--pretty=format:%H',
                stdout=self.m.raw_io.output_text(),
                step_test_data=lambda: self.test_api.m.raw_io.stream_output_text(
                    test_data,
                ),
            )

            # Sometimes test data has additional information, following a
            # newline. Only keep the part before the first newline.
            result = step.stdout.strip().split('\n')[0]
            step.presentation.step_summary_text = result
            return result