#!/usr/bin/env vpython3
# Copyright 2012 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Unit tests for gclient_scm.py."""

from io import StringIO
import json
import logging
import os
import re
from subprocess import Popen, PIPE, STDOUT
import sys
import tempfile
import unittest
from unittest import mock

import scm_mock

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import scm

import gclient_scm
import gclient_utils
import git_cache
import git_common
import subprocess2
from testing_support import fake_repos
from testing_support import test_case_utils

GIT = "git" if sys.platform != "win32" else "git.bat"

# Disable global git cache
git_cache.Mirror.SetCachePath(None)

# Shortcut since this function is used often
join = gclient_scm.os.path.join

TIMESTAMP_RE = re.compile(r"\[[0-9]{1,2}:[0-9]{2}:[0-9]{2}\] (.*)", re.DOTALL)


def strip_timestamps(value):
    lines = value.splitlines(True)
    for i in range(len(lines)):
        m = TIMESTAMP_RE.match(lines[i])
        if m:
            lines[i] = m.group(1)
    return "".join(lines)


class BasicTests(unittest.TestCase):
    def setUp(self) -> None:
        scm_mock.GIT(self)
        return super().setUp()

    def testGetFirstRemoteUrl(self):
        FAKE_PATH = "/fake/path"
        scm.GIT.SetConfig(
            FAKE_PATH, "remote.origin.url", "first-value", append=True
        )
        scm.GIT.SetConfig(
            FAKE_PATH, "remote.origin.url", "second-value", append=True
        )

        self.assertEqual(
            gclient_scm.SCMWrapper._get_first_remote_url(FAKE_PATH),
            "first-value",
        )

    def testCanonicalGitUrl(self):
        # Endpoint form folds cosmetic spellings of one fetch URL, but keeps
        # the /a/ auth path distinct so switching to it is still applied.
        canonical = "https://x.googlesource.com/repo"
        for url in (
            "https://x.googlesource.com/repo",
            "https://x.googlesource.com/repo/",
            "https://x.googlesource.com/repo.git",
            "https://x.googlesource.com/repo.git/",
            "git+https://x.googlesource.com/repo",
        ):
            self.assertEqual(gclient_scm._canonical_git_url(url), canonical)
        self.assertNotEqual(
            gclient_scm._canonical_git_url("https://x.googlesource.com/a/repo"),
            canonical,
        )

    def testCanonicalGitRepo(self):
        # Identity form additionally folds the /a/ auth path.
        canonical = "https://x.googlesource.com/repo"
        for url in (
            "https://x.googlesource.com/repo",
            "https://x.googlesource.com/a/repo",
            "https://x.googlesource.com/a/repo.git/",
            "git+https://x.googlesource.com/a/repo.git",
        ):
            self.assertEqual(gclient_scm._canonical_git_repo(url), canonical)

    def testDeleteOrMoveRoot(self):
        root_dir = tempfile.mkdtemp()
        try:
            with open(os.path.join(root_dir, "some_file"), "w") as f:
                f.write("foo")
            wrapper = gclient_scm.SCMWrapper(
                url="git://foo",
                root_dir=root_dir,
                relpath=".",
                out_fh=StringIO(),
            )
            wrapper._DeleteOrMove(False)

            bad_scm_dir = os.path.join(root_dir, "_bad_scm")
            self.assertTrue(os.path.exists(bad_scm_dir))

            subdirs = os.listdir(bad_scm_dir)
            self.assertEqual(len(subdirs), 1)
            dest_path = os.path.join(bad_scm_dir, subdirs[0])
            self.assertTrue(subdirs[0].startswith("gclient_root"))

            self.assertTrue(
                os.path.exists(os.path.join(dest_path, "some_file"))
            )
            self.assertFalse(
                os.path.exists(os.path.join(root_dir, "some_file"))
            )
        finally:
            gclient_utils.rmtree(root_dir)

    def testDeleteOrMoveNonRoot(self):
        root_dir = tempfile.mkdtemp()
        try:
            relpath = "some_dir"
            checkout_path = os.path.join(root_dir, relpath)
            os.makedirs(checkout_path)
            with open(os.path.join(checkout_path, "some_file"), "w") as f:
                f.write("foo")

            wrapper = gclient_scm.SCMWrapper(
                url="git://foo",
                root_dir=root_dir,
                relpath=relpath,
                out_fh=StringIO(),
            )
            wrapper._DeleteOrMove(False)

            bad_scm_dir = os.path.join(root_dir, "_bad_scm")
            self.assertTrue(os.path.exists(bad_scm_dir))

            subdirs = os.listdir(bad_scm_dir)
            self.assertEqual(len(subdirs), 1)
            dest_path = os.path.join(bad_scm_dir, subdirs[0])
            self.assertTrue(subdirs[0].startswith("some_dir"))

            self.assertTrue(
                os.path.exists(os.path.join(dest_path, relpath, "some_file"))
            )
            self.assertFalse(os.path.exists(checkout_path))
        finally:
            gclient_utils.rmtree(root_dir)


class BaseGitWrapperTestCase(unittest.TestCase, test_case_utils.TestCaseUtils):
    """This class doesn't use pymox."""

    class OptionsObject(object):
        def __init__(self, verbose=False, revision=None):
            self.auto_rebase = False
            self.verbose = verbose
            self.revision = revision
            self.deps_os = None
            self.force = False
            self.reset = False
            self.nohooks = False
            self.no_history = False
            self.upstream = False
            self.cache_dir = None
            self.merge = False
            self.jobs = 1
            self.break_repo_locks = False
            self.delete_unversioned_trees = False
            self.patch_ref = None
            self.patch_repo = None
            self.rebase_patch_ref = True
            self.reset_patch_ref = True

    sample_git_import = """blob
mark :1
data 6
Hello

blob
mark :2
data 4
Bye

reset refs/heads/main
commit refs/heads/main
mark :3
author Bob <bob@example.com> 1253744361 -0700
committer Bob <bob@example.com> 1253744361 -0700
data 8
A and B
M 100644 :1 a
M 100644 :2 b

blob
mark :4
data 10
Hello
You

blob
mark :5
data 8
Bye
You

commit refs/heads/origin
mark :6
author Alice <alice@example.com> 1253744424 -0700
committer Alice <alice@example.com> 1253744424 -0700
data 13
Personalized
from :3
M 100644 :4 a
M 100644 :5 b
M 160000 1111111111111111111111111111111111111111 submodule

blob
mark :7
data 5
Mooh

commit refs/heads/feature
mark :8
author Bob <bob@example.com> 1390311986 -0000
committer Bob <bob@example.com> 1390311986 -0000
data 6
Add C
from :3
M 100644 :7 c

reset refs/heads/main
from :3
"""

    def Options(self, *args, **kwargs):
        return self.OptionsObject(*args, **kwargs)

    def checkstdout(self, expected):
        value = sys.stdout.getvalue()
        sys.stdout.close()
        # Check that the expected output appears.
        self.assertIn(expected, strip_timestamps(value))

    @staticmethod
    def CreateGitRepo(git_import, path):
        """Do it for real."""
        try:
            Popen(
                [GIT, "init", "-q"], stdout=PIPE, stderr=STDOUT, cwd=path
            ).communicate()
        except OSError:
            # git is not available, skip this test.
            return False
        Popen(
            [GIT, "fast-import", "--quiet"],
            stdin=PIPE,
            stdout=PIPE,
            stderr=STDOUT,
            cwd=path,
        ).communicate(input=git_import.encode())
        Popen(
            [GIT, "checkout", "-q"], stdout=PIPE, stderr=STDOUT, cwd=path
        ).communicate()
        Popen(
            [GIT, "remote", "add", "-f", "origin", "."],
            stdout=PIPE,
            stderr=STDOUT,
            cwd=path,
        ).communicate()
        Popen(
            [GIT, "checkout", "-b", "new", "origin/main", "-q"],
            stdout=PIPE,
            stderr=STDOUT,
            cwd=path,
        ).communicate()
        Popen(
            [GIT, "push", "origin", "origin/origin:origin/main", "-q"],
            stdout=PIPE,
            stderr=STDOUT,
            cwd=path,
        ).communicate()
        Popen(
            [GIT, "config", "--unset", "remote.origin.fetch"],
            stdout=PIPE,
            stderr=STDOUT,
            cwd=path,
        ).communicate()
        Popen(
            [GIT, "config", "user.email", "someuser@chromium.org"],
            stdout=PIPE,
            stderr=STDOUT,
            cwd=path,
        ).communicate()
        Popen(
            [GIT, "config", "user.name", "Some User"],
            stdout=PIPE,
            stderr=STDOUT,
            cwd=path,
        ).communicate()
        # Set HEAD back to main
        Popen(
            [GIT, "checkout", "main", "-q"],
            stdout=PIPE,
            stderr=STDOUT,
            cwd=path,
        ).communicate()
        return True

    def _GetAskForDataCallback(self, expected_prompt, return_value):
        def AskForData(prompt, options):
            self.assertEqual(prompt, expected_prompt)
            return return_value

        return AskForData

    def setUp(self):
        unittest.TestCase.setUp(self)
        test_case_utils.TestCaseUtils.setUp(self)
        self.url = "git://foo"
        # The .git suffix allows gclient_scm to recognize the dir as a git repo
        # when cloning it locally
        self.root_dir = tempfile.mkdtemp(".git")
        self.relpath = "."
        self.base_path = join(self.root_dir, self.relpath)
        self.enabled = self.CreateGitRepo(
            self.sample_git_import, self.base_path
        )
        mock.patch("sys.stdout", StringIO()).start()
        self.addCleanup(mock.patch.stopall)
        self.addCleanup(gclient_utils.rmtree, self.root_dir)


