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

Most of the tests in this file reach out to a staging/test Gerrit server. By
default, this server is t3st-chr0m3-review.googlesource.com. These tests will
be skipped unless run_tests is passed the '--network' arg:
$ ./run_tests --network -- lib/gerrit_unittest.py
"""

import collections
import http.client
import http.cookiejar
import io
import json
import os
import re
import shutil
import stat
from unittest import mock
import urllib.parse

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 gob_util
from chromite.lib import osutils


class GerritTestCase(cros_test_lib.MockTempDirTestCase):
    """Test class for tests that interact with a Gerrit server.

    Configured by default to use a specially-configured test Gerrit server at
    t3st-chr0m3(-review).googlesource.com. The test server configuration may be
    altered by setting the following environment variables from the parent
    process:
        CROS_TEST_GIT_HOST: host name for git operations; defaults to
            t3st-chr0me.googlesource.com.
        CROS_TEST_GERRIT_HOST: host name for Gerrit operations; defaults to
            t3st-chr0me-review.googlesource.com.
        CROS_TEST_COOKIES_PATH: path to a cookies.txt file to use for git/Gerrit
            requests; defaults to ~/.gitcookies.
        CROS_TEST_COOKIE_NAMES: comma-separated list of cookie names from
            CROS_TEST_COOKIES_PATH to set on requests; defaults to none. The
            current implementation only sends cookies matching the exact host
            name and the empty path ("/").
    """

    # pylint: disable=protected-access

    TEST_USERNAME = "test-username"
    TEST_EMAIL = "test-username@test.org"

    GerritInstance = collections.namedtuple(
        "GerritInstance",
        [
            "cookie_names",
            "cookies_path",
            "gerrit_host",
            "gerrit_url",
            "git_host",
            "git_url",
            "project_prefix",
        ],
    )

    def _create_gerrit_instance(self, tmp_dir):
        default_host = "t3st-chr0m3"
        git_host = os.environ.get(
            "CROS_TEST_GIT_HOST", constants.GOB_HOST % default_host
        )
        gerrit_host = os.environ.get(
            "CROS_TEST_GERRIT_HOST", "%s-review.googlesource.com" % default_host
        )
        project_prefix = "test-%s/" % (cros_build_lib.GetRandomString(),)
        cookies_path = os.environ.get(
            "CROS_TEST_COOKIES_PATH", str(constants.GITCOOKIES_PATH)
        )
        # "o" is the cookie name that GoB uses in its instructions.
        cookie_names_str = os.environ.get("CROS_TEST_COOKIE_NAMES", "o")
        cookie_names = {c for c in cookie_names_str.split(",") if c}

        tmpcookies_path = os.path.join(tmp_dir, ".gitcookies")
        if os.path.exists(cookies_path):
            shutil.copy(cookies_path, tmpcookies_path)
        else:
            osutils.Touch(tmpcookies_path)

        return self.GerritInstance(
            cookie_names=cookie_names,
            cookies_path=tmpcookies_path,
            gerrit_host=gerrit_host,
            gerrit_url="https://%s/" % gerrit_host,
            git_host=git_host,
            git_url="https://%s/" % git_host,
            project_prefix=project_prefix,
        )

    def setUp(self) -> None:
        """Sets up the gerrit instances in a class-specific temp dir."""
        self.saved_params = {}
        os.environ["HOME"] = self.tempdir

        # Create gerrit instance.
        gi = self.gerrit_instance = self._create_gerrit_instance(self.tempdir)

        # This --global will use our tempdir $HOME we set above, not the real
        # ~/.
        cros_build_lib.dbg_run(
            ["git", "config", "--global", "http.cookiefile", gi.cookies_path],
            capture_output=True,
        )

        # If you're seeing "does not look like a Netscape format cookies file"
        # errors here, make sure the first line in your gitcookies file is:
        # "# HTTP Cookie File"
        # TODO(b/210490942): Detect and handle this automatically.
        jar = http.cookiejar.MozillaCookieJar(gi.cookies_path)
        jar.load(ignore_expires=True)

        def GetCookies(host, _path):
            return dict(
                (c.name, urllib.parse.unquote(c.value))
                for c in jar
                if c.domain == host
                and c.path == "/"
                and c.name in gi.cookie_names
            )

        self.PatchObject(gob_util, "GetCookies", GetCookies)

        site_params = config_lib.GetSiteParams()

        # Make all chromite code point to the test server.
        self.patched_params = {
            "EXTERNAL_GOB_HOST": gi.git_host,
            "EXTERNAL_GERRIT_HOST": gi.gerrit_host,
            "EXTERNAL_GOB_URL": gi.git_url,
            "EXTERNAL_GERRIT_URL": gi.gerrit_url,
            "INTERNAL_GOB_HOST": gi.git_host,
            "INTERNAL_GERRIT_HOST": gi.gerrit_host,
            "INTERNAL_GOB_URL": gi.git_url,
            "INTERNAL_GERRIT_URL": gi.gerrit_url,
            "AOSP_GOB_HOST": gi.git_host,
            "AOSP_GERRIT_HOST": gi.gerrit_host,
            "AOSP_GOB_URL": gi.git_url,
            "AOSP_GERRIT_URL": gi.gerrit_url,
            "MANIFEST_URL": "%s/%s"
            % (
                gi.git_url,
                site_params.MANIFEST_PROJECT,
            ),
            "MANIFEST_INT_URL": "%s/%s"
            % (
                gi.git_url,
                site_params.MANIFEST_INT_PROJECT,
            ),
            "GIT_REMOTES": {
                site_params.EXTERNAL_REMOTE: gi.gerrit_url,
                site_params.INTERNAL_REMOTE: gi.gerrit_url,
                site_params.CHROMIUM_REMOTE: gi.gerrit_url,
                site_params.CHROME_REMOTE: gi.gerrit_url,
            },
        }

        for k in self.patched_params.keys():
            self.saved_params[k] = site_params.get(k)

        site_params.update(self.patched_params)

    def tearDown(self) -> None:
        # Restore the 'patched' site parameters.
        site_params = config_lib.GetSiteParams()
        site_params.update(self.saved_params)

    def createProject(
        self,
        suffix,
        description="Test project",
        owners=None,
        submit_type="CHERRY_PICK",
    ):
        """Create a project on the test gerrit server."""
        name = self.gerrit_instance.project_prefix + suffix
        body = {
            "description": description,
            "submit_type": submit_type,
            "branches": ["main"],
        }
        if owners is not None:
            body["owners"] = owners
        path = "projects/%s" % urllib.parse.quote(name, "")
        response = gob_util.CreateHttpConn(
            self.gerrit_instance.gerrit_host, path, reqtype="PUT", body=body
        )
        self.assertEqual(
            201, response.status, "Expected 201, got %s" % response.status
        )
        s = io.BytesIO(response.read())
        self.assertEqual(b")]}'", s.readline().rstrip())
        jmsg = json.load(s)
        self.assertEqual(name, jmsg["name"])
        return name

    def _CloneProject(self, name, path):
        """Clone a project from the test gerrit server."""
        root = os.path.dirname(path)
        osutils.SafeMakedirs(root)
        url = "%s://%s/%s" % (
            gob_util.GIT_PROTOCOL,
            self.gerrit_instance.git_host,
            name,
        )
        git.RunGit(root, ["clone", url, path])
        # Install commit-msg hook.
        hook_path = os.path.join(path, ".git", "hooks", "commit-msg")
        hook_cmd = [
            "curl",
            "-n",
            "-o",
            hook_path,
            "-b",
            self.gerrit_instance.cookies_path,
        ]
        hook_cmd.append(
            "https://%s/a/tools/hooks/commit-msg"
            % self.gerrit_instance.gerrit_host
        )
        cros_build_lib.dbg_run(hook_cmd, capture_output=True)
        os.chmod(hook_path, stat.S_IRWXU)
        # Set git identity to test account
        cros_build_lib.dbg_run(
            ["git", "config", "user.email", self.TEST_EMAIL],
            cwd=path,
            capture_output=True,
        )
        return path

    def cloneProject(self, name, path=None):
        """Clone a project from the test gerrit server."""
        if path is None:
            path = os.path.basename(name)
            if path.endswith(".git"):
                path = path[:-4]
        path = os.path.join(self.tempdir, path)
        return self._CloneProject(name, path)

    @classmethod
    def _CreateCommit(
        cls, clone_path, filename=None, msg=None, text=None, amend=False
    ):
        """Create a commit in the given git checkout.

        Args:
            clone_path: The directory on disk of the git clone.
            filename: The name of the file to write. Optional.
            msg: The commit message. Optional.
            text: The text to append to the file. Optional.
            amend: Whether to amend an existing patch. If set, we will amend the
                HEAD commit in the checkout and upload that patch.

        Returns:
            (sha1, changeid) of the new commit.
        """
        if not filename:
            filename = "test-file.txt"
        if not msg:
            msg = "Test Message"
        if not text:
            text = "Another day, another dollar."
        fpath = os.path.join(clone_path, filename)
        osutils.WriteFile(fpath, "%s\n" % text, mode="a")
        cros_build_lib.dbg_run(
            ["git", "add", filename], cwd=clone_path, capture_output=True
        )
        cmd = ["git", "commit"]
        cmd += ["--amend", "-C", "HEAD"] if amend else ["-m", msg]
        cros_build_lib.dbg_run(cmd, cwd=clone_path, capture_output=True)
        return cls._GetCommit(clone_path)

    def createCommit(
        self, clone_path, filename=None, msg=None, text=None, amend=False
    ):
        """Create a commit in the given git checkout.

        Args:
            clone_path: The directory on disk of the git clone.
            filename: The name of the file to write. Optional.
            msg: The commit message. Optional.
            text: The text to append to the file. Optional.
            amend: Whether to amend an existing patch. If set, we will amend the
                HEAD commit in the checkout and upload that patch.
        """
        clone_path = os.path.join(self.tempdir, clone_path)
        return self._CreateCommit(clone_path, filename, msg, text, amend)

    @staticmethod
    def _GetCommit(clone_path, ref="HEAD"):
        log_proc = cros_build_lib.run(
            ["git", "log", "-n", "1", ref],
            cwd=clone_path,
            print_cmd=False,
            capture_output=True,
            encoding="utf-8",
        )
        sha1 = None
        change_id = None
        for line in log_proc.stdout.splitlines():
            match = re.match(r"^commit ([0-9a-fA-F]{40})$", line)
            if match:
                sha1 = match.group(1)
                continue
            match = re.match(r"^\s+Change-Id:\s*(\S+)$", line)
            if match:
                change_id = match.group(1)
                continue
        return (sha1, change_id)

    def getCommit(self, clone_path, ref="HEAD"):
        """Get the sha1 and change-id for the head commit in a git checkout."""
        clone_path = os.path.join(self.tempdir, clone_path)
        (sha1, change_id) = self._GetCommit(clone_path, ref)
        self.assertTrue(sha1)
        self.assertTrue(change_id)
        return (sha1, change_id)

    @staticmethod
    def _UploadChange(clone_path, branch="main", remote="origin") -> None:
        cros_build_lib.dbg_run(
            ["git", "push", remote, "HEAD:refs/for/%s" % branch],
            cwd=clone_path,
            capture_output=True,
        )

    def uploadChange(self, clone_path, branch="main", remote="origin") -> None:
        """Create a gerrit CL from the HEAD of a git checkout."""
        clone_path = os.path.join(self.tempdir, clone_path)
        self._UploadChange(clone_path, branch, remote)

    @staticmethod
    def _PushBranch(clone_path, branch="main") -> None:
        cros_build_lib.dbg_run(
            ["git", "push", "origin", "HEAD:refs/heads/%s" % branch],
            cwd=clone_path,
            capture_output=True,
        )

    def pushBranch(self, clone_path, branch="main") -> None:
        """Push a branch directly to gerrit, bypassing code review."""
        clone_path = os.path.join(self.tempdir, clone_path)
        self._PushBranch(clone_path, branch)

    def createAccount(
        self,
        name="Test User",
        email="test-user@test.org",
        password=None,
        groups=None,
    ) -> None:
        """Create a new user account on gerrit."""
        username = urllib.parse.quote(email.partition("@")[0])
        path = "accounts/%s" % username
        body = {
            "name": name,
            "email": email,
        }

        if password:
            body["http_password"] = password
        if groups:
            if isinstance(groups, str):
                groups = [groups]
            body["groups"] = groups
        response = gob_util.CreateHttpConn(
            self.gerrit_instance.gerrit_host, path, reqtype="PUT", body=body
        )
        self.assertEqual(201, response.status)
        s = io.BytesIO(response.read())
        self.assertEqual(b")]}'", s.readline().rstrip())
        jmsg = json.load(s)
        self.assertEqual(email, jmsg["email"])


@cros_test_lib.pytestmark_network_test
class GerritHelperTest(GerritTestCase):
    """Unittests for GerritHelper."""

    def _GetHelper(self, remote=config_lib.GetSiteParams().EXTERNAL_REMOTE):
        return gerrit.GetGerritHelper(remote)

    def createPatch(self, clone_path, project, remote="origin", **kwargs):
        """Create a patch in the given git checkout and upload it to gerrit.

        Args:
            clone_path: The directory on disk of the git clone.
            project: The associated project.
            remote: The remote to upload changes to.
            **kwargs: Additional keyword arguments to pass to createCommit.

        Returns:
            A GerritPatch object.
        """
        (revision, changeid) = self.createCommit(clone_path, **kwargs)
        helper = self._GetHelper()
        gpatch = helper.CreateGerritPatch(
            clone_path, remote, "main", project=project
        )
        self.assertEqual(gpatch.change_id, changeid)
        self.assertEqual(gpatch.revision, revision)
        return gpatch

    def testSimpleQuery(self) -> None:
        """Create and query one independent and three dependent changes."""
        project = self.createProject("testProject")
        clone_path = self.cloneProject(project)
        (head_sha1, head_changeid) = self.createCommit(clone_path)
        for idx in range(3):
            cros_build_lib.dbg_run(
                ["git", "checkout", head_sha1],
                cwd=clone_path,
                capture_output=True,
            )
            self.createCommit(clone_path, filename="test-file-%d.txt" % idx)
            self.uploadChange(clone_path)
        helper = self._GetHelper()
        changes = helper.Query(owner="self", project=project)
        self.assertEqual(len(changes), 4)
        changes = helper.Query(head_changeid, project=project, branch="main")
        self.assertEqual(len(changes), 1)
        self.assertEqual(changes[0].change_id, head_changeid)
        self.assertEqual(changes[0].sha1, head_sha1)
        change = helper.QuerySingleRecord(
            head_changeid, project=project, branch="main"
        )
        self.assertTrue(change)
        self.assertEqual(change.change_id, head_changeid)
        self.assertEqual(change.sha1, head_sha1)
        change = helper.GrabPatchFromGerrit(project, head_changeid, head_sha1)
        self.assertTrue(change)
        self.assertEqual(change.change_id, head_changeid)
        self.assertEqual(change.sha1, head_sha1)

    @mock.patch.object(gerrit.GerritHelper, "_GERRIT_MAX_QUERY_RETURN", 2)
    def testGerritQueryTruncation(self) -> None:
        """Verify that we detect gerrit truncating our query, and handle it."""
        project = self.createProject("testProject")
        clone_path = self.cloneProject(project)
        # Using a shell loop is markedly faster than running a python loop.
        num_changes = 5
        cmd = (
            "for ((i=0; i<%i; i=i+1)); do "
            'echo "Another day, another dollar." > test-file-$i.txt; '
            "git add test-file-$i.txt; "
            'git commit -m "Test commit $i."; '
            "done" % num_changes
        )
        cros_build_lib.dbg_run(
            cmd, shell=True, cwd=clone_path, capture_output=True
        )
        self.uploadChange(clone_path)
        helper = self._GetHelper()
        changes = helper.Query(project=project)
        self.assertEqual(len(changes), num_changes)

    def testIsChangeCommitted(self) -> None:
        """Tests that we can parse a json to check if a change is committed."""
        project = self.createProject("testProject")
        clone_path = self.cloneProject(project)
        gpatch = self.createPatch(clone_path, project)
        helper = self._GetHelper()
        helper.SetReview(gpatch.gerrit_number, labels={"Code-Review": "+2"})
        helper.SubmitChange(gpatch)
        self.assertTrue(helper.IsChangeCommitted(gpatch.gerrit_number))

        gpatch = self.createPatch(clone_path, project)
        self.assertFalse(helper.IsChangeCommitted(gpatch.gerrit_number))

    def testGetLatestSHA1ForBranch(self) -> None:
        """Verify we can query the tip-of-tree commit in a git repository."""
        project = self.createProject("testProject")
        clone_path = self.cloneProject(project)
        for _ in range(5):
            (main_sha1, _) = self.createCommit(clone_path)
        self.pushBranch(clone_path, "main")
        for _ in range(5):
            (testbranch_sha1, _) = self.createCommit(clone_path)
        self.pushBranch(clone_path, "testbranch")
        helper = self._GetHelper()
        self.assertEqual(
            helper.GetLatestSHA1ForBranch(project, "main"), main_sha1
        )
        self.assertEqual(
            helper.GetLatestSHA1ForBranch(project, "testbranch"),
            testbranch_sha1,
        )

    def testChangeEdit(self) -> None:
        """Verify CreateChange & ChangeEdit can create CLs with changes."""
        project = self.createProject("testProject")
        # Gerrit returns "Destination branch does not exist" errors if we don't
        # push something onto the new project's branch.
        clone_path = self.cloneProject(project)
        self.createCommit(clone_path)
        self.pushBranch(clone_path, "main")
        helper = self._GetHelper()
        # There should be no changes in our project that touch some_file.
        file_path = "some_file"
        self.assertEqual(len(helper.Query(project=project, path=file_path)), 0)
        change = helper.CreateChange(project, "main", "Test Change", True)
        helper.ChangeEdit(change.gerrit_number, file_path, "some file contents")
        # After creating the change and adding a file modification, there should
        # be a single change that touches some_file in our project.
        self.assertEqual(len(helper.Query(project=project, path=file_path)), 1)

    def _ChooseReviewers(self):
        # TODO(b/210507794): register some test accounts on test server. This
        #   fixed list of real IDs has a few problems, not limited to the
        #   following:
        #       * Some functions behave differently if the account is inactive;
        #       * Gerrit doesn't let you add yourself as a reviewer. So these
        #           accounts can't run the tests correctly. ;)
        return ["dborowitz@google.com", "jrn@google.com"]

    def testSetAttentionSet(self) -> None:
        """Verify that we can set the attention set on a CL."""
        project = self.createProject("testProject")
        clone_path = self.cloneProject(project)
        gpatch = self.createPatch(clone_path, project)
        emails = self._ChooseReviewers()
        helper = self._GetHelper()
        helper.SetReviewers(gpatch.gerrit_number, add=(emails[0], emails[1]))
        helper.SetAttentionSet(gpatch.gerrit_number, add=(emails[0], emails[1]))
        attention = gob_util.GetAttentionSet(helper.host, gpatch.gerrit_number)
        self.assertEqual(len(attention), 2)
        self.assertCountEqual(
            [r["account"]["email"] for r in attention], [emails[0], emails[1]]
        )
        helper.SetAttentionSet(gpatch.gerrit_number, remove=(emails[0],))
        attention = gob_util.GetAttentionSet(helper.host, gpatch.gerrit_number)
        self.assertEqual(len(attention), 1)
        self.assertEqual(attention[0]["account"]["email"], emails[1])

    def testSetReviewers(self) -> None:
        """Verify that we can set reviewers on a CL."""
        project = self.createProject("testProject")
        clone_path = self.cloneProject(project)
        gpatch = self.createPatch(clone_path, project)
        emails = self._ChooseReviewers()
        helper = self._GetHelper()
        helper.SetReviewers(gpatch.gerrit_number, add=(emails[0], emails[1]))
        reviewers = gob_util.GetReviewers(helper.host, gpatch.gerrit_number)
        self.assertEqual(len(reviewers), 2)
        self.assertCountEqual(
            [r["email"] for r in reviewers], [emails[0], emails[1]]
        )
        helper.SetReviewers(gpatch.gerrit_number, remove=(emails[0],))
        reviewers = gob_util.GetReviewers(helper.host, gpatch.gerrit_number)
        self.assertEqual(len(reviewers), 1)
        self.assertEqual(reviewers[0]["email"], emails[1])

    def testPatchNotFound(self) -> None:
        """Test case where ChangeID isn't found on the server."""
        changeids = ["I" + ("deadbeef" * 5), "I" + ("beadface" * 5)]
        self.assertRaises(
            gerrit.GerritException, gerrit.GetGerritPatchInfo, changeids
        )
        self.assertRaises(
            gerrit.GerritException,
            gerrit.GetGerritPatchInfo,
            ["*" + cid for cid in changeids],
        )
        # Change ID sequence starts at 1000.
        gerrit_numbers = ["123", "543"]
        self.assertRaises(
            gerrit.GerritException, gerrit.GetGerritPatchInfo, gerrit_numbers
        )
        self.assertRaises(
            gerrit.GerritException,
            gerrit.GetGerritPatchInfo,
            ["*" + num for num in gerrit_numbers],
        )

    def testVagueQuery(self) -> None:
        """Verify GerritHelper complains if an ID matches multiple changes."""
        project = self.createProject("testProject")
        clone_path = self.cloneProject(project)
        (sha1, _) = self.createCommit(clone_path)
        (_, changeid) = self.createCommit(clone_path)
        self.uploadChange(clone_path, "main")
        cros_build_lib.dbg_run(
            ["git", "checkout", sha1], cwd=clone_path, capture_output=True
        )
        self.createCommit(clone_path)
        self.pushBranch(clone_path, "testbranch")
        self.createCommit(
            clone_path, msg="Test commit.\n\nChange-Id: %s" % changeid
        )
        self.uploadChange(clone_path, "testbranch")
        self.assertRaises(
            gerrit.GerritException, gerrit.GetGerritPatchInfo, [changeid]
        )

    def testQueries(self) -> None:
        """Verify assorted query operations."""
        project = self.createProject("testProject")
        clone_path = self.cloneProject(project)
        gpatch = self.createPatch(clone_path, project)
        helper = self._GetHelper()

        # Multi-queries with one valid and one invalid term should raise.
        invalid_change_id = "I1234567890123456789012345678901234567890"
        self.assertRaises(
            gerrit.GerritException,
            gerrit.GetGerritPatchInfo,
            [invalid_change_id, gpatch.change_id],
        )
        self.assertRaises(
            gerrit.GerritException,
            gerrit.GetGerritPatchInfo,
            [gpatch.change_id, invalid_change_id],
        )
        self.assertRaises(
            gerrit.GerritException,
            gerrit.GetGerritPatchInfo,
            ["9876543", gpatch.gerrit_number],
        )
        self.assertRaises(
            gerrit.GerritException,
            gerrit.GetGerritPatchInfo,
            [gpatch.gerrit_number, "9876543"],
        )

        site_params = config_lib.GetSiteParams()
        # Simple query by project/changeid/sha1.
        patch_info = helper.GrabPatchFromGerrit(
            gpatch.project, gpatch.change_id, gpatch.sha1
        )
        self.assertEqual(patch_info.gerrit_number, gpatch.gerrit_number)
        self.assertEqual(patch_info.remote, site_params.EXTERNAL_REMOTE)

        # Simple query by gerrit number to external remote.
        patch_info = gerrit.GetGerritPatchInfo([gpatch.gerrit_number])
        self.assertEqual(patch_info[0].gerrit_number, gpatch.gerrit_number)
        self.assertEqual(patch_info[0].remote, site_params.EXTERNAL_REMOTE)

        # Simple query by gerrit number to internal remote.
        patch_info = gerrit.GetGerritPatchInfo(["*" + gpatch.gerrit_number])
        self.assertEqual(patch_info[0].gerrit_number, gpatch.gerrit_number)
        self.assertEqual(patch_info[0].remote, site_params.INTERNAL_REMOTE)

        # Query to external server by gerrit number and change-id which refer to
        # the same change should return one result.
        fq_changeid = "~".join((gpatch.project, "main", gpatch.change_id))
        patch_info = gerrit.GetGerritPatchInfo(
            [gpatch.gerrit_number, fq_changeid]
        )
        self.assertEqual(len(patch_info), 1)
        self.assertEqual(patch_info[0].gerrit_number, gpatch.gerrit_number)
        self.assertEqual(patch_info[0].remote, site_params.EXTERNAL_REMOTE)

        # Query to internal server by gerrit number and change-id which refer to
        # the same change should return one result.
        patch_info = gerrit.GetGerritPatchInfo(
            ["*" + gpatch.gerrit_number, "*" + fq_changeid]
        )
        self.assertEqual(len(patch_info), 1)
        self.assertEqual(patch_info[0].gerrit_number, gpatch.gerrit_number)
        self.assertEqual(patch_info[0].remote, site_params.INTERNAL_REMOTE)

    def testSubmitOutdatedCommit(self) -> None:
        """Tests that we can parse a json to check if a change is committed."""
        project = self.createProject("testProject")
        clone_path = self.cloneProject(project, "p1")

        # Create a change.
        gpatch1 = self.createPatch(clone_path, project)

        # Update the change.
        gpatch2 = self.createPatch(clone_path, project, amend=True)

        # Make sure we're talking about the same change.
        self.assertEqual(gpatch1.change_id, gpatch2.change_id)

        # Try submitting the out-of-date change.
        helper = self._GetHelper()
        helper.SetReview(gpatch1.gerrit_number, labels={"Code-Review": "+2"})
        with self.assertRaises(gob_util.GOBError) as ex:
            helper.SubmitChange(gpatch1)
        self.assertEqual(ex.exception.http_status, http.client.CONFLICT)
        self.assertFalse(helper.IsChangeCommitted(gpatch1.gerrit_number))

        # Try submitting the up-to-date change.
        helper.SubmitChange(gpatch2)
        helper.IsChangeCommitted(gpatch2.gerrit_number)

    def testResetReviewLabels(self) -> None:
        """Tests that we can remove a code review label."""
        project = self.createProject("testProject")
        helper = self._GetHelper()
        clone_path = self.cloneProject(project, "p1")
        gpatch = self.createPatch(clone_path, project, msg="Init")
        helper.SetReview(gpatch.gerrit_number, labels={"Code-Review": "+2"})
        gob_util.ResetReviewLabels(
            helper.host,
            gpatch.gerrit_number,
            label="Code-Review",
            notify="OWNER",
        )

    def testApprovalTime(self) -> None:
        """Approval timestamp should be reset when a new patchset is created."""
        # Create a change.
        project = self.createProject("testProject")
        helper = self._GetHelper()
        clone_path = self.cloneProject(project, "p1")
        gpatch = self.createPatch(clone_path, project, msg="Init")
        helper.SetReview(gpatch.gerrit_number, labels={"Code-Review": "+2"})

        # Update the change.
        new_msg = "New %s" % gpatch.commit_message
        cros_build_lib.dbg_run(
            ["git", "commit", "--amend", "-m", new_msg],
            cwd=clone_path,
            capture_output=True,
        )
        self.uploadChange(clone_path)
        gpatch2 = self._GetHelper().QuerySingleRecord(
            change=gpatch.change_id, project=gpatch.project, branch="main"
        )
        self.assertNotEqual(gpatch2.approval_timestamp, 0)
        self.assertNotEqual(gpatch2.commit_timestamp, 0)
        self.assertEqual(gpatch2.approval_timestamp, gpatch2.commit_timestamp)


