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

"""Unittests for chromite.lib.patch."""

import copy
import functools
import itertools
import os
import shutil
import tempfile
import time
from unittest import mock

from chromite.lib import config_lib
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import gerrit
from chromite.lib import git
from chromite.lib import osutils
from chromite.lib import patch as cros_patch


_GetNumber = functools.partial(next, itertools.count())

# Change-ID of a known open change in public gerrit.
GERRIT_OPEN_CHANGEID = "8366"
GERRIT_MERGED_CHANGEID = "3"
GERRIT_ABANDONED_CHANGEID = "2"

FAKE_PATCH_JSON = {
    "project": "tacos/chromite",
    "branch": "main",
    "id": "Iee5c89d929f1850d7d4e1a4ff5f21adda800025f",
    "currentPatchSet": {
        "number": "2",
        "ref": gerrit.GetChangeRef(1112, 2),
        "revision": "ff10979dd360e75ff21f5cf53b7f8647578785ef",
    },
    "number": "1112",
    "subject": "chromite commit",
    "owner": {
        "name": "Chromite main",
        "email": "chromite@chromium.org",
    },
    "url": "https://chromium-review.googlesource.com/1112",
    "lastUpdated": 1311024529,
    "sortKey": "00166e8700001052",
    "open": True,
    "status": "NEW",
}

# List of labels as seen in the top level desc by normal CrOS devs.
FAKE_LABELS_JSON = {
    "Code-Review": {
        "default_value": 0,
        "values": {
            " 0": "No score",
            "+1": "Looks good to me, but someone else must approve",
            "+2": "Looks good to me, approved",
            "-1": "I would prefer that you did not submit this",
            "-2": "Do not submit",
        },
    },
    "Commit-Queue": {
        "default_value": 0,
        "optional": True,
        "values": {
            " 0": "Not Ready",
            "+1": "Try ready",
            "+2": "Ready",
            "+3": "Ignore this label",
        },
    },
    "Verified": {
        "default_value": 0,
        "values": {
            " 0": "No score",
            "+1": "Verified",
            "-1": "Fails",
        },
    },
}

# List of label values as seen in the permitted_labels field.
# Note: The space before the 0 is intentional -- it's what gerrit returns.
FAKE_PERMITTED_LABELS_JSON = {
    "Code-Review": ["-2", "-1", " 0", "+1", "+2"],
    "Commit-Queue": [" 0", "+1", "+2"],
    "Verified": ["-1", " 0", "+1"],
}

# An account structure as seen in owners or reviewers lists.
FAKE_GERRIT_ACCT_JSON = {
    "_account_id": 991919291,
    "avatars": [
        {"height": 26, "url": "https://example.com/26/photo.jpg"},
        {"height": 32, "url": "https://example.com/32/photo.jpg"},
        {"height": 100, "url": "https://example.com/100/photo.jpg"},
    ],
    "email": "happy-funky-duck@chromium.org",
    "name": "Duckworth Duck",
}

# A valid change json result as returned by gerrit.
FAKE_CHANGE_JSON = {
    "_number": 8366,
    "branch": "main",
    "change_id": "I3a753d6bacfbe76e5675a6f2f7941fe520c095e5",
    "created": "2016-09-14 23:02:46.000000000",
    "current_revision": "be54f9935bedb157b078eefa26fc1885b8264da6",
    "deletions": 1,
    "hashtags": [],
    "id": "example%2Frepo~main~I3a753d6bacfbe76e5675a6f2f7941fe520c095e5",
    "insertions": 0,
    "labels": FAKE_LABELS_JSON,
    "mergeable": True,
    "owner": FAKE_GERRIT_ACCT_JSON,
    "permitted_labels": FAKE_PERMITTED_LABELS_JSON,
    "project": "example/repo",
    "removable_reviewers": [],
    "reviewers": {},
    "revisions": {
        "be54f9935bedb157b078eefa26fc1885b8264da6": {
            "_number": 1,
            "commit": {
                "author": {
                    "date": "2016-09-14 22:54:07.000000000",
                    "email": "vapier@chromium.org",
                    "name": "Mike Frysinger",
                    "tz": -240,
                },
                "committer": {
                    "date": "2016-09-14 23:02:03.000000000",
                    "email": "vapier@chromium.org",
                    "name": "Mike Frysinger",
                    "tz": -240,
                },
                "message": (
                    "my super great message\n\nChange-Id: "
                    "I3a753d6bacfbe76e5675a6f2f7941fe520c095e5\n"
                ),
                "parents": [
                    {
                        "commit": "3d54362a9b010330bae2dde973fc5c3efc4e5f44",
                        "subject": "some parent commit",
                    },
                ],
                "subject": "my super great message",
            },
            "created": "2016-09-14 23:02:46.000000000",
            "fetch": {
                "http": {
                    "ref": "refs/changes/15/8366/1",
                    "url": "https://chromium.googlesource.com/example/repo",
                },
                "repo": {
                    "ref": "refs/changes/15/8366/1",
                    "url": "example/repo",
                },
                "rpc": {
                    "ref": "refs/changes/15/8366/1",
                    "url": "rpc://chromium/example/repo",
                },
                "sso": {
                    "ref": "refs/changes/15/8366/1",
                    "url": "sso://chromium/example/repo",
                },
            },
            "kind": "REWORK",
            "ref": "refs/changes/15/8366/1",
            "uploader": FAKE_GERRIT_ACCT_JSON,
        },
    },
    "status": "NEW",
    "subject": "my super great message",
    "submit_type": "CHERRY_PICK",
    "topic": "some-topic",
    "updated": "2016-09-14 23:02:46.000000000",
}