class ManagedGitWrapperTestCase(BaseGitWrapperTestCase):
    def testRevertMissing(self):
        if not self.enabled:
            return
        options = self.Options()
        file_path = join(self.base_path, "a")
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_list = []
        git_wrapper.update(options, None, file_list)
        gclient_scm.os.remove(file_path)
        file_list = []
        git_wrapper.revert(options, self.args, file_list)
        self.assertEqual(file_list, [file_path])
        file_list = []
        git_wrapper.diff(options, self.args, file_list)
        self.assertEqual(file_list, [])
        sys.stdout.close()

    def testRevertNone(self):
        if not self.enabled:
            return
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_list = []
        git_wrapper.update(options, None, file_list)
        file_list = []
        git_wrapper.revert(options, self.args, file_list)
        self.assertEqual(file_list, [])
        self.assertEqual(
            git_wrapper.revinfo(options, self.args, None),
            "4091c7d010ca99d0f2dd416d4b70b758ae432187",
        )
        sys.stdout.close()

    def testRevertModified(self):
        if not self.enabled:
            return
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_list = []
        git_wrapper.update(options, None, file_list)
        file_path = join(self.base_path, "a")
        with open(file_path, "a") as f:
            f.writelines("touched\n")
        file_list = []
        git_wrapper.revert(options, self.args, file_list)
        self.assertEqual(file_list, [file_path])
        file_list = []
        git_wrapper.diff(options, self.args, file_list)
        self.assertEqual(file_list, [])
        self.assertEqual(
            git_wrapper.revinfo(options, self.args, None),
            "4091c7d010ca99d0f2dd416d4b70b758ae432187",
        )
        sys.stdout.close()

    def testRevertNew(self):
        if not self.enabled:
            return
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_list = []
        git_wrapper.update(options, None, file_list)
        file_path = join(self.base_path, "c")
        with open(file_path, "w") as f:
            f.writelines("new\n")
        Popen(
            [GIT, "add", "c"], stdout=PIPE, stderr=STDOUT, cwd=self.base_path
        ).communicate()
        file_list = []
        git_wrapper.revert(options, self.args, file_list)
        self.assertEqual(file_list, [file_path])
        file_list = []
        git_wrapper.diff(options, self.args, file_list)
        self.assertEqual(file_list, [])
        self.assertEqual(
            git_wrapper.revinfo(options, self.args, None),
            "4091c7d010ca99d0f2dd416d4b70b758ae432187",
        )
        sys.stdout.close()

    def testStatusRef(self):
        if not self.enabled:
            return
        options = self.Options()
        file_paths = [join(self.base_path, "a")]
        with open(file_paths[0], "a") as f:
            f.writelines("touched\n")
        git_wrapper = gclient_scm.GitWrapper(
            self.url + "@refs/heads/feature", self.root_dir, self.relpath
        )
        file_paths.append(join(self.base_path, "c"))  # feature branch touches c
        file_list = []
        git_wrapper.status(options, self.args, file_list)
        self.assertEqual(file_list, file_paths)
        self.checkstdout(
            (
                "\n________ running 'git -c core.quotePath=false diff --name-status "
                "refs/remotes/origin/feature' in '%s'\n\nM\ta\n"
            )
            % join(self.root_dir, ".")
        )

    def testStatusNew(self):
        if not self.enabled:
            return
        options = self.Options()
        file_path = join(self.base_path, "a")
        with open(file_path, "a") as f:
            f.writelines("touched\n")
        git_wrapper = gclient_scm.GitWrapper(
            self.url + "@069c602044c5388d2d15c3f875b057c852003458",
            self.root_dir,
            self.relpath,
        )
        file_list = []
        git_wrapper.status(options, self.args, file_list)
        self.assertEqual(file_list, [file_path])
        self.checkstdout(
            (
                "\n________ running 'git -c core.quotePath=false diff --name-status "
                "069c602044c5388d2d15c3f875b057c852003458' in '%s'\n\nM\ta\n"
            )
            % join(self.root_dir, ".")
        )

    def testStatusNewNoBaseRev(self):
        if not self.enabled:
            return
        options = self.Options()
        file_path = join(self.base_path, "a")
        with open(file_path, "a") as f:
            f.writelines("touched\n")
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_list = []
        git_wrapper.status(options, self.args, file_list)
        self.assertEqual(file_list, [file_path])
        self.checkstdout(
            (
                "\n________ running 'git -c core.quotePath=false diff --name-status"
                "' in '%s'\n\nM\ta\n"
            )
            % join(self.root_dir, ".")
        )

    def testStatus2New(self):
        if not self.enabled:
            return
        options = self.Options()
        expected_file_list = []
        for f in ["a", "b"]:
            file_path = join(self.base_path, f)
            with open(file_path, "a") as f:
                f.writelines("touched\n")
            expected_file_list.extend([file_path])
        git_wrapper = gclient_scm.GitWrapper(
            self.url + "@069c602044c5388d2d15c3f875b057c852003458",
            self.root_dir,
            self.relpath,
        )
        file_list = []
        git_wrapper.status(options, self.args, file_list)
        expected_file_list = [join(self.base_path, x) for x in ["a", "b"]]
        self.assertEqual(sorted(file_list), expected_file_list)
        self.checkstdout(
            (
                "\n________ running 'git -c core.quotePath=false diff --name-status "
                "069c602044c5388d2d15c3f875b057c852003458' in '%s'\n\nM\ta\nM\tb\n"
            )
            % join(self.root_dir, ".")
        )

    def testUpdateUpdate(self):
        if not self.enabled:
            return
        options = self.Options()
        expected_file_list = [join(self.base_path, x) for x in ["a", "b"]]
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_list = []

        # Set diff.ignoreSubmodules to something other than 'dirty'
        git_wrapper._Run(["config", "diff.ignoreSubmodules", "all"], options)
        git_wrapper.update(options, (), file_list)
        expected_warning = "diff.ignoreSubmodules is not set to 'dirty'"
        self.assertTrue(
            any(expected_warning in w for w in gclient_utils._WARNINGS),
            f"Expected warning not found. "
            f"New warnings: {gclient_utils._WARNINGS}",
        )
        self.assertEqual(file_list, expected_file_list)
        self.assertEqual(
            git_wrapper.revinfo(options, (), None),
            "4091c7d010ca99d0f2dd416d4b70b758ae432187",
        )
        self.assertEqual(
            git_wrapper._Capture(["config", "--get", "diff.ignoreSubmodules"]),
            "all",
        )
        self.assertEqual(
            git_wrapper._Capture(
                ["config", "--get", "fetch.recurseSubmodules"]
            ),
            "off",
        )
        self.assertEqual(
            git_wrapper._Capture(["config", "--get", "push.recurseSubmodules"]),
            "off",
        )
        os.environ["GCLIENT_SUPPRESS_SUBMODULE_WARNING"] = "1"
        gclient_utils._WARNINGS.clear()
        git_wrapper.update(options, (), file_list)
        self.assertEqual(
            len(gclient_utils._WARNINGS),
            0,
            "Warning was added despite being suppressed",
        )
        # Clean up
        del os.environ["GCLIENT_SUPPRESS_SUBMODULE_WARNING"]
        sys.stdout.close()

    def testUpdateMerge(self):
        if not self.enabled:
            return
        options = self.Options()
        options.merge = True
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        # This sets correct remote HEAD
        git_wrapper.update(options, (), [])

        git_wrapper._Run(["checkout", "-q", "feature"], options)
        rev = git_wrapper.revinfo(options, (), None)
        file_list = []
        git_wrapper.update(options, (), file_list)
        self.assertEqual(
            file_list,
            [join(self.base_path, x) for x in ["a", "b", "c", "submodule"]],
        )
        # The actual commit that is created is unstable, so we verify its tree
        # and parents instead.
        self.assertEqual(
            git_wrapper._Capture(["rev-parse", "HEAD:"]),
            "3a3ba72731fa208d37b06598a129ba93970325df",
        )
        self.assertEqual(git_wrapper._Capture(["rev-parse", "HEAD^1"]), rev)
        self.assertEqual(
            git_wrapper._Capture(["rev-parse", "HEAD^2"]),
            git_wrapper._Capture(["rev-parse", "origin/main"]),
        )
        sys.stdout.close()

    def testUpdateRebase(self):
        if not self.enabled:
            return
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        # This sets correct remote HEAD
        git_wrapper.update(options, (), [])

        git_wrapper._Run(["checkout", "-q", "feature"], options)
        # Fake a 'y' key press.
        git_wrapper._AskForData = self._GetAskForDataCallback(
            "Cannot fast-forward merge, attempt to rebase? "
            "(y)es / (q)uit / (s)kip : ",
            "y",
        )
        file_list = []
        git_wrapper.update(options, (), file_list)
        self.assertEqual(
            file_list,
            [join(self.base_path, x) for x in ["a", "b", "c", "submodule"]],
        )
        # The actual commit that is created is unstable, so we verify its tree
        # and parent instead.
        self.assertEqual(
            git_wrapper._Capture(["rev-parse", "HEAD:"]),
            "3a3ba72731fa208d37b06598a129ba93970325df",
        )
        self.assertEqual(
            git_wrapper._Capture(["rev-parse", "HEAD^1"]),
            git_wrapper._Capture(["rev-parse", "origin/main"]),
        )
        sys.stdout.close()

    def testUpdateReset(self):
        if not self.enabled:
            return
        options = self.Options()
        options.reset = True

        dir_path = join(self.base_path, "c")
        os.mkdir(dir_path)
        with open(join(dir_path, "nested"), "w") as f:
            f.writelines("new\n")

        file_path = join(self.base_path, "file")
        with open(file_path, "w") as f:
            f.writelines("new\n")

        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_list = []
        git_wrapper.update(options, (), file_list)
        self.assertTrue(gclient_scm.os.path.isdir(dir_path))
        self.assertTrue(gclient_scm.os.path.isfile(file_path))
        sys.stdout.close()

    def testUpdateResetUnsetsFetchConfig(self):
        if not self.enabled:
            return
        options = self.Options()
        options.reset = True

        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        git_wrapper._Run(
            [
                "config",
                "remote.origin.fetch",
                "+refs/heads/bad/ref:refs/remotes/origin/bad/ref",
            ],
            options,
        )

        file_list = []
        git_wrapper.update(options, (), file_list)
        self.assertEqual(
            git_wrapper.revinfo(options, (), None),
            "069c602044c5388d2d15c3f875b057c852003458",
        )
        sys.stdout.close()

    def testUpdateResetDeleteUnversionedTrees(self):
        if not self.enabled:
            return
        options = self.Options()
        options.reset = True
        options.delete_unversioned_trees = True

        dir_path = join(self.base_path, "dir")
        os.mkdir(dir_path)
        with open(join(dir_path, "nested"), "w") as f:
            f.writelines("new\n")

        file_path = join(self.base_path, "file")
        with open(file_path, "w") as f:
            f.writelines("new\n")

        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_list = []
        git_wrapper.update(options, (), file_list)
        self.assertTrue(not gclient_scm.os.path.isdir(dir_path))
        self.assertTrue(gclient_scm.os.path.isfile(file_path))
        sys.stdout.close()

    def testUpdateUnstagedConflict(self):
        if not self.enabled:
            return
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_path = join(self.base_path, "b")
        with open(file_path, "w") as f:
            f.writelines("conflict\n")
        try:
            git_wrapper.update(options, (), [])
            self.fail()
        except (
            gclient_scm.gclient_utils.Error,
            subprocess2.CalledProcessError,
        ):
            # The exact exception text varies across git versions so it's not
            # worth verifying it. It's fine as long as it throws.
            pass
        # Manually flush stdout since we can't verify it's content accurately
        # across git versions.
        sys.stdout.getvalue()
        sys.stdout.close()

    @unittest.skip("Skipping until crbug.com/670884 is resolved.")
    def testUpdateLocked(self):
        if not self.enabled:
            return
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_path = join(self.base_path, ".git", "index.lock")
        with open(file_path, "w"):
            pass
        with self.assertRaises(subprocess2.CalledProcessError):
            git_wrapper.update(options, (), [])
        sys.stdout.close()

    def testUpdateLockedBreak(self):
        if not self.enabled:
            return
        options = self.Options()
        options.break_repo_locks = True
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_path = join(self.base_path, ".git", "index.lock")
        with open(file_path, "w"):
            pass
        git_wrapper.update(options, (), [])
        self.assertRegex(
            sys.stdout.getvalue(), r"breaking lock.*\.git[/|\\]index\.lock"
        )
        self.assertFalse(os.path.exists(file_path))
        sys.stdout.close()

    def testUpdateConflict(self):
        if not self.enabled:
            return
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_path = join(self.base_path, "b")
        with open(file_path, "w") as f:
            f.writelines("conflict\n")
        git_wrapper._Run(["commit", "-am", "test"], options)
        git_wrapper._AskForData = self._GetAskForDataCallback(
            "Cannot fast-forward merge, attempt to rebase? "
            "(y)es / (q)uit / (s)kip : ",
            "y",
        )

        with self.assertRaises(gclient_scm.gclient_utils.Error) as e:
            git_wrapper.update(options, (), [])
        self.assertEqual(
            e.exception.args[0],
            "Conflict while rebasing this branch.\n"
            "Fix the conflict and run gclient again.\n"
            "See 'man git-rebase' for details.\n",
        )

        with self.assertRaises(gclient_scm.gclient_utils.Error) as e:
            git_wrapper.update(options, (), [])
        self.assertEqual(
            e.exception.args[0],
            "\n____ . at refs/remotes/origin/main\n"
            "\tYou have uncommitted changes.\n"
            "\tcd into ., run git status to see changes,\n"
            "\tand commit, stash, or reset.\n",
        )

        sys.stdout.close()

    def testRevinfo(self):
        if not self.enabled:
            return
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        rev_info = git_wrapper.revinfo(options, (), None)
        self.assertEqual(rev_info, "069c602044c5388d2d15c3f875b057c852003458")