class GerritParserTest(cros_test_lib.TestCase):
    """Unittests for GerritHelper."""

    # pylint: disable=protected-access

    def _GetHelper(self, remote=config_lib.GetSiteParams().EXTERNAL_REMOTE):
        return gerrit.GetGerritHelper(remote)

    def testGetChangeFromStdoutPass(self) -> None:
        """Verify the proper change number is returned from the git stdout."""
        stdout = (
            "remote:\nremote:\nremote:   "
            "https://example.com/c/some/project/repo/+/123 gerrit: test"
        )
        changenum = self._GetHelper()._get_changenumber_from_stdout(stdout)
        self.assertEqual(changenum, "123")

        stdout = (
            "remote:\nremote:   "
            "https://example.com/c/some/project3/repo/+/123 gerrit: test"
        )
        changenum = self._GetHelper()._get_changenumber_from_stdout(stdout)
        self.assertEqual(changenum, "123")

        stdout = (
            "remote:   "
            "https://example.com/c/some/project/repo/+/123 gerrit: test 456"
        )
        changenum = self._GetHelper()._get_changenumber_from_stdout(stdout)
        self.assertEqual(changenum, "123")

        stdout = (
            "remote:   "
            "https://example.com/c/some/project/repo/+/123 "
            "handle /+/124 in URI"
        )
        changenum = self._GetHelper()._get_changenumber_from_stdout(stdout)
        self.assertEqual(changenum, "123")

    def testGetChangeFromStdoutFail(self) -> None:
        """Verify the function returns None when an improper stdout is given."""

        # Fails because remote is not at the start of the text.
        stdout = """
    remote:
    remote:   https://example./c/some/project/repo/+/123 gerrit: test
    """
        changenum = self._GetHelper()._get_changenumber_from_stdout(stdout)
        self.assertIsNone(changenum)

        stdout = """
    remote:https://example./c/some/project/repo/+/123 gerrit: test
    """
        changenum = self._GetHelper()._get_changenumber_from_stdout(stdout)

        self.assertIsNone(changenum)