class GitRepoPatchTestCase(cros_test_lib.TempDirTestCase):
    """Helper TestCase class for writing test cases."""

    # No mock bits are to be used in this class's tests.
    # This needs to actually validate git output, and git behaviour, rather
    # than test our assumptions about git's behaviour/output.

    patch_kls = cros_patch.GitRepoPatch

    COMMIT_TEMPLATE = """\
commit abcdefgh

Author: Fake person
Date:  Tue Oct 99

I am the first commit.

%(extra)s

%(change-id)s
"""

    # Boolean controlling whether the target class natively knows its
    # ChangeId; only GerritPatches do.
    has_native_change_id = False

    DEFAULT_TRACKING = (
        "refs/remotes/%s/main" % config_lib.GetSiteParams().EXTERNAL_REMOTE
    )

    @staticmethod
    def _CreateSourceRepo(tmp_path):
        """Generate a new repo with a single commit."""
        bare_path = os.path.join(tmp_path, "bare.git")
        checkout_path = os.path.join(tmp_path, "checkout")
        cros_build_lib.run(
            [
                "git",
                "init",
                "--separate-git-dir",
                bare_path,
                "--initial-branch",
                "main",
                checkout_path,
            ],
            cwd=tmp_path,
            print_cmd=False,
            capture_output=True,
        )

        # Nerf any hooks the OS might have installed on us as they aren't going
        # to be useful to us, just slow things down.
        shutil.rmtree(os.path.join(bare_path, "hooks"))

        # Add an initial commit then wipe the working tree.
        cros_build_lib.run(
            ["git", "commit", "--allow-empty", "-m", "initial commit"],
            cwd=checkout_path,
            print_cmd=False,
            capture_output=True,
        )

        return bare_path

    @classmethod
    def setUpClass(cls):
        # Generate the same git tree once as it's a bit expensive.
        cls._cache_root = tempfile.mkdtemp(prefix="chromite-patch-unittest")
        cls._cache_bare = cls._CreateSourceRepo(cls._cache_root)

    @classmethod
    def tearDownClass(cls):
        shutil.rmtree(cls._cache_root)

    def setUp(self):
        # Create an empty repo to work from.
        self.source = self._cache_bare
        self.default_cwd = os.path.join(self.tempdir, "unwritable")
        # Disallow write so as to smoke out any invalid writes to cwd.
        os.mkdir(self.default_cwd, 0o500)
        os.chdir(self.default_cwd)

    def _MkPatch(self, source, sha1, ref="refs/heads/main", **kwargs):
        # This arg is used by inherited versions of _MkPatch. Pop it to make
        # this _MkPatch compatible with them.
        site_params = config_lib.GetSiteParams()
        kwargs.pop("suppress_branch", None)
        return self.patch_kls(
            source,
            "chromiumos/chromite",
            ref,
            "%s/main" % site_params.EXTERNAL_REMOTE,
            kwargs.pop("remote", site_params.EXTERNAL_REMOTE),
            sha1=sha1,
            **kwargs,
        )

    def _run(self, cmd, cwd=None, **kwargs):
        # This is to match git.RunGit behavior.
        kwargs.setdefault("print_cmd", False)
        if "capture_output" not in kwargs:
            kwargs.setdefault("stdout", True)
            kwargs.setdefault("stderr", True)
        kwargs.setdefault("encoding", "utf-8")

        # Note that cwd is intentionally set to a location the user can't write
        # to; this flushes out any bad usage in the tests that would work by
        # fluke of being invoked from w/in a git repo.
        if cwd is None:
            cwd = self.default_cwd

        return cros_build_lib.run(cmd, cwd=cwd, **kwargs).stdout.strip()

    def _GetSha1(self, cwd, refspec):
        return self._run(["git", "rev-list", "-n1", refspec], cwd=cwd)

    def _MakeRepo(self, name, clone, remote=None, alternates=True):
        path = os.path.join(self.tempdir, name)
        cmd = ["git", "clone", clone, path]
        if alternates:
            cmd += ["--reference", clone]
        if remote is None:
            remote = config_lib.GetSiteParams().EXTERNAL_REMOTE
        cmd += ["--origin", remote]
        self._run(cmd)
        # Nerf any hooks the OS might have installed on us as they aren't going
        # to be useful to us, just slow things down.
        shutil.rmtree(os.path.join(path, ".git", "hooks"))
        return path

    def _MakeCommit(self, repo, commit=None):
        if commit is None:
            commit = "commit at %s" % (time.time(),)
        self._run(["git", "commit", "-a", "-m", commit], repo)
        return self._GetSha1(repo, "HEAD")

    def CommitFile(self, repo, filename, content, commit=None, **kwargs):
        osutils.WriteFile(os.path.join(repo, filename), content)
        self._run(["git", "add", filename], repo)
        sha1 = self._MakeCommit(repo, commit=commit)
        if not self.has_native_change_id:
            kwargs.pop("ChangeId", None)
        patch = self._MkPatch(repo, sha1, **kwargs)
        self.assertEqual(patch.sha1, sha1)
        return patch

    def _CommonGitSetup(self):
        git1 = self._MakeRepo("git1", self.source)
        git2 = os.path.join(self.tempdir, "git2")
        # Avoid another git clone as it's a bit expensive.
        shutil.copytree(git1, git2)
        patch = self.CommitFile(git1, "monkeys", "foon")
        return git1, git2, patch

    def MakeChangeId(self, how_many=1):
        l = [cros_patch.MakeChangeId() for _ in range(how_many)]
        if how_many == 1:
            return l[0]
        return l

    def CommitChangeIdFile(
        self,
        repo,
        changeid=None,
        extra=None,
        filename="monkeys",
        content="flinging",
        raw_changeid_text=None,
        **kwargs,
    ):
        template = self.COMMIT_TEMPLATE
        if changeid is None:
            changeid = self.MakeChangeId()
        if raw_changeid_text is None:
            raw_changeid_text = "Change-Id: %s" % (changeid,)
        if extra is None:
            extra = ""
        commit = template % {"change-id": raw_changeid_text, "extra": extra}

        return self.CommitFile(
            repo, filename, content, commit=commit, ChangeId=changeid, **kwargs
        )