class ManagedGitWrapperTestCaseMock(unittest.TestCase):
    class OptionsObject(object):
        def __init__(self, verbose=False, revision=None, force=False):
            self.verbose = verbose
            self.revision = revision
            self.deps_os = None
            self.force = force
            self.reset = False
            self.nohooks = False
            self.break_repo_locks = False
            # TODO(maruel): Test --jobs > 1.
            self.jobs = 1
            self.patch_ref = None
            self.patch_repo = None
            self.rebase_patch_ref = True

    def Options(self, *args, **kwargs):
        return self.OptionsObject(*args, **kwargs)

    def checkstdout(self, expected):
        value = sys.stdout.getvalue()
        sys.stdout.close()
        # Check that the expected output appears.
        self.assertIn(expected, strip_timestamps(value))

    def setUp(self):
        self.fake_hash_1 = "t0ta11yf4k3"
        self.fake_hash_2 = "3v3nf4k3r"
        self.url = "git://foo"
        self.root_dir = "/tmp" if sys.platform != "win32" else "t:\\tmp"
        self.relpath = "fake"
        self.base_path = os.path.join(self.root_dir, self.relpath)
        self.backup_base_path = os.path.join(
            self.root_dir, "old_%s.git" % self.relpath
        )
        mock.patch("gclient_scm.scm.GIT.ApplyEnvVars").start()
        mock.patch("gclient_scm.GitWrapper._Fetch").start()
        mock.patch("gclient_scm.GitWrapper._DeleteOrMove").start()
        mock.patch("sys.stdout", StringIO()).start()
        self.addCleanup(mock.patch.stopall)

    @mock.patch("scm.GIT.IsValidRevision")
    @mock.patch("os.path.isdir", lambda _: True)
    def testGetUsableRevGit(self, mockIsValidRevision):
        options = self.Options(verbose=True)

        mockIsValidRevision.side_effect = lambda cwd, rev: rev != "1"

        git_scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath)
        # A [fake] git sha1 with a git repo should work (this is in the case
        # that the LKGR gets flipped to git sha1's some day).
        self.assertEqual(
            git_scm.GetUsableRev(self.fake_hash_1, options), self.fake_hash_1
        )
        # An SVN rev with an existing purely git repo should raise an exception.
        self.assertRaises(
            gclient_scm.gclient_utils.Error, git_scm.GetUsableRev, "1", options
        )

    @mock.patch("gclient_scm.GitWrapper._Clone")
    def testEnsureValidHeadObjectOrCheckout_CouldNotParseHEAD(self, mockClone):
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        git_cache.Mirror.SetCachePath("/tmp/cache")
        self.addCleanup(git_cache.Mirror.SetCachePath, None)
        url = "/tmp/cache/foo"

        mock_capture = mock.Mock(
            side_effect=subprocess2.CalledProcessError(
                returncode=128,
                cmd=["rev-list"],
                cwd=None,
                stdout=None,
                stderr=b"error: unable to normalize alternate object path...\nfatal: Could not parse object 'HEAD'.",
            )
        )
        git_wrapper._Capture = mock_capture

        git_wrapper._EnsureValidHeadObjectOrCheckout("revision", options, url)

        git_wrapper._DeleteOrMove.assert_called_with(options.force)
        mockClone.assert_called_with("revision", url, options)

    @mock.patch("gclient_scm.GitWrapper._Clone")
    def testEnsureValidHeadObjectOrCheckout_BadObjectHEAD(self, mockClone):
        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        git_cache.Mirror.SetCachePath("/tmp/cache")
        self.addCleanup(git_cache.Mirror.SetCachePath, None)
        url = "/tmp/cache/foo"

        mock_capture = mock.Mock(
            side_effect=subprocess2.CalledProcessError(
                returncode=128,
                cmd=["rev-list"],
                cwd=None,
                stdout=None,
                stderr=b"fatal: bad object HEAD\n",
            )
        )
        git_wrapper._Capture = mock_capture

        git_wrapper._EnsureValidHeadObjectOrCheckout("revision", options, url)

        git_wrapper._DeleteOrMove.assert_called_with(options.force)
        mockClone.assert_called_with("revision", url, options)

    @mock.patch("gclient_scm.GitWrapper._Clone")
    @mock.patch("os.path.isdir")
    @mock.patch("os.path.exists")
    @mock.patch("git_common.run")
    def testUpdateNoDotGit(self, mockRun, mockExists, mockIsdir, mockClone):
        mockIsdir.side_effect = lambda path: path == self.base_path
        mockExists.side_effect = lambda path: path == self.base_path
        mockRun.side_effect = ["refs/remotes/origin/main", "", ""]

        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        git_wrapper.update(options, None, [])

        env = gclient_scm.scm.GIT.ApplyEnvVars({})
        self.assertEqual(
            mockRun.mock_calls,
            [
                mock.call(
                    "symbolic-ref",
                    "refs/remotes/origin/HEAD",
                    autostrip=True,
                    cwd=self.base_path,
                    env=env,
                ),
                mock.call(
                    "-c",
                    "core.quotePath=false",
                    "ls-files",
                    cwd=self.base_path,
                    env=env,
                    stderr=-1,
                ),
                mock.call(
                    "rev-parse",
                    "--verify",
                    "HEAD",
                    cwd=self.base_path,
                    env=env,
                    stderr=-1,
                ),
            ],
        )
        mockClone.assert_called_with(
            "refs/remotes/origin/main", self.url, options
        )
        self.checkstdout("\n")

    @mock.patch("gclient_scm.GitWrapper._Clone")
    @mock.patch("os.path.isdir")
    @mock.patch("os.path.exists")
    @mock.patch("git_common.run")
    def testUpdateConflict(self, mockRun, mockExists, mockIsdir, mockClone):
        mockIsdir.side_effect = lambda path: path == self.base_path
        mockExists.side_effect = lambda path: path == self.base_path
        mockRun.side_effect = ["refs/remotes/origin/main", "", ""]
        mockClone.side_effect = [
            git_common.subprocess2.CalledProcessError(
                None, None, None, None, None
            ),
            None,
        ]

        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        git_wrapper.update(options, None, [])

        env = gclient_scm.scm.GIT.ApplyEnvVars({})
        self.assertEqual(
            mockRun.mock_calls,
            [
                mock.call(
                    "symbolic-ref",
                    "refs/remotes/origin/HEAD",
                    autostrip=True,
                    cwd=self.base_path,
                    env=env,
                ),
                mock.call(
                    "-c",
                    "core.quotePath=false",
                    "ls-files",
                    cwd=self.base_path,
                    env=env,
                    stderr=-1,
                ),
                mock.call(
                    "rev-parse",
                    "--verify",
                    "HEAD",
                    cwd=self.base_path,
                    env=env,
                    stderr=-1,
                ),
            ],
        )
        mockClone.assert_called_with(
            "refs/remotes/origin/main", self.url, options
        )
        self.checkstdout("\n")

    @mock.patch("git_cache.Mirror.get_bootstrap_default_branch")
    @mock.patch("gclient_scm.GitWrapper._UpdateMirrorIfNotContains")
    @mock.patch("gclient_scm.GitWrapper._Clone")
    @mock.patch("os.path.isdir")
    @mock.patch("os.path.exists")
    @mock.patch("git_common.run")
    def testUpdateFreshCheckoutUsesSnapshotDefaultBranch(
        self,
        mockRun,
        mockExists,
        mockIsdir,
        mockClone,
        mockUpdateMirror,
        mockBranch,
    ):
        # No local .git: the default branch is read from the bootstrap snapshot.
        mockIsdir.side_effect = lambda path: path == self.base_path
        mockExists.side_effect = lambda path: path == self.base_path
        mockRun.return_value = ""
        mockBranch.return_value = "main"
        git_cache.Mirror.SetCachePath("/tmp/cache")
        self.addCleanup(git_cache.Mirror.SetCachePath, None)

        options = self.Options()
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        git_wrapper.update(options, None, [])

        mockBranch.assert_called_once()
        # The snapshot's default branch becomes the checkout revision.
        self.assertEqual(mockClone.call_args[0][0], "refs/remotes/origin/main")

    @mock.patch("git_cache.Mirror.get_bootstrap_default_branch")
    @mock.patch("gclient_scm.GitWrapper._SetFetchConfig")
    @mock.patch("os.path.isdir")
    @mock.patch("os.path.exists")
    @mock.patch("git_common.run")
    def testUpdateExistingCheckoutSkipsSnapshotLookup(
        self, mockRun, mockExists, mockIsdir, mockSetFetch, mockBranch
    ):
        # A local .git already has the default branch in refs/remotes/*/HEAD, so
        # update() must not hit the (gsutil) bootstrap snapshot. An unmanaged
        # solution takes the early-return existing-checkout path.
        git_dir = os.path.join(self.base_path, ".git")
        mockIsdir.side_effect = lambda path: path == self.base_path
        mockExists.side_effect = lambda path: path in (self.base_path, git_dir)
        mockRun.return_value = "refs/remotes/origin/main"
        git_cache.Mirror.SetCachePath("/tmp/cache")
        self.addCleanup(git_cache.Mirror.SetCachePath, None)

        options = self.Options(revision="unmanaged")
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        git_wrapper.update(options, None, [])

        mockBranch.assert_not_called()

    @mock.patch("git_cache.Mirror.get_bootstrap_default_branch")
    @mock.patch("gclient_scm.scm.GIT.GetRemoteHeadRef")
    @mock.patch("gclient_scm.scm.GIT.GetConfig")
    @mock.patch("gclient_scm.GitWrapper._SetFetchConfig")
    @mock.patch("os.path.isdir")
    @mock.patch("os.path.exists")
    @mock.patch("git_common.run")
    def testUpdateUrlSwitchUsesNewMirrorDefaultBranch(
        self,
        mockRun,
        mockExists,
        mockIsdir,
        mockSetFetch,
        mockGetConfig,
        mockHeadRef,
        mockBranch,
    ):
        # The checkout still points at the old upstream, so its local
        # refs/remotes/*/HEAD describes the old repo and must not be trusted;
        # the default branch comes from the new URL's snapshot.
        url = "https://example.com/new.git"
        git_dir = os.path.join(self.base_path, ".git")
        mockIsdir.side_effect = lambda path: path == self.base_path
        mockExists.side_effect = lambda path: path in (self.base_path, git_dir)
        mockRun.return_value = ""
        mockGetConfig.return_value = "https://example.com/old.git"
        mockBranch.return_value = "master"
        git_cache.Mirror.SetCachePath("/tmp/cache")
        self.addCleanup(git_cache.Mirror.SetCachePath, None)

        options = self.Options(revision="unmanaged")
        git_wrapper = gclient_scm.GitWrapper(url, self.root_dir, self.relpath)
        git_wrapper.update(options, None, [])

        mockBranch.assert_called_once()
        mockHeadRef.assert_not_called()

    @mock.patch("git_cache.Mirror.get_bootstrap_default_branch")
    @mock.patch("gclient_scm.scm.GIT.GetRemoteHeadRef")
    @mock.patch("gclient_scm.scm.GIT.GetConfig")
    @mock.patch("gclient_scm.GitWrapper._SetFetchConfig")
    @mock.patch("os.path.isdir")
    @mock.patch("os.path.exists")
    @mock.patch("git_common.run")
    def testUpdateMatchingUrlTrustsLocalHead(
        self,
        mockRun,
        mockExists,
        mockIsdir,
        mockSetFetch,
        mockGetConfig,
        mockHeadRef,
        mockBranch,
    ):
        # The checkout already points at the requested upstream (its remote
        # URL is the mirror of the requested url), so the local
        # refs/remotes/*/HEAD wins and the snapshot is not consulted.
        url = "https://example.com/new.git"
        git_dir = os.path.join(self.base_path, ".git")
        mockIsdir.side_effect = lambda path: path == self.base_path
        mockExists.side_effect = lambda path: path in (self.base_path, git_dir)
        mockRun.return_value = ""
        git_cache.Mirror.SetCachePath("/tmp/cache")
        self.addCleanup(git_cache.Mirror.SetCachePath, None)
        mockGetConfig.return_value = os.path.join(
            "/tmp/cache", git_cache.Mirror.UrlToCacheDir(url)
        )
        mockHeadRef.return_value = "refs/remotes/origin/main"

        options = self.Options(revision="unmanaged")
        git_wrapper = gclient_scm.GitWrapper(url, self.root_dir, self.relpath)
        git_wrapper.update(options, None, [])

        mockBranch.assert_not_called()
        mockHeadRef.assert_called_once_with(
            self.base_path, url, "origin", use_local=True
        )

    @mock.patch("git_cache.Mirror.get_bootstrap_default_branch")
    @mock.patch("gclient_scm.scm.GIT.GetRemoteHeadRef")
    @mock.patch("gclient_scm.scm.GIT.GetConfig")
    @mock.patch("gclient_scm.GitWrapper._SetFetchConfig")
    @mock.patch("os.path.isdir")
    @mock.patch("os.path.exists")
    @mock.patch("git_common.run")
    def testUpdateUpstreamFetchUrlTrustsLocalHead(
        self,
        mockRun,
        mockExists,
        mockIsdir,
        mockSetFetch,
        mockGetConfig,
        mockHeadRef,
        mockBranch,
    ):
        # Unmanaged checkouts fetch from the upstream even with a cache
        # configured; the local HEAD still wins over the snapshot.
        url = "https://example.com/new.git"
        git_dir = os.path.join(self.base_path, ".git")
        mockIsdir.side_effect = lambda path: path == self.base_path
        mockExists.side_effect = lambda path: path in (self.base_path, git_dir)
        mockRun.return_value = ""
        git_cache.Mirror.SetCachePath("/tmp/cache")
        self.addCleanup(git_cache.Mirror.SetCachePath, None)
        mockGetConfig.return_value = url
        mockHeadRef.return_value = "refs/remotes/origin/main"

        options = self.Options(revision="unmanaged")
        git_wrapper = gclient_scm.GitWrapper(url, self.root_dir, self.relpath)
        git_wrapper.update(options, None, [])

        mockBranch.assert_not_called()
        mockHeadRef.assert_called_once_with(
            self.base_path, url, "origin", use_local=True
        )