@cros_test_lib.pytestmark_network_test
class DirectGerritHelperTest(cros_test_lib.TestCase):
    """Unittests for GerritHelper that use the real Chromium instance."""

    # A big list of real changes.
    CHANGES = ["235893", "*189165", "231790", "*190026", "231647", "234645"]

    def testMultipleChangeDetail(self) -> None:
        """Test ordering of results in GetMultipleChangeDetail"""
        changes = [x for x in self.CHANGES if not x.startswith("*")]
        helper = gerrit.GetCrosExternal()
        results = list(
            helper.GetMultipleChangeDetail([str(x) for x in changes])
        )
        gerrit_numbers = [str(x["_number"]) for x in results]
        self.assertEqual(changes, gerrit_numbers)

    def testQueryMultipleCurrentPatchset(self) -> None:
        """Test ordering of results in QueryMultipleCurrentPatchset"""
        changes = [x for x in self.CHANGES if not x.startswith("*")]
        helper = gerrit.GetCrosExternal()
        results = list(helper.QueryMultipleCurrentPatchset(changes))
        self.assertEqual(changes, [x.gerrit_number for _, x in results])
        self.assertEqual(changes, [x for x, _ in results])

    def testGetGerritPatchInfo(self) -> None:
        """Test ordering of results in GetGerritPatchInfo"""
        # Swizzle from our old syntax to the new syntax.
        changes = []
        for change in self.CHANGES:
            if change.startswith("*"):
                changes.append("chrome-internal:%s" % (change[1:],))
            else:
                changes.append("chromium:%s" % (change,))
        results = list(gerrit.GetGerritPatchInfo(changes))
        self.assertEqual(changes, [x.gerrit_number_str for x in results])