# pylint: disable=protected-access
class TestGitRepoPatch(GitRepoPatchTestCase):
    """Unittests for git patch related methods."""

    def testGetDiffStatus(self):
        git1, _, patch1 = self._CommonGitSetup()
        # Ensure that it can work on the first commit, even if it
        # doesn't report anything (no delta; it's the first files).
        patch1 = self._MkPatch(git1, self._GetSha1(git1, self.DEFAULT_TRACKING))
        self.assertEqual({}, patch1.GetDiffStatus(git1))
        patch2 = self.CommitFile(git1, "monkeys", "blah")
        self.assertEqual({"monkeys": "M"}, patch2.GetDiffStatus(git1))
        git.RunGit(git1, ["mv", "monkeys", "monkeys2"])
        patch3 = self._MkPatch(git1, self._MakeCommit(git1, commit="mv"))
        self.assertEqual(
            {"monkeys": "D", "monkeys2": "A"}, patch3.GetDiffStatus(git1)
        )
        patch4 = self.CommitFile(git1, "monkey2", "blah")
        self.assertEqual({"monkey2": "A"}, patch4.GetDiffStatus(git1))

    def testFetch(self):
        _, git2, patch = self._CommonGitSetup()
        patch.Fetch(git2)
        self.assertEqual(patch.sha1, self._GetSha1(git2, "FETCH_HEAD"))
        # Verify reuse; specifically that Fetch doesn't actually run since
        # the rev is already available locally via alternates.
        patch.project_url = "/dev/null"
        git3 = self._MakeRepo("git3", git2)
        patch.Fetch(git3)
        self.assertEqual(patch.sha1, self._GetSha1(git3, patch.sha1))

    def testFetchFirstPatchInSeries(self):
        git1, git2, patch = self._CommonGitSetup()
        self.CommitFile(git1, "monkeys", "foon2")
        patch.Fetch(git2)

    def testFetchWithoutSha1(self):
        git1, git2, _ = self._CommonGitSetup()
        patch2 = self.CommitFile(git1, "monkeys", "foon2")
        sha1, patch2.sha1 = patch2.sha1, None
        patch2.Fetch(git2)
        self.assertEqual(sha1, patch2.sha1)

    def testParentless(self):
        git1 = self._MakeRepo("git1", self.source)
        patch1 = self._MkPatch(git1, self._GetSha1(git1, "HEAD"))
        self.assertRaises2(
            cros_patch.PatchNoParents,
            patch1.Apply,
            git1,
            self.DEFAULT_TRACKING,
            check_attrs={"inflight": False},
        )

    def testNoParentOrAlreadyApplied(self):
        git1 = self._MakeRepo("git1", self.source)
        patch1 = self._MkPatch(git1, self._GetSha1(git1, "HEAD"))
        self.assertRaises2(
            cros_patch.PatchNoParents,
            patch1.Apply,
            git1,
            self.DEFAULT_TRACKING,
            check_attrs={"inflight": False},
        )
        patch2 = self.CommitFile(git1, "monkeys", "rule")
        self.assertRaises2(
            cros_patch.PatchIsEmpty,
            patch2.Apply,
            git1,
            self.DEFAULT_TRACKING,
            check_attrs={"inflight": True},
        )

    def testGetNoParents(self):
        git1 = self._MakeRepo("git1", self.source)
        sha1 = self._GetSha1(git1, "HEAD")
        patch = self._MkPatch(self.source, sha1)
        self.assertEqual(patch._GetParents(git1), [])

    def testGet1Parent(self):
        git1 = self._MakeRepo("git1", self.source)
        patch1 = self.CommitFile(git1, "foo", "foo")
        patch2 = self.CommitFile(git1, "bar", "bar")
        self.assertEqual(patch2._GetParents(git1), [patch1.sha1])

    def testGet2Parents(self):
        # Prepare a merge commit, then test that its two parents are correctly
        # calculated.
        git1 = self._MakeRepo("git1", self.source)
        patch_common = self.CommitFile(git1, "foo", "foo")

        patch_right = self.CommitFile(git1, "bar", "bar")

        git.RunGit(git1, ["reset", "--hard", patch_common.sha1])
        patch_left = self.CommitFile(git1, "baz", "baz")

        git.RunGit(git1, ["merge", patch_right.sha1])
        sha1 = self._GetSha1(git1, "HEAD")
        patch_merge = self._MkPatch(self.source, sha1, suppress_branch=True)

        self.assertEqual(
            patch_merge._GetParents(git1), [patch_left.sha1, patch_right.sha1]
        )

    def testIsAncestor(self):
        git1 = self._MakeRepo("git1", self.source)
        patch1 = self.CommitFile(git1, "foo", "foo")
        patch2 = self.CommitFile(git1, "bar", "bar")
        self.assertTrue(patch1._IsAncestorOf(git1, patch1))
        self.assertTrue(patch1._IsAncestorOf(git1, patch2))
        self.assertFalse(patch2._IsAncestorOf(git1, patch1))

    def testFromSha1(self):
        git1 = self._MakeRepo("git1", self.source)
        patch1 = self.CommitFile(git1, "foo", "foo")
        patch2 = self.CommitFile(git1, "bar", "bar")
        patch2_from_sha1 = patch1._FromSha1(patch2.sha1)
        patch2_from_sha1.Fetch(git1)
        patch2.Fetch(git1)
        self.assertEqual(patch2.tree_hash, patch2_from_sha1.tree_hash)

    def testValidateMerge(self):
        git1 = self._MakeRepo("git1", self.source)

        # Prepare history like this:
        #   * E (upstream)
        # * | D (merge being handled)
        # |\|
        # * | C
        # | * B
        # |/
        # *   A
        #
        # D is valid to merge into E.
        A = self.CommitFile(git1, "A", "A")
        B = self.CommitFile(git1, "B", "B")
        E = self.CommitFile(git1, "E", "E")
        git.RunGit(git1, ["reset", "--hard", A.sha1])
        C = self.CommitFile(git1, "C", "C")
        git.RunGit(git1, ["merge", B.sha1])
        sha1 = self._GetSha1(git1, "HEAD")
        D = self._MkPatch(self.source, sha1, suppress_branch=True)

        D._ValidateMergeCommit(git1, E.sha1, [C.sha1, B.sha1])

    def testValidateMergeFailure(self):
        git1 = self._MakeRepo("git1", self.source)
        # *     F (merge being handled)
        # |\
        # | *   E
        # * |   D
        # | | * C (upstream)
        # | |/
        # | *   B
        # |/
        # *     A
        #
        # F is not valid to merge into E.
        A = self.CommitFile(git1, "A", "A")
        B = self.CommitFile(git1, "B", "B")
        C = self.CommitFile(git1, "C", "C")
        git.RunGit(git1, ["reset", "--hard", B.sha1])
        E = self.CommitFile(git1, "E", "E")
        git.RunGit(git1, ["reset", "--hard", A.sha1])
        D = self.CommitFile(git1, "D", "D")
        git.RunGit(git1, ["merge", E.sha1])
        sha1 = self._GetSha1(git1, "HEAD")
        F = self._MkPatch(self.source, sha1, suppress_branch=True)

        with self.assertRaises(cros_patch.NonMainlineMerge):
            F._ValidateMergeCommit(git1, C.sha1, [D.sha1, E.sha1])

    def testDeleteEbuildTwice(self):
        """Test that double-deletes of ebuilds are flagged as conflicts."""
        # Create monkeys.ebuild for testing.
        git1 = self._MakeRepo("git1", self.source)
        patch1 = self.CommitFile(git1, "monkeys.ebuild", "rule")
        git.RunGit(git1, ["rm", "monkeys.ebuild"])
        patch2 = self._MkPatch(git1, self._MakeCommit(git1, commit="rm"))

        # Delete an ebuild that does not exist in TOT.
        check_attrs = {"inflight": False, "files": ("monkeys.ebuild",)}
        self.assertRaises2(
            cros_patch.EbuildConflict,
            patch2.Apply,
            git1,
            self.DEFAULT_TRACKING,
            check_attrs=check_attrs,
        )

        # Delete an ebuild that exists in TOT, but does not exist in the current
        # patch series.
        check_attrs["inflight"] = True
        self.assertRaises2(
            cros_patch.EbuildConflict,
            patch2.Apply,
            git1,
            patch1.sha1,
            check_attrs=check_attrs,
        )

    def testCleanlyApply(self):
        _, git2, patch = self._CommonGitSetup()
        # Clone git3 before we modify git2; else we'll just wind up
        # cloning its default branch.
        git3 = self._MakeRepo("git3", git2)
        patch.Apply(git2, self.DEFAULT_TRACKING)
        # Verify reuse; specifically that Fetch doesn't actually run since
        # the object is available in alternates.  testFetch partially
        # validates this; the Apply usage here fully validates it via
        # ensuring that the attempted Apply goes boom if it can't get the
        # required sha1.
        patch.project_url = "/dev/null"
        patch.Apply(git3, self.DEFAULT_TRACKING)

    def testFailsApply(self):
        _, git2, patch1 = self._CommonGitSetup()
        patch2 = self.CommitFile(git2, "monkeys", "not foon")
        # Note that Apply creates its own branch, resetting to the default,
        # thus we have to re-apply (even if it looks wrong, it's right).
        patch2.Apply(git2, self.DEFAULT_TRACKING)
        self.assertRaises2(
            cros_patch.ApplyPatchException,
            patch1.Apply,
            git2,
            self.DEFAULT_TRACKING,
            exact_kls=True,
            check_attrs={"inflight": True},
        )

    def testTrivial(self):
        _, git2, patch1 = self._CommonGitSetup()
        # Throw in a bunch of newlines so that content-merging would work.
        content = "not foon%s" % ("\n" * 100)
        patch1 = self._MkPatch(git2, self._GetSha1(git2, "HEAD"))
        patch1 = self.CommitFile(git2, "monkeys", content)
        git.RunGit(git2, ["update-ref", self.DEFAULT_TRACKING, patch1.sha1])
        patch2 = self.CommitFile(git2, "monkeys", "%sblah" % content)
        patch3 = self.CommitFile(git2, "monkeys", "%sblahblah" % content)
        # Get us a back to the basic, then derive from there; this is used to
        # verify that even if content merging works, trivial is flagged.
        self.CommitFile(git2, "monkeys", "foon")
        patch4 = self.CommitFile(git2, "monkeys", content)
        patch5 = self.CommitFile(git2, "monkeys", "%sfoon" % content)
        # Reset so we derive the next changes from patch1.
        git.RunGit(git2, ["reset", "--hard", patch1.sha1])
        patch6 = self.CommitFile(git2, "blah", "some-other-file")
        self.CommitFile(
            git2, "monkeys", "%sblah" % content.replace("not", "bot")
        )

        self.assertRaises2(
            cros_patch.PatchIsEmpty,
            patch1.Apply,
            git2,
            self.DEFAULT_TRACKING,
            trivial=True,
            check_attrs={"inflight": False, "trivial": False},
        )

        # Now test conflicts since we're still at ToT; note that this is an
        # actual conflict because the fuzz anchors have changed.
        self.assertRaises2(
            cros_patch.ApplyPatchException,
            patch3.Apply,
            git2,
            self.DEFAULT_TRACKING,
            trivial=True,
            check_attrs={"inflight": False, "trivial": False},
            exact_kls=True,
        )

        # Now test trivial conflict; this would've merged fine were it not for
        # trivial.
        self.assertRaises2(
            cros_patch.PatchIsEmpty,
            patch4.Apply,
            git2,
            self.DEFAULT_TRACKING,
            trivial=True,
            check_attrs={"inflight": False, "trivial": False},
            exact_kls=True,
        )

        # Move us into inflight testing.
        patch2.Apply(git2, self.DEFAULT_TRACKING, trivial=True)

        # Repeat the tests from above; should still be the same.
        self.assertRaises2(
            cros_patch.PatchIsEmpty,
            patch4.Apply,
            git2,
            self.DEFAULT_TRACKING,
            trivial=True,
            check_attrs={"inflight": False, "trivial": False},
        )

        # Actual conflict merge conflict due to inflight; non trivial induced.
        self.assertRaises2(
            cros_patch.ApplyPatchException,
            patch5.Apply,
            git2,
            self.DEFAULT_TRACKING,
            trivial=True,
            check_attrs={"inflight": True, "trivial": False},
            exact_kls=True,
        )

        self.assertRaises2(
            cros_patch.PatchIsEmpty,
            patch1.Apply,
            git2,
            self.DEFAULT_TRACKING,
            trivial=True,
            check_attrs={"inflight": False},
        )

        self.assertRaises2(
            cros_patch.ApplyPatchException,
            patch5.Apply,
            git2,
            self.DEFAULT_TRACKING,
            trivial=True,
            check_attrs={"inflight": True, "trivial": False},
            exact_kls=True,
        )

        # And this should apply without issue, despite the differing history.
        patch6.Apply(git2, self.DEFAULT_TRACKING, trivial=True)

    def _assertLookupAliases(self, remote):
        git1 = self._MakeRepo("git1", self.source)
        patch = self.CommitChangeIdFile(git1, remote=remote)
        prefix = "chrome-internal:" if patch.internal else "chromium:"
        vals = [
            patch.sha1,
            getattr(patch, "gerrit_number", None),
            getattr(patch, "original_sha1", None),
        ]
        # Append full Change-ID if it exists.
        if patch.project and patch.tracking_branch and patch.change_id:
            vals.append(
                "%s~%s~%s"
                % (patch.project, patch.tracking_branch, patch.change_id)
            )
        vals = [x for x in vals if x is not None]
        self.assertEqual(
            set(prefix + x for x in vals), set(patch.LookupAliases())
        )

    def testExternalLookupAliases(self):
        self._assertLookupAliases(config_lib.GetSiteParams().EXTERNAL_REMOTE)

    def testInternalLookupAliases(self):
        self._assertLookupAliases(config_lib.GetSiteParams().INTERNAL_REMOTE)

    def testChangeIdMetadata(self):
        """Verify Change-Id is set in git metadata."""
        git1, git2, _ = self._CommonGitSetup()
        changeid = "I%s" % ("1".rjust(40, "0"))
        patch = self.CommitChangeIdFile(
            git1, changeid=changeid, change_id=changeid, raw_changeid_text=""
        )
        patch.change_id = changeid
        patch.Fetch(git1)
        self.assertIn("Change-Id: %s\n" % changeid, patch.commit_message)
        patch = self.CommitChangeIdFile(
            git2, changeid=changeid, change_id=changeid
        )
        patch.Fetch(git2)
        self.assertEqual(patch.change_id, changeid)
        self.assertIn("Change-Id: %s\n" % changeid, patch.commit_message)