class UnmanagedGitWrapperTestCase(BaseGitWrapperTestCase):
    def checkInStdout(self, expected):
        value = sys.stdout.getvalue()
        sys.stdout.close()
        self.assertIn(expected, value)

    def checkNotInStdout(self, expected):
        value = sys.stdout.getvalue()
        sys.stdout.close()
        self.assertNotIn(expected, value)

    def getCurrentBranch(self):
        # Returns name of current branch or HEAD for detached HEAD
        branch = gclient_scm.scm.GIT.Capture(
            ["rev-parse", "--abbrev-ref", "HEAD"], cwd=self.base_path
        )
        if branch == "HEAD":
            return None
        return branch

    def testUpdateClone(self):
        if not self.enabled:
            return
        options = self.Options()

        origin_root_dir = self.root_dir
        self.addCleanup(gclient_utils.rmtree, origin_root_dir)

        self.root_dir = tempfile.mkdtemp()
        self.relpath = "."
        self.base_path = join(self.root_dir, self.relpath)

        git_wrapper = gclient_scm.GitWrapper(
            origin_root_dir, self.root_dir, self.relpath
        )

        expected_file_list = [
            join(self.base_path, "a"),
            join(self.base_path, "b"),
        ]
        file_list = []
        options.revision = "unmanaged"
        git_wrapper.update(options, (), file_list)

        self.assertEqual(file_list, expected_file_list)
        self.assertEqual(
            git_wrapper.revinfo(options, (), None),
            "069c602044c5388d2d15c3f875b057c852003458",
        )
        # indicates detached HEAD
        self.assertEqual(self.getCurrentBranch(), None)
        self.checkInStdout(
            "Checked out refs/remotes/origin/main to a detached HEAD"
        )

    def testUpdateCloneOnCommit(self):
        if not self.enabled:
            return
        options = self.Options()

        origin_root_dir = self.root_dir
        self.addCleanup(gclient_utils.rmtree, origin_root_dir)

        self.root_dir = tempfile.mkdtemp()
        self.relpath = "."
        self.base_path = join(self.root_dir, self.relpath)
        url_with_commit_ref = (
            origin_root_dir + "@4091c7d010ca99d0f2dd416d4b70b758ae432187"
        )

        git_wrapper = gclient_scm.GitWrapper(
            url_with_commit_ref, self.root_dir, self.relpath
        )

        expected_file_list = [
            join(self.base_path, "a"),
            join(self.base_path, "b"),
            join(self.base_path, "submodule"),
        ]
        file_list = []
        options.revision = "unmanaged"
        git_wrapper.update(options, (), file_list)

        self.assertEqual(file_list, expected_file_list)
        self.assertEqual(
            git_wrapper.revinfo(options, (), None),
            "4091c7d010ca99d0f2dd416d4b70b758ae432187",
        )
        # indicates detached HEAD
        self.assertEqual(self.getCurrentBranch(), None)
        self.checkInStdout(
            "Checked out 4091c7d010ca99d0f2dd416d4b70b758ae432187 to a detached HEAD"
        )

    def testUpdateCloneOnBranch(self):
        if not self.enabled:
            return
        options = self.Options()

        origin_root_dir = self.root_dir
        self.addCleanup(gclient_utils.rmtree, origin_root_dir)

        self.root_dir = tempfile.mkdtemp()
        self.relpath = "."
        self.base_path = join(self.root_dir, self.relpath)
        url_with_branch_ref = origin_root_dir + "@feature"

        git_wrapper = gclient_scm.GitWrapper(
            url_with_branch_ref, self.root_dir, self.relpath
        )

        expected_file_list = [
            join(self.base_path, "a"),
            join(self.base_path, "b"),
            join(self.base_path, "c"),
        ]
        file_list = []
        options.revision = "unmanaged"
        git_wrapper.update(options, (), file_list)

        self.assertEqual(file_list, expected_file_list)
        self.assertEqual(
            git_wrapper.revinfo(options, (), None),
            "9a51244740b25fa2ded5252ca00a3178d3f665a9",
        )
        # indicates detached HEAD
        self.assertEqual(self.getCurrentBranch(), None)
        self.checkInStdout(
            "Checked out 9a51244740b25fa2ded5252ca00a3178d3f665a9 "
            "to a detached HEAD"
        )

    def testUpdateCloneOnFetchedRemoteBranch(self):
        if not self.enabled:
            return
        options = self.Options()

        origin_root_dir = self.root_dir
        self.addCleanup(gclient_utils.rmtree, origin_root_dir)

        self.root_dir = tempfile.mkdtemp()
        self.relpath = "."
        self.base_path = join(self.root_dir, self.relpath)
        url_with_branch_ref = origin_root_dir + "@refs/remotes/origin/feature"

        git_wrapper = gclient_scm.GitWrapper(
            url_with_branch_ref, self.root_dir, self.relpath
        )

        expected_file_list = [
            join(self.base_path, "a"),
            join(self.base_path, "b"),
            join(self.base_path, "c"),
        ]
        file_list = []
        options.revision = "unmanaged"
        git_wrapper.update(options, (), file_list)

        self.assertEqual(file_list, expected_file_list)
        self.assertEqual(
            git_wrapper.revinfo(options, (), None),
            "9a51244740b25fa2ded5252ca00a3178d3f665a9",
        )
        # indicates detached HEAD
        self.assertEqual(self.getCurrentBranch(), None)
        self.checkInStdout(
            "Checked out refs/remotes/origin/feature to a detached HEAD"
        )

    def testUpdateCloneOnTrueRemoteBranch(self):
        if not self.enabled:
            return
        options = self.Options()

        origin_root_dir = self.root_dir
        self.addCleanup(gclient_utils.rmtree, origin_root_dir)

        self.root_dir = tempfile.mkdtemp()
        self.relpath = "."
        self.base_path = join(self.root_dir, self.relpath)
        url_with_branch_ref = origin_root_dir + "@refs/heads/feature"

        git_wrapper = gclient_scm.GitWrapper(
            url_with_branch_ref, self.root_dir, self.relpath
        )

        expected_file_list = [
            join(self.base_path, "a"),
            join(self.base_path, "b"),
            join(self.base_path, "c"),
        ]
        file_list = []
        options.revision = "unmanaged"
        git_wrapper.update(options, (), file_list)

        self.assertEqual(file_list, expected_file_list)
        self.assertEqual(
            git_wrapper.revinfo(options, (), None),
            "9a51244740b25fa2ded5252ca00a3178d3f665a9",
        )
        # @refs/heads/feature is AKA @refs/remotes/origin/feature in the clone,
        # so should be treated as such by gclient. TODO(mmoss): Though really,
        # we should only allow DEPS to specify branches as they are known in the
        # upstream repo, since the mapping into the local repo can be modified
        # by users (or we might even want to change the gclient defaults at some
        # point). But that will take more work to stop using refs/remotes/
        # everywhere that we do (and to stop assuming a DEPS ref will always
        # resolve locally, like when passing them to show-ref or rev-list).
        self.assertEqual(self.getCurrentBranch(), None)
        self.checkInStdout(
            "Checked out refs/remotes/origin/feature to a detached HEAD"
        )

    def testUpdateUpdate(self):
        if not self.enabled:
            return
        options = self.Options()
        expected_file_list = []
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        file_list = []
        options.revision = "unmanaged"
        git_wrapper.update(options, (), file_list)
        self.assertEqual(file_list, expected_file_list)
        self.assertEqual(
            git_wrapper.revinfo(options, (), None),
            "069c602044c5388d2d15c3f875b057c852003458",
        )
        self.checkstdout("________ unmanaged solution; skipping .\n")