class TestGerritFetchOnlyPatch(cros_test_lib.MockTestCase):
    """Test of GerritFetchOnlyPatch."""

    def testFromAttrDict(self):
        """Test whether FromAttrDict can handle with commit message."""
        attr_dict_without_msg = {
            cros_patch.ATTR_PROJECT_URL: "https://host/chromite/tacos",
            cros_patch.ATTR_PROJECT: "chromite/tacos",
            cros_patch.ATTR_REF: "refs/changes/11/12345/4",
            cros_patch.ATTR_BRANCH: "main",
            cros_patch.ATTR_REMOTE: "cros-internal",
            cros_patch.ATTR_COMMIT: "7181e4b5e182b6f7d68461b04253de095bad74f9",
            cros_patch.ATTR_CHANGE_ID: (
                "I47ea30385af60ae4cc2acc5d1a283a46423bc6e1"
            ),
            cros_patch.ATTR_GERRIT_NUMBER: "12345",
            cros_patch.ATTR_PATCH_NUMBER: "4",
            cros_patch.ATTR_OWNER_EMAIL: "foo@chromium.org",
            cros_patch.ATTR_FAIL_COUNT: 1,
            cros_patch.ATTR_PASS_COUNT: 1,
            cros_patch.ATTR_TOTAL_FAIL_COUNT: 3,
        }

        attr_dict_with_msg = {
            cros_patch.ATTR_PROJECT_URL: "https://host/chromite/tacos",
            cros_patch.ATTR_PROJECT: "chromite/tacos",
            cros_patch.ATTR_REF: "refs/changes/11/12345/4",
            cros_patch.ATTR_BRANCH: "main",
            cros_patch.ATTR_REMOTE: "cros-internal",
            cros_patch.ATTR_COMMIT: "7181e4b5e182b6f7d68461b04253de095bad74f9",
            cros_patch.ATTR_CHANGE_ID: (
                "I47ea30385af60ae4cc2acc5d1a283a46423bc6e1"
            ),
            cros_patch.ATTR_GERRIT_NUMBER: "12345",
            cros_patch.ATTR_PATCH_NUMBER: "4",
            cros_patch.ATTR_OWNER_EMAIL: "foo@chromium.org",
            cros_patch.ATTR_FAIL_COUNT: 1,
            cros_patch.ATTR_PASS_COUNT: 1,
            cros_patch.ATTR_TOTAL_FAIL_COUNT: 3,
            cros_patch.ATTR_COMMIT_MESSAGE: "commit message",
        }

        self.PatchObject(
            cros_patch.GitRepoPatch,
            "_AddFooters",
            return_value="commit message",
        )

        result_1 = cros_patch.GerritFetchOnlyPatch.FromAttrDict(
            attr_dict_without_msg
        ).commit_message
        result_2 = cros_patch.GerritFetchOnlyPatch.FromAttrDict(
            attr_dict_with_msg
        ).commit_message
        self.assertEqual(None, result_1)
        self.assertEqual("commit message", result_2)

    def testGetAttributeDict(self):
        """Test Whether GetAttributeDict can get the commit message properly."""
        change = cros_patch.GerritFetchOnlyPatch(
            "https://host/chromite/tacos",
            "chromite/tacos",
            "refs/changes/11/12345/4",
            "main",
            "cros-internal",
            "7181e4b5e182b6f7d68461b04253de095bad74f9",
            "I47ea30385af60ae4cc2acc5d1a283a46423bc6e1",
            "12345",
            "4",
            "foo@chromium.org",
            1,
            1,
            3,
        )

        expected = {
            cros_patch.ATTR_PROJECT_URL: "https://host/chromite/tacos",
            cros_patch.ATTR_PROJECT: "chromite/tacos",
            cros_patch.ATTR_REF: "refs/changes/11/12345/4",
            cros_patch.ATTR_BRANCH: "main",
            cros_patch.ATTR_REMOTE: "cros-internal",
            cros_patch.ATTR_COMMIT: "7181e4b5e182b6f7d68461b04253de095bad74f9",
            cros_patch.ATTR_CHANGE_ID: (
                "I47ea30385af60ae4cc2acc5d1a283a46423bc6e1"
            ),
            cros_patch.ATTR_GERRIT_NUMBER: "12345",
            cros_patch.ATTR_PATCH_NUMBER: "4",
            cros_patch.ATTR_OWNER_EMAIL: "foo@chromium.org",
            cros_patch.ATTR_FAIL_COUNT: "1",
            cros_patch.ATTR_PASS_COUNT: "1",
            cros_patch.ATTR_TOTAL_FAIL_COUNT: "3",
            cros_patch.ATTR_COMMIT_MESSAGE: None,
        }
        self.assertEqual(change.GetAttributeDict(), expected)

        self.PatchObject(
            cros_patch.GitRepoPatch,
            "_AddFooters",
            return_value="commit message",
        )
        change.commit_message = "commit message"
        expected[cros_patch.ATTR_COMMIT_MESSAGE] = "commit message"
        self.assertEqual(change.GetAttributeDict(), expected)


class TestGetOptionLinesFromCommitMessage(cros_test_lib.TestCase):
    """Tests of GetOptionFromCommitMessage."""

    _M1 = """jabberwocky: by Lewis Carroll

'Twas brillig, and the slithy toves
did gyre and gimble in the wabe.
"""

    _M2 = """jabberwocky: by Lewis Carroll

All mimsy were the borogroves,
And the mome wraths outgrabe.
jabberwocky: Charles Lutwidge Dodgson
"""

    _M3 = """jabberwocky: by Lewis Carroll

He took his vorpal sword in hand:
Long time the manxome foe he sought
jabberwocky:
"""

    _M4 = """the poem continues...

jabberwocky: O frabjuous day!
jabberwocky: Calloh! Callay!
"""

    def testNoMessage(self):
        o = cros_patch.GetOptionLinesFromCommitMessage("", "jabberwocky:")
        self.assertEqual(None, o)

    def testNoOption(self):
        o = cros_patch.GetOptionLinesFromCommitMessage(self._M1, "jabberwocky:")
        self.assertEqual(None, o)

    def testYesOption(self):
        o = cros_patch.GetOptionLinesFromCommitMessage(self._M2, "jabberwocky:")
        self.assertEqual(["Charles Lutwidge Dodgson"], o)

    def testEmptyOption(self):
        o = cros_patch.GetOptionLinesFromCommitMessage(self._M3, "jabberwocky:")
        self.assertEqual([], o)

    def testMultiOption(self):
        o = cros_patch.GetOptionLinesFromCommitMessage(self._M4, "jabberwocky:")
        self.assertEqual(["O frabjuous day!", "Calloh! Callay!"], o)


class TestApplyAgainstManifest(
    GitRepoPatchTestCase, cros_test_lib.MockTestCase
):
    """Test applying a patch against a manifest"""

    MANIFEST_TEMPLATE = """\
<?xml version="1.0" encoding="UTF-8"?>
<manifest>
  <remote name="cros" />
  <default revision="refs/heads/main" remote="cros" />
  %(projects)s
</manifest>
"""

    def _CommonRepoSetup(self, *projects):
        basedir = self.tempdir
        repodir = os.path.join(basedir, ".repo")
        manifest_file = os.path.join(repodir, "manifest.xml")
        proj_pieces = []
        for project in projects:
            proj_pieces.append("<project")
            for key, val in project.items():
                if key == "path":
                    val = os.path.relpath(
                        os.path.realpath(val), os.path.realpath(self.tempdir)
                    )
                proj_pieces.append(' %s="%s"' % (key, val))
            proj_pieces.append(" />\n  ")
        proj_str = "".join(proj_pieces)
        content = self.MANIFEST_TEMPLATE % {"projects": proj_str}
        os.mkdir(repodir)
        osutils.WriteFile(manifest_file, content)
        return basedir

    def testApplyAgainstManifest(self):
        git1, git2, _ = self._CommonGitSetup()

        readme_text = "Stub README text."
        readme1 = self.CommitFile(git1, "README", readme_text)
        readme_text += " Even more stub README text."
        readme2 = self.CommitFile(git1, "README", readme_text)
        readme_text += " Even more README text."
        readme3 = self.CommitFile(git1, "README", readme_text)

        git1_proj = {
            "path": git1,
            "name": "chromiumos/chromite",
            "revision": str(readme1.sha1),
            "upstream": "refs/heads/main",
        }
        git2_proj = {
            "path": git2,
            "name": "git2",
        }
        basedir = self._CommonRepoSetup(git1_proj, git2_proj)

        self.PatchObject(
            git.ManifestCheckout, "_GetManifestsBranch", return_value=None
        )
        manifest = git.ManifestCheckout(basedir)

        readme2.ApplyAgainstManifest(manifest)
        readme3.ApplyAgainstManifest(manifest)

        # Verify that both readme2 and readme3 are on the patch branch.
        cmd = [
            "git",
            "log",
            "--format=%T",
            "%s..%s" % (readme1.sha1, constants.PATCH_BRANCH),
        ]
        trees = self._run(cmd, git1).splitlines()
        self.assertEqual(
            trees, [str(readme3.tree_hash), str(readme2.tree_hash)]
        )


class TestLocalPatchGit(GitRepoPatchTestCase):
    """Test Local patch handling."""

    patch_kls = cros_patch.LocalPatch

    def setUp(self):
        self.sourceroot = os.path.join(self.tempdir, "sourceroot")

    def _MkPatch(self, source, sha1, ref="refs/heads/main", **kwargs):
        remote = kwargs.pop(
            "remote", config_lib.GetSiteParams().EXTERNAL_REMOTE
        )
        return self.patch_kls(
            source,
            "chromiumos/chromite",
            ref,
            "%s/main" % remote,
            remote,
            sha1,
            **kwargs,
        )

    def testUpload(self):
        def ProjectDirMock(_sourceroot):
            return git1

        git1, git2, patch = self._CommonGitSetup()

        git2_sha1 = self._GetSha1(git2, "HEAD")

        patch.ProjectDir = ProjectDirMock
        # First suppress carbon copy behaviour so we verify pushing plain works.
        sha1 = patch.sha1
        patch._GetCarbonCopy = lambda: sha1  # pylint: disable=protected-access
        patch.Upload(git2, "refs/testing/test1")
        self.assertEqual(self._GetSha1(git2, "refs/testing/test1"), patch.sha1)

        # Enable CarbonCopy behaviour; verify it lands a different
        # sha1.  Additionally verify it didn't corrupt the patch's sha1 locally.
        del patch._GetCarbonCopy
        patch.Upload(git2, "refs/testing/test2")
        self.assertNotEqual(
            self._GetSha1(git2, "refs/testing/test2"), patch.sha1
        )
        self.assertEqual(patch.sha1, sha1)
        # Ensure the carbon creation didn't damage the target repo.
        self.assertEqual(self._GetSha1(git1, "HEAD"), sha1)

        # Ensure we didn't damage the target repo's state at all.
        self.assertEqual(git2_sha1, self._GetSha1(git2, "HEAD"))
        # Ensure the content is the same.
        base = ["git", "show"]
        self.assertEqual(
            self._run(base + ["refs/testing/test1:monkeys"], git2),
            self._run(base + ["refs/testing/test2:monkeys"], git2),
        )
        base = ["git", "log", "--format=%B", "-n1"]
        self.assertEqual(
            self._run(base + ["refs/testing/test1"], git2),
            self._run(base + ["refs/testing/test2"], git2),
        )