class CipdWrapperTestCase(unittest.TestCase):
    def setUp(self):
        # Create this before setting up mocks.
        self._cipd_root_dir = tempfile.mkdtemp()
        self._workdir = tempfile.mkdtemp()

        self._cipd_instance_url = "https://chrome-infra-packages.appspot.com"
        self._cipd_root = gclient_scm.CipdRoot(
            self._cipd_root_dir, self._cipd_instance_url
        )
        self._cipd_packages = [
            self._cipd_root.add_package("f", "foo_package", "foo_version"),
            self._cipd_root.add_package("b", "bar_package", "bar_version"),
            self._cipd_root.add_package("b", "baz_package", "baz_version"),
        ]
        mock.patch("tempfile.mkdtemp", lambda: self._workdir).start()
        mock.patch("gclient_scm.CipdRoot.add_package").start()
        mock.patch("gclient_scm.CipdRoot.clobber").start()
        mock.patch("gclient_scm.CipdRoot.ensure_file_resolve").start()
        mock.patch("gclient_scm.CipdRoot.ensure").start()
        self.addCleanup(mock.patch.stopall)
        self.addCleanup(gclient_utils.rmtree, self._cipd_root_dir)
        self.addCleanup(gclient_utils.rmtree, self._workdir)

    def createScmWithPackageThatSatisfies(self, condition):
        return gclient_scm.CipdWrapper(
            url=self._cipd_instance_url,
            root_dir=self._cipd_root_dir,
            relpath="fake_relpath",
            root=self._cipd_root,
            package=self.getPackageThatSatisfies(condition),
        )

    def getPackageThatSatisfies(self, condition):
        for p in self._cipd_packages:
            if condition(p):
                return p

        self.fail("Unable to find a satisfactory package.")

    def testRevert(self):
        """Checks that revert does nothing."""
        git_wrapper = self.createScmWithPackageThatSatisfies(lambda _: True)
        git_wrapper.revert(None, (), [])

    @mock.patch("gclient_scm.gclient_utils.CheckCallAndFilter")
    @mock.patch("gclient_scm.gclient_utils.rmtree")
    def testRevinfo(self, mockRmtree, mockCheckCallAndFilter):
        """Checks that revinfo uses the JSON from cipd describe."""
        git_wrapper = self.createScmWithPackageThatSatisfies(lambda _: True)

        expected_revinfo = "0123456789abcdef0123456789abcdef01234567"
        json_contents = {
            "result": {
                "pin": {
                    "instance_id": expected_revinfo,
                }
            }
        }
        describe_json_path = join(self._workdir, "describe.json")
        with open(describe_json_path, "w") as describe_json:
            json.dump(json_contents, describe_json)

        revinfo = git_wrapper.revinfo(None, (), [])
        self.assertEqual(revinfo, expected_revinfo)

        mockRmtree.assert_called_with(self._workdir)
        mockCheckCallAndFilter.assert_called_with(
            [
                "cipd",
                "describe",
                "foo_package",
                "-log-level",
                "error",
                "-version",
                "foo_version",
                "-json-output",
                describe_json_path,
            ]
        )

    def testUpdate(self):
        """Checks that update does nothing."""
        git_wrapper = self.createScmWithPackageThatSatisfies(lambda _: True)
        git_wrapper.update(None, (), [])