class UploadedLocalPatchTestCase(GitRepoPatchTestCase):
    """Test uploading of local git patches."""

    PROJECT = "chromiumos/chromite"
    ORIGINAL_BRANCH = "original_branch"
    ORIGINAL_SHA1 = "ffffffff".ljust(40, "0")

    patch_kls = cros_patch.UploadedLocalPatch

    def _MkPatch(self, source, sha1, ref="refs/heads/main", **kwargs):
        site_params = config_lib.GetSiteParams()
        return self.patch_kls(
            source,
            self.PROJECT,
            ref,
            "%s/main" % site_params.EXTERNAL_REMOTE,
            self.ORIGINAL_BRANCH,
            kwargs.pop("original_sha1", self.ORIGINAL_SHA1),
            kwargs.pop("remote", site_params.EXTERNAL_REMOTE),
            carbon_copy_sha1=sha1,
            **kwargs,
        )


class TestUploadedLocalPatch(UploadedLocalPatchTestCase):
    """Test uploading of local git patches."""

    def testStringRepresentation(self):
        _, _, patch = self._CommonGitSetup()
        str_rep = str(patch).split(":")
        for element in [
            self.PROJECT,
            self.ORIGINAL_BRANCH,
            self.ORIGINAL_SHA1[:8],
        ]:
            self.assertTrue(
                element in str_rep,
                msg="Couldn't find %s in %s" % (element, str_rep),
            )


class TestGerritPatch(TestGitRepoPatch):
    """Test Gerrit patch handling."""

    has_native_change_id = True

    class patch_kls(cros_patch.GerritPatch):
        """Test helper class to suppress pointing to actual gerrit."""

        # Suppress the behaviour pointing the project url at actual gerrit,
        # instead slaving it back to a local repo for tests.
        def __init__(self, *args, **kwargs):
            cros_patch.GerritPatch.__init__(self, *args, **kwargs)
            assert hasattr(self, "patch_dict")
            self.project_url = self.patch_dict["_unittest_url_bypass"]

    @property
    def test_json(self):
        return copy.deepcopy(FAKE_PATCH_JSON)

    def _MkPatch(self, source, sha1, ref="refs/heads/main", **kwargs):
        site_params = config_lib.GetSiteParams()
        json = self.test_json
        remote = kwargs.pop("remote", site_params.EXTERNAL_REMOTE)
        url_prefix = kwargs.pop("url_prefix", site_params.EXTERNAL_GERRIT_URL)
        suppress_branch = kwargs.pop("suppress_branch", False)
        change_id = kwargs.pop("ChangeId", None)
        if change_id is None:
            change_id = self.MakeChangeId()
        json.update(kwargs)
        change_num, patch_num = _GetNumber(), _GetNumber()
        # Note we intentionally use a gerrit like refspec here; we want to
        # ensure that none of our common code pathways puke on a non head/tag.
        refspec = gerrit.GetChangeRef(change_num + 1000, patch_num)
        json["currentPatchSet"].update(
            dict(number=patch_num, ref=refspec, revision=sha1)
        )
        json["branch"] = os.path.basename(ref)
        json["_unittest_url_bypass"] = source
        json["id"] = change_id

        obj = self.patch_kls(json.copy(), remote, url_prefix)
        self.assertEqual(obj.patch_dict, json)
        self.assertEqual(obj.remote, remote)
        self.assertEqual(obj.url_prefix, url_prefix)
        self.assertEqual(obj.project, json["project"])
        self.assertEqual(obj.ref, refspec)
        self.assertEqual(obj.change_id, change_id)
        self.assertEqual(
            obj.id,
            "%s%s~%s~%s"
            % (
                site_params.CHANGE_PREFIX[remote],
                json["project"],
                json["branch"],
                change_id,
            ),
        )

        # Now make the fetching actually work, if desired.
        if not suppress_branch:
            # Note that a push is needed here, rather than a branch; branch
            # will just make it under refs/heads, we want it literally in
            # refs/changes/
            self._run(
                ["git", "push", source, "%s:%s" % (sha1, refspec)], source
            )
        return obj

    def testApprovalTimestamp(self):
        """Test that the approval timestamp is correctly extracted from JSON."""
        repo = self._MakeRepo("git", self.source)
        for approvals, expected in [
            (None, 0),
            ([], 0),
            ([1], 1),
            ([1, 3, 2], 3),
        ]:
            currentPatchSet = copy.deepcopy(FAKE_PATCH_JSON["currentPatchSet"])
            if approvals is not None:
                currentPatchSet["approvals"] = [
                    {"grantedOn": x} for x in approvals
                ]
            patch = self._MkPatch(
                repo,
                self._GetSha1(repo, self.DEFAULT_TRACKING),
                currentPatchSet=currentPatchSet,
            )
            msg = "Expected %r, but got %r (approvals=%r)" % (
                expected,
                patch.approval_timestamp,
                approvals,
            )
            self.assertEqual(patch.approval_timestamp, expected, msg)

    def _assertGerritDependencies(self, remote=None):
        if remote is None:
            remote = config_lib.GetSiteParams().EXTERNAL_REMOTE

        convert = str
        if remote == config_lib.GetSiteParams().INTERNAL_REMOTE:
            convert = lambda val: "chrome-internal:%s" % (val,)
        elif remote == config_lib.GetSiteParams().EXTERNAL_REMOTE:
            convert = lambda val: "chromium:%s" % (val,)
        git1 = self._MakeRepo("git1", self.source, remote=remote)
        patch = self._MkPatch(git1, self._GetSha1(git1, "HEAD"), remote=remote)
        cid1, cid2 = "1", "2"

        # Test cases with no dependencies, 1 dependency, and 2 dependencies.
        self.assertEqual(patch.GerritDependencies(), [])
        patch.patch_dict["dependsOn"] = [{"number": cid1}]
        self.assertEqual(
            [
                cros_patch.AddPrefix(x, x.gerrit_number)
                for x in patch.GerritDependencies()
            ],
            [convert(cid1)],
        )
        patch.patch_dict["dependsOn"].append({"number": cid2})
        self.assertEqual(
            [
                cros_patch.AddPrefix(x, x.gerrit_number)
                for x in patch.GerritDependencies()
            ],
            [convert(cid1), convert(cid2)],
        )

    def testExternalGerritDependencies(self):
        self._assertGerritDependencies()

    def testInternalGerritDependencies(self):
        self._assertGerritDependencies(
            config_lib.GetSiteParams().INTERNAL_REMOTE
        )

    def testReviewedOnMetadata(self):
        """Verify Change-Id and Reviewed-On are set in git metadata."""
        git1, _, patch = self._CommonGitSetup()
        patch.Apply(git1, self.DEFAULT_TRACKING)
        reviewed_on = "/".join(
            [
                config_lib.GetSiteParams().EXTERNAL_GERRIT_URL,
                patch.gerrit_number,
            ]
        )
        self.assertIn("Reviewed-on: %s\n" % reviewed_on, patch.commit_message)

    def _MakeFooters(self):
        return (
            (),
            (("Footer-1", "foo"),),
            (("Change-id", "42"),),
            (("Footer-1", "foo"), ("Change-id", "42")),
        )

    def _MakeCommitMessages(self):
        headers = (
            "A standard commit message header",
            "",
            "Footer-1: foo",
            "Change-id: 42",
        )

        bodies = ("", "\n", "Lots of comments\n about the commit\n" * 100)

        for header, body, preexisting in itertools.product(
            headers, bodies, self._MakeFooters()
        ):
            yield "\n".join(
                (
                    header,
                    body,
                    "\n".join("%s: %s" for tag, ident in preexisting),
                )
            )

    def testAddFooters(self):
        repo = self._MakeRepo("git", self.source)
        patch = self._MkPatch(repo, self._GetSha1(repo, "HEAD"))
        approval = {"type": "VRIF", "value": "1", "grantedOn": 1391733002}

        for msg in self._MakeCommitMessages():
            for footers in self._MakeFooters():
                with mock.patch(
                    "chromite.lib.patch.FooterForApproval",
                    new=mock.Mock(side_effect=itertools.cycle(footers)),
                ), mock.patch.object(
                    patch, "_approvals", new=[approval] * len(footers)
                ):
                    patch._commit_message = msg

                    # Idempotence
                    self.assertEqual(
                        patch._AddFooters(msg),
                        patch._AddFooters(patch._AddFooters(msg)),
                    )

                    # there may be pre-existing footers.  This asserts that we
                    # can Get all the footers after we Set them.
                    self.assertFalse(
                        bool(
                            set(footers)
                            - set(patch._GetFooters(patch._AddFooters(msg)))
                        )
                    )

                    if set(footers) - set(patch._GetFooters(msg)):
                        self.assertNotEqual(msg, patch._AddFooters(msg))

    def testConvertQueryResults(self):
        """Verify basic ConvertQueryResults behavior."""
        j = FAKE_CHANGE_JSON
        exp = {
            "project": j["project"],
            "url": "https://host/#/c/8366/",
            "status": j["status"],
            "branch": j["branch"],
            "owner": {
                "username": "Duckworth Duck",
                "name": "Duckworth Duck",
                "email": "happy-funky-duck@chromium.org",
            },
            "createdOn": 1473894166,
            "commitMessage": (
                "my super great message\n\nChange-Id: "
                "I3a753d6bacfbe76e5675a6f2f7941fe520c095e5\n"
            ),
            "currentPatchSet": {
                "approvals": [],
                "date": 1473894123,
                "draft": False,
                "number": "1",
                "ref": "refs/changes/15/8366/1",
                "revision": j["current_revision"],
            },
            "dependsOn": [
                {"revision": "3d54362a9b010330bae2dde973fc5c3efc4e5f44"}
            ],
            "subject": j["subject"],
            "id": j["change_id"],
            "lastUpdated": 1473894166,
            "number": str(j["_number"]),
            "private": False,
            "topic": "some-topic",
        }
        ret = cros_patch.GerritPatch.ConvertQueryResults(j, "host")
        self.assertEqual(ret, exp)

    def testConvertQueryResultsProtoNoHttp(self):
        """Verify ConvertQueryResults handling of non-http protos."""
        j = copy.deepcopy(FAKE_CHANGE_JSON)
        fetch = j["revisions"][j["current_revision"]]["fetch"]
        fetch.pop("http", None)
        fetch.pop("https", None)
        # Mostly just verifying it still parses.
        ret = cros_patch.GerritPatch.ConvertQueryResults(j, "host")
        self.assertNotEqual(ret, None)