class GcsWrapperTestCase(unittest.TestCase):
    def setUp(self):
        self.workdir = tempfile.mkdtemp()
        self.url = "gs://123bucket/path_to_tar.gz"

    def createScm(self):
        return gclient_scm.GcsWrapper(
            url=self.url, root_dir=self.workdir, relpath="fake_relpath"
        )

    def testRevert(self):
        """Checks that revert does nothing."""
        git_wrapper = self.createScm()
        git_wrapper.revert(None, (), [])

    def testRevinfo(self):
        """Checks that revinfo does nothing."""
        git_wrapper = self.createScm()
        git_wrapper.revinfo(None, (), [])

    def testUpdate(self):
        """Checks that update does nothing."""
        git_wrapper = self.createScm()
        git_wrapper.update(None, (), [])


class BranchHeadsFakeRepo(fake_repos.FakeReposBase):
    def populateGit(self):
        # Creates a tree that looks like this:
        #
        #    5 refs/branch-heads/5
        #    |
        #    4
        #    |
        # 1--2--3 refs/heads/main
        self._commit_git("repo_1", {"commit 1": "touched"})
        self._commit_git("repo_1", {"commit 2": "touched"})
        self._commit_git("repo_1", {"commit 3": "touched"})
        self._create_ref("repo_1", "refs/heads/main", 3)

        self._commit_git("repo_1", {"commit 4": "touched"}, base=2)
        self._commit_git("repo_1", {"commit 5": "touched"}, base=2)
        self._create_ref("repo_1", "refs/branch-heads/5", 5)


class BranchHeadsTest(fake_repos.FakeReposTestBase):
    FAKE_REPOS_CLASS = BranchHeadsFakeRepo

    def setUp(self):
        super(BranchHeadsTest, self).setUp()
        self.enabled = self.FAKE_REPOS.set_up_git()
        self.options = BaseGitWrapperTestCase.OptionsObject()
        self.url = self.git_base + "repo_1"
        self.mirror = None
        mock.patch("sys.stdout", StringIO()).start()
        self.addCleanup(mock.patch.stopall)

    def setUpMirror(self):
        self.mirror = tempfile.mkdtemp("mirror")
        git_cache.Mirror.SetCachePath(self.mirror)
        self.addCleanup(gclient_utils.rmtree, self.mirror)
        self.addCleanup(git_cache.Mirror.SetCachePath, None)

    def testCheckoutBranchHeads(self):
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        self.options.revision = "refs/branch-heads/5"
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 5), self.gitrevparse(self.root_dir)
        )

    def testCheckoutUpdatedBranchHeads(self):
        # Travel back in time, and set refs/branch-heads/5 to its parent.
        subprocess2.check_call(
            [
                "git",
                "update-ref",
                "refs/branch-heads/5",
                self.githash("repo_1", 4),
            ],
            cwd=self.url,
        )

        # Sync to refs/branch-heads/5
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        self.options.revision = "refs/branch-heads/5"
        git_wrapper.update(self.options, None, [])

        # Set refs/branch-heads/5 back to its original value.
        subprocess2.check_call(
            [
                "git",
                "update-ref",
                "refs/branch-heads/5",
                self.githash("repo_1", 5),
            ],
            cwd=self.url,
        )

        # Attempt to sync to refs/branch-heads/5 again.
        self.testCheckoutBranchHeads()

    def testCheckoutBranchHeadsMirror(self):
        self.setUpMirror()
        self.testCheckoutBranchHeads()

    def testCheckoutUpdatedBranchHeadsMirror(self):
        self.setUpMirror()
        self.testCheckoutUpdatedBranchHeads()


class GerritChangesFakeRepo(fake_repos.FakeReposBase):
    def populateGit(self):
        # Creates a tree that looks like this:
        #
        #       6 refs/changes/35/1235/1
        #       |
        #       5 refs/changes/34/1234/1
        #       |
        # 1--2--3--4 refs/heads/main
        #    |  |
        #    |  11(5)--12 refs/heads/main-with-5
        #    |
        #    7--8--9 refs/heads/feature
        #       |
        #       10 refs/changes/36/1236/1
        #

        self._commit_git("repo_1", {"commit 1": "touched"})
        self._commit_git("repo_1", {"commit 2": "touched"})
        self._commit_git("repo_1", {"commit 3": "touched"})
        self._commit_git("repo_1", {"commit 4": "touched"})
        self._create_ref("repo_1", "refs/heads/main", 4)

        # Create a change on top of commit 3 that consists of two commits.
        self._commit_git(
            "repo_1", {"commit 5": "touched", "change": "1234"}, base=3
        )
        self._create_ref("repo_1", "refs/changes/34/1234/1", 5)
        self._commit_git("repo_1", {"commit 6": "touched", "change": "1235"})
        self._create_ref("repo_1", "refs/changes/35/1235/1", 6)

        # Create a refs/heads/feature branch on top of commit 2, consisting of
        # three commits.
        self._commit_git("repo_1", {"commit 7": "touched"}, base=2)
        self._commit_git("repo_1", {"commit 8": "touched"})
        self._commit_git("repo_1", {"commit 9": "touched"})
        self._create_ref("repo_1", "refs/heads/feature", 9)

        # Create a change of top of commit 8.
        self._commit_git(
            "repo_1", {"commit 10": "touched", "change": "1236"}, base=8
        )
        self._create_ref("repo_1", "refs/changes/36/1236/1", 10)

        # Create a refs/heads/main-with-5 on top of commit 3 which is a branch
        # where refs/changes/34/1234/1 (commit 5) has already landed as commit
        # 11.
        self._commit_git(
            "repo_1",
            # This is really commit 11, but has the changes of commit 5
            {"commit 5": "touched", "change": "1234"},
            base=3,
        )
        self._commit_git("repo_1", {"commit 12": "touched"})
        self._create_ref("repo_1", "refs/heads/main-with-5", 12)


class GerritChangesTest(fake_repos.FakeReposTestBase):
    FAKE_REPOS_CLASS = GerritChangesFakeRepo

    def setUp(self):
        super(GerritChangesTest, self).setUp()
        self.enabled = self.FAKE_REPOS.set_up_git()
        self.options = BaseGitWrapperTestCase.OptionsObject()
        self.url = self.git_base + "repo_1"
        self.mirror = None
        mock.patch("sys.stdout", StringIO()).start()
        self.addCleanup(mock.patch.stopall)

    def setUpMirror(self):
        self.mirror = tempfile.mkdtemp()
        git_cache.Mirror.SetCachePath(self.mirror)
        self.addCleanup(gclient_utils.rmtree, self.mirror)
        self.addCleanup(git_cache.Mirror.SetCachePath, None)

    def assertCommits(self, commits):
        """Check that all, and only |commits| are present in the current checkout."""
        for i in commits:
            name = os.path.join(self.root_dir, "commit " + str(i))
            self.assertTrue(os.path.exists(name), "Commit not found: %s" % name)

        all_commits = set(range(1, len(self.FAKE_REPOS.git_hashes["repo_1"])))
        for i in all_commits - set(commits):
            name = os.path.join(self.root_dir, "commit " + str(i))
            self.assertFalse(
                os.path.exists(name), "Unexpected commit: %s" % name
            )

    def testCanCloneGerritChange(self):
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        self.options.revision = "refs/changes/35/1235/1"
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 6), self.gitrevparse(self.root_dir)
        )

    def testCanSyncToGerritChange(self):
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        self.options.revision = self.githash("repo_1", 1)
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 1), self.gitrevparse(self.root_dir)
        )

        self.options.revision = "refs/changes/35/1235/1"
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 6), self.gitrevparse(self.root_dir)
        )

    def testCanCloneGerritChangeMirror(self):
        with mock.patch(
            "git_cache.Mirror.contains_revision",
            side_effect=lambda r: r == "refs/changes/35/1235/1",
        ):
            self.testCanCloneGerritChange()

    def testCanSyncToGerritChangeMirror(self):
        with mock.patch(
            "git_cache.Mirror.contains_revision",
            side_effect=lambda r: r == "refs/changes/35/1235/1",
        ):
            self.testCanSyncToGerritChange()

    def testMirrorPushUrl(self):
        self.setUpMirror()

        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []
        self.assertIsNotNone(git_wrapper._GetMirror(self.url, self.options))

        git_wrapper.update(self.options, None, file_list)

        fetch_url = git_wrapper._Capture(["remote", "get-url", "origin"])
        self.assertTrue(
            fetch_url.startswith(self.mirror),
            msg="\n".join(
                [
                    "Repository fetch url should be in the git cache mirror directory.",
                    "  fetch_url: %s" % fetch_url,
                    "  mirror:    %s" % self.mirror,
                ]
            ),
        )
        push_url = git_wrapper._Capture(
            ["remote", "get-url", "--push", "origin"]
        )
        self.assertEqual(push_url, self.url)

    def setUpBootstrapMode(self):
        self.setUpMirror()
        gclient_scm.GitWrapper.cache_mode = gclient_scm.CacheMode.BOOTSTRAP
        self.addCleanup(
            setattr,
            gclient_scm.GitWrapper,
            "cache_mode",
            gclient_scm.CacheMode.SHARED,
        )

    def testBootstrapModeDetachesCheckoutFromMirror(self):
        self.setUpBootstrapMode()

        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        self.assertIsNotNone(git_wrapper._GetMirror(self.url, self.options))

        git_wrapper.update(self.options, None, [])

        # Configured like a checkout made without a cache: both remote URLs
        # on the real remote, no alternates borrowing from the mirror.
        fetch_url = git_wrapper._Capture(["remote", "get-url", "origin"])
        self.assertEqual(fetch_url, self.url)
        push_url = git_wrapper._Capture(
            ["remote", "get-url", "--push", "origin"]
        )
        self.assertEqual(push_url, self.url)
        self.assertFalse(
            os.path.exists(
                os.path.join(
                    self.root_dir, ".git", "objects", "info", "alternates"
                )
            )
        )

        # The checkout must survive deletion of the cache.
        gclient_utils.rmtree(self.mirror)
        git_wrapper._Capture(["fsck", "--no-progress"])

    def testBootstrapModeExistingCheckoutIgnoresMirror(self):
        self.setUpBootstrapMode()

        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        git_wrapper.update(self.options, None, [])

        # Only a checkout that does not exist yet may use the mirror.
        self.assertIsNone(git_wrapper._GetMirror(self.url, self.options))

        # A later sync keeps the detached configuration.
        git_wrapper.update(self.options, None, [])
        fetch_url = git_wrapper._Capture(["remote", "get-url", "origin"])
        self.assertEqual(fetch_url, self.url)

    def testBootstrapModeKeepsExistingSharedCheckout(self):
        self.setUpMirror()

        # Sync in the default shared topology: origin on the mirror, objects
        # borrowed via alternates.
        shared = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        shared.update(self.options, None, [])
        alternates = os.path.join(
            self.root_dir, ".git", "objects", "info", "alternates"
        )
        self.assertTrue(os.path.exists(alternates))

        gclient_scm.GitWrapper.cache_mode = gclient_scm.CacheMode.BOOTSTRAP
        self.addCleanup(
            setattr,
            gclient_scm.GitWrapper,
            "cache_mode",
            gclient_scm.CacheMode.SHARED,
        )

        # Bootstrap must not detach it, or deleting the cache corrupts it.
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        self.assertIsNotNone(git_wrapper._GetMirror(self.url, self.options))

        git_wrapper.update(self.options, None, [])
        self.assertTrue(os.path.exists(alternates))
        fetch_url = git_wrapper._Capture(["remote", "get-url", "origin"])
        self.assertTrue(fetch_url.startswith(self.mirror))

    def testAppliesPatchOnTopOfMasterByDefault(self):
        """Test the default case, where we apply a patch on top of main."""
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        # Make sure we don't specify a revision.
        self.options.revision = None
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 4), self.gitrevparse(self.root_dir)
        )

        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/35/1235/1",
            "refs/heads/main",
            self.options,
            file_list,
        )

        self.assertCommits([1, 2, 3, 4, 5, 6])
        self.assertEqual(
            self.githash("repo_1", 4), self.gitrevparse(self.root_dir)
        )

    def testCheckoutOlderThanPatchBase(self):
        """Test applying a patch on an old checkout.

        We first checkout commit 1, and try to patch refs/changes/35/1235/1, which
        contains commits 5 and 6, and is based on top of commit 3.
        The final result should contain commits 1, 5 and 6, but not commits 2 or 3.
        """
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        # Sync to commit 1
        self.options.revision = self.githash("repo_1", 1)
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 1), self.gitrevparse(self.root_dir)
        )

        # Apply the change on top of that.
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/35/1235/1",
            "refs/heads/main",
            self.options,
            file_list,
        )

        self.assertCommits([1, 5, 6])
        self.assertEqual(
            self.githash("repo_1", 1), self.gitrevparse(self.root_dir)
        )

    def testCheckoutOriginFeature(self):
        """Tests that we can apply a patch on a branch other than main."""
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        # Sync to remote's refs/heads/feature
        self.options.revision = "refs/heads/feature"
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 9), self.gitrevparse(self.root_dir)
        )

        # Apply the change on top of that.
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/36/1236/1",
            "refs/heads/feature",
            self.options,
            file_list,
        )

        self.assertCommits([1, 2, 7, 8, 9, 10])
        self.assertEqual(
            self.githash("repo_1", 9), self.gitrevparse(self.root_dir)
        )

    def testCheckoutOriginFeatureOnOldRevision(self):
        """Tests that we can apply a patch on an old checkout, on a branch other
        than main."""
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        # Sync to remote's refs/heads/feature on an old revision
        self.options.revision = self.githash("repo_1", 7)
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 7), self.gitrevparse(self.root_dir)
        )

        # Apply the change on top of that.
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/36/1236/1",
            "refs/heads/feature",
            self.options,
            file_list,
        )

        # We shouldn't have rebased on top of 2 (which is the merge base between
        # remote's main branch and the change) but on top of 7 (which is the
        # merge base between remote's feature branch and the change).
        self.assertCommits([1, 2, 7, 10])
        self.assertEqual(
            self.githash("repo_1", 7), self.gitrevparse(self.root_dir)
        )

    def testCheckoutOriginFeaturePatchBranch(self):
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        # Sync to the hash instead of remote's refs/heads/feature.
        self.options.revision = self.githash("repo_1", 9)
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 9), self.gitrevparse(self.root_dir)
        )

        # Apply refs/changes/34/1234/1, created for remote's main branch on top
        # of remote's feature branch.
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/35/1235/1",
            "refs/heads/main",
            self.options,
            file_list,
        )

        # Commits 5 and 6 are part of the patch, and commits 1, 2, 7, 8 and 9
        # are part of remote's feature branch.
        self.assertCommits([1, 2, 5, 6, 7, 8, 9])
        self.assertEqual(
            self.githash("repo_1", 9), self.gitrevparse(self.root_dir)
        )

    def testDoesntRebasePatchMaster(self):
        """Tests that we can apply a patch without rebasing it."""
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        self.options.rebase_patch_ref = False
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 4), self.gitrevparse(self.root_dir)
        )

        # Apply the change on top of that.
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/35/1235/1",
            "refs/heads/main",
            self.options,
            file_list,
        )

        self.assertCommits([1, 2, 3, 5, 6])
        self.assertEqual(
            self.githash("repo_1", 5), self.gitrevparse(self.root_dir)
        )

    def testDoesntRebasePatchOldCheckout(self):
        """Tests that we can apply a patch without rebasing it on an old checkout."""
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        # Sync to commit 1
        self.options.revision = self.githash("repo_1", 1)
        self.options.rebase_patch_ref = False
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 1), self.gitrevparse(self.root_dir)
        )

        # Apply the change on top of that.
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/35/1235/1",
            "refs/heads/main",
            self.options,
            file_list,
        )

        self.assertCommits([1, 2, 3, 5, 6])
        self.assertEqual(
            self.githash("repo_1", 5), self.gitrevparse(self.root_dir)
        )

    def testDoesntSoftResetIfNotAskedTo(self):
        """Test that we can apply a patch without doing a soft reset."""
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        self.options.reset_patch_ref = False
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 4), self.gitrevparse(self.root_dir)
        )

        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/35/1235/1",
            "refs/heads/main",
            self.options,
            file_list,
        )

        self.assertCommits([1, 2, 3, 4, 5, 6])
        # The commit hash after cherry-picking is not known, but it must be
        # different from what the repo was synced at before patching.
        self.assertNotEqual(
            self.githash("repo_1", 4), self.gitrevparse(self.root_dir)
        )

    @mock.patch("gerrit_util.GetChange", return_value={"topic": "test_topic"})
    @mock.patch(
        "gerrit_util.QueryChanges",
        return_value=[
            {"_number": 1234},
            {
                "_number": 1235,
                "current_revision": "abc",
                "revisions": {"abc": {"ref": "refs/changes/35/1235/1"}},
            },
        ],
    )
    def testDownloadTopics(self, query_changes_mock, get_change_mock):
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        self.options.revision = "refs/changes/34/1234/1"
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 5), self.gitrevparse(self.root_dir)
        )

        self.options.download_topics = True
        git_wrapper.url = "https://test-repo.googlesource.com/repo_1.git"
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/34/1234/1",
            "refs/heads/main",
            self.options,
            file_list,
        )

        get_change_mock.assert_called_once_with(mock.ANY, "1234")
        query_changes_mock.assert_called_once_with(
            mock.ANY,
            [("topic", "test_topic"), ("status", "open"), ("repo", "repo_1")],
            o_params=["ALL_REVISIONS"],
        )

        self.assertCommits([1, 2, 3, 5, 6])
        # The commit hash after the two cherry-picks is not known, but it must
        # be different from what the repo was synced at before patching.
        self.assertNotEqual(
            self.githash("repo_1", 4), self.gitrevparse(self.root_dir)
        )

    def testRecoversAfterPatchFailure(self):
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        self.options.revision = "refs/changes/34/1234/1"
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 5), self.gitrevparse(self.root_dir)
        )

        # Checkout 'refs/changes/34/1234/1' modifies the 'change' file, so
        # trying to patch 'refs/changes/36/1236/1' creates a patch failure.
        with self.assertRaises(subprocess2.CalledProcessError) as cm:
            git_wrapper.apply_patch_ref(
                self.url,
                "refs/changes/36/1236/1",
                "refs/heads/main",
                self.options,
                file_list,
            )
        self.assertEqual(cm.exception.cmd[3], "cherry-pick")
        self.assertIn(b"error: could not apply", cm.exception.stderr)

        # Try to apply 'refs/changes/35/1235/1', which doesn't have a merge
        # conflict.
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/35/1235/1",
            "refs/heads/main",
            self.options,
            file_list,
        )
        self.assertCommits([1, 2, 3, 5, 6])
        self.assertEqual(
            self.githash("repo_1", 5), self.gitrevparse(self.root_dir)
        )

    def testIgnoresAlreadyMergedCommits(self):
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        self.options.revision = "refs/heads/main-with-5"
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 12), self.gitrevparse(self.root_dir)
        )

        # When we try 'refs/changes/35/1235/1' on top of 'refs/heads/feature',
        # 'refs/changes/34/1234/1' will be an empty commit, since the changes
        # were already present in the tree as commit 11. Make sure we deal with
        # this gracefully.
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/35/1235/1",
            "refs/heads/feature",
            self.options,
            file_list,
        )
        self.assertCommits([1, 2, 3, 5, 6, 12])
        self.assertEqual(
            self.githash("repo_1", 12), self.gitrevparse(self.root_dir)
        )

    def testRecoversFromExistingCherryPick(self):
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        self.options.revision = "refs/changes/34/1234/1"
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 5), self.gitrevparse(self.root_dir)
        )

        # Checkout 'refs/changes/34/1234/1' modifies the 'change' file, so
        # trying to cherry-pick 'refs/changes/36/1236/1' raises an error.
        git_wrapper._Run(
            ["fetch", "origin", "refs/changes/36/1236/1"], self.options
        )
        with self.assertRaises(subprocess2.CalledProcessError) as cm:
            git_wrapper._Run(["cherry-pick", "FETCH_HEAD"], self.options)
        self.assertEqual(cm.exception.cmd[:2], ["git", "cherry-pick"])

        # Try to apply 'refs/changes/35/1235/1', which doesn't have a merge
        # conflict.
        git_wrapper.apply_patch_ref(
            self.url,
            "refs/changes/35/1235/1",
            "refs/heads/main",
            self.options,
            file_list,
        )
        self.assertCommits([1, 2, 3, 5, 6])
        self.assertEqual(
            self.githash("repo_1", 5), self.gitrevparse(self.root_dir)
        )