class PrepareRemotePatchesTest(cros_test_lib.TestCase):
    """Test preparing remote patches."""

    def MkRemote(
        self,
        project="my/project",
        original_branch="my-local",
        ref="refs/tryjobs/elmer/patches",
        tracking_branch="main",
        internal=False,
    ):
        l = [
            project,
            original_branch,
            ref,
            tracking_branch,
            getattr(
                constants,
                ("%s_PATCH_TAG" % ("INTERNAL" if internal else "EXTERNAL")),
            ),
        ]
        return ":".join(l)

    def assertRemote(
        self,
        patch,
        project="my/project",
        original_branch="my-local",
        ref="refs/tryjobs/elmer/patches",
        tracking_branch="main",
        internal=False,
    ):
        self.assertEqual(patch.project, project)
        self.assertEqual(patch.original_branch, original_branch)
        self.assertEqual(patch.ref, ref)
        self.assertEqual(patch.tracking_branch, tracking_branch)
        self.assertEqual(patch.internal, internal)

    def test(self):
        # Check handling of a single patch...
        patches = cros_patch.PrepareRemotePatches([self.MkRemote()])
        self.assertEqual(len(patches), 1)
        self.assertRemote(patches[0])

        # Check handling of a multiple...
        patches = cros_patch.PrepareRemotePatches(
            [self.MkRemote(), self.MkRemote(project="foon")]
        )
        self.assertEqual(len(patches), 2)
        self.assertRemote(patches[0])
        self.assertRemote(patches[1], project="foon")

        # Ensure basic validation occurs:
        chunks = self.MkRemote().split(":")
        self.assertRaises(
            ValueError, cros_patch.PrepareRemotePatches, ":".join(chunks[:-1])
        )
        self.assertRaises(
            ValueError,
            cros_patch.PrepareRemotePatches,
            ":".join(chunks[:-1] + ["monkeys"]),
        )
        self.assertRaises(
            ValueError,
            cros_patch.PrepareRemotePatches,
            ":".join(chunks + [":"]),
        )


class PrepareLocalPatchesTests(cros_test_lib.RunCommandTestCase):
    """Test preparing local patches."""

    def setUp(self):
        self.path, self.project, self.branch = "mydir", "my/project", "mybranch"
        self.tracking_branch = "kernel"
        self.patches = ["%s:%s" % (self.project, self.branch)]
        self.manifest = mock.MagicMock()
        attrs = dict(
            tracking_branch=self.tracking_branch,
            local_path=self.path,
            remote="cros",
        )
        checkout = git.ProjectCheckout(attrs)
        self.PatchObject(
            self.manifest, "FindCheckouts", return_value=[checkout]
        )

    def PrepareLocalPatches(self, output):
        """Check the returned GitRepoPatchInfo against golden values."""
        output_obj = mock.MagicMock()
        output_obj.stdout = output
        self.PatchObject(
            cros_patch.LocalPatch, "Fetch", return_value=output_obj
        )
        self.PatchObject(git, "RunGit", return_value=output_obj)
        patch_info = cros_patch.PrepareLocalPatches(
            self.manifest, self.patches
        )[0]
        self.assertEqual(patch_info.project, self.project)
        self.assertEqual(patch_info.ref, self.branch)
        self.assertEqual(patch_info.tracking_branch, self.tracking_branch)

    def testBranchSpecifiedSuccessRun(self):
        """Test success with branch specified by user."""
        self.PrepareLocalPatches("12345".rjust(40, "0"))

    def testBranchSpecifiedNoChanges(self):
        """Test when no changes on the branch specified by user."""
        self.assertRaises(SystemExit, self.PrepareLocalPatches, "")


class TestFormatting(cros_test_lib.TestCase):
    """Test formatting of output."""

    VALID_CHANGE_ID = "I47ea30385af60ae4cc2acc5d1a283a46423bc6e1"

    def _assertResult(
        self, functor, value, expected=None, raises=False, **kwargs
    ):
        if raises:
            self.assertRaises2(
                ValueError,
                functor,
                value,
                msg="%s(%r) did not throw a ValueError"
                % (functor.__name__, value),
                **kwargs,
            )
        else:
            self.assertEqual(
                functor(value, **kwargs),
                expected,
                msg="failed: %s(%r) != %r"
                % (functor.__name__, value, expected),
            )

    def _assertBad(self, functor, values, **kwargs):
        for value in values:
            self._assertResult(functor, value, raises=True, **kwargs)

    def _assertGood(self, functor, values, **kwargs):
        for value, expected in values:
            self._assertResult(functor, value, expected, **kwargs)

    def testGerritNumber(self):
        """Tests that we can pasre a Gerrit number."""
        self._assertGood(
            cros_patch.ParseGerritNumber,
            [("12345",) * 2, ("12",) * 2, ("123",) * 2],
        )

        self._assertBad(
            cros_patch.ParseGerritNumber,
            ["is", "i1325", "01234567", "012345a", "**12345", "+123", "/0123"],
            error_ok=False,
        )

    def testChangeID(self):
        """Tests that we can parse a change-ID."""
        self._assertGood(
            cros_patch.ParseChangeID, [(self.VALID_CHANGE_ID,) * 2]
        )

        # Change-IDs too short/long, with unexpected characters in it.
        self._assertBad(
            cros_patch.ParseChangeID,
            [
                "is",
                "**i1325",
                "i134".ljust(41, "0"),
                "I1234+".ljust(41, "0"),
                "I123".ljust(42, "0"),
            ],
            error_ok=False,
        )

    def testSHA1(self):
        """Tests that we can parse a SHA1 hash."""
        self._assertGood(
            cros_patch.ParseSHA1,
            [("1" * 40,) * 2, ("a" * 40,) * 2, ("1a7e034".ljust(40, "0"),) * 2],
        )

        self._assertBad(
            cros_patch.ParseSHA1,
            ["0abcg", "Z", "**a", "+123", "1234ab" * 10],
            error_ok=False,
        )

    def testFullChangeID(self):
        """Tests that we can parse a full change-ID."""
        change_id = self.VALID_CHANGE_ID
        self._assertGood(
            cros_patch.ParseFullChangeID,
            (
                (
                    "foo~bar~%s" % change_id,
                    cros_patch.FullChangeId("foo", "bar", change_id),
                ),
                (
                    "foo/bar/baz~refs/heads/_my-branch_~%s" % change_id,
                    cros_patch.FullChangeId(
                        "foo/bar/baz", "refs/heads/_my-branch_", change_id
                    ),
                ),
            ),
        )

    def testInvalidFullChangeID(self):
        """Should throw an error on bad inputs."""
        change_id = self.VALID_CHANGE_ID
        self._assertBad(
            cros_patch.ParseFullChangeID,
            ["foo", "foo~bar", "foo~bar~baz", "foo~refs/bar~%s" % change_id],
            error_ok=False,
        )

    def testParsePatchDeps(self):
        """Tests that we can parse the dependency specified by the user."""
        change_id = self.VALID_CHANGE_ID
        vals = [
            "CL:12345",
            "project~branch~%s" % change_id,
            change_id,
            change_id[1:],
        ]
        for val in vals:
            self.assertTrue(cros_patch.ParsePatchDep(val) is not None)

        self._assertBad(
            cros_patch.ParsePatchDep,
            ["145462399", "I47ea3", "i47ea3".ljust(41, "0")],
        )


class MockPatchFactory:
    """Helper class to create patches or series of them, for unit tests."""

    def __init__(self, patch_mock=None):
        """Constructor for factory.

        patch_mock: Optional PatchMock instance.
        """
        self.patch_mock = patch_mock
        self._patch_counter = functools.partial(next, itertools.count(1))

    def MockPatch(
        self,
        change_id=None,
        patch_number=None,
        is_merged=False,
        project="chromiumos/chromite",
        remote=config_lib.GetSiteParams().EXTERNAL_REMOTE,
        tracking_branch="refs/heads/main",
        is_draft=False,
        approvals=(),
        commit_message=None,
    ):
        """Helper function to create mock GerritPatch objects."""
        if change_id is None:
            # TODO(b/236161656): Fix.
            # pylint: disable-next=assignment-from-no-return
            change_id = self._patch_counter()
        subject = f"PatchSet: {change_id}"
        gerrit_number = str(change_id)
        change_id = hex(change_id)[2:].rstrip("L").lower()
        change_id = "I%s" % change_id.rjust(40, "0")
        sha1 = hex(_GetNumber())[2:].rstrip("L").lower().rjust(40, "0")
        if patch_number is None:
            # TODO(b/236161656): Fix.
            # pylint: disable-next=assignment-from-no-return
            patch_number = _GetNumber()
        fake_url = "http://foo/bar"
        if not approvals:
            approvals = [
                {"type": "VRIF", "value": "1", "grantedOn": 1391733002},
                {"type": "CRVW", "value": "2", "grantedOn": 1391733002},
                {"type": "COMR", "value": "2", "grantedOn": 1391733002},
            ]

        current_patch_set = {
            "number": patch_number,
            "revision": sha1,
            "draft": is_draft,
            "approvals": approvals,
        }
        patch_dict = {
            "currentPatchSet": current_patch_set,
            "id": change_id,
            "number": gerrit_number,
            "project": project,
            "branch": tracking_branch,
            "subject": subject,
            "owner": {"email": "elmer.fudd@chromium.org"},
            "remote": remote,
            "status": "MERGED" if is_merged else "NEW",
            "url": "%s/%s" % (fake_url, change_id),
        }

        patch = cros_patch.GerritPatch(patch_dict, remote, fake_url)
        patch.pass_count = 0
        patch.fail_count = 1
        patch.total_fail_count = 3
        patch.commit_message = commit_message

        return patch

    def GetPatches(self, how_many=1, always_use_list=False, **kwargs):
        """Get a sequential list of patches.

        Args:
            how_many: How many patches to return.
            always_use_list: Whether to use a list for a single item list.
            **kwargs: Keyword arguments for self.MockPatch.
        """
        patches = [self.MockPatch(**kwargs) for _ in range(how_many)]
        if self.patch_mock:
            for i, patch in enumerate(patches):
                self.patch_mock.SetGerritDependencies(patch, patches[: i + 1])
        if how_many == 1 and not always_use_list:
            return patches[0]
        return patches


class DependencyErrorTests(cros_test_lib.MockTestCase):
    """Tests for DependencyError."""

    def testGetRootError(self):
        """Test GetRootError on nested DependencyError."""
        p_1, p_2, p_3 = MockPatchFactory().GetPatches(how_many=3)
        ex_1 = cros_patch.ApplyPatchException(p_1)
        ex_2 = cros_patch.DependencyError(p_2, ex_1)
        ex_3 = cros_patch.DependencyError(p_3, ex_2)

        self.assertEqual(ex_3.GetRootError(), ex_1)

    def testGetRootErrorOnCircurlarError(self):
        """Test GetRootError on circular."""
        p_1, p_2, p_3 = MockPatchFactory().GetPatches(how_many=3)
        ex_1 = cros_patch.DependencyError(
            p_2, cros_patch.ApplyPatchException(p_1)
        )
        ex_2 = cros_patch.DependencyError(p_2, ex_1)
        ex_3 = cros_patch.DependencyError(p_3, ex_2)
        ex_1.error = ex_3

        self.assertIsNone(ex_3.GetRootError())