class DepsChangesFakeRepo(fake_repos.FakeReposBase):
    def populateGit(self):
        self._commit_git("repo_1", {"DEPS": "versionA", "doesnotmatter": "B"})
        self._commit_git("repo_1", {"DEPS": "versionA", "doesnotmatter": "C"})

        self._commit_git("repo_1", {"DEPS": "versionB"})
        self._commit_git("repo_1", {"DEPS": "versionA", "doesnotmatter": "C"})
        self._create_ref("repo_1", "refs/heads/main", 4)


class CheckDiffTest(fake_repos.FakeReposTestBase):
    FAKE_REPOS_CLASS = DepsChangesFakeRepo

    def setUp(self):
        super(CheckDiffTest, self).setUp()
        self.enabled = self.FAKE_REPOS.set_up_git()
        self.options = BaseGitWrapperTestCase.OptionsObject()
        self.url = self.git_base + "repo_1"
        self.mirror = None
        mock.patch("sys.stdout", StringIO()).start()
        self.addCleanup(mock.patch.stopall)

    def setUpMirror(self):
        self.mirror = tempfile.mkdtemp()
        git_cache.Mirror.SetCachePath(self.mirror)
        self.addCleanup(gclient_utils.rmtree, self.mirror)
        self.addCleanup(git_cache.Mirror.SetCachePath, None)

    def testCheckDiff(self):
        """Correctly check for diffs."""
        git_wrapper = gclient_scm.GitWrapper(self.url, self.root_dir, ".")
        file_list = []

        # Make sure we don't specify a revision.
        self.options.revision = None
        git_wrapper.update(self.options, None, file_list)
        self.assertEqual(
            self.githash("repo_1", 4), self.gitrevparse(self.root_dir)
        )

        self.assertFalse(
            git_wrapper.check_diff(self.githash("repo_1", 1), files=["DEPS"])
        )
        self.assertTrue(git_wrapper.check_diff(self.githash("repo_1", 1)))
        self.assertTrue(
            git_wrapper.check_diff(self.githash("repo_1", 3), files=["DEPS"])
        )

        self.assertFalse(
            git_wrapper.check_diff(
                self.githash("repo_1", 2), files=["DEPS", "doesnotmatter"]
            )
        )
        self.assertFalse(git_wrapper.check_diff(self.githash("repo_1", 2)))


class Submodules(BaseGitWrapperTestCase):
    submodule_hash = "1111111111111111111111111111111111111111"

    def testGetSubmoduleClean(self):
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        options = self.Options()
        git_wrapper.update(options, None, [])
        self.assertEqual(
            git_wrapper.GetSubmoduleStateFromIndex(),
            {"submodule": self.submodule_hash},
        )
        self.assertEqual(git_wrapper.GetSubmoduleDiff(), {})

    def testGetSubmoduleModified(self):
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        options = self.Options()
        git_wrapper.update(options, None, [])

        # Create submodule diff
        submodule_dir = os.path.join(self.root_dir, "submodule")
        subprocess2.check_output(["git", "-C", submodule_dir, "init"])
        subprocess2.check_output(
            ["git", "-C", submodule_dir, "commit", "-m", "foo", "--allow-empty"]
        )
        new_rev = (
            subprocess2.check_output(
                ["git", "-C", submodule_dir, "rev-parse", "HEAD"]
            )
            .decode("utf-8")
            .strip()
        )

        # And file diff
        with open(os.path.join(self.root_dir, "a"), "w") as f:
            f.write("foo")

        self.assertEqual(
            git_wrapper.GetSubmoduleStateFromIndex(),
            {"submodule": self.submodule_hash},
        )

        self.assertEqual(
            git_wrapper.GetSubmoduleDiff(),
            {"submodule": (self.submodule_hash, new_rev)},
        )

    def testGetSubmoduleDeleted(self):
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        options = self.Options()
        git_wrapper.update(options, None, [])
        subprocess2.check_output(
            ["git", "-C", self.root_dir, "rm", "submodule"]
        )

        # When git removes submodule, it's autmatically staged and content is
        # unavailable. Therefore, the index shouldn't have any entries and diff
        # should be empty.
        self.assertEqual(git_wrapper.GetSubmoduleStateFromIndex(), {})

        self.assertEqual(git_wrapper.GetSubmoduleDiff(), {})

    def testGetSubmoduleAssumeUnchanged(self):
        git_wrapper = gclient_scm.GitWrapper(
            self.url, self.root_dir, self.relpath
        )
        options = self.Options()
        git_wrapper.update(options, None, [])
        subprocess2.check_output(
            [
                "git",
                "-C",
                self.root_dir,
                "update-index",
                "--assume-unchanged",
                "submodule",
            ]
        )
        self.assertEqual(git_wrapper.GetSubmoduleStateFromIndex(), {})
        self.assertEqual(git_wrapper.GetSubmoduleDiff(), {})


if "unittest.util" in __import__("sys").modules:
    # Show full diff in self.assertEqual.
    __import__("sys").modules["unittest.util"]._MAX_LENGTH = 999999999

if __name__ == "__main__":
    level = logging.DEBUG if "-v" in sys.argv else logging.FATAL
    logging.basicConfig(
        level=level,
        format="%(asctime).19s %(levelname)s %(filename)s:"
        "%(lineno)s %(message)s",
    )
    unittest.main()

# vim: ts=2:sw=2:tw=80:et: